tools/startup/generate_startup_descriptors.py

#!/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 adb_utils
import profile_utils

import argparse
import os
import sys
import time


class Device:

    def __init__(self, device_id, device_pin):
        self.device_id = device_id
        self.device_pin = device_pin


def extend_startup_descriptors(startup_descriptors, iteration, device, options):
    (logcat, profile, profile_classes_and_methods) = \
        generate_startup_profile(device, options)
    if options.logcat:
        write_tmp_logcat(logcat, iteration, options)
        current_startup_descriptors = get_r8_startup_descriptors_from_logcat(
            logcat, options)
    else:
        write_tmp_profile(profile, iteration, options)
        write_tmp_profile_classes_and_methods(profile_classes_and_methods,
                                              iteration, options)
        current_startup_descriptors = \
            profile_utils.transform_art_profile_to_r8_startup_list(
                profile_classes_and_methods, options.generalize_synthetics)
    write_tmp_startup_descriptors(current_startup_descriptors, iteration,
                                  options)
    new_startup_descriptors = add_r8_startup_descriptors(
        startup_descriptors, current_startup_descriptors)
    number_of_new_startup_descriptors = \
        len(new_startup_descriptors) - len(startup_descriptors)
    if options.out is not None:
        print('Found %i new startup descriptors in iteration %i' %
              (number_of_new_startup_descriptors, iteration + 1))
    return new_startup_descriptors


def generate_startup_profile(device, options):
    logcat = None
    profile = None
    profile_classes_and_methods = None
    if options.use_existing_profile:
        # Verify presence of profile.
        adb_utils.check_app_has_profile_data(options.app_id, device.device_id)
        profile = adb_utils.get_profile_data(options.app_id, device.device_id)
        profile_classes_and_methods = \
            adb_utils.get_classes_and_methods_from_app_profile(
                options.app_id, device.device_id)
    else:
        # Unlock device.
        tear_down_options = adb_utils.prepare_for_interaction_with_device(
            device.device_id, device.device_pin)

        logcat_process = None
        if options.logcat:
            # Clear logcat and start capturing logcat.
            adb_utils.clear_logcat(device.device_id)
            logcat_process = adb_utils.start_logcat(
                device.device_id,
                format='tag',
                filter='R8:I ActivityTaskManager:I *:S')
        else:
            # Clear existing profile data.
            adb_utils.clear_profile_data(options.app_id, device.device_id)

        # Launch activity to generate startup profile on device.
        adb_utils.launch_activity(options.app_id, options.main_activity,
                                  device.device_id)

        # Wait for activity startup.
        time.sleep(options.startup_duration)

        if options.logcat:
            # Get startup descriptors from logcat.
            logcat = adb_utils.stop_logcat(logcat_process)
        else:
            # Capture startup profile.
            adb_utils.capture_app_profile_data(options.app_id, device.device_id)
            profile = adb_utils.get_profile_data(options.app_id,
                                                 device.device_id)
            profile_classes_and_methods = \
                adb_utils.get_classes_and_methods_from_app_profile(
                    options.app_id, device.device_id)

        # Shutdown app.
        adb_utils.stop_app(options.app_id, device.device_id)
        adb_utils.teardown_after_interaction_with_device(
            tear_down_options, device.device_id)

    return (logcat, profile, profile_classes_and_methods)


def get_r8_startup_descriptors_from_logcat(logcat, options):
    post_startup = False
    startup_descriptors = {}
    for line in logcat:
        line_elements = parse_logcat_line(line)
        if line_elements is None:
            continue
        (priority, tag, message) = line_elements
        if tag == 'ActivityTaskManager':
            if message.startswith('START') \
                or message.startswith('Activity pause timeout for') \
                or message.startswith('Activity top resumed state loss timeout for') \
                or message.startswith('Force removing') \
                or message.startswith(
                    'Launch timeout has expired, giving up wake lock!'):
                continue
            elif message.startswith('Displayed %s/' % options.app_id):
                print('Entering post startup: %s' % message)
                post_startup = True
                continue
        elif tag == 'R8':
            if is_startup_descriptor(message):
                startup_descriptors[message] = {
                    'conditional_startup': False,
                    'hot': False,
                    'post_startup': post_startup,
                    'startup': True
                }
                continue
        # Reaching here means we didn't expect this line.
        report_unrecognized_logcat_line(line)
    return startup_descriptors


