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

server/server_job_utils.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.
 
 """
 Utility classes used by server_job.distribute_across_machines().
 
 test_item: extends the basic test tuple to add include/exclude attributes.
 
 machine_worker: is a thread that manages running tests on a host.  It
@@ skipping 62 matching lines @@
         if self.inc_set is not None:
             if not self.inc_set <= machine_attributes: return False
         if self.exc_set is not None:
             if self.exc_set & machine_attributes: return False
         return True
 
 
 class machine_worker(threading.Thread):
     """Thread that runs tests on a remote host machine."""
 
-    def __init__(self, server_job, machine, work_dir, test_queue, queue_lock):
+    def __init__(self, server_job, machine, work_dir, test_queue, queue_lock,
+                 continuous_parsing=False):
         """Creates an instance of machine_worker to run tests on a remote host.
 
         Retrieves that host attributes for this machine and creates the set of
         True attributes to validate against test include/exclude attributes.
 
         Creates a directory to hold the log files for tests run and writes the
         hostname and tko parser version into keyvals file.
 
         Args:
             server_job: run tests for this server_job.
             machine: name of remote host.
             work_dir: directory server job is using.
             test_queue: queue of tests.
             queue_lock: lock protecting test_queue.
+            continuous_parsing: bool, enable continuous parsing.
         """
         threading.Thread.__init__(self)
         self._server_job = server_job
         self._test_queue = test_queue
         self._test_queue_lock = queue_lock
+        self._continuous_parsing = continuous_parsing
         self._tests_run = 0
         self._machine = machine
         self._host = hosts.create_host(self._machine)
         self._client_at = autotest.Autotest(self._host)
         client_attributes = host_attributes.host_attributes(machine)
         self.attribute_set = set([key for key, value in
                                   client_attributes.__dict__.items() if value])
         self._results_dir = os.path.join(work_dir, self._machine)
         if not os.path.exists(self._results_dir):
             os.makedirs(self._results_dir)
@@ skipping 33 matching lines @@
 
                 except Queue.Empty:
                     break
 
             # Return any skipped tests to the queue.
             for st in skipped_tests:
                 self._test_queue.put(st)
 
         return good_test
 
-    def run_subcommand(self, active_test):
-        """Use subcommand to fork process and execute test."""
-        sub_cmd = subcommand.subcommand(self.subcommand_wrapper,
-                                        [active_test],
+    def run(self):
+        """Use subcommand to fork process and execute tests.
+
+        The forked processes prevents log files from simultaneous tests
+        interweaving with each other. Logging doesn't communicate host autotest
+        to client autotest, it communicates host module to client autotest.  So
+        different server side autotest instances share the same module and
+        require split processes to have clean logging.
+        """
+        sub_cmd = subcommand.subcommand(self._run,
+                                        [],
                                         self._results_dir)
         sub_cmd.fork_start()
         sub_cmd.fork_waitfor()
 
-    def subcommand_wrapper(self, active_test):
-        """Callback for subcommand to call into with the test parameter.
+    def _run(self):
+        """Executes tests on the host machine.
 
-        When this function executes it has forked from the main process so it
-        is safe to modify state on the server_job object. These changes enable
-        continuous parsing which communicates results back to the database
-        while the server_job is running instead of only when the server_job is
-        complete.
+        If continuous parsing was requested, start the parser before running
+        tests.
         """
-        self._server_job._parse_job += "/" + self._machine
-        self._server_job._using_parser = True
-        self._server_job.machines = [self._machine]
-        self._server_job.push_execution_context(self._machine)
-        self._server_job.init_parser()
-        self._client_at.run_test(active_test.test_name,
-                                 results_dir=self._results_dir,
-                                 **active_test.test_args)
-        self._server_job.cleanup_parser()
+        if self._continuous_parsing:
+            self._server_job._parse_job += "/" + self._machine
+            self._server_job._using_parser = True
+            self._server_job.machines = [self._machine]
+            self._server_job.push_execution_context(self._machine)
+            self._server_job.init_parser()
 
-    def run(self):
-        """Executes tests on host machine.
-
-        Uses subprocess to fork the process when running tests so unique client
-        jobs talk to unique server jobs which prevents log files from
-        simultaneous tests interweaving with each other.
-        """
         while True:
             active_test = self.get_test()
             if active_test is None:
                 break
 
-            # Install autoest on host before running tests.  Do this before
-            # the subcommand fork so all future forks see that it has been
-            # installed on the host.
-            if not self._client_at.installed:
-                self._client_at.install()
-
             logging.info('%s running %s', self._machine, active_test)
             try:
-                self.run_subcommand(active_test)
+                self._client_at.run_test(active_test.test_name,
+                                         results_dir=self._results_dir,
+                                         **active_test.test_args)
             except (error.AutoservError, error.AutotestError):
                 logging.exception('Error running test "%s".', active_test)
             except Exception:
                 logging.exception('Exception running test "%s".', active_test)
                 raise
             finally:
                 self._test_queue.task_done()
                 self._tests_run += 1
 
+        if self._continuous_parsing:
+            self._server_job.cleanup_parser()
         logging.info('%s completed %d tests.', self._machine, self._tests_run)