feat: add concurrency option to run and exec (#1312)

To speed things up, add a `concurrency` option to `aegir run` and
`aegir exec` - this will run commands up to the cocurrency limit
while still taking monorepo sibling dependencies into account.

---------

Co-authored-by: Cayman <caymannava@gmail.com>

package.json

@@ -286,6 +286,7 @@
     "npm-package-json-lint": "^7.0.0",
     "nyc": "^15.1.0",
     "p-map": "^6.0.0",
+    "p-queue": "^7.3.4",
     "p-retry": "^6.0.0",
     "pascalcase": "^2.0.0",
     "path": "^0.12.7",

src/cmds/exec.js

@@ -34,6 +34,11 @@ export default {
           type: 'boolean',
           describe: 'Prefix output with the package name',
           default: userConfig.exec.prefix
+        },
+        concurrency: {
+          type: 'number',
+          describe: 'How many commands to run at the same time',
+          default: userConfig.exec.concurrency
         }
       })
   },

src/cmds/release-rc.js

@@ -40,6 +40,11 @@ export default {
           type: 'boolean',
           describe: 'Prefix output with the package name',
           default: userConfig.releaseRc.prefix
+        },
+        concurrency: {
+          type: 'number',
+          describe: 'How many modules to release at the same time',
+          default: userConfig.releaseRc.concurrency
         }
       })
   },

src/cmds/run.js

@@ -34,6 +34,11 @@ export default {
           type: 'boolean',
           describe: 'Prefix output with the package name',
           default: userConfig.run.prefix
+        },
+        concurrency: {
+          type: 'number',
+          describe: 'How many scripts to run at the same time',
+          default: userConfig.run.concurrency
         }
       })
       .positional('script', {

src/exec.js

@@ -31,6 +31,8 @@ export default {
 
         console.info(kleur.red(err.stack)) // eslint-disable-line no-console
       }
+    }, {
+      concurrency: ctx.concurrency
     })
   }
 }

src/release-rc.js

@@ -29,6 +29,8 @@ async function releaseMonorepoRcs (commit, ctx) {
     }
 
     versions[project.manifest.name] = `${project.manifest.version}-${commit}`
+  }, {
+    concurrency: ctx.concurrency
   })
 
   console.info('Will release the following packages:')
@@ -97,6 +99,8 @@ async function releaseMonorepoRcs (commit, ctx) {
     })
 
     console.info('')
+  }, {
+    concurrency: ctx.concurrency
   })
 }
 

src/run.js

@@ -44,6 +44,8 @@ export default {
           console.info(kleur.red(err.stack)) // eslint-disable-line no-console
         }
       }
+    }, {
+      concurrency: ctx.concurrency
     })
   }
 }

src/types.ts

