test: run tests in parallel, common improvements

* Allow running tests in mixed parallel/sequential modes
* Add -J flag for running tests on all available CPUs
* Support TEST_THREAD_ID in test/common.js and use it for tmpDir and PORT
* make: use -J flag

Reviewed-By: Ben Noordhuis <[email protected]>
PR-URL: #172
Fix: #139

Author: indutny

.gitignore

@@ -9,6 +9,7 @@ tags
 *.pyc
 doc/api.xml
 tmp/
+test/tmp*/
 node
 node_g
 *.swp

Makefile

@@ -90,7 +90,7 @@ distclean:
 	-rm -rf node_modules
 
 test: all
-	$(PYTHON) tools/test.py --mode=release message parallel sequential
+	$(PYTHON) tools/test.py --mode=release message parallel sequential -J
 	$(MAKE) jslint
 	$(MAKE) cpplint
 

test/common.js

@@ -27,9 +27,18 @@ var os = require('os');
 exports.testDir = path.dirname(__filename);
 exports.fixturesDir = path.join(exports.testDir, 'fixtures');
 exports.libDir = path.join(exports.testDir, '../lib');
-exports.tmpDir = path.join(exports.testDir, 'tmp');
+exports.tmpDirName = 'tmp';
 exports.PORT = +process.env.NODE_COMMON_PORT || 12346;
 
