xref: /aosp_15_r20/external/chromium-trace/catapult/systrace/systrace/tracing_agents/walt_agent.py (revision 1fa4b3da657c0e9ad43c0220bacf9731820715a5) 
1# Copyright 2017 The Chromium Authors. All rights reserved.
2# Use of this source code is governed by a BSD-style license that can be
3# found in the LICENSE file.
4
5import optparse
6import threading
7
8import py_utils
9
10from devil.android import device_utils
11from systrace import trace_result
12from systrace import tracing_agents
13from py_trace_event import trace_time as trace_time_module
14
15TRACE_FILE_PATH = \
16    '/sdcard/Android/data/org.chromium.latency.walt/files/trace.txt'
17
18CLOCK_DOMAIN_MARKER = '# clock_type=LINUX_CLOCK_MONOTONIC\n'
19
20
21def try_create_agent(options):
22  if options.is_walt_enabled:
23    return WaltAgent()
24  return None
25
26
27class WaltConfig(tracing_agents.TracingConfig):
28  def __init__(self, device_serial_number, is_walt_enabled):
29    tracing_agents.TracingConfig.__init__(self)
30    self.device_serial_number = device_serial_number
31    self.is_walt_enabled = is_walt_enabled
32
33
34def add_options(parser):
35  options = optparse.OptionGroup(parser, 'WALT trace options')
36  options.add_option('--walt', dest='is_walt_enabled', default=False,
37                    action='store_true', help='Use the WALT tracing agent. '
38                    'WALT is a device for measuring latency of physical '
39                    'sensors on phones and computers. '
40                    'See https://github.com/google/walt')
41  return options
42
43
44def get_config(options):
45  return WaltConfig(options.device_serial_number, options.is_walt_enabled)
46
47
48class WaltAgent(tracing_agents.TracingAgent):
49  """
50  This tracing agent requires the WALT app to be installed on the Android phone,
51  and requires the WALT device to be attached to the phone. WALT is a device
52  for measuring latency of physical sensors and outputs on phones and
53  computers. For more information, visit https://github.com/google/walt
54  """
55  def __init__(self):
56    super(WaltAgent, self).__init__()
57    self._trace_contents = None
58    self._config = None
59    self._device_utils = None
60    self._clock_sync_marker = None
61    self._collection_thread = None
62
63  def __repr__(self):
64    return 'WaltAgent'
65
66  @py_utils.Timeout(tracing_agents.START_STOP_TIMEOUT)
67  def StartAgentTracing(self, config, timeout=None):
68    del timeout  # unused
69    self._config = config
70    self._device_utils = device_utils.DeviceUtils(
71        self._config.device_serial_number)
72    if self._device_utils.PathExists(TRACE_FILE_PATH):
73      # clear old trace events so they are not included in the current trace
74      self._device_utils.WriteFile(TRACE_FILE_PATH, '')
75    return True
76
77  @py_utils.Timeout(tracing_agents.START_STOP_TIMEOUT)
78  def StopAgentTracing(self, timeout=None):
79    """Stops tracing and starts collecting results.
80
81    To synchronously retrieve the results after calling this function,
82    call GetResults().
83    """
84    del timeout  # unused
85    self._collection_thread = threading.Thread(
86        target=self._collect_trace_data)
87    self._collection_thread.start()
88    return True
89
90  def _collect_trace_data(self):
91    self._trace_contents = self._device_utils.ReadFile(TRACE_FILE_PATH)
92
93  def SupportsExplicitClockSync(self):
94    return True
95
96  def RecordClockSyncMarker(self, sync_id, did_record_clock_sync_callback):
97    cmd = 'cat /proc/timer_list | grep now'
98    t1 = trace_time_module.Now()
99    command_result = self._device_utils.RunShellCommand(cmd, shell=True)
100    nsec = command_result[0].split()[2]
101    self._clock_sync_marker = format_clock_sync_marker(sync_id, nsec)
102    did_record_clock_sync_callback(t1, sync_id)
103
104  @py_utils.Timeout(tracing_agents.GET_RESULTS_TIMEOUT)
105  def GetResults(self, timeout=None):
106    del timeout  # unused
107    self._collection_thread.join()
108    self._collection_thread = None
109    return trace_result.TraceResult('waltTrace', self._get_trace_result())
110
111  def _get_trace_result(self):
112    result = '# tracer: \n' + CLOCK_DOMAIN_MARKER + self._trace_contents
113    if self._clock_sync_marker is not None:
114      result += self._clock_sync_marker
115    return result
116
117
118def format_clock_sync_marker(sync_id, nanosec_time):
119  return ('<0>-0  (-----) [001] ...1  ' + str(float(nanosec_time) / 1e9)
120          + ': tracing_mark_write: trace_event_clock_sync: name='
121          + sync_id + '\n')
122