Implement cloneConcurrency config (#258)

Add new acceptance test that did not exist before for concurrency
behavior in git clone flow.

Author: cijiugechu

packages/zpm-config/schema.json

@@ -14,6 +14,11 @@
       },
       "default": ["main", "master"]
     },
+    "cloneConcurrency": {
+      "type": "usize",
+      "description": "The number of concurrent git clone operations Yarn can perform",
+      "default": 2
+    },
     "compressionLevel": {
       "type": ["zpm_formats::CompressionAlgorithm", "null"],
       "description": "The compression level to use for the packed file"

packages/zpm/src/git.rs

@@ -265,6 +265,12 @@ pub async fn clone_repository(context: &InstallContext<'_>, source: &GitSource,
         return Ok(clone_dir);
     }
 
+    let _clone_permit
+        = project.clone_limiter
+            .acquire()
+            .await
+            .expect("The clone limiter semaphore should not be closed");
+
     git_clone_into(source, commit, &clone_dir, &project.http_client.config).await?;
     Ok(clone_dir)
 }

packages/zpm/src/project.rs

@@ -73,6 +73,7 @@ pub struct Project {
     pub last_modified_at: LastModifiedAt,
     pub install_state: Option<InstallState>,
     pub http_client: std::sync::Arc<HttpClient>,
+    pub clone_limiter: std::sync::Arc<tokio::sync::Semaphore>,
 }
 
 impl Project {
@@ -144,6 +145,16 @@ impl Project {
             config.settings.enable_global_cache.source = config.settings.enable_migration_mode.source;
         }
 
+        let clone_concurrency
+            = config.settings.clone_concurrency.value;
+
+        if clone_concurrency == 0 {
+            return Err(Error::InvalidConfigValue("cloneConcurrency".to_string(), "must be >= 1".to_string()));
+        }
+
+        let clone_limiter
+            = Arc::new(tokio::sync::Semaphore::new(clone_concurrency));
+
         let root_workspace
             = Workspace::from_root_path(&project_cwd)?;
 
@@ -181,6 +192,7 @@ impl Project {
             last_modified_at,
             install_state: None,
             http_client,
+            clone_limiter,
         })
     }
 

tests/acceptance-tests/pkg-tests-specs/sources/commands/config/get.test.js

@@ -43,6 +43,37 @@ describe(`Commands`, () => {
       }),
     );
 
