Limit the number of parallel mongodump workers based on CPU

Author: timvaillancourt

MongoBackup/Methods/Dump.py

@@ -12,7 +12,7 @@
 # noinspection PyStringFormat
 class Dump(Process):
     def __init__(self, response_queue, backup_name, host_port, user, password, authdb, base_dir, binary,
-                 dump_gzip=False, verbose=False):
+                 threads=0, dump_gzip=False, verbose=False):
         Process.__init__(self)
         self.host, port     = host_port.split(":")
         self.host_port      = host_port
@@ -24,6 +24,7 @@ def __init__(self, response_queue, backup_name, host_port, user, password, authd
         self.authdb         = authdb
         self.base_dir       = base_dir
         self.binary         = binary
+        self.threads        = threads
         self.dump_gzip      = dump_gzip
         self.verbose        = verbose
 
@@ -52,6 +53,8 @@ def run(self):
         ))
 
         mongodump_flags = ["-h", self.host_port, "--oplog", "-o", "%s/dump" % self.backup_dir]
+        if self.threads > 0:
+            mongodump_flags.extend(["--numParallelCollections="+str(self.threads)])
         if self.dump_gzip:
             mongodump_flags.extend(["--gzip"])
         if self.authdb and self.authdb != "admin":
@@ -68,6 +71,7 @@ def run(self):
             commands.append([self.binary, mongodump_flags])
 
             for (command, command_flags) in commands:
+                print command_flags
                 self._command = LocalCommand(command, command_flags, self.verbose)
                 self._command.run()
         except Exception, e:

MongoBackup/Methods/Dumper.py

@@ -1,7 +1,8 @@
 import logging
 
 from fabric.api import hide, settings, local
-from multiprocessing import Queue
+from math import floor
+from multiprocessing import Queue, cpu_count
 from time import sleep
 
 
@@ -22,6 +23,7 @@ def __init__(self, secondaries, base_dir, binary, dump_gzip=False, user=None, pa
         self.verbose       = verbose
 
         self.config_replset = False
+        self.cpu_count      = cpu_count()
         self.response_queue = Queue()
         self.threads        = []
         self._summary       = {}
@@ -65,6 +67,16 @@ def wait(self):
             raise Exception, "Not all mongodump threads completed successfully!", None
 
     def run(self):
+        # decide how many parallel dump workers to use based on cpu count vs # of shards (if 3.2+)
+        self.threads_per_dump = 1
+	if tuple(self.version.split(".")) >= tuple("3.2.0".split(".")): 
+            self.threads_per_dump = 1
+            if self.cpu_count > len(self.secondaries):
+                self.threads_per_dump = int(floor(self.cpu_count / len(self.secondaries)))
+        else:
+            self.threads_per_dump = 0
+            logging.warn("Threading unsupported by mongodump version %s. Use mongodump 3.2.0 or greater to enable per-dump threading." % self.version)
+
         # backup a secondary from each shard:
         for shard in self.secondaries:
             secondary = self.secondaries[shard]
@@ -77,6 +89,7 @@ def run(self):
                 self.authdb,
                 self.base_dir,
                 self.binary,
+                self.threads_per_dump,
                 self.dump_gzip,
                 self.verbose
             )
@@ -87,7 +100,7 @@ def run(self):
 
         # start all threads and wait
         logging.info(
-            "Starting backups in threads using mongodump %s (inline gzip: %s)" % (self.version, str(self.dump_gzip)))
+		"Starting backups using mongodump %s (inline gzip: %s, threads per dump: %i)" % (self.version, str(self.dump_gzip), self.threads_per_dump))
         for thread in self.threads:
             thread.start()
         self.wait()
@@ -104,6 +117,7 @@ def run(self):
                 self.authdb,
                 self.base_dir,
                 self.binary,
+                self.threads_per_dump,
                 self.dump_gzip,
                 self.verbose
             )]

VERSION

@@ -1 +1 @@
-0.3.4
+0.3.5