From b5d075d04f1e555f85aad27e74f16073a50b2ae6 Mon Sep 17 00:00:00 2001 From: Mike Frysinger Date: Mon, 1 Mar 2021 00:56:38 -0500 Subject: command: add a helper for the parallel execution boilerplate Now that we have a bunch of subcommands doing parallel execution, a common pattern arises that we can factor out for most of them. We leave forall alone as it's a bit too complicated atm to cut over. Change-Id: I3617a4f7c66142bcd1ab030cb4cca698a65010ac Reviewed-on: https://gerrit-review.googlesource.com/c/git-repo/+/301942 Tested-by: Mike Frysinger Reviewed-by: Chris Mcdonald --- subcmds/start.py | 26 ++++++++++---------------- 1 file changed, 10 insertions(+), 16 deletions(-) (limited to 'subcmds/start.py') diff --git a/subcmds/start.py b/subcmds/start.py index aa2f915a..ff2bae56 100644 --- a/subcmds/start.py +++ b/subcmds/start.py @@ -13,11 +13,10 @@ # limitations under the License. import functools -import multiprocessing import os import sys -from command import Command, DEFAULT_LOCAL_JOBS, WORKER_BATCH_SIZE +from command import Command, DEFAULT_LOCAL_JOBS from git_config import IsImmutable from git_command import git import gitc_utils @@ -55,7 +54,7 @@ revision specified in the manifest. if not git.check_ref_format('heads/%s' % nb): self.OptionParser.error("'%s' is not a valid name" % nb) - def _ExecuteOne(self, opt, nb, project): + def _ExecuteOne(self, revision, nb, project): """Start one project.""" # If the current revision is immutable, such as a SHA1, a tag or # a change, then we can't push back to it. Substitute with @@ -69,7 +68,7 @@ revision specified in the manifest. try: ret = project.StartBranch( - nb, branch_merge=branch_merge, revision=opt.revision) + nb, branch_merge=branch_merge, revision=revision) except Exception as e: print('error: unable to checkout %s: %s' % (project.name, e), file=sys.stderr) ret = False @@ -123,23 +122,18 @@ revision specified in the manifest. pm.update() pm.end() - def _ProcessResults(results): + def _ProcessResults(_pool, pm, results): for (result, project) in results: if not result: err.append(project) pm.update() - pm = Progress('Starting %s' % nb, len(all_projects), quiet=opt.quiet) - # NB: Multiprocessing is heavy, so don't spin it up for one job. - if len(all_projects) == 1 or opt.jobs == 1: - _ProcessResults(self._ExecuteOne(opt, nb, x) for x in all_projects) - else: - with multiprocessing.Pool(opt.jobs) as pool: - results = pool.imap_unordered( - functools.partial(self._ExecuteOne, opt, nb), all_projects, - chunksize=WORKER_BATCH_SIZE) - _ProcessResults(results) - pm.end() + self.ExecuteInParallel( + opt.jobs, + functools.partial(self._ExecuteOne, opt.revision, nb), + all_projects, + callback=_ProcessResults, + output=Progress('Starting %s' % (nb,), len(all_projects), quiet=opt.quiet)) if err: for p in err: -- cgit v1.2.3-54-g00ecf