Enable continuous parsing and add host names to global autoserv.DEBUG entries.

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, machine, work_dir, test_queue, queue_lock):
+    def __init__(self, server_job, machine, work_dir, test_queue, queue_lock):
         """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.
         """
         threading.Thread.__init__(self)
+        self._server_job = server_job
         self._test_queue = test_queue
         self._test_queue_lock = queue_lock
         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)
@@ skipping 37 matching lines @@
                     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])
+        sub_cmd = subcommand.subcommand(self.subcommand_wrapper,
+                                        [active_test],
+                                        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."""
+        """Callback for subcommand to call into with the test parameter.
+
+        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.
+        """
+        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()
 
     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
 
             logging.info('%s running %s', self._machine, active_test)
             try:
                 self.run_subcommand(active_test)
             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
 
         logging.info('%s completed %d tests.', self._machine, self._tests_run)