+if (process.env.TEST_THREAD_ID) {
+  // Distribute ports in parallel tests
+  if (!process.env.NODE_COMMON_PORT)
+    exports.PORT += +process.env.TEST_THREAD_ID * 100;
+
+  exports.tmpDirName += '.' + process.env.TEST_THREAD_ID;
+}
+exports.tmpDir = path.join(exports.testDir, exports.tmpDirName);
+
 exports.opensslCli = path.join(path.dirname(process.execPath), 'openssl-cli');
 if (process.platform === 'win32') {
   exports.PIPE = '\\\\.\\pipe\\libuv-test';

test/testpy/__init__.py

@@ -49,30 +49,34 @@ def __init__(self, path, file, arch, mode, context, config, additional=[]):
     self.mode = mode
     self.tmpdir = join(dirname(self.config.root), 'tmp')
     self.additional_flags = additional
+
+  def GetTmpDir(self):
+    return "%s.%d" % (self.tmpdir, self.thread_id)
+
 
   def AfterRun(self, result):
     # delete the whole tmp dir
     try:
-      rmtree(self.tmpdir)
+      rmtree(self.GetTmpDir())
     except:
       pass
     # make it again.
     try:
-      mkdir(self.tmpdir)
+      mkdir(self.GetTmpDir())
     except:
       pass
 
   def BeforeRun(self):
     # delete the whole tmp dir
     try:
-      rmtree(self.tmpdir)
+      rmtree(self.GetTmpDir())
     except:
       pass
     # make it again.
     # intermittently fails on win32, so keep trying
-    while not os.path.exists(self.tmpdir):
+    while not os.path.exists(self.GetTmpDir()):
       try:
-        mkdir(self.tmpdir)
+        mkdir(self.GetTmpDir())
       except:
         pass
 
@@ -105,7 +109,6 @@ def GetCommand(self):
   def GetSource(self):
     return open(self.file).read()
 
-
 class SimpleTestConfiguration(test.TestConfiguration):
 
   def __init__(self, context, root, section, additional=[]):
@@ -136,6 +139,18 @@ def GetTestStatus(self, sections, defs):
     if exists(status_file):
       test.ReadConfigurationInto(status_file, sections, defs)
 
+class ParallelTestConfiguration(SimpleTestConfiguration):
+  def __init__(self, context, root, section, additional=[]):
+    super(ParallelTestConfiguration, self).__init__(context, root, section,
+                                                    additional)
+
+  def ListTests(self, current_path, path, arch, mode):
+    result = super(ParallelTestConfiguration, self).ListTests(
+         current_path, path, arch, mode)
+    for test in result:
+      test.parallel = True
+    return result
+
 class AddonTestConfiguration(SimpleTestConfiguration):
   def __init__(self, context, root, section, additional=[]):
     super(AddonTestConfiguration, self).__init__(context, root, section)

tools/test.py

@@ -40,6 +40,7 @@
 import time
 import threading
 import utils
+import multiprocessing
 
 from os.path import join, dirname, abspath, basename, isdir, exists
 from datetime import datetime
@@ -58,9 +59,13 @@ class ProgressIndicator(object):
   def __init__(self, cases, flaky_tests_mode):
     self.cases = cases
     self.flaky_tests_mode = flaky_tests_mode
-    self.queue = Queue(len(cases))
+    self.parallel_queue = Queue(len(cases))
+    self.sequential_queue = Queue(len(cases))
     for case in cases:
-      self.queue.put_nowait(case)
+      if case.parallel:
+        self.parallel_queue.put_nowait(case)
+      else:
+        self.sequential_queue.put_nowait(case)
     self.succeeded = 0
     self.remaining = len(cases)
     self.total = len(cases)
@@ -87,11 +92,11 @@ def Run(self, tasks):
     # That way -j1 avoids threading altogether which is a nice fallback
     # in case of threading problems.
     for i in xrange(tasks - 1):
-      thread = threading.Thread(target=self.RunSingle, args=[])
+      thread = threading.Thread(target=self.RunSingle, args=[True, i + 1])
       threads.append(thread)
       thread.start()
     try:
-      self.RunSingle()
+      self.RunSingle(False, 0)
       # Wait for the remaining threads
       for thread in threads:
         # Use a timeout so that signals (ctrl-c) will be processed.
@@ -105,13 +110,19 @@ def Run(self, tasks):
     self.Done()
     return not self.failed
 
-  def RunSingle(self):
+  def RunSingle(self, parallel, thread_id):
     while not self.terminate:
       try:
-        test = self.queue.get_nowait()
+        test = self.parallel_queue.get_nowait()
       except Empty:
-        return
+        if parallel:
+          return
+        try:
+          test = self.sequential_queue.get_nowait()
+        except Empty:
+          return
       case = test.case
+      case.thread_id = thread_id
       self.lock.acquire()
       self.AboutToRun(case)
       self.lock.release()
@@ -381,6 +392,8 @@ def __init__(self, context, path, arch, mode):
     self.duration = None
     self.arch = arch
     self.mode = mode
+    self.parallel = False
+    self.thread_id = 0
 
   def IsNegative(self):
     return False
@@ -399,11 +412,12 @@ def IsFailureOutput(self, output):
   def GetSource(self):
     return "(no source available)"
 
-  def RunCommand(self, command):
+  def RunCommand(self, command, env):
     full_command = self.context.processor(command)
     output = Execute(full_command,
                      self.context,
-                     self.context.GetTimeout(self.mode))
+                     self.context.GetTimeout(self.mode),
+                     env)
     self.Cleanup()
     return TestOutput(self,
                       full_command,
@@ -420,7 +434,9 @@ def Run(self):
     self.BeforeRun()
 
     try:
-      result = self.RunCommand(self.GetCommand())
+      result = self.RunCommand(self.GetCommand(), {
+        "TEST_THREAD_ID": "%d" % self.thread_id
+      })
     finally:
       # Tests can leave the tty in non-blocking mode. If the test runner
       # tries to print to stdout/stderr after that and the tty buffer is
@@ -559,15 +575,22 @@ def CheckedUnlink(name):
     PrintError("os.unlink() " + str(e))
 
 
-def Execute(args, context, timeout=None):
+def Execute(args, context, timeout=None, env={}):
   (fd_out, outname) = tempfile.mkstemp()
   (fd_err, errname) = tempfile.mkstemp()
+
+  # Extend environment
+  env_copy = os.environ.copy()
+  for key, value in env.iteritems():
+    env_copy[key] = value
+
   (process, exit_code, timed_out) = RunProcess(
     context,
     timeout,
     args = args,
     stdout = fd_out,
     stderr = fd_err,
+    env = env_copy
   )
   os.close(fd_out)
   os.close(fd_err)
@@ -1068,6 +1091,7 @@ class ClassifiedTest(object):
   def __init__(self, case, outcomes):
     self.case = case
     self.outcomes = outcomes
+    self.parallel = self.case.parallel
 
 
 class Configuration(object):
@@ -1224,6 +1248,8 @@ def BuildOptions():
       default=False, action="store_true")
   result.add_option("-j", help="The number of parallel tasks to run",
       default=1, type="int")
+  result.add_option("-J", help="Run tasks in parallel on all cores",
+      default=False, action="store_true")
   result.add_option("--time", help="Print timing information after running",
       default=False, action="store_true")
   result.add_option("--suppress-dialogs", help="Suppress Windows dialogs for crashing tests",
@@ -1245,6 +1271,8 @@ def ProcessOptions(options):
   VERBOSE = options.verbose
   options.arch = options.arch.split(',')
   options.mode = options.mode.split(',')
+  if options.J:
+    options.j = multiprocessing.cpu_count()
   def CheckTestMode(name, option):
     if not option in ["run", "skip", "dontcare"]:
       print "Unknown %s mode %s" % (name, option)

vcbuild.bat

@@ -163,7 +163,7 @@ if "%test%"=="" goto exit
 if "%config%"=="Debug" set test_args=--mode=debug
 if "%config%"=="Release" set test_args=--mode=release
 
-if "%test%"=="test" set test_args=%test_args% sequential parallel message
+if "%test%"=="test" set test_args=%test_args% sequential parallel message -J
 if "%test%"=="test-internet" set test_args=%test_args% internet
 if "%test%"=="test-pummel" set test_args=%test_args% pummel
 if "%test%"=="test-simple" set test_args=%test_args% sequential parallel