SVN: zope.testing/branches/benji-parallelize-subprocesses/src/zope/testing/testrunner/ checkpoint

Log message for revision 87998:
  checkpoint
  

Changed:
  U   zope.testing/branches/benji-parallelize-subprocesses/src/zope/testing/testrunner/layer.py
  U   zope.testing/branches/benji-parallelize-subprocesses/src/zope/testing/testrunner/options.py
  U   zope.testing/branches/benji-parallelize-subprocesses/src/zope/testing/testrunner/runner.py

-=-
Modified: zope.testing/branches/benji-parallelize-subprocesses/src/zope/testing/testrunner/layer.py
===================================================================
--- zope.testing/branches/benji-parallelize-subprocesses/src/zope/testing/testrunner/layer.py	2008-07-04 08:40:27 UTC (rev 87997)
+++ zope.testing/branches/benji-parallelize-subprocesses/src/zope/testing/testrunner/layer.py	2008-07-04 13:45:39 UTC (rev 87998)
@@ -19,7 +19,3 @@
 
 class UnitTests(object):
     """A layer for gathering all unit tests."""
-
-    @staticmethod
-    def tearDown():
-        raise NotImplementedError

Modified: zope.testing/branches/benji-parallelize-subprocesses/src/zope/testing/testrunner/options.py
===================================================================
--- zope.testing/branches/benji-parallelize-subprocesses/src/zope/testing/testrunner/options.py	2008-07-04 08:40:27 UTC (rev 87997)
+++ zope.testing/branches/benji-parallelize-subprocesses/src/zope/testing/testrunner/options.py	2008-07-04 13:45:39 UTC (rev 87998)
@@ -407,6 +407,20 @@
 other = optparse.OptionGroup(parser, "Other", "Other options")
 
 other.add_option(
+    '--exit-with-status', action="store_true", dest='exitwithstatus',
+    help="""\
+Return an error exit status if the tests failed.  This can be useful for
+an invoking process that wants to monitor the result of a test run.
+""")
+
+other.add_option(
+    '-j', action="store", type="int", dest='processes', default=1,
+    help="""\
+Use up to given number of parallel processes to execute tests.  May decrease
+test run time substantially.  Defaults to %default.
+""")
+
+other.add_option(
     '--keepbytecode', '-k', action="store_true", dest='keepbytecode',
     help="""\
 Normally, the test runner scans the test paths and the test
@@ -431,13 +445,6 @@
 compilation to .pyc/.pyo.  Use of this option implies --keepbytecode.
 """)
 
-other.add_option(
-    '--exit-with-status', action="store_true", dest='exitwithstatus',
-    help="""\
-Return an error exit status if the tests failed.  This can be useful for
-an invoking process that wants to monitor the result of a test run.
-""")
-
 parser.add_option_group(other)
 
 ######################################################################

Modified: zope.testing/branches/benji-parallelize-subprocesses/src/zope/testing/testrunner/runner.py
===================================================================
--- zope.testing/branches/benji-parallelize-subprocesses/src/zope/testing/testrunner/runner.py	2008-07-04 08:40:27 UTC (rev 87997)
+++ zope.testing/branches/benji-parallelize-subprocesses/src/zope/testing/testrunner/runner.py	2008-07-04 13:45:39 UTC (rev 87998)
@@ -206,8 +206,10 @@
         """
         setup_layers = {}
         layers_to_run = list(self.ordered_layers())
+        should_resume = False
 
-        for layer_name, layer, tests in layers_to_run:
+        while layers_to_run:
+            layer_name, layer, tests = layers_to_run.pop(0)
             for feature in self.features:
                 feature.layer_setup(layer)
             try:
@@ -217,12 +219,19 @@
                 self.failed = True
                 return
             except CanNotTearDown:
-                setup_layers = None
                 if not self.options.resume_layer:
-                    self.ran += resume_tests(self.options, layer_name, layers_to_run,
-                                             self.failures, self.errors)
+                    should_resume = True
                     break
 
+            if self.options.processes > 1:
+                should_resume = True
+                break
+
+        if should_resume:
+            setup_layers = None
+            self.ran += resume_tests(self.options, self.features,
+                layers_to_run, self.failures, self.errors)
+
         if setup_layers:
             if self.options.resume_layer == None:
                 self.options.output.info("Tearing down left over layers:")
@@ -359,8 +368,8 @@
     def runTest(self):
         "Layer set up failure."
 
-def resume_a_layer(queue, options, layer_name, failures, errors,
-        resume_number):
+def spawn_layer_in_subprocess(queue, options, features, layer_name, layer,
+        failures, errors, resume_number):
     args = [sys.executable,
             sys.argv[0],
             '--resume-layer', layer_name, str(resume_number),
@@ -381,6 +390,9 @@
             for a in args[1:]
             ])
 
+    for feature in features:
+        feature.layer_setup(layer)
+
     subin, subout, suberr = os.popen3(args)
     while True:
         try:
@@ -416,24 +428,21 @@
     queue.put(ran)
 
 
-def resume_tests(options, layer_name, layers, failures, errors):
-    output = options.output
+def resume_tests(options, features, layers, failures, errors):
     queue = Queue.Queue()
-    layers = [l for (l, _, _) in layers]
-    layers = layers[layers.index(layer_name):]
     resume_number = 0
     ready_threads = []
-    for layer_name in layers:
+    for layer_name, layer, tests in layers:
         ready_threads.append(threading.Thread(
-            target=resume_a_layer,
-            args=(queue, options, layer_name, failures, errors, resume_number)))
+            target=spawn_layer_in_subprocess,
+            args=(queue, options, features, layer_name, layer, failures,
+                errors, resume_number)))
         resume_number += 1
 
     # Now start a few threads at a time.
-    num_threads = 3
     running_threads = []
     while ready_threads or running_threads:
-        while len(running_threads) < num_threads and ready_threads:
+        while len(running_threads) < options.processes and ready_threads:
             thread = ready_threads.pop(0)
             thread.start()
             running_threads.append(thread)
@@ -441,7 +450,7 @@
         for index, thread in list(enumerate(running_threads)):
             if not thread.isAlive():
                 del running_threads[index]
-        time.sleep(0.1)
+        time.sleep(0.01) # keep the loop from being too tight
 
     # Gather up all the results.
     rantotal = 0