# 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()