Add option to parallelize full test cases. (#37)

Adds --serialize_test_cases to run tests from the same test case
sequentially. This adds a compromise that permits running binaries where
at least the test cases are independent.

It should also be significantly easier to get binaries into a state
where test cases are fully independent than all individual tests.

README.md

diff --git a/README.md b/README.md
index 3a50bce..9ef0051 100644
--- a/README.md
+++ b/README.md
@@ -56,3 +56,16 @@
 as such they have problem writing to hard-coded files, even if they are only
 used by that single test. `tmpfile()` and similar library functions are often
 your friends here.
+
+## Running Tests Within Test Cases Sequentially
+
+Sometimes tests within a single test case use globally-shared resources
+(hard-coded file paths, sockets, etc.) and cannot be run in parallel. Running
+such tests in parallel will either fail or be flaky (if they happen to not
+overlap during execution, they pass). So long as these resources are only shared
+within the same test case `gtest-parallel` can still provide some parallelism.
+
+For such binaries where test cases are independent, `gtest-parallel` provides
+`--serialize_test_cases` that runs tests within the same test case sequentially.
+While generally not providing as much speedup as fully parallel test execution,
+this permits such binaries to partially benefit from parallel execution.

gtest_parallel.py

diff --git a/gtest_parallel.py b/gtest_parallel.py
index e9f23d2..a059bf8 100755
--- a/gtest_parallel.py
+++ b/gtest_parallel.py
@@ -458,23 +458,41 @@
   return tasks
 
 
-def execute_tasks(tasks, pool_size, task_manager, timeout):
+def execute_tasks(tasks, pool_size, task_manager,
+                  timeout, serialize_test_cases):
   class WorkerFn(object):
-    def __init__(self, tasks):
-      self.task_id = 0
+    def __init__(self, tasks, running_groups):
       self.tasks = tasks
+      self.running_groups = running_groups
       self.task_lock = threading.Lock()
 
     def __call__(self):
       while True:
         with self.task_lock:
-          if self.task_id < len(self.tasks):
-            task = self.tasks[self.task_id]
-            self.task_id += 1
+          for task_id in range(len(self.tasks)):
+            task = self.tasks[task_id]
+
+            if self.running_groups is not None:
+              test_group = task.test_name.split('.')[0]
+              if test_group in self.running_groups:
+                # Try to find other non-running test group.
+                continue
+              else:
+                self.running_groups.add(test_group)
+
+            del self.tasks[task_id]
+            break
           else:
+            # Either there is no tasks left or number or remaining test
+            # cases (groups) is less than number or running threads.
             return
+
         task_manager.run_task(task)
 
+        if self.running_groups is not None:
+          with self.task_lock:
+            self.running_groups.remove(test_group)
+
   def start_daemon(func):
     t = threading.Thread(target=func)
     t.daemon = True
@@ -484,7 +502,8 @@
   try:
     if timeout:
       timeout.start()
-    worker_fn = WorkerFn(tasks)
+    running_groups = set() if serialize_test_cases else None
+    worker_fn = WorkerFn(tasks, running_groups)
     workers = [start_daemon(worker_fn) for _ in range(pool_size)]
     for worker in workers:
       worker.join()
@@ -539,6 +558,9 @@
   parser.add_option('--timeout', type='int', default=None,
                     help='Interrupt all remaining processes after the given '
                          'time (in seconds).')
+  parser.add_option('--serialize_test_cases', action='store_true',
+                    default=False, help='Do not run tests from the same test '
+                                        'case in parallel.')
 
   (options, binaries) = parser.parse_args()
 
@@ -590,7 +612,8 @@
 
   tasks = find_tests(binaries, additional_args, options, times)
   logger.log_tasks(len(tasks))
-  execute_tasks(tasks, options.workers, task_manager, timeout)
+  execute_tasks(tasks, options.workers, task_manager,
+                timeout, options.serialize_test_cases)
 
   print_try_number = options.retry_failed > 0 or options.repeat > 1
   if task_manager.passed:

gtest_parallel_tests.py

diff --git a/gtest_parallel_tests.py b/gtest_parallel_tests.py
index 409a381..127cb5d 100755
--- a/gtest_parallel_tests.py
+++ b/gtest_parallel_tests.py
@@ -16,9 +16,12 @@
 import contextlib
 import gtest_parallel
 import os.path
+import random
 import shutil
 import sys
 import tempfile
+import threading
+import time
 import unittest
 
 
@@ -300,5 +303,77 @@
                          gtest_parallel.get_save_file_path())
 
 
+class TestSerializeTestCases(unittest.TestCase):
+  class TaskManagerMock(object):
+    def __init__(self):
+      self.running_groups = []
+      self.check_lock = threading.Lock()
+
+      self.had_running_parallel_groups = False
+      self.total_tasks_run = 0
+
+    def run_task(self, task):
+      test_group = task.test_name.split('.')[0]
+
+      with self.check_lock:
+        self.total_tasks_run += 1
+        if test_group in self.running_groups:
+          self.had_running_parallel_groups = True
+        self.running_groups.append(test_group)
+
+      # Delay as if real test were run.
+      time.sleep(0.001)
+
+      with self.check_lock:
+        self.running_groups.remove(test_group)
+
+  def _execute_tasks(self, max_number_of_test_cases,
+                     max_number_of_tests_per_test_case,
+                     max_number_of_repeats, max_number_of_workers,
+                     serialize_test_cases):
+    tasks = []
+    for test_case in range(max_number_of_test_cases):
+      for test_name in range(max_number_of_tests_per_test_case):
+        # All arguments for gtest_parallel.Task except for test_name are fake.
+        test_name = 'TestCase{}.test{}'.format(test_case, test_name)
+
+        for execution_number in range(random.randint(1, max_number_of_repeats)):
+          tasks.append(gtest_parallel.Task(
+            'path/to/binary', test_name, ['path/to/binary', '--gtest_filter=*'],
+            execution_number + 1, None, 'path/to/output'))
+
+    expected_tasks_number = len(tasks)
+
+    task_manager = TestSerializeTestCases.TaskManagerMock()
+
+    gtest_parallel.execute_tasks(tasks, max_number_of_workers,
+                                 task_manager, None, serialize_test_cases)
+
+    self.assertEqual(serialize_test_cases,
+                     not task_manager.had_running_parallel_groups)
+    self.assertEqual(expected_tasks_number, task_manager.total_tasks_run)
+
+  def test_running_parallel_test_cases_without_repeats(self):
+    self._execute_tasks(max_number_of_test_cases=4,
+                        max_number_of_tests_per_test_case=32,
+                        max_number_of_repeats=1,
+                        max_number_of_workers=16,
+                        serialize_test_cases=True)
+
+  def test_running_parallel_test_cases_with_repeats(self):
+    self._execute_tasks(max_number_of_test_cases=4,
+                        max_number_of_tests_per_test_case=32,
+                        max_number_of_repeats=4,
+                        max_number_of_workers=16,
+                        serialize_test_cases=True)
+
+  def test_running_parallel_tests(self):
+    self._execute_tasks(max_number_of_test_cases=4,
+                        max_number_of_tests_per_test_case=128,
+                        max_number_of_repeats=1,
+                        max_number_of_workers=16,
+                        serialize_test_cases=False)
+
+
 if __name__ == '__main__':
   unittest.main()