implement parallel testing using Python's Queue class

6 files changed, 104 insertions, 34 deletions

diff --git a/regtest/Printer.py b/regtest/Printer.py
index 008f46bf..22931785 100644
--- a/regtest/Printer.py
+++ b/regtest/Printer.py
@@ -19,6 +19,8 @@
 import sys
 from Config import Config
 
+from threading import RLock
+
 class Printer:
 
     __single = None
@@ -32,6 +34,8 @@ class Printer:
         self._rewrite = self._stream.isatty() and not self._verbose
         self._current_line = None
 
+        self._lock = RLock()
+
         Printer.__single = self
 
     def _erase_current_line(self):
@@ -52,27 +56,31 @@ class Printer:
         self._stream.flush()
 
     def printout(self, msg):
-        self._erase_current_line()
-        self._print(msg)
-        self._current_line = msg[msg.rfind('\n') + 1:]
+        with self._lock:
+            self._erase_current_line()
+            self._print(msg)
+            self._current_line = msg[msg.rfind('\n') + 1:]
 
     def printout_update(self, msg):
-        if self._rewrite and self._current_line is not None:
-            msg = self._current_line + msg
-        elif not self._rewrite:
-            msg = self._ensure_new_line(msg)
-        self.printout(msg)
+        with self._lock:
+            if self._rewrite and self._current_line is not None:
+                msg = self._current_line + msg
+            elif not self._rewrite:
+                msg = self._ensure_new_line(msg)
+            self.printout(msg)
 
     def printout_ln(self, msg):
-        if self._current_line is not None:
-            self._current_line = None
-            msg = '\n' + msg
+        with self._lock:
+            if self._current_line is not None:
+                self._current_line = None
+                msg = '\n' + msg
 
-        self._print(self._ensure_new_line(msg))
+            self._print(self._ensure_new_line(msg))
 
     def printerr(self, msg):
-        self.stderr.write(self._ensure_new_line(msg))
-        self.stderr.flush()
+        with self._lock:
+            self.stderr.write(self._ensure_new_line(msg))
+            self.stderr.flush()
 
     def print_test_start(self, msg):
         self.printout(msg)
diff --git a/regtest/TestReferences.py b/regtest/TestReferences.py
index 9a9e9233..1d7b005a 100644
--- a/regtest/TestReferences.py
+++ b/regtest/TestReferences.py
@@ -23,6 +23,9 @@ from Config import Config
 from Printer import get_printer
 from Utils import get_document_paths_from_dir, get_skipped_tests
 
+from Queue import Queue
+from threading import Thread
+
 class TestReferences:
 
     def __init__(self, docsdir, refsdir):
@@ -32,6 +35,8 @@ class TestReferences:
         self.config = Config()
         self.printer = get_printer()
 
+        self._queue = Queue()
+
         try:
             os.makedirs(self._refsdir)
         except OSError as e:
@@ -68,9 +73,25 @@ class TestReferences:
             if backend.create_refs(doc_path, refs_path):
                 backend.create_checksums(refs_path, self.config.checksums_only)
 
+    def _worker_thread(self):
+        while True:
+            doc, n_doc, total_docs = self._queue.get()
+            self.create_refs_for_file(doc, n_doc, total_docs)
+            self._queue.task_done()
+    
     def create_refs(self):
         docs, total_docs = get_document_paths_from_dir(self._docsdir)
+        
+        self.printer.printout_ln('Process %d is spawning %d worker threads...' % (os.getpid(), self.config.threads))
+        
+        for n_thread in range(self.config.threads):
+            thread = Thread(target=self._worker_thread)
+            thread.daemon = True
+            thread.start()
+        
         n_doc = 0
         for doc in docs:
             n_doc += 1
-            self.create_refs_for_file(doc, n_doc, total_docs)
+            self._queue.put( (doc, n_doc, total_docs) )
+        
+        self._queue.join()
diff --git a/regtest/TestRun.py b/regtest/TestRun.py
index f4e50519..648ea741 100644
--- a/regtest/TestRun.py
+++ b/regtest/TestRun.py
@@ -24,6 +24,9 @@ import sys
 import os
 import errno
 
+from Queue import Queue
+from threading import Thread, RLock
+
 class TestRun:
 
     def __init__(self, docsdir, refsdir, outdir):