def is_startup_descriptor(string):
    # The descriptor should start with the holder (possibly prefixed with 'S').
    if not any(string.startswith('%sL' % flags) for flags in ['', 'S']):
        return False
    # The descriptor should end with ';', a primitive type, or void.
    if not string.endswith(';') \
        and not any(string.endswith(c) for c in get_primitive_descriptors()) \
        and not string.endswith('V'):
        return False
    return True


def get_primitive_descriptors():
    return ['Z', 'B', 'S', 'C', 'I', 'F', 'J', 'D']


def parse_logcat_line(line):
    if line == '--------- beginning of kernel':
        return None
    if line == '--------- beginning of main':
        return None
    if line == '--------- beginning of system':
        return None

    priority = None
    tag = None

    try:
        priority_end = line.index('/')
        priority = line[0:priority_end]
        line = line[priority_end + 1:]
    except ValueError:
        return report_unrecognized_logcat_line(line)

    try:
        tag_end = line.index(':')
        tag = line[0:tag_end].strip()
        line = line[tag_end + 1:]
    except ValueError:
        return report_unrecognized_logcat_line(line)

    message = line.strip()
    return (priority, tag, message)


def report_unrecognized_logcat_line(line):
    print('Unrecognized line in logcat: %s' % line)


def add_r8_startup_descriptors(old_startup_descriptors,
                               startup_descriptors_to_add):
    new_startup_descriptors = {}
    if len(old_startup_descriptors) == 0:
        for startup_descriptor, flags in startup_descriptors_to_add.items():
            new_startup_descriptors[startup_descriptor] = flags.copy()
    else:
        # Merge the new startup descriptors with the old descriptors in a way so
        # that new startup descriptors are added next to the startup descriptors
        # they are close to in the newly generated list of startup descriptors.
        startup_descriptors_to_add_after_key = {}
        startup_descriptors_to_add_in_the_end = {}
        closest_seen_startup_descriptor = None
        for startup_descriptor, flags in startup_descriptors_to_add.items():
            if startup_descriptor in old_startup_descriptors:
                closest_seen_startup_descriptor = startup_descriptor
            else:
                if closest_seen_startup_descriptor is None:
                    # Insert this new startup descriptor in the end of the result.
                    startup_descriptors_to_add_in_the_end[
                        startup_descriptor] = flags
                else:
                    # Record that this should be inserted after
                    # closest_seen_startup_descriptor.
                    pending_startup_descriptors = \
                        startup_descriptors_to_add_after_key.setdefault(
                            closest_seen_startup_descriptor, {})
                    pending_startup_descriptors[startup_descriptor] = flags
        for startup_descriptor, flags in old_startup_descriptors.items():
            # Merge flags if this also exists in startup_descriptors_to_add.
            if startup_descriptor in startup_descriptors_to_add:
                merged_flags = flags.copy()
                other_flags = startup_descriptors_to_add[startup_descriptor]
                assert not other_flags['conditional_startup']
                merged_flags['hot'] = \
                    merged_flags['hot'] or other_flags['hot']
                merged_flags['startup'] = \
                    merged_flags['startup'] or other_flags['startup']
                merged_flags['post_startup'] = \
                    merged_flags['post_startup'] or other_flags['post_startup']
                new_startup_descriptors[startup_descriptor] = merged_flags
            else:
                new_startup_descriptors[startup_descriptor] = flags.copy()
            # Flush startup descriptors that followed this item in the new trace.
            if startup_descriptor in startup_descriptors_to_add_after_key:
                pending_startup_descriptors = \
                    startup_descriptors_to_add_after_key[startup_descriptor]
                for pending_startup_descriptor, pending_flags \
                    in pending_startup_descriptors.items():
                    new_startup_descriptors[pending_startup_descriptor] = \
                        pending_flags.copy()
        # Insert remaining new startup descriptors in the end.
        for startup_descriptor, flags \
            in startup_descriptors_to_add_in_the_end.items():
            assert startup_descriptor not in new_startup_descriptors
            new_startup_descriptors[startup_descriptor] = flags.copy()
    return new_startup_descriptors


