Fix bug where multiprocessing.Queue hangs on large outputs.

au_test_harness/parallel_test_job.py

 # Copyright (c) 2011 The Chromium OS Authors. All rights reserved.
 # Use of this source code is governed by a BSD-style license that can be
 # found in the LICENSE file.
 
 """Module containing methods/classes related to running parallel test jobs."""
 
 import multiprocessing
 import sys
 import time
 
@@ skipping 71 matching lines @@
       if parallel_job.is_alive():
         parallel_job.terminate()
 
     raise ParallelJobTimeoutError('Exceeded max time of %d seconds to wait for '
                                   'job completion.' % cls.MAX_TIMEOUT_SECONDS)
 
   def __str__(self):
     return '%s(%s)' % (self._target, self._args)
 
 
-def RunParallelJobs(number_of_simultaneous_jobs, jobs, jobs_args,
-                    print_status):
+def RunParallelJobs(number_of_simultaneous_jobs, jobs, jobs_args):
   """Runs set number of specified jobs in parallel.
 
+  Note that there is a bug in Python Queue implementation that doesn't
+  allow arbitrary sizes to be returned.  Instead, the process will just
+  appear to have hung.  Be careful when accepting output.
+
   Args:
     number_of_simultaneous_jobs:  Max number of parallel_jobs to be run in
       parallel.
     jobs:  Array of methods to run.
     jobs_args:  Array of args associated with method calls.
-    print_status: True if you'd like this to print out .'s as it runs jobs.
   Returns:
-    Returns an array of results corresponding to each parallel_job.
+    Returns an array of results corresponding to each parallel_job's output.
   """
-  def ProcessOutputWrapper(func, args, output):
+  def ProcessOutputWrapper(func, args, output_queue):
     """Simple function wrapper that puts the output of a function in a queue."""
     try:
-      output.put(func(*args))
+      output_queue.put(func(*args))
     except:
-      output.put('')
+      output_queue.put(None)
       raise
     finally:
-      output.close()
+      output_queue.close()
 
   assert len(jobs) == len(jobs_args), 'Length of args array is wrong.'
   # Cache sudo access.
   cros_lib.RunCommand(['sudo', 'echo', 'Caching sudo credentials'],
                       print_cmd=False, redirect_stdout=True,
                       redirect_stderr=True)
 
   parallel_jobs = []
   output_array = []
 
@@ skipping 10 matching lines @@
     output_array.append(output)
 
   # We use a semaphore to ensure we don't run more jobs than required.
   # After each parallel_job finishes, it releases (increments semaphore).
   for next_parallel_job in parallel_jobs:
     job_start_semaphore.acquire(block=True)
     next_parallel_job.start()
 
   ParallelJob.WaitUntilJobsComplete(parallel_jobs)
   return [output.get() for output in output_array]