run_isolated.py: install CIPD packages

client/utils/subprocess42.py

 # Copyright 2013 The LUCI Authors. All rights reserved.
 # Use of this source code is governed under the Apache License, Version 2.0
 # that can be found in the LICENSE file.
 
 """subprocess42 is the answer to life the universe and everything.
 
 It has the particularity of having a Popen implementation that can yield output
 as it is produced while implementing a timeout and NOT requiring the use of
 worker threads.
 
 Example:
   Wait for a child process with a timeout, send SIGTERM, wait a grace period
   then send SIGKILL:
 
     def wait_terminate_then_kill(proc, timeout, grace):
       try:
         return proc.wait(timeout)
       except subprocess42.TimeoutExpired:
         proc.terminate()
         try:
           return proc.wait(grace)
         except subprocess42.TimeoutExpired:
           proc.kill()
         return proc.wait()
 
 
 TODO(maruel): Add VOID support like subprocess2.
 """
 
+import collections
 import contextlib
 import errno
 import os
 import signal
 import sys
 import threading
 import time
 
 import subprocess
 
@@ skipping 340 matching lines @@
       # communicate() uses wait() internally.
       self.end = time.time()
     return self.returncode
 
   def poll(self):
     ret = super(Popen, self).poll()
     if ret is not None and not self.end:
       self.end = time.time()
     return ret
 
+  def yield_any_line(self, **kwargs):
+    """Yields lines until the process terminates.
+
+    Like yield_any, but yields lines.
+    """
+    return split(self.yield_any(**kwargs))
+
   def yield_any(self, maxsize=None, timeout=None):
     """Yields output until the process terminates.
 
     Unlike wait(), does not raise TimeoutExpired.
 
     Yields:
       (pipename, data) where pipename is either 'stdout', 'stderr' or None in
       case of timeout or when the child process closed one of the pipe(s) and
       all pending data on the pipe was read.
 
@@ skipping 11 matching lines @@
         old_timeout = timeout
         timeout = lambda: old_timeout
       else:
         assert callable(timeout), timeout
 
     if maxsize is not None and not callable(maxsize):
       assert isinstance(maxsize, (int, float)), maxsize
 
     last_yield = time.time()
     while self.poll() is None:
-      to = (None if timeout is None
-            else max(timeout() - (time.time() - last_yield), 0))
+      to = timeout() if timeout else None
+      if to is not None:
+        to = max(to - (time.time() - last_yield), 0)
       t, data = self.recv_any(
           maxsize=maxsize() if callable(maxsize) else maxsize, timeout=to)
       if data or to is 0:
         yield t, data
         last_yield = time.time()
 
     # Read all remaining output in the pipes.
     # There is 3 cases:
     # - pipes get closed automatically by the calling process before it exits
     # - pipes are closed automated by the OS
     # - pipes are kept open due to grand-children processes outliving the
     #   children process.
     while True:
       ms = maxsize
       if callable(maxsize):
         ms = maxsize()
       # timeout=0 is mainly to handle the case where a grand-children process
       # outlives the process started.
       t, data = self.recv_any(maxsize=ms, timeout=0)
       if not data:
         break
       yield t, data
 
   def recv_any(self, maxsize=None, timeout=None):
     """Reads from the first pipe available from stdout and stderr.
 
-    Unlike wait(), it does not throw TimeoutExpired.
+    Unlike wait(), does not throw TimeoutExpired.
 
     Arguments:
     - maxsize: Maximum number of bytes to return. Defaults to MAX_SIZE.
     - timeout: If None, it is blocking. If 0 or above, will return None if no
           data is available within |timeout| seconds.
 
     Returns:
       tuple(pipename or None, str(data)). pipename is one of 'stdout' or
       'stderr'.
     """
@@ skipping 168 matching lines @@
       SEM_NOGPFAULTERRORBOX = 2
       SEM_NOALIGNMENTFAULTEXCEPT = 0x8000
       ctypes.windll.kernel32.SetErrorMode(
           SEM_FAILCRITICALERRORS|SEM_NOGPFAULTERRORBOX|
             SEM_NOALIGNMENTFAULTEXCEPT)
   # TODO(maruel): Other OSes.
   # - OSX, need to figure out a way to make the following process tree local:
   #     defaults write com.apple.CrashReporter UseUNC 1
   #     defaults write com.apple.CrashReporter DialogType none
   # - Ubuntu, disable apport if needed.
+
+
+def split(data, sep='\n'):
+  """Splits pipe data by |sep|. Does some buffering.
+
+  For example, [('stdout', 'a\nb'), ('stdout', '\n'), ('stderr', 'c\n')] ->
+  [('stdout', 'a'), ('stdout', 'b'), ('stderr', 'c')].
+
+  Args:
+    data: iterable of tuples (pipe_name, bytes).
+
+  Returns:
+    An iterator of tuples (pipe_name, bytes) where bytes is the input data
+    but split by sep into separate tuples.
+  """
+  # A dict {pipe_name -> list of pending chunks without separators}
+  pending_chunks = collections.defaultdict(list)
+  for pipe_name, chunk in data:
+    if chunk is None:
+      # Happens if a pipe is closed.
+      continue
+
+    pending = pending_chunks[pipe_name]
+    start = 0  # offset in chunk to start |sep| search from
+    while start < len(chunk):
+      j = chunk.find(sep, start)
+      if j == -1:
+        pending_chunks[pipe_name].append(chunk[start:])
+        break
+
+      to_emit = chunk[start:j]
+      start = j + 1
+      if pending:
+        # prepend and forget
+        to_emit = ''.join(pending) + to_emit
+        pending = []
+        pending_chunks[pipe_name] = pending
+      yield pipe_name, to_emit
+
+  # Emit remaining chunks that don't end with separators as is.
+  for pipe_name, chunks in sorted(pending_chunks.iteritems()):
+    if chunks:
+      yield pipe_name, ''.join(chunks)