@@ -341,6 +341,11 @@ interface ReleaseRcOptions {
    * Prefix output with the package name
    */
   prefix?: boolean
+
+  /**
+   * Release modules in parallel up to this limit
+   */
+  concurrency?: number
 }
 
 interface DependencyCheckOptions {
@@ -365,18 +370,28 @@ interface ExecOptions {
    * Prefix output with the package name
    */
   prefix?: boolean
+
+  /**
+   * Run commands in parallel up to this limit
+   */
+  concurrency?: number
 }
 
 interface RunOptions {
   /**
-   * If false, the command will continue to be run in other packages
+   * If false, the script will continue to be run in other packages
    */
   bail?: boolean
 
   /**
    * Prefix output with the package name
    */
   prefix?: boolean
+
+  /**
+   * Run scripts in parallel up to this limit
+   */
+  concurrency?: number
 }
 
 export type {

src/utils.js

@@ -18,6 +18,7 @@ import fs from 'fs-extra'
 import kleur from 'kleur'
 import Listr from 'listr'
 import { minimatch } from 'minimatch'
+import PQueue from 'p-queue'
 import lockfile from 'proper-lockfile'
 import { readPackageUpSync } from 'read-pkg-up'
 import stripBom from 'strip-bom'
@@ -313,14 +314,15 @@ export function findBinary (bin) {
  * @property {string} dir
  * @property {string[]} siblingDependencies
  * @property {string[]} dependencies
- * @property {boolean} run
  */
 
 /**
  * @param {string} projectDir
  * @param {(project: Project) => Promise<void>} fn
+ * @param {object} [opts]
+ * @param {number} [opts.concurrency]
  */
-export async function everyMonorepoProject (projectDir, fn) {
+export async function everyMonorepoProject (projectDir, fn, opts) {
   const manifest = fs.readJSONSync(path.join(projectDir, 'package.json'))
   const workspaces = manifest.workspaces
 
@@ -334,27 +336,40 @@ export async function everyMonorepoProject (projectDir, fn) {
   checkForCircularDependencies(projects)
 
   /**
-   * @param {Project} project
+   * @type {Map<string, number>} Track the number of outstanding dependencies of each project
+   *
+   * This is mutated (decremented and deleted) as tasks are run for dependencies
    */
-  async function run (project) {
-    if (project.run) {
-      return
-    }
+  const inDegree = new Map()
+  for (const [name, project] of Object.entries(projects)) {
+    inDegree.set(name, project.siblingDependencies.length)
+  }
 
-    for (const siblingDep of project.siblingDependencies) {
-      await run(projects[siblingDep])
-    }
+  const queue = new PQueue({
+    concurrency: opts?.concurrency ?? os.availableParallelism?.() ?? os.cpus().length
+  })
 
-    if (project.run) {
-      return
+  while (inDegree.size) {
+    /** @type {string[]} */
+    const toRun = []
+
+    for (const [name, d] of inDegree) {
+      // when there are no more dependencies
+      // the project can be added to the queue
+      // and removed from the tracker
+      if (d === 0) {
+        toRun.push(name)
+        inDegree.delete(name)
+      }
     }
 
-    project.run = true
-    await fn(project)
-  }
+    await Promise.all(toRun.map((name) => queue.add(() => fn(projects[name]))))
 
-  for (const project of Object.values(projects)) {
-    await run(project)
+    // decrement projects whose dependencies were just run
+    for (const [name, d] of inDegree) {
+      const decrement = projects[name].siblingDependencies.filter(dep => toRun.includes(dep)).length
+      inDegree.set(name, d - decrement)
+    }
   }
 }
 
@@ -390,7 +405,6 @@ export function parseProjects (projectDir, workspaces) {
         manifest: pkg,
         dir: subProjectDir,
         siblingDependencies: [],
-        run: false,
         dependencies: [
           ...Object.keys(pkg.dependencies ?? {}),
           ...Object.keys(pkg.devDependencies ?? {}),

test/fixtures/projects/a-large-monorepo/package.json

@@ -0,0 +1,12 @@
+{
+  "name": "a-large-monorepo",
+  "version": "1.0.0",
+  "description": "",
+  "homepage": "https://github.com/ipfs/aegir#readme",
+  "scripts": {
+    "docs": "aegir docs"
+  },
+  "workspaces": [
+    "packages/*"
+  ]
+}

test/fixtures/projects/a-large-monorepo/packages/a/package.json

@@ -0,0 +1,17 @@
+{
+  "name": "a",
+  "version": "1.0.0",
+  "description": "",
+  "homepage": "https://github.com/ipfs/aegir#readme",
+  "exports": {
+    ".": {
+      "import": "./src/index.js"
+    }
+  },
+  "scripts": {
+    "test": "echo very test"
+  },
+  "type": "module",
+  "author": "",
+  "license": "ISC"
+}

test/fixtures/projects/a-large-monorepo/packages/a/src/index.js

@@ -0,0 +1,16 @@
+/**
+ * @typedef {import('./types').ExportedButNotInExports} ExportedButNotInExports
+ */
+
+/**
+ * @typedef {object} AnExportedInterface
+ * @property {() => void} AnExportedInterface.aMethod
+ */
+
+export const useHerp = () => {
+
+}
+
+export const useDerp = () => {
+
+}

test/fixtures/projects/a-large-monorepo/packages/a/src/types.ts

@@ -0,0 +1,3 @@
+export interface ExportedButNotInExports {
+  aMethod(): void
+}

test/fixtures/projects/a-large-monorepo/packages/a/tsconfig.json

@@ -0,0 +1,39 @@
+{
+  "compilerOptions": {
+      "strict": true,
+      // project options
+      "outDir": "dist",
+      "allowJs": true,
+      "checkJs": true,
+      "target": "ES2020",
+      "module": "ES2020",
+      "lib": ["ES2021", "ES2021.Promise", "ES2021.String", "ES2020.BigInt", "DOM", "DOM.Iterable"],
+      "noEmit": false,
+      "noEmitOnError": true,
+      "emitDeclarationOnly": true,
+      "declaration": true,
+      "declarationMap": true,
+      "incremental": true,
+      "composite": true,
+      "isolatedModules": true,
+      "removeComments": false,
+      "sourceMap": true,
+      // module resolution
+      "esModuleInterop": true,
+      "moduleResolution": "node",
+      // linter checks
+      "noImplicitReturns": false,
+      "noFallthroughCasesInSwitch": true,
+      "noUnusedLocals": true,
+      "noUnusedParameters": false,
+      // advanced
+      "verbatimModuleSyntax": true,
+      "forceConsistentCasingInFileNames": true,
+      "skipLibCheck": true,
+      "stripInternal": true,
+      "resolveJsonModule": true
+  },
+  "include": [
+      "src"
+  ]
+}

test/fixtures/projects/a-large-monorepo/packages/a/typedoc.json

@@ -0,0 +1,3 @@
+{
+  "entryPoints": ["./src/index.js"]
+}

test/fixtures/projects/a-large-monorepo/packages/b/package.json

@@ -0,0 +1,20 @@
+{
+  "name": "b",
+  "version": "1.0.0",
+  "description": "",
+  "homepage": "https://github.com/ipfs/aegir#readme",
+  "exports": {
+    ".": {
+      "import": "./src/index.js"
+    }
+  },
+  "dependencies": {
+    "a": "1.0.0"
+  },
+  "scripts": {
+    "test": "echo very test"
+  },
+  "type": "module",
+  "author": "",
+  "license": "ISC"
+}

test/fixtures/projects/a-large-monorepo/packages/b/src/dir/index.js

@@ -0,0 +1,7 @@
+/**
+ * @param {string} arg
+ * @returns {boolean}
+ */
+export function garply (arg) {
+  return true
+}

test/fixtures/projects/a-large-monorepo/packages/b/src/index.js

@@ -0,0 +1,18 @@
+/**
+ * @typedef {import('./types.js').ExportedButNotInExports} ExportedButNotInExports
+ */
+
+/**
+ * @typedef {object} AnExportedInterface
+ * @property {() => void} AnExportedInterface.aMethod
+ */
+
+export const useHerp = () => {
+
+}
+
+export const useDerp = () => {
+
+}
+
+export { garply } from './dir/index.js'

test/fixtures/projects/a-large-monorepo/packages/b/src/types.ts

@@ -0,0 +1,3 @@
+export interface ExportedButNotInExports {
+  aMethod(): void
+}