@@ -43,6 +46,9 @@ class TestRun:
         self._stderr = []
         self._skipped = []
 
+        self._queue = Queue()
+        self._lock = RLock()
+
         try:
             os.makedirs(self._outdir);
         except OSError as e:
@@ -57,25 +63,31 @@ class TestRun:
         ref_is_crashed = backend.is_crashed(refs_path)
         ref_is_failed = backend.is_failed(refs_path)
         if not ref_has_md5 and not ref_is_crashed and not ref_is_failed:
-            self._skipped.append("%s (%s)" % (doc_path, backend.get_name()))
+            with self._lock:
+                self._skipped.append("%s (%s)" % (doc_path, backend.get_name()))
             self.printer.print_default("Reference files not found, skipping '%s' for %s backend" % (doc_path, backend.get_name()))
             return
 
-        self._n_tests += 1
+        with self._lock:
+            self._n_tests += 1
+
         self.printer.print_test_start("Testing '%s' using %s backend (%d/%d): " % (doc_path, backend.get_name(), n_doc, total_docs))
         test_has_md5 = backend.create_refs(doc_path, test_path)
 
         if backend.has_stderr(test_path):
-            self._stderr.append("%s (%s)" % (doc_path, backend.get_name()))
+            with self._lock:
+                self._stderr.append("%s (%s)" % (doc_path, backend.get_name()))
 
         if ref_has_md5 and test_has_md5:
             if backend.compare_checksums(refs_path, test_path, not self.config.keep_results, self.config.create_diffs, self.config.update_refs):
                 # FIXME: remove dir if it's empty?
                 self.printer.print_test_result("PASS")
-                self._n_passed += 1
+                with self._lock:
+                    self._n_passed += 1
             else:
                 self.printer.print_test_result_ln("FAIL")
-                self._failed.append("%s (%s)" % (doc_path, backend.get_name()))
+                with self._lock:
+                    self._failed.append("%s (%s)" % (doc_path, backend.get_name()))
             return
         elif test_has_md5:
             if ref_is_crashed:
@@ -87,30 +99,35 @@ class TestRun:
         test_is_crashed = backend.is_crashed(test_path)
         if ref_is_crashed and test_is_crashed:
             self.printer.print_test_result("PASS (Expected crash)")
-            self._n_passed += 1
+            with self._lock:
+                self._n_passed += 1
             return
 
         test_is_failed = backend.is_failed(test_path)
         if ref_is_failed and test_is_failed:
             # FIXME: compare status errors
             self.printer.print_test_result("PASS (Expected fail with status error %d)" % (test_is_failed))
-            self._n_passed += 1
+            with self._lock:
+                self._n_passed += 1
             return
 
         if test_is_crashed:
             self.printer.print_test_result_ln("CRASH")
-            self._crashed.append("%s (%s)" % (doc_path, backend.get_name()))
+            with self._lock:
+                self._crashed.append("%s (%s)" % (doc_path, backend.get_name()))
             return
 
         if test_is_failed:
             self.printer.print_test_result_ln("FAIL (status error %d)" % (test_is_failed))
-            self._failed_status_error("%s (%s)" % (doc_path, backend.get_name()))
+            with self._lock:
+                self._failed_status_error("%s (%s)" % (doc_path, backend.get_name()))
             return
 
     def run_test(self, filename, n_doc = 1, total_docs = 1):
         if filename in self._skip:
             doc_path = os.path.join(self._docsdir, filename)
-            self._skipped.append("%s" % (doc_path))
+            with self._lock:
+                self._skipped.append("%s" % (doc_path))
             self.printer.print_default("Skipping test '%s' (%d/%d)" % (doc_path, n_doc, total_docs))
             return
 
@@ -126,7 +143,8 @@ class TestRun:
         refs_path = os.path.join(self._refsdir, filename)
 
         if not os.path.isdir(refs_path):
-            self._skipped.append("%s" % (doc_path))
+            with self._lock:
+                self._skipped.append("%s" % (doc_path))
             self.printer.print_default("Reference dir not found for %s, skipping (%d/%d)" % (doc_path, n_doc, total_docs))
             return
 
@@ -138,12 +156,28 @@ class TestRun:
         for backend in backends:
             self.test(refs_path, doc_path, out_path, backend, n_doc, total_docs)
 
