blob: d2f53b4b0f2f3f7f0c46f30a3cd379d1bb4cedff [file] [log] [blame]
#!/usr/bin/env python3
# Copyright (c) 2022, the R8 project authors. Please see the AUTHORS file
# for details. All rights reserved. Use of this source code is governed by a
# BSD-style license that can be found in the LICENSE file.
import os
import subprocess
import sys
from perfetto.trace_processor import TraceProcessor
except ImportError:
'Unable to analyze perfetto trace without the perfetto library. '
'Install instructions:\n'
' sudo apt install python3-pip\n'
' pip3 install perfetto')
def ensure_record_android_trace(tmp_dir):
record_android_trace_path = os.path.join(tmp_dir, 'record_android_trace')
if not os.path.exists(record_android_trace_path):
cmd = [
assert os.path.exists(record_android_trace_path)
return record_android_trace_path
def record_android_trace(out_dir, tmp_dir, device_id=None):
record_android_trace_path = ensure_record_android_trace(tmp_dir)
config_path = os.path.join(os.path.dirname(__file__), 'config.pbtx')
perfetto_trace_path = os.path.join(out_dir, 'trace.perfetto-trace')
cmd = [
if device_id is not None:
cmd.extend(['--serial', device_id])
perfetto_process = subprocess.Popen(
cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
lines = []
for line in perfetto_process.stdout:
line = line.decode('utf-8')
if 'enabled ftrace' in line.strip():
return perfetto_process, perfetto_trace_path
raise ValueError(
'Expected to find line containing: enabled ftrace, got: %s' % lines)
def stop_record_android_trace(perfetto_process, out_dir):
if perfetto_process.poll() is not None:
raise ValueError('Expected perfetto process to be running')
# perfetto should terminate in at most 15 seconds,
stdout, stderr = perfetto_process.communicate(
stdout = stdout.decode('utf-8')
stderr = stderr.decode('utf-8')
assert perfetto_process.returncode == 0
assert os.path.exists(os.path.join(out_dir, 'trace.perfetto-trace'))
def find_slices_by_name(slice_name, options, trace_processor):
return trace_processor.query(
'SELECT slice.dur, slice.ts FROM slice'
' INNER JOIN thread_track ON (slice.track_id ='
' INNER JOIN thread using (utid)'
' INNER JOIN process using (upid)'
' WHERE = "%s"'
' AND = "%s"'
' ORDER BY slice.ts ASC'
% (slice_name, options.app_id))
def find_unique_slice_by_name(slice_name, options, trace_processor):
query_it = find_slices_by_name(slice_name, options, trace_processor)
assert len(query_it) == 1
return next(query_it)
def get_slice_end_since_start(slice, initial_slice):
return (slice.ts + slice.dur - initial_slice.ts) / 1000000