parallelize repository mirroring (#2744)

fixes #2743

Author: Vladimir Kotal

opengrok-tools/src/main/python/opengrok_tools/mirror.py

@@ -34,6 +34,7 @@
 import os
 import sys
 import tempfile
+from multiprocessing import Pool, cpu_count
 
 from filelock import Timeout, FileLock
 
@@ -52,7 +53,22 @@
     print("Need Python 3, you are running {}".format(major_version))
     sys.exit(1)
 
-__version__ = "0.7"
+__version__ = "0.8"
+
+
+def worker(args):
+    project_name, logdir, loglevel, backupcount, config, incoming, uri, \
+        source_root, batch = args
+
+    if batch:
+        get_batch_logger(logdir, project_name,
+                         loglevel,
+                         backupcount,
+                         get_class_basename())
+
+    return mirror_project(config, project_name,
+                          incoming,
+                          uri, source_root)
 
 
 def main():
@@ -77,6 +93,9 @@ def main():
     parser.add_argument('-I', '--incoming', action='store_true',
                         help='Check for incoming changes, terminate the '
                              'processing if not found.')
+    parser.add_argument('-w', '--workers', default=cpu_count(),
+                        help='Number of worker processes')
+
     try:
         args = parser.parse_args()
     except ValueError as e:
@@ -150,18 +169,22 @@ def main():
     lock = FileLock(os.path.join(tempfile.gettempdir(), lockfile + ".lock"))
     try:
         with lock.acquire(timeout=0):
-            for project_name in projects:
-                if args.batch:
-                    get_batch_logger(logdir, project_name,
-                                     args.loglevel,
-                                     args.backupcount,
-                                     get_class_basename())
-
-                project_result = mirror_project(config, project_name,
-                                                args.incoming,
-                                                args.uri, source_root)
-                if project_result == 1:
-                    ret = 1
+            with Pool(processes=int(args.workers)) as pool:
+                worker_args = []
+                for x in projects:
+                    worker_args.append([x, logdir, args.loglevel,
+                                        args.backupcount, config,
+                                        args.incoming,
+                                        args.uri, source_root,
+                                        args.batch])
+                print(worker_args)
+                try:
+                    project_results = pool.map(worker, worker_args, 1)
+                except KeyboardInterrupt:
+                    sys.exit(1)
+                else:
+                    if any([True for x in project_results if x == 1]):
+                        ret = 1
     except Timeout:
         logger.warning("Already running, exiting.")
         sys.exit(1)

opengrok-tools/src/main/python/opengrok_tools/sync.py

@@ -183,21 +183,19 @@ def main():
                 cmds_base.append(cmd_base)
 
             # Map the commands into pool of workers so they can be processed.
-            pool = Pool(processes=int(args.workers))
-            try:
-                cmds_base_results = pool.map(worker, cmds_base, 1)
-            except KeyboardInterrupt:
-                pool.close()
-                pool.terminate()
-                sys.exit(1)
-            else:
-                for cmds_base in cmds_base_results:
-                    logger.debug("Checking results of project {}".
-                                 format(cmds_base))
-                    cmds = CommandSequence(cmds_base)
-                    cmds.fill(cmds_base.retcodes, cmds_base.outputs,
-                              cmds_base.failed)
-                    cmds.check(ignore_errors)
+            with Pool(processes=int(args.workers)) as pool:
+                try:
+                    cmds_base_results = pool.map(worker, cmds_base, 1)
+                except KeyboardInterrupt:
+                    sys.exit(1)
+                else:
+                    for cmds_base in cmds_base_results:
+                        logger.debug("Checking results of project {}".
+                                     format(cmds_base))
+                        cmds = CommandSequence(cmds_base)
+                        cmds.fill(cmds_base.retcodes, cmds_base.outputs,
+                                  cmds_base.failed)
+                        cmds.check(ignore_errors)
     except Timeout:
         logger.warning("Already running, exiting.")
         sys.exit(1)

opengrok-tools/src/main/python/opengrok_tools/utils/log.py

@@ -147,8 +147,6 @@ def get_batch_logger(logdir, project_name, loglevel, backupcount,
     logger = logging.getLogger(name)
 
     logfile = os.path.join(logdir, project_name + ".log")
-    logger.debug("Switching logging to the {} file".
-                 format(logfile))
 
     handler = RotatingFileHandler(logfile, maxBytes=0, mode='a',
                                   backupCount=backupcount)