Check status of pooled threads properly

Author: timvaillancourt

mongodb_consistent_backup/Archive/Tar/Tar.py

@@ -27,7 +27,23 @@ def __init__(self, manager, config, timer, base_dir, backup_dir, **kwargs):
         self.compression_method = self.config.archive.tar.compression
         self.binary             = "tar"
 
-        self._pool  = None
+        self._pool   = None
+        self._pooled = []
+
+    def wait(self):
+        if len(self._pooled) > 0:
+            self._pool.close()
+            logging.debug("Waiting for tar threads to stop")
+            while len(self._pooled) > 0:
+                try:
+                    item = self._pooled[0]
+                    path, result = item
+                    result.get(1)
+		    logging.debug("Archiving completed for directory: %s" % path)
+                    self._pooled.remove(item)
+                except TimeoutError:
+                    continue
+            self.stopped = True
 
     def run(self):
         try:
@@ -48,18 +64,17 @@ def run(self):
                     output_file = "%s.tar" % subdir_name
                     if self.do_gzip():
                         output_file  = "%s.tgz" % subdir_name
-                    self._pool.apply_async(TarThread(subdir_name, output_file, self.do_gzip(), self.verbose, self.binary).run)
+                    result = self._pool.apply_async(TarThread(subdir_name, output_file, self.do_gzip(), self.verbose, self.binary).run)
+                    self._pooled.append((subdir_name, result))
             except Exception, e:
                 self._pool.terminate()
                 logging.fatal("Could not create tar archiving thread! Error: %s" % e)
                 raise Error(e)
             finally:
-                self._pool.close()
-                self._pool.join()
-                self.stopped = True
+                self.wait()
             self.completed = True
 
-    def close(self):
+    def close(self, code=None, frame=None):
         logging.debug("Stopping tar archiving threads")
         if not self.stopped and self._pool is not None:
             self._pool.terminate()

mongodb_consistent_backup/Archive/Tar/TarThread.py

@@ -34,11 +34,11 @@ def run(self):
                      backup_base_dir  = os.path.dirname(self.backup_dir)
                      backup_base_name = os.path.basename(self.backup_dir)
 
-                     log_msg   = "Archiving and compressing directory: %s" % self.backup_dir
+                     log_msg   = "Archiving directory: %s" % self.backup_dir
                      cmd_flags = ["-C", backup_base_dir, "-cf", self.output_file, "--remove-files", backup_base_name]
 
                      if self.do_gzip:
-                         log_msg   = "Archiving directory: %s" % self.backup_dir
+                         log_msg   = "Archiving and compressing directory: %s" % self.backup_dir
                          cmd_flags = ["-C", backup_base_dir, "-czf", self.output_file, "--remove-files", backup_base_name]
 
                      logging.info(log_msg)

mongodb_consistent_backup/Main.py

@@ -460,5 +460,4 @@ def run(self):
         logging.info("Completed %s in %.2f sec" % (self.program_name, self.timer.duration(self.timer_name)))
 
         self.logger.rotate()
-        self.logger.close()
         self.release_lock()

mongodb_consistent_backup/Pipeline/Stage.py

@@ -56,9 +56,11 @@ def has_task(self):
         return False
 
     def close(self):
-        if self.has_task():
+        if self.has_task() and not self.stopped:
             logging.debug("Calling close on backup stage %s with task %s" % (self.stage, self.task.capitalize()))
             self._task.close()
+            self.running = False
+            self.stopped = True
 
     def is_compressed(self):
         if self.has_task() and hasattr(self._task, "is_compressed"):

mongodb_consistent_backup/Pipeline/Task.py

@@ -58,5 +58,5 @@ def threads(self, thread_count=None, default_cpu_multiply=1):
     def run(self):
         raise Error("Must define a .run() method when using %s class!" % self.__class__.__name__)
 
-    def close(self):
+    def close(self, code=None, frame=None):
         raise Error("Must define a .close() method when using %s class!" % self.__class__.__name__)