+    def _worker_thread(self):
+        while True:
+            doc, n_doc, total_docs = self._queue.get()
+            self.run_test(doc, n_doc, total_docs)
+            self._queue.task_done()
+    
     def run_tests(self):
         docs, total_docs = get_document_paths_from_dir(self._docsdir)
+        
+        self.printer.printout_ln('Process %d is spawning %d worker threads...' % (os.getpid(), self.config.threads))
+        
+        for n_thread in range(self.config.threads):
+            thread = Thread(target=self._worker_thread)
+            thread.daemon = True
+            thread.start()
+        
         n_doc = 0
         for doc in docs:
             n_doc += 1
-            self.run_test(doc, n_doc, total_docs)
+            self._queue.put( (doc, n_doc, total_docs) )
+        
+        self._queue.join()
 
     def summary(self):
         if not self._n_tests:
diff --git a/regtest/backends/cairo.py b/regtest/backends/cairo.py
index 304783ea..3593342e 100644
--- a/regtest/backends/cairo.py
+++ b/regtest/backends/cairo.py
@@ -28,9 +28,8 @@ class Cairo(Backend):
 
     def create_refs(self, doc_path, refs_path):
         out_path = os.path.join(refs_path, 'cairo')
-        p1 = subprocess.Popen([self._pdftocairo, '-cropbox', '-r', '72', '-e', '-png', doc_path, out_path], stderr = subprocess.PIPE)
-        p2 = subprocess.Popen([self._pdftocairo, '-cropbox', '-r', '72', '-o', '-png', doc_path, out_path], stderr = subprocess.PIPE)
-        return self._check_exit_status2(p1, p2, out_path)
+        p = subprocess.Popen([self._pdftocairo, '-cropbox', '-r', '72', '-png', doc_path, out_path], stderr = subprocess.PIPE)
+        return self._check_exit_status(p, out_path)
 
     def _create_diff(self, ref_path, result_path):
         self._diff_png(ref_path, result_path)
diff --git a/regtest/backends/splash.py b/regtest/backends/splash.py
index 3144bc75..aadbdec9 100644
--- a/regtest/backends/splash.py
+++ b/regtest/backends/splash.py
@@ -28,9 +28,8 @@ class Splash(Backend):
 
     def create_refs(self, doc_path, refs_path):
         out_path = os.path.join(refs_path, 'splash')
-        p1 = subprocess.Popen([self._pdftoppm, '-cropbox', '-r', '72', '-e', '-png', doc_path, out_path], stderr = subprocess.PIPE)
-        p2 = subprocess.Popen([self._pdftoppm, '-cropbox', '-r', '72', '-o', '-png', doc_path, out_path], stderr = subprocess.PIPE)
-        return self._check_exit_status2(p1, p2, out_path)
+        p = subprocess.Popen([self._pdftoppm, '-cropbox', '-r', '72', '-png', doc_path, out_path], stderr = subprocess.PIPE)
+        return self._check_exit_status(p, out_path)
 
     def _create_diff(self, ref_path, result_path):
         self._diff_png(ref_path, result_path)
diff --git a/regtest/main.py b/regtest/main.py
index 290c8bc6..9ac4c14e 100644
--- a/regtest/main.py
+++ b/regtest/main.py
@@ -23,6 +23,8 @@ import backends
 import os
 from Config import Config
 
+from multiprocessing import cpu_count
+
 class ListAction(argparse.Action):
     def __call__(self, parser, namespace, values, option_string = None):
         setattr(namespace, self.dest, values.split(','))
@@ -61,13 +63,20 @@ def main(args):
     parser.add_argument('--skip', metavar = 'FILE',
                         action = 'store', dest = 'skipped_file',
                         help = 'File containing tests to skip')
+    parser.add_argument('-t', '--threads',
+                        action = 'store', dest = 'threads', type = int, default = 1,
+                        help = 'Number of worker threads')
 
     ns, args = parser.parse_known_args(args)
     if not args:
         parser.print_help()
         sys.exit(0)
 
-    Config(vars(ns))
+    c = Config(vars(ns))
+    
+    if c.threads <= 0:
+        c.threads = cpu_count() - c.threads         
+    
     try:
         commands.run(args)
     except commands.UnknownCommandError: