Configuring all Task sub-classes to use their respected threads configuration item and using the Task threads() functionality (#273)

This commit implements the thread system as described in the documentation.
And actually make threading configurable.

Author: corey-hammerton

mongodb_consistent_backup/Archive/Tar/Tar.py

@@ -31,6 +31,8 @@ def __init__(self, manager, config, timer, base_dir, backup_dir, **kwargs):
         self._pool   = None
         self._pooled = []
 
+        self.threads(self.config.archive.tar.threads)
+
     def done(self, done_dir):
         if done_dir in self._pooled:
             logging.debug("Archiving completed for: %s" % done_dir)
@@ -51,9 +53,8 @@ def wait(self):
 
     def run(self):
         try:
-            thread_count = self.threads()
-            self._pool   = Pool(processes=thread_count)
-            logging.info("Archiving backup directories with pool of %i thread(s)" % thread_count)
+            self._pool   = Pool(processes=self.threads())
+            logging.info("Archiving backup directories with pool of %i thread(s)" % self.threads())
         except Exception, e:
             logging.fatal("Could not start pool! Error: %s" % e)
             raise Error(e)

mongodb_consistent_backup/Archive/Zbackup/Zbackup.py

@@ -18,8 +18,7 @@ def __init__(self, manager, config, timer, base_dir, backup_dir, **kwargs):
         self.zbackup_cache_mb    = self.config.archive.zbackup.cache_mb
         self.zbackup_passwd_file = self.config.archive.zbackup.password_file
 
-        if self.config.archive.zbackup.threads and self.config.archive.zbackup.threads > 0:
-            self.threads(self.config.archive.zbackup.threads)
+        self.threads(self.config.archive.zbackup.threads)
 
         # only lzma compression supported (for now)
         self.compression_method    = 'lzma'

mongodb_consistent_backup/Oplog/Resolver/Resolver.py

@@ -41,8 +41,11 @@ def __init__(self, manager, config, timer, base_dir, backup_dir, tailed_oplogs,
         self._pool     = None
         self._pooled   = []
         self._results  = {}
+
+        self.threads(self.config.oplog.resolver.threads)
+
         try:
-            self._pool = Pool(processes=self.threads(None, 2))
+            self._pool = Pool(processes=self.threads())
         except Exception, e:
             logging.fatal("Could not start oplog resolver pool! Error: %s" % e)
             raise Error(e)

mongodb_consistent_backup/Upload/Rsync/Rsync.py

@@ -32,7 +32,6 @@ def __init__(self, manager, config, timer, base_dir, backup_dir, **kwargs):
         self.backup_name     = self.config.backup.name
         self.remove_uploaded = self.config.upload.remove_uploaded
         self.retries         = self.config.upload.retries
-        self.thread_count    = self.config.upload.threads
         self.rsync_path      = self.config.upload.rsync.path
         self.rsync_user      = self.config.upload.rsync.user
         self.rsync_host      = self.config.upload.rsync.host
@@ -44,6 +43,7 @@ def __init__(self, manager, config, timer, base_dir, backup_dir, **kwargs):
         self.rsync_version = None
         self._rsync_info   = None
 
+        self.threads(self.config.upload.threads)
         self._pool = Pool(processes=self.threads())
 
     def init(self):

mongodb_consistent_backup/Upload/S3/S3.py

@@ -13,7 +13,6 @@ def __init__(self, manager, config, timer, base_dir, backup_dir, **kwargs):
         super(S3, self).__init__(self.__class__.__name__, manager, config, timer, base_dir, backup_dir, **kwargs)
         self.remove_uploaded     = self.config.upload.remove_uploaded
         self.retries             = self.config.upload.retries
-        self.thread_count        = self.config.upload.threads
         self.region              = self.config.upload.s3.region
         self.bucket_name         = getattr(self.config.upload.s3, 'bucket_name', None)
         self.bucket_prefix       = getattr(self.config.upload.s3, 'bucket_prefix', None)
@@ -25,6 +24,7 @@ def __init__(self, manager, config, timer, base_dir, backup_dir, **kwargs):
         self.s3_acl              = self.config.upload.s3.acl
         self.key_prefix          = base_dir
 
+        self.threads(self.config.upload.threads)
         self._pool = None
 
         if self.region is None:
@@ -35,7 +35,7 @@ def __init__(self, manager, config, timer, base_dir, backup_dir, **kwargs):
             self.region,
             self.access_key,
             self.secret_key,
-            self.thread_count,
+            self.threads(),
             self.remove_uploaded,
             self.chunk_size,
             self.s3_acl
@@ -59,7 +59,7 @@ def run(self):
             self.timer.start(self.timer_name)
             logging.info("Starting AWS S3 upload to %s (%i threads, %imb multipart chunks, %i retries)" % (
                 self.bucket_name,
-                self.thread_count,
+                self.threads(),
                 self.chunk_size_mb,
                 self.retries
             ))

mongodb_consistent_backup/Upload/S3/S3UploadPool.py

@@ -29,7 +29,6 @@ def _reduce_method(m):
 
 class S3UploadPool():
     def __init__(self, bucket_name, region, access_key, secret_key, threads=4, remove_uploaded=False, chunk_bytes=50 * 1024 * 1024, key_acl=None):
-        self.threads         = threads
         self.bucket_name     = bucket_name
         self.region          = region
         self.access_key      = access_key