[Android] Implement perf tests to platform mode.

build/android/pylib/local/device/local_device_perf_test_run.py

+# Copyright 2015 The Chromium 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 io
+import json
+import logging
+import os
+import pickle
+import re
+import shutil
+import sys
+import tempfile
+import threading
+import time
+import zipfile
+
+from devil.android import battery_utils
+from devil.android import device_errors
+from devil.android import device_list
+from devil.android import device_utils
+from devil.android import forwarder
+from devil.utils import cmd_helper
+from devil.utils import reraiser_thread
+from devil.utils import watchdog_timer
+from pylib import constants
+from pylib.base import base_test_result
+from pylib.constants import host_paths
+from pylib.local.device import local_device_test_run
+
+
+# Regex for the master branch commit position.
+_GIT_CR_POS_RE = re.compile(r'^Cr-Commit-Position: refs/heads/master@{#(\d+)}$')
+
+
+class _HeartBeatLogger(object):
+  # How often to print the heartbeat on flush().
+  _PRINT_INTERVAL = 30.0
+
+  def __init__(self):
+    """A file-like class for keeping the buildbot alive."""

[jbudorick, 2016/06/01 20:46:06]

What calls are taking so long that this is necessary?

timeout_retry.py does something like this already, so perhaps this isn't
necessary any more?
https://code.google.com/p/chromium/codesearch#chromium/src/third_party/catapu...

[rnephew (Reviews Here), 2016/06/01 21:01:39]

This was part of the old test runner, it might not be required. I'll take it out
and see how it reacts. My guess is that the page cycler tests were causing
problems.

[rnephew (Reviews Here), 2016/06/01 22:14:15]

Done.

+    self._len = 0
+    self._tick = time.time()
+    self._stopped = threading.Event()
+    self._timer = threading.Thread(target=self._runner)
+    self._timer.start()
+
+  def _runner(self):
+    while not self._stopped.is_set():
+      self.flush()
+      self._stopped.wait(_HeartBeatLogger._PRINT_INTERVAL)
+
+  def write(self, data):
+    self._len += len(data)
+
+  def flush(self):
+    now = time.time()
+    if now - self._tick >= _HeartBeatLogger._PRINT_INTERVAL:
+      self._tick = now
+      print '--single-step output length %d' % self._len
+      sys.stdout.flush()
+
+  def stop(self):
+    self._stopped.set()
+
+
+def _GetChromiumRevision():
+  # pylint: disable=line-too-long
+  """Get the git hash and commit position of the chromium master branch.
+
+  See: https://chromium.googlesource.com/chromium/tools/build/+/master/scripts/slave/runtest.py#212
+
+  Returns:
+    A dictionary with 'revision' and 'commit_pos' keys.
+  """
+  # pylint: enable=line-too-long
+  status, output = cmd_helper.GetCmdStatusAndOutput(
+      ['git', 'log', '-n', '1', '--pretty=format:%H%n%B', 'HEAD'],
+      host_paths.DIR_SOURCE_ROOT)
+  revision = None
+  commit_pos = None
+  if not status:
+    lines = output.splitlines()
+    revision = lines[0]
+    for line in reversed(lines):
+      m = _GIT_CR_POS_RE.match(line.strip())
+      if m:
+        commit_pos = int(m.group(1))
+        break
+  return {'revision': revision, 'commit_pos': commit_pos}
+
+
+class TestShard(object):
+  def __init__(self, test_instance, device, index, tests, results, watcher=None,
+               retries=3):
+    logging.info('Create shard %s for device %s to run the following tests:',
+                 index, device)
+    for t in tests:
+      logging.info('  %s', t)
+    self._battery = battery_utils.BatteryUtils(device)
+    self._device = device
+    self._index = index
+    self._tests = tests
+    self._watcher = watcher
+    self._test_instance = test_instance
+    self._output_dir = None
+    self._results = results
+    self._retries = retries
+
+  def _WriteBuildBotJson(self):

[jbudorick, 2016/06/01 20:46:06]

I don't think this should be part of the default operating mode.

[rnephew (Reviews Here), 2016/06/01 21:01:39]

It currently is, and for this CL I would like runtime behavior to match old
behavior as close as possible. I will add a flag to this though, that defaults
to True, to write the buildbot json. Then we can make changes to the recipes to
enable it then I can default it to not write the buildbot json.

[jbudorick, 2016/06/01 21:03:20]

