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.
 
@@ skipping 369 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

[M-A Ruel, 2016/06/06 23:34:51]

if you want to make this more readable, do:

to = None
if timeout:
  to = max(to - (time.time() - last_yield), 0)

[nodir, 2016/06/07 18:46:35]

to be clear: the purpose of the change was to allow timeout() function to return
None, not to make it more readable. Still I've adapted "if" statement; it looks
better now

+      to = max(to - (time.time() - last_yield), 0) if to is not None else None
       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
@@ skipping 195 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')].
+
+  If a chunk of data from one pipe does not end with the separator, and next

[Vadim Sh., 2016/06/06 21:32:14]

I don't think that's a good idea (see below, also the code that uses yield_lines
doesn't seem to rely on this).

Also consider cleanly separating it into two processors:

def split(....):
  splitters = defaultdict(splitter)
  for pipe, chunk in data:
    for line in splitters[pipe].feed():
      yield pipe, line
  for pipe, splitter in sorted(splitters.iteritems()):
    for line in splitter.flush():
      yield pipe, line

[nodir, 2016/06/07 18:46:35]

the only reason I did this is to buffer less, but I guess it is fine.

done

+  chunk comes from another pipe, the end of the first chunk is returned as is.
+  E.g. [('stdout', 'a\nb'), ('stderr', 'c')] ->
+  [('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.
+  """
+  pending_chunks = None
+  pending_chunks_pipe_name = None
+
+  for pipe_name, chunk in data:
+    start = 0  # offset in chunk to start |sep| search from
+
+    if pending_chunks:
+      if pending_chunks_pipe_name != pipe_name:

[Vadim Sh., 2016/06/06 21:32:14]

why? interleaving of stdout and stderr in general is "random" (depends on CPU
scheduling delays), it will produce "random" splits in the output

[nodir, 2016/06/07 18:46:35]

removed, simplified

+        # don't buffer more than one pipe at a time.
+        # yield pending chunk as is and forget.
+        yield pending_chunks_pipe_name, ''.join(pending_chunks)
+        pending_chunks = None
+        pending_chunks_pipe_name = None
+      else:
+        # it is the same pipe
+
+        j = chunk.find(sep)
+        if j == -1:
+          # this chunk is incomplete either
+          pending_chunks.append(chunk)
+          continue
+
+        # beginning of this chunk is a continuation of the pending one.
+        pending_chunks.append(chunk[:j])
+        # now the pending chunk is complete: yield and forget
+        yield pending_chunks_pipe_name, ''.join(pending_chunks)
+        pending_chunks = None
+        pending_chunks_pipe_name = None
+
+        start = j + 1
+
+    # split the chunk and yield parts that are followed by sep
+    while start < len(chunk):
+      j = chunk.find(sep, start)
+      if j != -1:
+        yield pipe_name, chunk[start:j]
+        start = j + 1
+      else:
+        # last part is incomplete
+        pending_chunks = [chunk[start:]]
+        pending_chunks_pipe_name = pipe_name
+        break
+  # data is exhausted
+
+  if pending_chunks:
+    yield pending_chunks_pipe_name, ''.join(pending_chunks)