SERVER-22142 join() FlushThread in resmoke.py after running all tests.

This prevents the FlushThread from running while the interpreter is
shutting down and the import machinery is uninitialized.

Also fixed an issue where a KeyboardInterrupt while running multiple
suites with --continueOnFailure would skip to the next suite, rather
than exiting resmoke.py.

(cherry picked from commit 00c8c64f6cd9b0d236d50a8a1cf7152e0367a5cd)

4 files changed, 55 insertions, 10 deletions

diff --git a/buildscripts/resmoke.py b/buildscripts/resmoke.py
index 9275b87f6fe..a6cb03cb620 100755
--- a/buildscripts/resmoke.py
+++ b/buildscripts/resmoke.py
@@ -23,6 +23,9 @@ if __name__ == "__main__" and __package__ is None:
 def _execute_suite(suite, logging_config):
     """
     Executes each test group of 'suite', failing fast if requested.
+
+    Returns true if the execution of the suite was interrupted by the
+    user, and false otherwise.
     """
 
     logger = resmokelib.logging.loggers.EXECUTOR
@@ -64,15 +67,15 @@ def _execute_suite(suite, logging_config):
             executor.run()
             if resmokelib.config.FAIL_FAST and group.return_code != 0:
                 suite.return_code = group.return_code
-                return
-        except resmokelib.errors.StopExecution:
+                return False
+        except resmokelib.errors.UserInterrupt:
             suite.return_code = 130  # Simulate SIGINT as exit code.
-            return
+            return True
         except:
             logger.exception("Encountered an error when running %ss of suite %s.",
                              group.test_kind, suite.get_name())
             suite.return_code = 2
-            return
+            return False
 
 
 def _log_summary(logger, suites, time_taken):
@@ -146,20 +149,21 @@ def main():
         resmoke_logger.info("Suites available to execute:\n%s", "\n".join(suite_names))
         sys.exit(0)
 
+    interrupted = False
     suites = resmokelib.parser.get_suites(values, args)
     try:
         for suite in suites:
             resmoke_logger.info(_dump_suite_config(suite, logging_config))
 
             suite.record_start()
-            _execute_suite(suite, logging_config)
+            interrupted = _execute_suite(suite, logging_config)
             suite.record_end()
 
             resmoke_logger.info("=" * 80)
             resmoke_logger.info("Summary of %s suite: %s",
                                 suite.get_name(), _summarize_suite(suite))
 
-            if resmokelib.config.FAIL_FAST and suite.return_code != 0:
+            if interrupted or (resmokelib.config.FAIL_FAST and suite.return_code != 0):
                 time_taken = time.time() - start_time
                 _log_summary(resmoke_logger, suites, time_taken)
                 sys.exit(suite.return_code)
@@ -171,6 +175,9 @@ def main():
         exit_code = max(suite.return_code for suite in suites)
         sys.exit(exit_code)
     finally:
+        if not interrupted:
+            resmokelib.logging.flush.stop_thread()
+
         if resmokelib.config.REPORT_FILE is not None:
             _write_report_file(suites, resmokelib.config.REPORT_FILE)
 
diff --git a/buildscripts/resmokelib/errors.py b/buildscripts/resmokelib/errors.py
index d07dd2078e6..6d2a704e390 100644
--- a/buildscripts/resmokelib/errors.py
+++ b/buildscripts/resmokelib/errors.py
@@ -18,6 +18,14 @@ class StopExecution(ResmokeError):
     pass
 
 
+class UserInterrupt(StopExecution):
+    """
+    Exception that is raised when a user signals resmoke.py to
+    unconditionally stop executing tests.
+    """
+    pass
+
+
 class TestFailure(ResmokeError):
     """
     Exception that is raised by a hook in the after_test method if it
diff --git a/buildscripts/resmokelib/logging/flush.py b/buildscripts/resmokelib/logging/flush.py
index e49d3d9f4f2..c45533f1e13 100644
--- a/buildscripts/resmokelib/logging/flush.py
+++ b/buildscripts/resmokelib/logging/flush.py
@@ -17,12 +17,36 @@ from ..utils import queue
 
 _LOGGER_QUEUE = queue.Queue()
 
+_FLUSH_THREAD_LOCK = threading.Lock()
+_FLUSH_THREAD = None
+
 
 def start_thread():
     """
     Starts the flush thread.
     """
-    _FlushThread().start()
+
+    global _FLUSH_THREAD
+    with _FLUSH_THREAD_LOCK:
+        if _FLUSH_THREAD is not None:
+            raise ValueError("FlushThread has already been started")
+
+        _FLUSH_THREAD = _FlushThread()
+        _FLUSH_THREAD.start()
+
+
+def stop_thread():
+    """
+    Signals the flush thread to stop and waits until it does.
+    """
+
+    with _FLUSH_THREAD_LOCK:
+        if _FLUSH_THREAD is None:
+            raise ValueError("FlushThread hasn't been started")
+
+    # Add sentinel value to indicate when there are no more loggers to process.
+    _LOGGER_QUEUE.put(None)
+    _FLUSH_THREAD.join()
 
 
 def close_later(logger):
@@ -44,7 +68,7 @@ class _FlushThread(threading.Thread):
         """
 
         threading.Thread.__init__(self, name="FlushThread")
-        # atexit handler is already set up to flush any loggers still in the queue when exiting.
+        # Do not wait to flush the logs if interrupted by the user.
         self.daemon = True
 
     def run(self):
@@ -54,7 +78,13 @@ class _FlushThread(threading.Thread):
 
         while True:
             logger = _LOGGER_QUEUE.get()
-            _FlushThread._shutdown_logger(logger)
+            try:
+                if logger is None:
+                    # Sentinel value received, so exit.
+                    break
+                _FlushThread._shutdown_logger(logger)
+            finally:
+                _LOGGER_QUEUE.task_done()
 
     @staticmethod
     def _shutdown_logger(logger):
diff --git a/buildscripts/resmokelib/testing/executor.py b/buildscripts/resmokelib/testing/executor.py
index fc118ddacbf..5d79abd6ac6 100644
--- a/buildscripts/resmokelib/testing/executor.py
+++ b/buildscripts/resmokelib/testing/executor.py
@@ -84,7 +84,7 @@ class TestGroupExecutor(object):
 
                 # If the user triggered a KeyboardInterrupt, then we should stop.
                 if interrupted:
-                    raise errors.StopExecution("Received interrupt from user")
+                    raise errors.UserInterrupt("Received interrupt from user")
 
                 sb = []  # String builder.
                 self._test_group.summarize_latest(sb)