| # Copyright 2018 The Chromium OS Authors. All rights reserved. |
| # Use of this source code is governed by a BSD-style license that can be |
| # found in the LICENSE file. |
| |
| import dbus, os, time |
| |
| from autotest_lib.client.bin import test |
| from autotest_lib.client.common_lib import error |
| from autotest_lib.client.cros import debugd_util |
| |
| class platform_DebugDaemonGetPerfOutputFd(test.test): |
| """ |
| This autotest tests the collection of perf data. It calls perf indirectly |
| through debugd -> quipper -> perf. This also tests stopping the perf |
| session. |
| |
| The perf data is read from a pipe that is redirected to stdout of the |
| quipper process. |
| """ |
| |
| version = 1 |
| |
| def check_perf_output(self, perf_data): |
| """ |
| Utility function to validate the perf data that was previously read |
| from the pipe. |
| """ |
| if len(perf_data) < 10: |
| raise error.TestFail('Perf output (%s) too small' % perf_data) |
| |
| # Perform basic sanity checks of the perf data: it should contain |
| # [kernel.kallsyms] and /usr/bin/perf |
| if (perf_data.find('[kernel.kallsyms]') == -1 or |
| perf_data.find('/usr/bin/perf') == -1): |
| raise error.TestFail('Quipper failed: %s' % perf_data) |
| |
| def call_get_perf_output_fd(self, duration): |
| """ |
| Utility function to call DBus method GetPerfOutputFd with the given |
| duration. |
| """ |
| pipe_r, pipe_w = os.pipe() |
| perf_command = ['perf', 'record', '-a', '-F', '100'] |
| |
| session_id = self.dbus_iface.GetPerfOutputFd( |
| duration, perf_command, dbus.types.UnixFd(pipe_w), signature="uash") |
| |
| # pipe_w is dup()'d in calling dbus. Close in this process. |
| os.close(pipe_w) |
| |
| if session_id == 0: |
| raise error.TestFail('Invalid session ID from GetPerfOutputFd') |
| |
| # Don't explicitly os.close(pipe_r) since it will be closed |
| # automatically when the file object returned by os.fdopen() is closed. |
| return session_id, os.fdopen(pipe_r, 'r') |
| |
| def call_stop_perf(self, session_id, real_duration): |
| """ |
| Utility function to call DBus method StopPerf to collect perf data |
| collected within the given duration. |
| """ |
| # Sleep for real_duration seconds and then stop the perf session. |
| time.sleep(real_duration) |
| self.dbus_iface.StopPerf(session_id, signature='t') |
| |
| def test_full_duration(self): |
| """ |
| Test GetPerfOutpuFd to collect a profile of 2 seconds. |
| """ |
| |
| session_id, result_file = self.call_get_perf_output_fd(2) |
| |
| # This performs synchronous read until perf exits. |
| result = result_file.read() |
| |
| self.check_perf_output(result) |
| |
| def test_stop_perf(self): |
| """ |
| Test StopPerf by calling GetPerfOutputFd to collect a profile of 30 |
| seconds. After the perf session is started for 2 seconds, call StopPerf |
| to stop the profiling session. The net result is a profile of 2 |
| seconds. Verify StopPerf working by timing the test case: the test case |
| shouldn't run for 30 seconds or longer. |
| """ |
| start = time.time() |
| |
| # Default duration is 30 sec. |
| session_id, result_file = self.call_get_perf_output_fd(30) |
| |
| # Get a profile of 2 seconds by premature stop. |
| self.call_stop_perf(session_id, 2) |
| |
| # This performs synchronous read until perf exits. |
| result = result_file.read() |
| |
| self.check_perf_output(result) |
| |
| end = time.time() |
| if (end - start) >= 30: |
| raise error.TestFail('Unable to stop the perf tool') |
| |
| def test_start_after_previous_finished(self): |
| """ |
| Test consecutive GetPerfOutputFd calls that there is no undesirable |
| side effect left in the previous profiling session. |
| """ |
| self.test_full_duration() |
| self.test_full_duration() |
| |
| def test_stop_without_start(self): |
| """ |
| Test unmatched StopPerf call by checking the returned DBusException. |
| """ |
| dbus_message = None |
| try: |
| self.call_stop_perf(0, 1) |
| except dbus.exceptions.DBusException as dbus_exception: |
| dbus_message = dbus_exception.get_dbus_message() |
| |
| if dbus_message is None: |
| raise error.TestFail('DBusException expected') |
| if dbus_message.find('Perf tool not started') == -1: |
| raise error.TestFail('Unexpected DBus message: %s' % dbus_message) |
| |
| def test_stop_using_wrong_id(self): |
| """ |
| Test calling StopPerf with an invalid session ID by checking the |
| returned DBusException. |
| """ |
| start = time.time() |
| |
| # Default duration is 30 sec. |
| session_id, result_file = self.call_get_perf_output_fd(30) |
| |
| dbus_message = None |
| try: |
| # Use session_id - 1 to trigger the error condition. |
| self.call_stop_perf(session_id - 1, 1) |
| except dbus.exceptions.DBusException as dbus_exception: |
| dbus_message = dbus_exception.get_dbus_message() |
| |
| if dbus_message is None: |
| raise error.TestFail('DBusException expected') |
| if dbus_message.find('Invalid profile session id') == -1: |
| raise error.TestFail('Unexpected DBus message: %s' % dbus_message) |
| |
| # Get a profile of 1 second by premature stop. |
| self.call_stop_perf(session_id, 1) |
| |
| # This performs synchronous read until perf exits. |
| result = result_file.read() |
| |
| self.check_perf_output(result) |
| |
| end = time.time() |
| if (end - start) >= 30: |
| raise error.TestFail('Unable to stop the perf tool') |
| |
| def test_start_2nd_time(self): |
| """ |
| Test calling GetPerfOutputFd when an existing profiling session is |
| running: the 2nd call should yield a DBusException without affecting |
| the 1st call. |
| """ |
| # Default duration is 30 sec. |
| session_id, result_file = self.call_get_perf_output_fd(30) |
| |
| dbus_message = None |
| try: |
| self.call_get_perf_output_fd(60) |
| except dbus.exceptions.DBusException as dbus_exception: |
| dbus_message = dbus_exception.get_dbus_message() |
| |
| if dbus_message is None: |
| raise error.TestFail('DBusException expected') |
| if dbus_message.find('Existing perf tool running') == -1: |
| raise error.TestFail('Unexpected DBus message: %s' % dbus_message) |
| |
| # Get a profile of 1 second by premature stop. |
| self.call_stop_perf(session_id, 1) |
| |
| # This performs synchronous read until perf exits. |
| result = result_file.read() |
| |
| self.check_perf_output(result) |
| |
| def run_once(self, *args, **kwargs): |
| """ |
| Primary autotest function. |
| """ |
| # Setup. |
| self.dbus_iface = debugd_util.iface() |
| |
| # Test normal cases. |
| self.test_full_duration() |
| self.test_start_after_previous_finished() |
| self.test_stop_perf() |
| |
| # Test error cases. |
| self.test_stop_without_start() |
| self.test_stop_using_wrong_id() |
| self.test_start_2nd_time() |