Reduce load on autotest by cutting number of forks and parsers.

server/server_job.py

 """
 The main job wrapper for the server side.
 
 This is the core infrastructure. Derived from the client side job.py
 
 Copyright Martin J. Bligh, Andy Whitcroft 2007
 """
 
 import getpass, os, sys, re, stat, tempfile, time, select, subprocess, platform
-import Queue, threading
+import multiprocessing
 import traceback, shutil, warnings, fcntl, pickle, logging, itertools, errno
 from autotest_lib.client.bin import sysinfo
 from autotest_lib.client.common_lib import base_job
 from autotest_lib.client.common_lib import error, log, utils, packages
 from autotest_lib.client.common_lib import logging_manager
 from autotest_lib.server import test, subcommand, profilers, server_job_utils
 from autotest_lib.server.hosts import abstract_ssh
 from autotest_lib.tko import db as tko_db, status_lib, utils as tko_utils
 
 
@@ skipping 434 matching lines @@
         """
         results = self.parallel_simple(function, machines, timeout=timeout,
                                        return_results=True)
         success_machines = []
         for result, machine in itertools.izip(results, machines):
             if not isinstance(result, Exception):
                 success_machines.append(machine)
         return success_machines
 
 
-    def distribute_across_machines(self, tests, machines):
+    def distribute_across_machines(self, tests, machines,
+                                   continuous_parsing=False):
         """Run each test in tests once using machines.
 
         Instead of running each test on each machine like parallel_on_machines,
         run each test once across all machines. Put another way, the total
         number of tests run by parallel_on_machines is len(tests) *
         len(machines). The number of tests run by distribute_across_machines is
         len(tests).
 
         Args:
             tests: List of tests to run.
-            machines: list of machines to use.
+            machines: List of machines to use.
+            continuous_parsing: Bool, if true parse job while running.
         """
         # The Queue is thread safe, but since a machine may have to search
         # through the queue to find a valid test the lock provides exclusive
         # queue access for more than just the get call.
-        test_queue = Queue.Queue()
-        test_queue_lock = threading.Lock()
+        test_queue = multiprocessing.JoinableQueue()
+        test_queue_lock = multiprocessing.Lock()
 
         machine_workers = [server_job_utils.machine_worker(self,
                                                            machine,
                                                            self.resultdir,
                                                            test_queue,
-                                                           test_queue_lock)
+                                                           test_queue_lock,
+                                                           continuous_parsing)
                            for machine in machines]
 
         # To (potentially) speed up searching for valid tests create a list of
         # unique attribute sets present in the machines for this job. If sets
         # were hashable we could just use a dictionary for fast verification.
         # This at least reduces the search space from the number of machines to
         # the number of unique machines.
         unique_machine_attributes = []
         for mw in machine_workers:
             if not mw.attribute_set in unique_machine_attributes:
@@ skipping 708 matching lines @@
         intervals = self.disabled_warnings.setdefault(warning_type, [])
         if not intervals or intervals[-1][1] is not None:
             intervals.append((int(current_time_func()), None))
 
 
     def enable_warnings(self, warning_type, current_time_func=time.time):
         """As of now, enables all further warnings of this type."""
         intervals = self.disabled_warnings.get(warning_type, [])
         if intervals and intervals[-1][1] is None:
             intervals[-1] = (intervals[-1][0], int(current_time_func()))