def write_tmp_binary_artifact(artifact, iteration, options, name):
    if not options.tmp_dir:
        return
    out_dir = os.path.join(options.tmp_dir, str(iteration))
    os.makedirs(out_dir, exist_ok=True)
    path = os.path.join(out_dir, name)
    with open(path, 'wb') as f:
        f.write(artifact)


def write_tmp_textual_artifact(artifact,
                               iteration,
                               options,
                               name,
                               item_to_string=None):
    if not options.tmp_dir:
        return
    out_dir = os.path.join(options.tmp_dir, str(iteration))
    os.makedirs(out_dir, exist_ok=True)
    path = os.path.join(out_dir, name)
    with open(path, 'w') as f:
        for item in artifact:
            f.write(item if item_to_string is None else item_to_string(item))
            f.write('\n')


def write_tmp_logcat(logcat, iteration, options):
    write_tmp_textual_artifact(logcat, iteration, options, 'logcat.txt')


def write_tmp_profile(profile, iteration, options):
    write_tmp_binary_artifact(profile, iteration, options, 'primary.prof')


def write_tmp_profile_classes_and_methods(profile_classes_and_methods,
                                          iteration, options):

    def item_to_string(item):
        (descriptor, flags) = item
        return '%s%s%s%s' % ('H' if flags.get('hot') else '',
                             'S' if flags.get('startup') else '', 'P'
                             if flags.get('post_startup') else '', descriptor)

    write_tmp_textual_artifact(profile_classes_and_methods.items(), iteration,
                               options, 'profile.txt', item_to_string)


def write_tmp_startup_descriptors(startup_descriptors, iteration, options):
    lines = [
        startup_descriptor_to_string(startup_descriptor, flags)
        for startup_descriptor, flags in startup_descriptors.items()
    ]
    write_tmp_textual_artifact(lines, iteration, options,
                               'startup-descriptors.txt')


def startup_descriptor_to_string(startup_descriptor, flags):
    result = ''
    if flags['hot']:
        result += 'H'
    if flags['startup']:
        result += 'S'
    if flags['post_startup']:
        result += 'P'
    result += startup_descriptor
    return result


def should_include_startup_descriptor(descriptor, flags, options):
    if flags.get('conditional_startup') \
        and not options.include_conditional_startup:
        return False
    if flags.get('post_startup') \
        and not flags.get('startup') \
        and not options.include_post_startup:
        return False
    return True


