Add workerPoolConcurrency as a configuration option

.yarn/versions/85334687.yml

@@ -0,0 +1,34 @@
+releases:
+  "@yarnpkg/cli": minor
+  "@yarnpkg/core": minor
+  "@yarnpkg/plugin-file": minor
+  "@yarnpkg/plugin-git": minor
+  "@yarnpkg/plugin-github": minor
+  "@yarnpkg/plugin-http": minor
+  "@yarnpkg/plugin-npm": minor
+
+declined:
+  - "@yarnpkg/plugin-compat"
+  - "@yarnpkg/plugin-constraints"
+  - "@yarnpkg/plugin-dlx"
+  - "@yarnpkg/plugin-essentials"
+  - "@yarnpkg/plugin-exec"
+  - "@yarnpkg/plugin-init"
+  - "@yarnpkg/plugin-interactive-tools"
+  - "@yarnpkg/plugin-link"
+  - "@yarnpkg/plugin-nm"
+  - "@yarnpkg/plugin-npm-cli"
+  - "@yarnpkg/plugin-pack"
+  - "@yarnpkg/plugin-patch"
+  - "@yarnpkg/plugin-pnp"
+  - "@yarnpkg/plugin-pnpm"
+  - "@yarnpkg/plugin-stage"
+  - "@yarnpkg/plugin-typescript"
+  - "@yarnpkg/plugin-version"
+  - "@yarnpkg/plugin-workspace-tools"
+  - "@yarnpkg/builder"
+  - "@yarnpkg/doctor"
+  - "@yarnpkg/extensions"
+  - "@yarnpkg/nm"
+  - "@yarnpkg/pnpify"
+  - "@yarnpkg/sdks"

packages/docusaurus/static/configuration/yarnrc.json

@@ -383,6 +383,13 @@
       "type": "number",
       "default": 50
     },
+    "workerPoolConcurrency": {
+      "_package": "@yarnpkg/core",
+      "title": "Amount of workers that are allowed to run at the same time.",
+      "description": "We default to the platform parallelism, but for some CI, `os.cpus` may not report accurate values and may overwhelm their containers",
+      "type": "number",
+      "default": "os.availableParallelism()"
+    },
     "networkSettings": {
       "_package": "@yarnpkg/core",
       "title": "Additional network settings, per hostname",

packages/gatsby/static/configuration/yarnrc.json

@@ -355,6 +355,13 @@
       "default": 50,
       "examples": [8]
     },
+    "workerPoolConcurrency": {
+      "_package": "@yarnpkg/core",
+      "description": "Defines how many workers that are allowed to run at the same time. Yarn defaults to platform parallelism but for some CI, `os.cpus` may not report accurate values and may overwhelm their containers.",
+      "type": "number",
+      "default": "os.availableParallelism()",
+      "examples": [8]
+    },
     "networkSettings": {
       "_package": "@yarnpkg/core",
       "description": "Additional network settings, per hostname",

packages/plugin-file/sources/TarballFileFetcher.ts

@@ -45,6 +45,7 @@ export class TarballFileFetcher implements Fetcher {
       compressionLevel: opts.project.configuration.get(`compressionLevel`),
       prefixPath: structUtils.getIdentVendorPath(locator),
       stripComponents: 1,
+      poolSize: opts.project.configuration.getLimit(`workerPoolConcurrency`),
     });
   }
 }

packages/plugin-git/sources/GitFetcher.ts

@@ -67,6 +67,7 @@ export class GitFetcher implements Fetcher {
         compressionLevel: opts.project.configuration.get(`compressionLevel`),
         prefixPath: structUtils.getIdentVendorPath(locator),
         stripComponents: 1,
+        poolSize: opts.project.configuration.getLimit(`workerPoolConcurrency`),
       });
     });
   }

packages/plugin-github/sources/GithubFetcher.ts

@@ -64,6 +64,7 @@ export class GithubFetcher implements Fetcher {
         compressionLevel: opts.project.configuration.get(`compressionLevel`),
         prefixPath: structUtils.getIdentVendorPath(locator),
         stripComponents: 1,
+        poolSize: opts.project.configuration.getLimit(`workerPoolConcurrency`),
       });
     });
   }

packages/plugin-http/sources/TarballHttpFetcher.ts

@@ -46,6 +46,7 @@ export class TarballHttpFetcher implements Fetcher {
       compressionLevel: opts.project.configuration.get(`compressionLevel`),
       prefixPath: structUtils.getIdentVendorPath(locator),
       stripComponents: 1,
+      poolSize: opts.project.configuration.getLimit(`workerPoolConcurrency`),
     });
   }
 }

packages/plugin-npm/sources/NpmHttpFetcher.ts

@@ -58,6 +58,7 @@ export class NpmHttpFetcher implements Fetcher {
       compressionLevel: opts.project.configuration.get(`compressionLevel`),
       prefixPath: structUtils.getIdentVendorPath(locator),
       stripComponents: 1,
+      poolSize: opts.project.configuration.getLimit(`workerPoolConcurrency`),
     });
   }
 }

packages/plugin-npm/sources/NpmSemverFetcher.ts

@@ -68,6 +68,7 @@ export class NpmSemverFetcher implements Fetcher {
       compressionLevel: opts.project.configuration.get(`compressionLevel`),
       prefixPath: structUtils.getIdentVendorPath(locator),
       stripComponents: 1,
+      poolSize: opts.project.configuration.getLimit(`workerPoolConcurrency`),
     });
   }
 

