perf: optimize the amount of workers (#9289)

ref #9286

.changeset/tame-hoops-sink.md

@@ -0,0 +1,6 @@
+---
+"@pnpm/worker": minor
+"pnpm": minor
+---
+
+The max amount of workers running for linking packages from the store has been reduced to 4 to achieve optimal results [#9286](https://github.com/pnpm/pnpm/issues/9286). The workers are performing many file system operations, so increasing the number of CPUs doesn't help performance after some point.

pnpm-lock.yaml

@@ -7931,6 +7931,9 @@ importers:
       load-json-file:
         specifier: 'catalog:'
         version: 6.2.0
+      p-limit:
+        specifier: 'catalog:'
+        version: 3.1.0
       shell-quote:
         specifier: 'catalog:'
         version: 1.8.2

worker/package.json

@@ -42,6 +42,7 @@
     "@rushstack/worker-pool": "catalog:",
     "is-windows": "catalog:",
     "load-json-file": "catalog:",
+    "p-limit": "catalog:",
     "shell-quote": "catalog:"
   },
   "peerDependencies": {

worker/src/index.ts

@@ -7,6 +7,7 @@ import { execSync } from 'child_process'
 import isWindows from 'is-windows'
 import { type PackageFilesIndex } from '@pnpm/store.cafs'
 import { type DependencyManifest } from '@pnpm/types'
+import pLimit from 'p-limit'
 import { quote as shellQuote } from 'shell-quote'
 import {
   type TarballExtractMessage,
@@ -29,7 +30,7 @@ export async function finishWorkers (): Promise<void> {
 }
 
 function createTarballWorkerPool (): WorkerPool {
-  const maxWorkers = Math.max(2, (os.availableParallelism?.() ?? os.cpus().length) - Math.abs(process.env.PNPM_WORKERS ? parseInt(process.env.PNPM_WORKERS) : 0)) - 1
+  const maxWorkers = calcMaxWorkers()
   const workerPool = new WorkerPool({
     id: 'pnpm',
     maxWorkers,
@@ -51,6 +52,18 @@ function createTarballWorkerPool (): WorkerPool {
   return workerPool
 }
 
+function calcMaxWorkers () {
+  if (process.env.PNPM_WORKERS) {
+    const idleCPUs = Math.abs(parseInt(process.env.PNPM_WORKERS))
+    return Math.max(2, availableParallelism() - idleCPUs) - 1
+  }
+  return Math.max(1, availableParallelism() - 1)
+}
+
+function availableParallelism (): number {
+  return os.availableParallelism?.() ?? os.cpus().length
+}
+
 interface AddFilesResult {
   filesIndex: Record<string, string>
   manifest: DependencyManifest
@@ -186,25 +199,33 @@ export async function readPkgFromCafs (
   })
 }
 
+// The workers are doing lots of file system operations
+// so, running them in parallel helps only to a point.
+// With local experimenting it was discovered that running 4 workers gives the best results.
+// Adding more workers actually makes installation slower.
+const limitImportingPackage = pLimit(4)
+
 export async function importPackage (
   opts: Omit<LinkPkgMessage, 'type'>
 ): Promise<{ isBuilt: boolean, importMethod: string | undefined }> {
-  if (!workerPool) {
-    workerPool = createTarballWorkerPool()
-  }
-  const localWorker = await workerPool.checkoutWorkerAsync(true)
-  return new Promise<{ isBuilt: boolean, importMethod: string | undefined }>((resolve, reject) => {
-    localWorker.once('message', ({ status, error, value }: any) => { // eslint-disable-line @typescript-eslint/no-explicit-any
-      workerPool!.checkinWorker(localWorker)
-      if (status === 'error') {
-        reject(new PnpmError(error.code ?? 'LINKING_FAILED', error.message as string))
-        return
-      }
-      resolve(value)
-    })
-    localWorker.postMessage({
-      type: 'link',
-      ...opts,
+  return limitImportingPackage(async () => {
+    if (!workerPool) {
+      workerPool = createTarballWorkerPool()
+    }
+    const localWorker = await workerPool.checkoutWorkerAsync(true)
+    return new Promise<{ isBuilt: boolean, importMethod: string | undefined }>((resolve, reject) => {
+      localWorker.once('message', ({ status, error, value }: any) => { // eslint-disable-line @typescript-eslint/no-explicit-any
+        workerPool!.checkinWorker(localWorker)
+        if (status === 'error') {
+          reject(new PnpmError(error.code ?? 'LINKING_FAILED', error.message as string))
+          return
+        }
+        resolve(value)
+      })
+      localWorker.postMessage({
+        type: 'link',
+        ...opts,
+      })
     })
   })
 }