Reland of 'Make allow_subannotations more robust'

scripts/common/annotator.py

Index: scripts/common/annotator.py
diff --git a/scripts/common/annotator.py b/scripts/common/annotator.py
index 497f7c131169bc12dee48772604c91ab43f6ed04..eb2a06446ed51973980624b8c3473209767ab5ed 100755
--- a/scripts/common/annotator.py
+++ b/scripts/common/annotator.py
@@ -464,21 +464,25 @@ def triggerBuilds(step, trigger_specs):
 
 
 def run_step(stream, name, cmd,
+             step_annotation,
              cwd=None, env=None,
-             allow_subannotations=False,
+             subannotator=None,
              trigger_specs=None,
              **kwargs):
   """Runs a single step.
 
   Context:
     stream: StructuredAnnotationStream to use to emit step
+    step_annotation: optional StructuredAnnotationStep to use instead
+        of creating one
 
   Step parameters:
     name: name of the step, will appear in buildbots waterfall
     cmd: command to run, list of one or more strings
     cwd: absolute path to working directory for the command
     env: dict with overrides for environment variables
-    allow_subannotations: if True, lets the step emit its own annotations
+    subannotator: a callback_implementor class used to parse subannotations
+        that the command emits; if None, subannotations will be suppressed.
     trigger_specs: a list of trigger specifications, which are dict with keys:
         properties: a dict of properties.
             Buildbot requires buildername property.
@@ -503,13 +507,10 @@ def run_step(stream, name, cmd,
       'cmd': cmd,
       'cwd': cwd,
       'env': env,
-      'allow_subannotations': allow_subannotations,
+      'allow_subannotations': subannotator is not None,
       })
   step_env = _merge_envs(os.environ, env)
 
-  step_annotation = stream.step(name)
-  step_annotation.step_started()
-
   print_step(step_dict, step_env, stream)
   returncode = 0
   if cmd:
@@ -557,19 +558,22 @@ def run_step(stream, name, cmd,
           handle.close()
 
       outlock = threading.Lock()
-      def filter_lines(lock, allow_subannotations, inhandle, outhandle):
+      def filter_lines(inhandle, outhandle):
         while True:
           line = inhandle.readline()
           if not line:
             break
-          lock.acquire()
-          try:
-            if not allow_subannotations and line.startswith('@@@'):
-              outhandle.write('!')
-            outhandle.write(line)
+          with outlock:
+            if line.startswith('@@@'):
+              if subannotator:
+                # The subannotator might write to the handle, thus the lock.
+                MatchAnnotation(line.strip(), subannotator)
+              else:
+                outhandle.write('!')
+                outhandle.write(line)
+            else:
+              outhandle.write(line)
             outhandle.flush()
-          finally:
-            lock.release()
 
       # Pump piped stdio through filter_lines. IO going to files on disk is
       # not filtered.
@@ -580,7 +584,7 @@ def run_step(stream, name, cmd,
           outhandle = getattr(sys, key)
           threads.append(threading.Thread(
               target=filter_lines,
-              args=(outlock, allow_subannotations, inhandle, outhandle)))
+              args=(inhandle, outhandle)))
 
       for th in threads:
         th.start()
@@ -598,7 +602,7 @@ def run_step(stream, name, cmd,
   if trigger_specs:
     triggerBuilds(step_annotation, trigger_specs)
 
-  return step_annotation, returncode
+  return returncode
 
 def update_build_failure(failure, retcode, **_kwargs):
   """Potentially moves failure from False to True, depending on returncode of
@@ -629,7 +633,9 @@ def run_steps(steps, build_failure):
     if build_failure and not step.get('always_run', False):
       ret = None
     else:
-      prev_annotation, ret = run_step(stream, **step)
+      prev_annotation = stream.step(step['name'])
+      prev_annotation.step_started()
+      ret = run_step(stream, step_annotation=prev_annotation, **step)
       stream = prev_annotation.annotation_stream
       if ret > 0:
         stream.step_cursor(stream.current_step)