+    test(
+      `it should print cloneConcurrency default value`,
+      makeTemporaryEnv({}, async ({path, run, source}) => {
+        await expect(run(`config`, `get`, `--json`, `cloneConcurrency`)).resolves.toMatchObject({
+          stdout: `2\n`,
+        });
+      }),
+    );
+
+    test(
+      `it should print cloneConcurrency configured value`,
+      makeTemporaryEnv({}, async ({path, run, source}) => {
+        await xfs.writeFilePromise(`${path}/.yarnrc.yml`, `cloneConcurrency: 7\n`);
+
+        await expect(run(`config`, `get`, `--json`, `cloneConcurrency`)).resolves.toMatchObject({
+          stdout: `7\n`,
+        });
+      }),
+    );
+
+    test(
+      `it should reject cloneConcurrency lower than 1`,
+      makeTemporaryEnv({}, async ({path, run, source}) => {
+        await xfs.writeFilePromise(`${path}/.yarnrc.yml`, `cloneConcurrency: 0\n`);
+
+        await expect(run(`config`, `get`, `cloneConcurrency`)).rejects.toMatchObject({
+          stdout: expect.stringContaining(`Invalid config value for cloneConcurrency (must be >= 1)`),
+        });
+      }),
+    );
+
     test(
       `it should support printing sub-keys`,
       makeTemporaryEnv({}, async ({path, run, source}) => {

tests/acceptance-tests/pkg-tests-specs/sources/protocols/git.test.ts

@@ -1,5 +1,5 @@
 import {execUtils, semverUtils} from '@yarnpkg/core';
-import {npath}                  from '@yarnpkg/fslib';
+import {npath, ppath, xfs}      from '@yarnpkg/fslib';
 import {tests}                  from 'pkg-tests-core';
 
 const TESTED_URLS = {
@@ -21,6 +21,110 @@ const TESTED_URLS = {
   [`https://github.com/yarnpkg/util-deprecate.git#b3562c2798507869edb767da869cd7b85487726d`]: {version: `1.0.0`, runOnCI: true},
 };
 
+const makeCloneMetricsWrapper = ({realGitPath, metricsDir}: {realGitPath: string, metricsDir: string}) => `
+#!/usr/bin/env node
+const fs = require('fs');
+const path = require('path');
+const {spawn} = require('child_process');
+
+const realGitPath = ${JSON.stringify(realGitPath)};
+const metricsDir = ${JSON.stringify(metricsDir)};
+const stateFile = path.join(metricsDir, 'state');
+const lockDir = path.join(metricsDir, 'lock');
+
+const sleep = ms => new Promise(resolve => setTimeout(resolve, ms));
+
+async function withLock(fn) {
+  while (true) {
+    try {
+      await fs.promises.mkdir(lockDir);
+      break;
+    } catch (error) {
+      if (error && error.code === 'EEXIST') {
+        await sleep(10);
+        continue;
+      }
+
+      throw error;
+    }
+  }
+
+  try {
+    return await fn();
+  } finally {
+    await fs.promises.rmdir(lockDir);
+  }
+}
+
+async function readState() {
+  const content = await fs.promises.readFile(stateFile, 'utf8');
+  const [currentLine = '0', maxLine = '0'] = content.trim().split(/\\r\\n|\\r|\\n/);
+
+  return {
+    current: Number(currentLine),
+    max: Number(maxLine),
+  };
+}
+
+async function writeState(current, max) {
+  await fs.promises.writeFile(stateFile, \`\${current}\\n\${max}\\n\`);
+}
+
+async function incrementCounter() {
+  await withLock(async () => {
+    const {current, max} = await readState();
+    const nextCurrent = current + 1;
+    const nextMax = Math.max(max, nextCurrent);
+
+    await writeState(nextCurrent, nextMax);
+  });
+}
+
+async function decrementCounter() {
+  await withLock(async () => {
+    const {current, max} = await readState();
+    await writeState(current - 1, max);
+  });
+}
+
+function runGit(args) {
+  return new Promise((resolve, reject) => {
+    const child = spawn(realGitPath, args, {stdio: 'inherit'});
+
+    child.on('error', reject);
+    child.on('exit', code => {
+      resolve(typeof code === 'number' ? code : 1);
+    });
+  });
+}
+
+async function main() {
+  const args = process.argv.slice(2);
+
+  if (args[0] === 'clone') {
+    await incrementCounter();
+    await sleep(200);
+
+    const exitCode = await (async () => {
+      try {
+        return await runGit(args);
+      } finally {
+        await decrementCounter();
+      }
+    })();
+
+    process.exit(exitCode);
+  }
+
+  process.exit(await runGit(args));
+}
+
+main().catch(error => {
+  console.error(error);
+  process.exit(1);
+});
+`.trimStart();
+
 describe(`Protocols`, () => {
   describe(`git:`, () => {
     for (const [url, {version, runOnCI}] of Object.entries(TESTED_URLS)) {
@@ -156,6 +260,51 @@ describe(`Protocols`, () => {
       ),
     );
 
+    tests.testIf(
+      () => process.platform !== `win32`,
+      `it should respect cloneConcurrency when cloning git repositories`,
+      makeTemporaryEnv(
+        {
+          dependencies: {
+            [`pkg-a`]: tests.startPackageServer().then(url => `${url}/repositories/deep-projects.git#cwd=projects/pkg-a`),
+            [`pkg-b`]: tests.startPackageServer().then(url => `${url}/repositories/deep-projects.git#cwd=projects/pkg-b`),
+            [`lib-a`]: tests.startPackageServer().then(url => `${url}/repositories/deep-projects.git#cwd=projects/pkg-a&workspace=lib`),
+            [`lib-b`]: tests.startPackageServer().then(url => `${url}/repositories/deep-projects.git#cwd=projects/pkg-b&workspace=lib`),
+          },
+        },
+        {
+          cloneConcurrency: 1,
+        },
+        async ({path, run, source}) => {
+          const {stdout: gitPathStdout} = await execUtils.execvp(`which`, [`git`], {cwd: path});
+          const realGitPath = gitPathStdout.trim();
+
+          const binDir = ppath.join(path, `bin`);
+          const metricsDir = ppath.join(path, `.clone-metrics`);
+          const stateFile = ppath.join(metricsDir, `state`);
+          const wrapperPath = ppath.join(binDir, `git`);
+
+          await xfs.mkdirPromise(binDir, {recursive: true});
+          await xfs.mkdirPromise(metricsDir, {recursive: true});
+          await xfs.writeFilePromise(stateFile, `0\n0\n`);
+          await xfs.writeFilePromise(wrapperPath, makeCloneMetricsWrapper({
+            realGitPath,
+            metricsDir: npath.fromPortablePath(metricsDir),
+          }));
+
+          await xfs.chmodPromise(wrapperPath, 0o755);
+
+          await run(`install`);
+
+          const [currentLine, maxLine] = (await xfs.readFilePromise(stateFile, `utf8`)).trim().split(/\r\n|\r|\n/);
+
+          expect(Number(currentLine)).toBe(0);
+          expect(Number(maxLine)).toBeGreaterThan(0);
+          expect(Number(maxLine)).toBeLessThanOrEqual(1);
+        },
+      ),
+    );
+
     test(
       `it should use Yarn Classic to setup classic repositories`,
       makeTemporaryEnv(