to reiterate: maintaining old crufty behavior is a non-goal of platform mode.
It's going to be behind a flag to begin with anyway.

[rnephew (Reviews Here), 2016/06/01 21:12:13]

By my last comment I just meant I wanted the flag to be on by default at first.
Make a flag for the output.
Land this.
Make changes to recipes to add the flag.
Default the flag to off.

That way we do not cause any problems on the bots.

[jbudorick, 2016/06/01 21:16:09]

and I meant that this entire CL should be behind --enable-platform-mode, which
will also not cause bot problems.

[rnephew (Reviews Here), 2016/06/01 21:23:16]

Oh, yeah. I planned to back out the changes in test_runner.py that make it go
live before landing. 
I can land the flag in this cl and have it be a no-op until platform mode is
enabled.

+    """Write metadata about the buildbot environment to the output dir."""
+    if not self._output_dir:
+      return
+    data = {
+      'chromium': _GetChromiumRevision(),
+      'environment': dict(os.environ)
+    }
+    with open(os.path.join(self._output_dir, 'buildbot.json'), 'w') as f:
+      json.dump(data, f, sort_keys=True, indent=2, separators=(',', ': '))
+
+  def _TestSetUp(self):
+    self._ResetWatcher()
+    try:
+      logging.info('Unmapping device ports.')

[jbudorick, 2016/06/01 20:46:06]

We should be doing unmapping at TearDown time, not SetUp time.

[rnephew (Reviews Here), 2016/06/01 22:14:15]

Done.

+      forwarder.Forwarder.UnmapAllDevicePorts(self._device)
+      self._device.RestartAdbd()

[jbudorick, 2016/06/01 20:46:06]

I'm wondering if this is still necessary.

[rnephew (Reviews Here), 2016/06/01 21:01:39]

I'll experiment locally.

[rnephew (Reviews Here), 2016/06/02 18:43:26]

Local experimentation shows no detriment by getting rid of it. Moved it to run
only when a test fails.

+    except Exception: # pylint: disable=broad-except
+      logging.exception('Exception when resetting ports.')
+
+    self._BatteryLevelCheck()
+    self._BatteryTempCheck()
+    self._ScreenCheck()
+
+    if not self._device.IsOnline():
+      msg = 'Device %s is unresponsive.' % str(self._device)
+      logging.warning(msg)

[jbudorick, 2016/06/01 20:46:06]

This should not log if you're already raising an exception. It should instead be
logged where the exception is caught.

[rnephew (Reviews Here), 2016/06/01 22:14:15]

Done.

