common: add jobs source configuration option

By default, repo sync will run as many jobs as the amount of CPU cores on the
host machine.

This is fine for smaller development machines, but for some pipelines might want
to adjust this. For example, when rate-limiting is applied.

Add a 'jobs' source configuration field to do so.

Signed-off-by: Mattijs Korpershoek <[email protected]>

README.md

@@ -27,6 +27,9 @@ Track changes in a [repo](https://gerrit.googlesource.com/git-repo/+/master/#rep
 * `depth`: *Optional.* shallow clone with a history truncated to the specified number of commits.
     Defaults to full git clone for each project.
 
+* `jobs`: *Optional.* number of jobs to run in parallel (default: 0; based on number of CPU cores)
+   Reduce this if you observe network errors.
+
 ### Example
 
 Resource configuration for a public project using repo (Android)
@@ -47,6 +50,7 @@ resources:
     branch: master
     name: default.xml
     depth: 1 # use shallow clone for faster syncing
+    jobs: 4  # run with -j4
 ```
 
 ## Behavior

repo_resource/check.py

@@ -39,7 +39,7 @@ def check(instream) -> list:
     try:
         repo = common.Repo()
         repo.init(config.url, config.revision, config.name, config.depth)
-        repo.sync()
+        repo.sync(jobs=config.jobs)
         version = repo.currentVersion()
     except Exception as e:
         raise e

repo_resource/common.py

@@ -66,6 +66,7 @@ class SourceConfiguration(NamedTuple):
     name: str = 'default.xml'
     private_key: str = '_invalid'
     depth: int = -1
+    jobs: int = 0
 
 
 def source_config_from_payload(payload):
@@ -171,7 +172,7 @@ def init(self, url, revision='HEAD', name='default.xml', depth=-1):
         finally:
             self.__restore_oldpwd()
 
-    def sync(self, version: Version = None):
+    def sync(self, version: Version = None, jobs: int = 0):
         self.__change_to_workdir()
         try:
             with redirect_stdout(sys.stderr):
@@ -180,6 +181,10 @@ def sync(self, version: Version = None):
                     '--current-branch', '--detach', '--no-tags',
                     '--fail-fast', '--force-sync'
                 ]
+
+                if jobs > 0:
+                    repo_cmd.append('--jobs={}'.format(jobs))
+
                 if version is None:
                     repo._Main(repo_cmd)
                 else:

repo_resource/in_.py

@@ -40,7 +40,7 @@ def in_(instream, dest_dir='.'):
         repo = common.Repo(workdir=Path(dest_dir))
 
         repo.init(config.url, config.revision, config.name, config.depth)
-        repo.sync(requested_version)
+        repo.sync(requested_version, config.jobs)
         fetched_version = repo.currentVersion()
     except Exception as e:
         raise e

repo_resource/test_check.py

@@ -8,6 +8,7 @@
 from io import StringIO
 import unittest
 from pathlib import Path
+from timeit import default_timer as timer
 import shutil
 import repo
 
@@ -37,6 +38,13 @@ def setUp(self):
                 'name': 'baylibre_ssh_project.xml',
             }
         }
+        self.demo_multiple_aosp_device_source = {
+            'source': {
+                'url': 'https://github.com/makohoek/demo-manifests.git',
+                'revision': 'main',
+                'name': 'aosp_multiple_device_fixed.xml'
+            }
+        }
 
     def tearDown(self):
         p = common.CACHEDIR
@@ -300,6 +308,31 @@ def test_ssh_private_key_without_manifest_access(self):
 
         self.assertEquals(len(versions), 0)
 
+    # test that we can specify an amount of jobs
+    # This is a little flaky because it depends on network
+    def test_jobs_limit(self):
+        data = self.demo_multiple_aosp_device_source
+
+        data['source']['jobs'] = 24
+        start = timer()
+        instream = StringIO(json.dumps(data))
+        check.check(instream)
+        end = timer()
+        fast_duration = end - start
+
+        # call tearDown() manually to clear the CACHE dir
+        self.tearDown()
+
+        data['source']['jobs'] = 1
+        start = timer()
+        instream = StringIO(json.dumps(data))
+        check.check(instream)
+        end = timer()
+        slow_duration = end - start
+
+        print('fast: {} slow: {}'.format(fast_duration, slow_duration))
+        self.assertTrue(fast_duration < slow_duration)
+
 
 if __name__ == '__main__':
     unittest.main()