From 23736e5863e0a8c618eff540ea730fca7930d890 Mon Sep 17 00:00:00 2001
From: Daniel Stockman <daniel.stockman@gmail.com>
Date: Mon, 13 May 2019 15:22:50 -0700
Subject: [PATCH] feat(exec): Add just-in-time queue management

---
 commands/exec/index.js     | 34 ++++++++++++++++++++++------------
 commands/exec/package.json |  6 +++---
 package-lock.json          |  6 +++---
 3 files changed, 28 insertions(+), 18 deletions(-)

diff --git a/commands/exec/index.js b/commands/exec/index.js
index c26a3e532e..72bf93f9eb 100644
--- a/commands/exec/index.js
+++ b/commands/exec/index.js
@@ -1,9 +1,10 @@
 "use strict";
 
+const pMap = require("p-map");
+
 const ChildProcessUtilities = require("@lerna/child-process");
 const Command = require("@lerna/command");
-const batchPackages = require("@lerna/batch-packages");
-const runParallelBatches = require("@lerna/run-parallel-batches");
+const runTopologically = require("@lerna/run-topologically");
 const ValidationError = require("@lerna/validation-error");
 const { getFilteredPackages } = require("@lerna/filter-options");
 
@@ -47,10 +48,6 @@ class ExecCommand extends Command {
       this.count = this.filteredPackages.length;
       this.packagePlural = this.count === 1 ? "package" : "packages";
       this.joinedCommand = [this.command].concat(this.args).join(" ");
-
-      this.batchedPackages = this.toposort
-        ? batchPackages(this.filteredPackages, this.options.rejectCycles)
-        : [this.filteredPackages];
     });
   }
 
@@ -67,8 +64,10 @@ class ExecCommand extends Command {
 
     if (this.options.parallel) {
       chain = chain.then(() => this.runCommandInPackagesParallel());
+    } else if (this.toposort) {
+      chain = chain.then(() => this.runCommandInPackagesTopological());
     } else {
-      chain = chain.then(() => this.runCommandInPackagesBatched());
+      chain = chain.then(() => this.runCommandInPackagesLexical());
     }
 
     if (this.bail) {
@@ -120,18 +119,29 @@ class ExecCommand extends Command {
     };
   }
 
-  runCommandInPackagesBatched() {
+  runCommandInPackagesTopological() {
     const runner = this.options.stream
       ? pkg => this.runCommandInPackageStreaming(pkg)
       : pkg => this.runCommandInPackageCapturing(pkg);
 
-    return runParallelBatches(this.batchedPackages, this.concurrency, runner).then(batchedResults =>
-      batchedResults.reduce((arr, batch) => arr.concat(batch), [])
-    );
+    return runTopologically({
+      packages: this.filteredPackages,
+      concurrency: this.concurrency,
+      rejectCycles: this.options.rejectCycles,
+      runner,
+    });
   }
 
   runCommandInPackagesParallel() {
-    return Promise.all(this.filteredPackages.map(pkg => this.runCommandInPackageStreaming(pkg)));
+    return pMap(this.filteredPackages, pkg => this.runCommandInPackageStreaming(pkg));
+  }
+
+  runCommandInPackagesLexical() {
+    const runner = this.options.stream
+      ? pkg => this.runCommandInPackageStreaming(pkg)
+      : pkg => this.runCommandInPackageCapturing(pkg);
+
+    return pMap(this.filteredPackages, runner, { concurrency: this.concurrency });
   }
 
   runCommandInPackageStreaming(pkg) {
diff --git a/commands/exec/package.json b/commands/exec/package.json
index 4181118815..5b601aa16c 100644
--- a/commands/exec/package.json
+++ b/commands/exec/package.json
@@ -35,11 +35,11 @@
     "populate--": true
   },
   "dependencies": {
-    "@lerna/batch-packages": "file:../../utils/batch-packages",
     "@lerna/child-process": "file:../../core/child-process",
     "@lerna/command": "file:../../core/command",
     "@lerna/filter-options": "file:../../core/filter-options",
-    "@lerna/run-parallel-batches": "file:../../utils/run-parallel-batches",
-    "@lerna/validation-error": "file:../../core/validation-error"
+    "@lerna/run-topologically": "file:../../utils/run-topologically",
+    "@lerna/validation-error": "file:../../core/validation-error",
+    "p-map": "^1.2.0"
   }
 }
diff --git a/package-lock.json b/package-lock.json
index 49708cbcf9..f02bf08f87 100644
--- a/package-lock.json
+++ b/package-lock.json
@@ -869,12 +869,12 @@
     "@lerna/exec": {
       "version": "file:commands/exec",
       "requires": {
-        "@lerna/batch-packages": "file:utils/batch-packages",
         "@lerna/child-process": "file:core/child-process",
         "@lerna/command": "file:core/command",
         "@lerna/filter-options": "file:core/filter-options",
-        "@lerna/run-parallel-batches": "file:utils/run-parallel-batches",
-        "@lerna/validation-error": "file:core/validation-error"
+        "@lerna/run-topologically": "file:utils/run-topologically",
+        "@lerna/validation-error": "file:core/validation-error",
+        "p-map": "^1.2.0"
       }
     },
     "@lerna/filter-options": {