def parse_options(argv):
    result = argparse.ArgumentParser(
        description='Generate a perfetto trace file.')
    result.add_argument('--apk', help='Path to the .apk')
    result.add_argument('--apks', help='Path to the .apks')
    result.add_argument('--app-id',
                        help='The application ID of interest',
                        required=True)
    result.add_argument('--bundle', help='Path to the .aab')
    result.add_argument('--device-id',
                        help='Device id (e.g., emulator-5554).',
                        action='append')
    result.add_argument('--device-pin',
                        help='Device pin code (e.g., 1234)',
                        action='append')
    result.add_argument(
        '--generalize-synthetics',
        help='Whether synthetics should be abstracted into their '
        'synthetic contexts',
        action='store_true',
        default=False)
    result.add_argument('--grant-post-notification-permission',
                        help='Grants the android.permission.POST_NOTIFICATIONS '
                        'permission before launching the app',
                        default=False,
                        action='store_true')
    result.add_argument('--logcat', action='store_true', default=False)
    result.add_argument(
        '--include-conditional-startup',
        help='Include conditional startup classes and methods in '
        'the R8 startup descriptors',
        action='store_true',
        default=False)
    result.add_argument(
        '--include-post-startup',
        help='Include post startup classes and methods in the R8 '
        'startup descriptors',
        action='store_true',
        default=False)
    result.add_argument('--iterations',
                        help='Number of profiles to generate',
                        default=1,
                        type=int)
    result.add_argument('--main-activity', help='Main activity class name')
    result.add_argument(
        '--out',
        help='File where to store startup descriptors (defaults '
        'to stdout)')
    result.add_argument('--startup-duration',
                        help='Duration in seconds before shutting down app',
                        default=15,
                        type=int)
    result.add_argument('--tmp-dir',
                        help='Directory where to store intermediate artifacts'
                        ' (by default these are not emitted)')
    result.add_argument('--until-stable',
                        help='Repeat profile generation until no new startup '
                        'descriptors are found',
                        action='store_true',
                        default=False)
    result.add_argument(
        '--until-stable-iterations',
        help='Number of times that profile generation must must '
        'not find new startup descriptors before exiting',
        default=1,
        type=int)
    result.add_argument('--use-existing-profile',
                        help='Do not launch app to generate startup profile',
                        action='store_true',
                        default=False)
    options, args = result.parse_known_args(argv)

    # Read the device pins.
    device_pins = options.device_pin or []
    del options.device_pin

    # Convert the device ids and pins into a list of devices.
    options.devices = []
    if options.device_id is None:
        # Assume a single device is attached.
        options.devices.append(
            Device(None, device_pins[0] if len(device_pins) > 0 else None))
    else:
        for i in range(len(options.device_id)):
            device_id = options.device_id[i]
            device_pin = device_pins[i] if i < len(device_pins) else None
            options.devices.append(Device(device_id, device_pin))
    del options.device_id

    paths = [
        path for path in [options.apk, options.apks, options.bundle]
        if path is not None
    ]
    assert len(paths) <= 1, 'Expected at most one .apk, .apks, or .aab file.'
    assert options.main_activity is not None or options.use_existing_profile, \
        'Argument --main-activity is required except when running with ' \
        '--use-existing-profile.'

    return options, args


def run_on_device(device, options, startup_descriptors):
    adb_utils.root(device.device_id)
    if options.apk:
        adb_utils.uninstall(options.app_id, device.device_id)
        adb_utils.install(options.apk, device.device_id)
    elif options.apks:
        adb_utils.uninstall(options.app_id, device.device_id)
        adb_utils.install_apks(options.apks, device.device_id)
    elif options.bundle:
        adb_utils.uninstall(options.app_id, device.device_id)
        adb_utils.install_bundle(options.bundle, device.device_id)
    # Grant notifications.
    if options.grant_post_notification_permission:
        adb_utils.grant(options.app_id, 'android.permission.POST_NOTIFICATIONS',
                        device.device_id)
    if options.until_stable:
        iteration = 0
        stable_iterations = 0
        while True:
            old_startup_descriptors = startup_descriptors
            startup_descriptors = extend_startup_descriptors(
                old_startup_descriptors, iteration, device, options)
            diff = len(startup_descriptors) - len(old_startup_descriptors)
            if diff == 0:
                stable_iterations = stable_iterations + 1
                if stable_iterations == options.until_stable_iterations:
                    break
            else:
                stable_iterations = 0
            iteration = iteration + 1
    else:
        for iteration in range(options.iterations):
            startup_descriptors = extend_startup_descriptors(
                startup_descriptors, iteration, device, options)
    return startup_descriptors


def main(argv):
    (options, args) = parse_options(argv)
    startup_descriptors = {}
    for device in options.devices:
        startup_descriptors = run_on_device(device, options,
                                            startup_descriptors)
    if options.out is not None:
        with open(options.out, 'w') as f:
            for startup_descriptor, flags in startup_descriptors.items():
                if should_include_startup_descriptor(startup_descriptor, flags,
                                                     options):
                    f.write(
                        startup_descriptor_to_string(startup_descriptor, flags))
                    f.write('\n')
    else:
        for startup_descriptor, flags in startup_descriptors.items():
            if should_include_startup_descriptor(startup_descriptor, flags,
                                                 options):
                print(startup_descriptor_to_string(startup_descriptor, flags))


if __name__ == '__main__':
    sys.exit(main(sys.argv[1:]))