Merge pull request #12 from timvaillancourt/thread_completed_fix

Moving thread-complete check to response queue

Author: dbmurphy

MongoBackup/Mongodump.py

@@ -1,7 +1,7 @@
 import os
 import logging
 
-from multiprocessing import Process, Event
+from multiprocessing import Process
 from signal import signal, SIGINT, SIGTERM
 from time import time
 
@@ -28,7 +28,7 @@ def __init__(self, response_queue, backup_name, host_port, user, password, authd
         self.verbose        = verbose
 
         self._command   = None
-        self._complete  = Event()
+        self.completed  = False 
         self.backup_dir = "%s/%s" % (self.base_dir, self.backup_name)
         self.dump_dir   = "%s/dump" % self.backup_dir
         self.oplog_file = "%s/oplog.bson" % self.dump_dir
@@ -37,9 +37,6 @@ def __init__(self, response_queue, backup_name, host_port, user, password, authd
         signal(SIGINT, self.close)
         signal(SIGTERM, self.close)
 
-    def complete(self):
-        return self._complete.is_set()
-
     def close(self, exit_code=None, frame=None):
         if self._command:
             logging.debug("Killing running subprocess/command: %s" % self._command.command)
@@ -77,15 +74,16 @@ def run(self):
             return None
 
         oplog = OplogInfo(self.oplog_file, self.dump_gzip)
+        self.completed = True
         self.response_queue.put({
             'host': self.host,
             'port': self.port,
             'file': self.oplog_file,
             'count': oplog.count(),
             'last_ts': oplog.last_ts(),
-            'first_ts': oplog.first_ts()
+            'first_ts': oplog.first_ts(),
+            'completed': self.completed
         })
-        self._complete.set()
 
         time_diff = time() - self.start_time
         logging.info("Backup for %s/%s:%s completed in %s sec with %i oplog changes captured to: %s" % (

MongoBackup/Mongodumper.py

@@ -113,25 +113,30 @@ def run(self):
         for thread in self.threads:
             thread.start()
 
-        # wait for all threads to finish and check their status
+        # wait for all threads to finish
         for thread in self.threads:
             thread.join()
-            if not thread.complete():
-                raise Exception, "Not all mongodump threads completed successfully!", None
 
         # sleep for 3 sec to fix logging order
         sleep(3)
 
         # get oplog summaries from the queue
+        completed = 0
         while not self.response_queue.empty():
             backup = self.response_queue.get()
             host = backup['host']
             port = backup['port']
             if host not in self._summary:
                 self._summary[host] = {}
             self._summary[host][port] = backup
+            if backup['completed']:
+                completed += 1
 
-        logging.info("All mongodump backups completed")
+        # check if all threads completed
+        if completed == len(self.threads):
+            logging.info("All mongodump backups completed")
+        else:
+            raise Exception, "Not all mongodump threads completed successfully!", None
 
         return self._summary