forall: rewrite parallel logic

This fixes intermingling of parallel jobs and simplifies the code
by switching to subprocess.run.  This also provides stable output
in the order of projects by returning the output as a string that
the main loop outputs.

This drops support for interactive commands, but it's unclear if
anyone was relying on that, and the default behavior (-j2) made
that unreliable.  If it turns out someone still wants this, we can
look at readding it.

Change-Id: I7555b4e7a15aad336667292614f730fb7a90bd26
Reviewed-on: https://gerrit-review.googlesource.com/c/git-repo/+/297482
Reviewed-by: Chris Mcdonald <cjmcdonald@google.com>
Tested-by: Mike Frysinger <vapier@google.com>

subcmds/forall.py

@@ -13,6 +13,7 @@
 # limitations under the License.
 
 import errno
+import io
 import multiprocessing
 import re
 import os
@@ -22,7 +23,6 @@ import subprocess
 
 from color import Coloring
 from command import DEFAULT_LOCAL_JOBS, Command, MirrorSafeCommand, WORKER_BATCH_SIZE
-import platform_utils
 
 _CAN_COLOR = [
     'branch',
@@ -241,7 +241,18 @@ without iterating through the remaining projects.
             DoWorkWrapper,
             self.ProjectArgs(projects, mirror, opt, cmd, shell, config),
             chunksize=WORKER_BATCH_SIZE)
-        for r in results_it:
+        first = True
+        for (r, output) in results_it:
+          if output:
+            if first:
+              first = False
+            elif opt.project_header:
+              print()
+            # To simplify the DoWorkWrapper, take care of automatic newlines.
+            end = '\n'
+            if output[-1] == '\n':
+              end = ''
+            print(output, end=end)
           rc = rc or r
           if r != 0 and opt.abort_on_errors:
             raise Exception('Aborting due to previous error')
@@ -326,73 +337,36 @@ def DoWork(project, mirror, opt, cmd, shell, cnt, config):
     # Allow the user to silently ignore missing checkouts so they can run on
     # partial checkouts (good for infra recovery tools).
     if opt.ignore_missing:
-      return 0
+      return (0, '')
+
+    output = ''
     if ((opt.project_header and opt.verbose)
             or not opt.project_header):
-      print('skipping %s/' % project['relpath'], file=sys.stderr)
+      output = 'skipping %s/' % project['relpath']
-    return 1
+    return (1, output)
 
-  if opt.project_header:
+  if opt.verbose:
-    stdin = subprocess.PIPE
+    stderr = subprocess.STDOUT
-    stdout = subprocess.PIPE
-    stderr = subprocess.PIPE
   else:
-    stdin = None
+    stderr = subprocess.DEVNULL
-    stdout = None
-    stderr = None
 
-  p = subprocess.Popen(cmd,
+  result = subprocess.run(
-                       cwd=cwd,
+      cmd, cwd=cwd, shell=shell, env=env, check=False,
-                       shell=shell,
+      encoding='utf-8', errors='replace',
-                       env=env,
+      stdin=subprocess.DEVNULL, stdout=subprocess.PIPE, stderr=stderr)
-                       stdin=stdin,
-                       stdout=stdout,
-                       stderr=stderr)
 
+  output = result.stdout
   if opt.project_header:
-    out = ForallColoring(config)
+    if output:
-    out.redirect(sys.stdout)
+      buf = io.StringIO()
-    empty = True
+      out = ForallColoring(config)
-    errbuf = ''
+      out.redirect(buf)
-
+      if mirror:
-    p.stdin.close()
+        project_header_path = project['name']
-    s_in = platform_utils.FileDescriptorStreams.create()
+      else:
-    s_in.add(p.stdout, sys.stdout, 'stdout')
+        project_header_path = project['relpath']
-    s_in.add(p.stderr, sys.stderr, 'stderr')
+      out.project('project %s/' % project_header_path)
-
+      out.nl()
-    while not s_in.is_done:
+      buf.write(output)
-      in_ready = s_in.select()
+      output = buf.getvalue()
-      for s in in_ready:
+  return (result.returncode, output)
-        buf = s.read().decode()
-        if not buf:
-          s_in.remove(s)
-          s.close()
-          continue
-
-        if not opt.verbose:
-          if s.std_name == 'stderr':
-            errbuf += buf
-            continue
-
-        if empty and out:
-          if not cnt == 0:
-            out.nl()
-
-          if mirror:
-            project_header_path = project['name']
-          else:
-            project_header_path = project['relpath']
-          out.project('project %s/', project_header_path)
-          out.nl()
-          out.flush()
-          if errbuf:
-            sys.stderr.write(errbuf)
-            sys.stderr.flush()
-            errbuf = ''
-          empty = False
-
-        s.dest.write(buf)
-        s.dest.flush()
-
-  r = p.wait()
-  return r