SERVER-27158 shutdown fixtures when a job finishes

2 files changed, 33 insertions, 6 deletions

diff --git a/buildscripts/resmokelib/testing/executor.py b/buildscripts/resmokelib/testing/executor.py
index c17b3bb1c7b..1ea67a4dfbe 100644
--- a/buildscripts/resmokelib/testing/executor.py
+++ b/buildscripts/resmokelib/testing/executor.py
@@ -71,6 +71,7 @@ class TestGroupExecutor(object):
         self.logger.info("Starting execution of %ss...", self._test_group.test_kind)
 
         return_code = 0
+        teardown_flag = None
         try:
             if not self._setup_fixtures():
                 return_code = 2
@@ -83,13 +84,22 @@ class TestGroupExecutor(object):
                 partial_reports = [job.report for job in self._jobs]
                 self._test_group.record_start(partial_reports)
 
-                (report, interrupted) = self._run_tests(test_queue)
+                # Have the Job threads destroy their fixture during the final repetition after they
+                # finish running their last test. This avoids having a large number of processes
+                # still running if an Evergreen task were to time out from a hang/deadlock being
+                # triggered.
+                teardown_flag = threading.Event() if num_repeats == 1 else None
+                (report, interrupted) = self._run_tests(test_queue, teardown_flag)
+
                 self._test_group.record_end(report)
 
                 # If the user triggered a KeyboardInterrupt, then we should stop.
                 if interrupted:
                     raise errors.UserInterrupt("Received interrupt from user")
 
+                if teardown_flag and teardown_flag.is_set():
+                    return_code = 2
+
                 sb = []  # String builder.
                 self._test_group.summarize_latest(sb)
                 self.logger.info("Summary: %s", "\n    ".join(sb))
@@ -104,8 +114,9 @@ class TestGroupExecutor(object):
                     job.report.reset()
                 num_repeats -= 1
         finally:
-            if not self._teardown_fixtures():
-                return_code = 2
+            if not teardown_flag:
+                if not self._teardown_fixtures():
+                    return_code = 2
             self._test_group.return_code = return_code
 
     def _setup_fixtures(self):
@@ -131,7 +142,7 @@ class TestGroupExecutor(object):
 
         return True
 
-    def _run_tests(self, test_queue):
+    def _run_tests(self, test_queue, teardown_flag):
         """
         Starts a thread for each Job instance and blocks until all of
         the tests are run.
@@ -147,7 +158,9 @@ class TestGroupExecutor(object):
         try:
             # Run each Job instance in its own thread.
             for job in self._jobs:
-                t = threading.Thread(target=job, args=(test_queue, interrupt_flag))
+                t = threading.Thread(target=job,
+                                     args=(test_queue, interrupt_flag),
+                                     kwargs=dict(teardown_flag=teardown_flag))
                 # Do not wait for tests to finish executing if interrupted by the user.
                 t.daemon = True
                 t.start()
diff --git a/buildscripts/resmokelib/testing/job.py b/buildscripts/resmokelib/testing/job.py
index 312d775414a..3f3b84337bf 100644
--- a/buildscripts/resmokelib/testing/job.py
+++ b/buildscripts/resmokelib/testing/job.py
@@ -28,10 +28,15 @@ class Job(object):
         self.hooks = hooks
         self.report = report
 
-    def __call__(self, queue, interrupt_flag):
+    def __call__(self, queue, interrupt_flag, teardown_flag=None):
         """
         Continuously executes tests from 'queue' and records their
         details in 'report'.
+
+        If 'teardown_flag' is not None, then 'self.fixture.teardown()'
+        will be called before this method returns. If an error occurs
+        while destroying the fixture, then the 'teardown_flag' will be
+        set.
         """
 
         should_stop = False
@@ -52,6 +57,15 @@ class Job(object):
             # Drain the queue to unblock the main thread.
             Job._drain_queue(queue)
 
+        if teardown_flag is not None:
+            try:
+                if not self.fixture.teardown():
+                    self.logger.warn("Teardown of %s was not successful.", self.fixture)
+                    teardown_flag.set()
+            except:
+                self.logger.exception("Encountered an error while tearing down %s.", self.fixture)
+                teardown_flag.set()
+
     def _run(self, queue, interrupt_flag):
         """
         Calls the before/after suite hooks and continuously executes