[tools/perf] Factor out common code into util.ApplyInParallel

Bug: 981349
Change-Id: I3a9231abd80cfb9098ac3b8e7d261e70547a0c88
Reviewed-on: https://chromium-review.googlesource.com/c/chromium/src/+/1852245Reviewed-by: Juan Antonio Navarro Pérez <perezju@chromium.org>
Commit-Queue: Mikhail Khokhlov <khokhlov@google.com>
Cr-Commit-Position: refs/heads/master@{#704573}

tools/perf/core/results_processor/compute_metrics.py

@@ -25,43 +25,35 @@ HISTOGRAM_DICTS_FILE = 'histogram_dicts.json'
 
 
 def _PoolWorker(test_result):
-  try:
-    metrics = [tag['value'] for tag in test_result['tags']
-               if tag['key'] == 'tbmv2']
-    html_trace = test_result['outputArtifacts'][HTML_TRACE_NAME]
-    html_local_path = html_trace['filePath']
-    html_remote_url = html_trace['remoteUrl']
-
-    logging.info('%s: Starting to compute metrics on trace.',
-                 test_result['testPath'])
-    start = time.time()
-    # The timeout needs to be coordinated with the Swarming IO timeout for the
-    # task that runs this code. If this timeout is longer or close in length
-    # to the swarming IO timeout then we risk being forcibly killed for not
-    # producing any output. Note that this could be fixed by periodically
-    # outputting logs while waiting for metrics to be calculated.
-    TEN_MINUTES = 60 * 10
-    mre_result = metric_runner.RunMetricOnSingleTrace(
-        html_local_path, metrics, canonical_url=html_remote_url,
-        timeout=TEN_MINUTES,
-        extra_import_options={'trackDetailedModelStats': True})
-    logging.info('%s: Computing metrics took %.3f seconds.' % (
-        test_result['testPath'], time.time() - start))
-
-    if mre_result.failures:
-      test_result['status'] = 'FAIL'
-      for f in mre_result.failures:
-        logging.error('Failure recorded for test %s: %s',
-                      test_result['testPath'], f)
-
-    return mre_result.pairs.get('histograms', [])
-  except Exception:  # pylint: disable=broad-except
-    # logging exception here is the only way to get a stack trace since
-    # multiprocessing's pool implementation does not save that data. See
-    # crbug.com/953365.
-    logging.exception('%s: Exception while calculating metric' %
-                      test_result['testPath'])
-    raise
+  metrics = [tag['value'] for tag in test_result['tags']
+             if tag['key'] == 'tbmv2']
+  html_trace = test_result['outputArtifacts'][HTML_TRACE_NAME]
+  html_local_path = html_trace['filePath']
+  html_remote_url = html_trace.get('remoteUrl')
+
+  logging.info('%s: Starting to compute metrics on trace.',
+               test_result['testPath'])
+  start = time.time()
+  # The timeout needs to be coordinated with the Swarming IO timeout for the
+  # task that runs this code. If this timeout is longer or close in length
+  # to the swarming IO timeout then we risk being forcibly killed for not
+  # producing any output. Note that this could be fixed by periodically
+  # outputting logs while waiting for metrics to be calculated.
+  TEN_MINUTES = 60 * 10
+  mre_result = metric_runner.RunMetricOnSingleTrace(
+      html_local_path, metrics, canonical_url=html_remote_url,
+      timeout=TEN_MINUTES,
+      extra_import_options={'trackDetailedModelStats': True})
+  logging.info('%s: Computing metrics took %.3f seconds.' % (
+      test_result['testPath'], time.time() - start))
+
+  if mre_result.failures:
+    test_result['status'] = 'FAIL'
+    for f in mre_result.failures:
+      logging.error('Failure recorded for test %s: %s',
+                    test_result['testPath'], f)
+
+  return mre_result.pairs.get('histograms', [])
 
 
 def ComputeTBMv2Metrics(intermediate_results):
@@ -104,9 +96,6 @@ def ComputeTBMv2Metrics(intermediate_results):
 
     work_list.append(test_result)
 
-  if not work_list:
-    return histogram_dicts
-
   for dicts in util.ApplyInParallel(_PoolWorker, work_list):
     histogram_dicts += dicts
 

tools/perf/core/results_processor/processor.py

@@ -155,9 +155,6 @@ def UploadArtifacts(intermediate_results, upload_bucket, results_label):
       remote_name = '/'.join([run_identifier, result['testPath'], name])
       work_list.append((artifact, remote_name))
 
-  if not work_list:
-    return
-
   def PoolUploader(work_item):
     artifact, remote_name = work_item
     artifact['remoteUrl'] = cloud_storage.Insert(

tools/perf/core/results_processor/util.py

@@ -18,6 +18,9 @@ def ApplyInParallel(function, work_list):
     A generator over results. The order of results might not match the
     order of the arguments in the work_list.
   """
+  if not work_list:
+    return
+
   try:
     # Note that this is speculatively halved as an attempt to fix
     # crbug.com/953365.
@@ -28,8 +31,18 @@ def ApplyInParallel(function, work_list):
     cpu_count = 4
   pool = ThreadPool(min(cpu_count, len(work_list)))
 
+  def function_with_try(arg):
+    try:
+      return function(arg)
+    except Exception:  # pylint: disable=broad-except
+      # logging exception here is the only way to get a stack trace since
+      # multiprocessing's pool implementation does not save that data. See
+      # crbug.com/953365.
+      logging.exception('Exception while running %s' % function.__name__)
+      raise
+
   try:
-    for result in pool.imap_unordered(function, work_list):
+    for result in pool.imap_unordered(function_with_try, work_list):
       yield result
     pool.close()
     pool.join()

tools/perf/core/results_processor/util_unittest.py

@@ -13,3 +13,11 @@ class UtilTests(unittest.TestCase):
     fun = lambda x: x * x
     result = set(util.ApplyInParallel(fun, work_list))
     self.assertEqual(result, set([1, 4, 9]))
+
+  def testApplyInParallelExceptionRaised(self):
+    work_list = [1, 2, 3]
+    def fun(x):
+      if x == 3:
+        raise RuntimeError()
+    with self.assertRaises(RuntimeError):
+      list(util.ApplyInParallel(fun, work_list))