packages/yarnpkg-core/sources/Configuration.ts

@@ -395,6 +395,11 @@ export const coreDefinitions: {[coreSettingName: string]: SettingsDefinition} =
     type: SettingsType.NUMBER,
     default: 50,
   },
+  workerPoolConcurrency: {
+    description: `Maximal number of workers in the worker pool`,
+    type: SettingsType.NUMBER,
+    default: nodeUtils.availableParallelism(),
+  },
   networkSettings: {
     description: `Network settings per hostname (glob patterns are supported)`,
     type: SettingsType.MAP,
@@ -636,6 +641,7 @@ export interface ConfigurationValueMap {
   httpsKeyFilePath: PortablePath | null;
   httpsCertFilePath: PortablePath | null;
   enableStrictSsl: boolean;
+  workerPoolConcurrency: number;
 
   logFilters: Array<miscUtils.ToMapValue<{code?: string, text?: string, pattern?: string, level?: formatUtils.LogLevel | null}>>;
 

packages/yarnpkg-core/sources/WorkerPool.ts

@@ -1,20 +1,28 @@
-import PLimit         from 'p-limit';
-import {Worker}       from 'worker_threads';
+import PLimit, {Limit} from 'p-limit';
+import {Worker}        from 'worker_threads';
 
-import * as nodeUtils from './nodeUtils';
+import * as nodeUtils  from './nodeUtils';
 
 const kTaskInfo = Symbol(`kTaskInfo`);
 
 type PoolWorker<TOut> = Worker & {
   [kTaskInfo]: null | { resolve: (value: TOut) => void, reject: (reason?: any) => void };
 };
 
+type WorkerPoolOptions = {
+  poolSize?: Limit;
+};
+
 export class WorkerPool<TIn, TOut> {
   private workers: Array<PoolWorker<TOut>> = [];
-  private limit = PLimit(nodeUtils.availableParallelism());
+
   private cleanupInterval: ReturnType<typeof setInterval>;
 
-  constructor(private source: string) {
+  private limit: Limit;
+
+  constructor(private source: string, opts: WorkerPoolOptions = {}) {
+    this.limit = opts.poolSize ?? PLimit(nodeUtils.availableParallelism());
+
     this.cleanupInterval = setInterval(() => {
       if (this.limit.pendingCount === 0 && this.limit.activeCount === 0) {
         // Start terminating one worker at a time when there are no tasks left.

packages/yarnpkg-core/sources/tgzUtils.ts

@@ -1,5 +1,6 @@
 import {FakeFS, PortablePath, NodeFS, ppath, xfs, npath, constants} from '@yarnpkg/fslib';
 import {ZipCompression, ZipFS}                                      from '@yarnpkg/libzip';
+import {Limit}                                                      from 'p-limit';
 import {PassThrough, Readable}                                      from 'stream';
 import tar                                                          from 'tar';
 
@@ -37,15 +38,25 @@ export interface ExtractBufferOptions {
   stripComponents?: number;
 }
 
+export interface ConvertToZipOptions extends ExtractBufferOptions {
+  poolSize?: Limit;
+}
+
 let workerPool: WorkerPool<ConvertToZipPayload, PortablePath> | null;
 
-export async function convertToZip(tgz: Buffer, opts: ExtractBufferOptions) {
+export async function convertToZip(tgz: Buffer, opts: ConvertToZipOptions) {
   const tmpFolder = await xfs.mktempPromise();
   const tmpFile = ppath.join(tmpFolder, `archive.zip`);
 
-  workerPool ||= new WorkerPool(getZipWorkerSource());
+  workerPool ||= new WorkerPool(getZipWorkerSource(), {poolSize: opts.poolSize});
+
+  const bufferOpts: ExtractBufferOptions = {
+    compressionLevel: opts.compressionLevel,
+    prefixPath: opts.prefixPath,
+    stripComponents: opts.stripComponents,
+  };
 
-  await workerPool.run({tmpFile, tgz, opts});
+  await workerPool.run({tmpFile, tgz, opts: bufferOpts});
 
   return new ZipFS(tmpFile, {level: opts.compressionLevel});
 }

packages/yarnpkg-core/tests/tgzUtils.test.ts

@@ -1,4 +1,5 @@
 import {npath, xfs}  from '@yarnpkg/fslib';
+import pLimit        from 'p-limit';
 
 import * as tgzUtils from '../sources/tgzUtils';
 
@@ -10,7 +11,7 @@ describe(`tgzUtils`, () => {
           npath.join(__dirname, `fixtures/carbon-icons-svelte-10.21.0.tgz`),
         ),
       );
-      await expect(tgzUtils.convertToZip(data, {compressionLevel: 0})).resolves.toBeTruthy();
+      await expect(tgzUtils.convertToZip(data, {compressionLevel: 0, poolSize: pLimit(2)})).resolves.toBeTruthy();
     });
 
     it(`should be able to convert a tgz without compression`, async () => {
@@ -19,7 +20,7 @@ describe(`tgzUtils`, () => {
           npath.join(__dirname, `fixtures/parse5-0.0.28.tgz`),
         ),
       );
-      await expect(tgzUtils.convertToZip(data, {compressionLevel: 0})).resolves.toBeTruthy();
+      await expect(tgzUtils.convertToZip(data, {compressionLevel: 0, poolSize: pLimit(2)})).resolves.toBeTruthy();
     });
   });
 });