Make test runner more chatty to avoid it getting killed by buildbot.

tools/testrunner/local/pool.py

 #!/usr/bin/env python
 # Copyright 2014 the V8 project authors. All rights reserved.
 # Use of this source code is governed by a BSD-style license that can be
 # found in the LICENSE file.
 
+from Queue import Empty
 from multiprocessing import Event, Process, Queue
+import sys
 
 class NormalResult():
   def __init__(self, result):
     self.result = result
     self.exception = False
     self.break_now = False
 
 
 class ExceptionResult():
   def __init__(self):
     self.exception = True
     self.break_now = False
 
 
 class BreakResult():
   def __init__(self):
     self.exception = False
     self.break_now = True
 
 
+class Heartbeat():

[Jakob Kummerow, 2015/04/08 09:00:34]

I'm not too happy with this kind of polymorphism. I'd use a single class
("MaybeResult"?), that always contains the same set of fields, and has
convenience factory functions to construct "heartbeat" and "actual result"
instances. This would be similar to e.g. MaybeHandles in our C++ code.
I guess it's fine to keep this as is though, if you prefer.

+  def __init__(self):
+    self.heartbeat = True
+
+
+class PoolResult():
+  def __init__(self, value):
+    self.heartbeat = False
+    self.value = value
+
+
 def Worker(fn, work_queue, done_queue, done):
   """Worker to be run in a child process.
   The worker stops on two conditions. 1. When the poison pill "STOP" is
   reached or 2. when the event "done" is set."""
   try:
     for args in iter(work_queue.get, "STOP"):
       if done.is_set():
         break
       try:
         done_queue.put(NormalResult(fn(*args)))
       except Exception, e:
         print(">>> EXCEPTION: %s" % e)
         done_queue.put(ExceptionResult())
   except KeyboardInterrupt:
     done_queue.put(BreakResult())
 
 
 class Pool():
   """Distributes tasks to a number of worker processes.
   New tasks can be added dynamically even after the workers have been started.
   Requirement: Tasks can only be added from the parent process, e.g. while
   consuming the results generator."""
 
   # Factor to calculate the maximum number of items in the work/done queue.
   # Necessary to not overflow the queue's pipe if a keyboard interrupt happens.
   BUFFER_FACTOR = 4
 
-  def __init__(self, num_workers):
+  def __init__(self, num_workers, heartbeat_timeout=30):
     self.num_workers = num_workers
     self.processes = []
     self.terminated = False
 
     # Invariant: count >= #work_queue + #done_queue. It is greater when a
     # worker takes an item from the work_queue and before the result is
     # submitted to the done_queue. It is equal when no worker is working,
     # e.g. when all workers have finished, and when no results are processed.
     # Count is only accessed by the parent process. Only the parent process is
     # allowed to remove items from the done_queue and to add items to the
     # work_queue.
     self.count = 0
     self.work_queue = Queue()
     self.done_queue = Queue()
     self.done = Event()
+    self.heartbeat_timeout = heartbeat_timeout
 
   def imap_unordered(self, fn, gen):
     """Maps function "fn" to items in generator "gen" on the worker processes
     in an arbitrary order. The items are expected to be lists of arguments to
-    the function. Returns a results iterator."""
+    the function. Returns a results iterator. A result value can either be a
+    hearbeat of the runner, i.e. indicating that the runner is still waiting

[Jakob Kummerow, 2015/04/08 09:00:34]

nit: "heartbeat"

+    for the result to be computed, or it can be a result wrapper of type
+    PoolResult."""
     try:
       gen = iter(gen)
       self.advance = self._advance_more
 
       for w in xrange(self.num_workers):
         p = Process(target=Worker, args=(fn,
                                          self.work_queue,
                                          self.done_queue,
                                          self.done))
         self.processes.append(p)
         p.start()
 
       self.advance(gen)
       while self.count > 0:
-        result = self.done_queue.get()
+        while True:
+          try:
+            result = self.done_queue.get(timeout=self.heartbeat_timeout)
+            break
+          except Empty:
+            # Indicate a heartbeat. The iterater will continue fetching the

[Jakob Kummerow, 2015/04/08 09:00:34]

nit: "iterator"

+            # next result.
+            yield Heartbeat()
         self.count -= 1
         if result.exception:
           # Ignore items with unexpected exceptions.
           continue
         elif result.break_now:
           # A keyboard interrupt happened in one of the worker processes.
           raise KeyboardInterrupt
         else:
-          yield result.result
+          yield PoolResult(result.result)
         self.advance(gen)
     finally:
       self.terminate()
 
   def _advance_more(self, gen):
     while self.count < self.num_workers * self.BUFFER_FACTOR:
       try:
         self.work_queue.put(gen.next())
         self.count += 1
       except StopIteration:
@@ skipping 28 matching lines @@
 
     # Drain the queues to prevent failures when queues are garbage collected.
     try:
       while True: self.work_queue.get(False)
     except:
       pass
     try:
       while True: self.done_queue.get(False)
     except:
       pass