+      raise device_errors.DeviceUnreachableError(msg)
+
+  def _CleanupOutputDirectory(self):
+    if self._output_dir:
+      shutil.rmtree(self._output_dir, ignore_errors=True)
+      self._output_dir = None
+
+  def _CreateCmd(self, test):
+    cmd = '%s --device %s' % (self._tests[test]['cmd'], str(self._device))
+    if (self._test_instance.collect_chartjson_data
+        or self._tests[test].get('archive_output_dir')):
+      self._output_dir = tempfile.mkdtemp()
+      cmd = cmd + ' --output-dir=%s' % self._output_dir
+    if self._test_instance.dry_run:
+      cmd = 'echo %s' % cmd
+    return cmd
+
+  def _RunSingleTest(self, test):
+
+    logging.info('Running %s on shard %s', test, self._index)
+    timeout = (
+        None if self._test_instance.no_timeout
+        else self._tests[test].get('timeout', 3600))
+    logging.info('Timeout for %s test: %s', test, timeout)
+
+    logfile = sys.stdout
+    if self._test_instance.single_step:
+      logfile = _HeartBeatLogger()
+    cmd = self._CreateCmd(test)
+    self._WriteBuildBotJson()
+    cwd = os.path.abspath(host_paths.DIR_SOURCE_ROOT)
+    if cmd.startswith('src/'):
+      cwd = os.path.abspath(os.path.join(host_paths.DIR_SOURCE_ROOT, os.pardir))
+
+    try:
+      logging.debug('Running test with command \'%s\'', cmd)
+      exit_code, output = cmd_helper.GetCmdStatusAndOutputWithTimeout(
+          cmd, timeout, cwd=cwd, shell=True, logfile=logfile)
+      json_output = self._test_instance.ReadChartjsonOutput(self._output_dir)
+    except cmd_helper.TimeoutError as e:
+      exit_code = -1
+      output = e.output
+      json_output = ''
+    finally:
+      if self._test_instance.single_step:
+        logfile.stop()
+    return cmd, exit_code, output, json_output
+
+  def _ProcessTestResult(
+      self, test, cmd, start_time, end_time, exit_code, output, json_output):
+    if exit_code is None:
+      exit_code = -1
+    logging.info('%s : exit_code=%d in %d secs on device %s',
+                 test, exit_code, end_time - start_time,
+                 str(self._device))
+    if exit_code == 0:
+      result_type = base_test_result.ResultType.PASS
+    else:
+      result_type = base_test_result.ResultType.FAIL
+      # TODO(rnephew): Improve device recovery logic.
+      try:
+        self._device.WaitUntilFullyBooted(timeout=120)
+      except device_errors.CommandTimeoutError:
+        logging.exception('Device failed to return after %s.', test)
+    actual_exit_code = exit_code
+    if (self._test_instance.flaky_steps
+        and test in self._test_instance.flaky_steps):
+      exit_code = 0
+    archive_bytes = (self._ArchiveOutputDir()
+                     if self._tests[test].get('archive_output_dir')
+                     else None)
+    persisted_result = {
+        'name': test,
+        'output': [output],
+        'chartjson': json_output,
+        'archive_bytes': archive_bytes,
+        'exit_code': exit_code,
+        'actual_exit_code': actual_exit_code,
+        'result_type': result_type,
+        'start_time': start_time,
+        'end_time': end_time,
+        'total_time': end_time - start_time,
+        'device': str(self._device),
+        'cmd': cmd,
+    }
+    self._SaveResult(persisted_result)
+    return result_type
+
+  def RunTestsOnShard(self):
+    for test in self._tests:
+      self._TestSetUp()
+
+      try:
+        exit_code = None
+        tries_left = self._retries
+
+        while exit_code != 0 and tries_left > 0:
+          self._ResetWatcher()
+          tries_left = tries_left - 1
+          start_time = time.time()
+          cmd, exit_code, output, json_output = self._RunSingleTest(test)
+          end_time = time.time()
+          result_type = self._ProcessTestResult(
+              test, cmd, start_time, end_time, exit_code, output, json_output)
+
+        result = base_test_result.TestRunResults()
+        result.AddResult(base_test_result.BaseTestResult(test, result_type))
+        self._results.append(result)
+      finally:
+        self._CleanupOutputDirectory()
+
+  @staticmethod
+  def _SaveResult(result):
+    pickled = os.path.join(constants.PERF_OUTPUT_DIR, result['name'])
+    if os.path.exists(pickled):
+      with file(pickled, 'r') as f:
+        previous = pickle.loads(f.read())
+        result['output'] = previous['output'] + result['output']
+    with file(pickled, 'w') as f:
+      f.write(pickle.dumps(result))
+
+  def _ArchiveOutputDir(self):
+    """Archive all files in the output dir, and return as compressed bytes."""
+    with io.BytesIO() as archive:
+      with zipfile.ZipFile(archive, 'w', zipfile.ZIP_DEFLATED) as contents:
+        num_files = 0
+        for absdir, _, files in os.walk(self._output_dir):
+          reldir = os.path.relpath(absdir, self._output_dir)
+          for filename in files:
+            src_path = os.path.join(absdir, filename)
+            # We use normpath to turn './file.txt' into just 'file.txt'.
+            dst_path = os.path.normpath(os.path.join(reldir, filename))
+            contents.write(src_path, dst_path)
+            num_files += 1
+      if num_files:
+        logging.info('%d files in the output dir were archived.', num_files)
+      else:
+        logging.warning('No files in the output dir. Archive is empty.')
+      return archive.getvalue()
+
+  def _ResetWatcher(self):
+    if self._watcher:
+      self._watcher.Reset()
+
+  def _BatteryLevelCheck(self):
+    logging.info('Charge level: %s%%',
+                 str(self._battery.GetBatteryInfo().get('level')))
+    if self._test_instance.min_battery_level:
+      self._battery.ChargeDeviceToLevel(self._test_instance.min_battery_level)
+
+  def _ScreenCheck(self):
+    if not self._device.IsScreenOn():
+      self._device.SetScreen(True)
+
+  def _BatteryTempCheck(self):
+    logging.info('temperature: %s (0.1 C)',
+                 str(self._battery.GetBatteryInfo().get('temperature')))
+    if self._test_instance.max_battery_temp:
+      self._battery.LetBatteryCoolToTemperature(
+          self._test_instance.max_battery_temp)
+
+
+class LocalDevicePerfTestRun(local_device_test_run.LocalDeviceTestRun):
+  def __init__(self, env, test_instance):
+    super(LocalDevicePerfTestRun, self).__init__(env, test_instance)
+    self._test_instance = test_instance
+    self._env = env
+    self._timeout = 10 * 60 # Ten minutes
+    self._devices = None
+    self._test_buckets = []
+    self._watcher = None
+
+  def SetUp(self):
+    self._devices = self._GetAllDevices(self._env.devices,
+                                        self._test_instance.known_devices_file)
+    self._watcher = watchdog_timer.WatchdogTimer(self._timeout)
+
+  def TearDown(self):
+    pass
+
+  def _GetStepsFromDict(self):
+    if self._test_instance.single_step:
+      return {
+          'version': 1,
+          'steps': {
+              'single_step': {
+                'device_affinity': 0,
+                'cmd': self._test_instance.single_step
+              },
+          }
+      }
+    if self._test_instance.steps:
+      with file(self._test_instance.steps, 'r') as f:
+        steps = json.load(f)
+        assert steps['version'] == 1
+        return steps
+
+  def _SplitTestsByAffinity(self):
+    test_dict = self._GetStepsFromDict()
+    for test in test_dict['steps']:
+      affinity = test_dict['steps'][test]['device_affinity']
+      if len(self._test_buckets) < affinity + 1:
+        while len(self._test_buckets) != affinity + 1:
+          self._test_buckets.append({})
+      self._test_buckets[affinity][test] = test_dict['steps'][test]
+    return self._test_buckets
+
+  @staticmethod
+  def _GetAllDevices(active_devices, devices_path):
+    if not devices_path:

[jbudorick, 2016/06/01 20:46:06]

What is going on here with the double devices_path check?

[rnephew (Reviews Here), 2016/06/01 22:14:15]

Done.

+      logging.warning('Known devices file path not being passed. For device '
+                      'affinity to work properly, it must be passed.')
+    try:
+      if devices_path:
+        devices = [device_utils.DeviceUtils(s)
+                   for s in device_list.GetPersistentDeviceList(devices_path)]
+      else:
+        logging.warning('Known devices file path not being passed. For device '
+                        'affinity to work properly, it must be passed.')
+        devices = active_devices
+    except IOError as e:
+      logging.error('Unable to find %s [%s]', devices_path, e)
+      devices = active_devices
+    return sorted(devices)
+
+
+  def RunTests(self):
+    # Option selected for saving a json file with a list of test names.
+    if self._test_instance.output_json_list:
+      return self._test_instance.OutputJsonList()
+
+    # Just print the results from a single previously executed step.
+    if self._test_instance.print_step:
+      return self._test_instance.PrintTestOutput()
+
+    # Affinitize the tests.
+    test_buckets = self._SplitTestsByAffinity()
+    if not test_buckets:
+      raise NotImplementedError('No tests found!')

[jbudorick, 2016/06/01 20:46:06]

NotImplementedError is the wrong exception type here.

[rnephew (Reviews Here), 2016/06/01 22:14:15]

Done.

+
+    threads = []
+    results = []
+    for x in xrange(min(len(self._devices), len(test_buckets))):
+      new_shard = TestShard(self._test_instance, self._devices[x], x,
+                            test_buckets[x], results, watcher=self._watcher)
+      threads.append(reraiser_thread.ReraiserThread(new_shard.RunTestsOnShard))
+
+    workers = reraiser_thread.ReraiserThreadGroup(threads)
+    workers.StartAll()
+
+    try:
+      workers.JoinAll(self._watcher)
+    except device_errors.CommandFailedError:
+      logging.exception('Command failed on device.')
+    except device_errors.CommandTimeoutError:
+      logging.exception('Command timed out on device.')
+    except device_errors.DeviceUnreachableError:
+      logging.exception('Device became unreachable.')
+    return results
+
+  # override
+  def TestPackage(self):
+    return 'Perf'
+
+  # override
+  def _CreateShards(self, _tests):
+    raise NotImplementedError
+
+  # override
+  def _GetTests(self):
+    return self._test_buckets
+
+  # override
+  def _RunTest(self, _device, _test):
+    raise NotImplementedError
+
+  # override
+  def _ShouldShard(self):
+    return False