[test-runner] Move test case processing beyond the multi-process boundary.

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 traceback
+
 
 class NormalResult():
   def __init__(self, result):
     self.result = result
     self.exception = False
     self.break_now = False
 
 
 class ExceptionResult():
   def __init__(self):
@@ skipping 14 matching lines @@
 
   @staticmethod
   def create_heartbeat():
     return MaybeResult(True, None)
 
   @staticmethod
   def create_result(value):
     return MaybeResult(False, value)
 
 
-def Worker(fn, work_queue, done_queue, done):
+def Worker(fn, work_queue, done_queue, done,
+           process_context_fn=None, process_context_args=None):
   """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:
+    kwargs = {}
+    if process_context_fn and process_context_args is not None:
+      kwargs.update(process_context=process_context_fn(*process_context_args))
     for args in iter(work_queue.get, "STOP"):
       if done.is_set():
         break
       try:
-        done_queue.put(NormalResult(fn(*args)))
+        done_queue.put(NormalResult(fn(*args, **kwargs)))
       except Exception, e:
+        traceback.print_exc()
         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
@@ skipping 14 matching lines @@
     # 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):
+  def imap_unordered(self, fn, gen,
+                     process_context_fn=None, process_context_args=None):
     """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. A result value of type
     MaybeResult either indicates a heartbeat of the runner, i.e. indicating
     that the runner is still waiting for the result to be computed, or it wraps
-    the real result."""
+    the real result.
+
+    Args:
+      process_context_fn: Function executed once by each worker. Expected to
+          return a process-context object. If present, this object is passed
+          as additional argument to each call to fn.
+      process_context_args: List of arguments for the invocation of
+          process_context_fn. All arguments will be pickled and sent beyond the
+          process boundary.
+    """
     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.done,
+                                         process_context_fn,
+                                         process_context_args))
         self.processes.append(p)
         p.start()
 
       self.advance(gen)
       while self.count > 0:
         while True:
           try:
             result = self.done_queue.get(timeout=self.heartbeat_timeout)
             break
           except Empty:
@@ skipping 50 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