batch-cluster - npm Package Compare versions

Comparing version 5.3.1 to 5.4.0

CHANGELOG.md

@@ -20,2 +20,18 @@ # Changelog
 
+## v5.4.0
+
+- ✨ "wear-leveling" for processes. Previously, only the first-spawned child
+  process would service most task requests, but that caused issues with (very)
+  long-running tasks where the other child processes would be spooled off ram,
+  and could time out when requested later.
+- 🐞 `maxProcs` is respected again. In prior builds, if tasks were enqueued all
+  at once, prior dispatch code would only spin 1 concurrent task at a time.
+- 🐞 Multiple calls to `BatchProcess.end` would result in different promise
+  resolution targets: the second call to `.end()` would resolve before the
+  first. This was fixed.
+- ✨
+  [BatchProcessOptions](https://batch-cluster.js.org/classes/batchclusteroptions.html)'s
+  `minDelayBetweenSpawnMillis` was added, to help relieve undue system load on
+  startup. It defaults to 1.5 seconds and can be disabled by setting it to 0.
+
 ## v5.3.1
@@ -22,0 +38,0 @@

dist/Array.d.ts

@@ -8,1 +8,9 @@ /**
 export declare function sortNumeric(arr: number[]): number[];
+/**
+ * Treat an array as a round-robin list, starting from `startIndex`.
+ */
+export declare function rrFind<T>(arr: T[], startIndex: number, predicate: (t: T, arrIdx: number, iter: number) => boolean): undefined | T;
+export declare function rrFindResult<T>(arr: T[], startIndex: number, predicate: (t: T, arrIdx: number, iter: number) => boolean): undefined | {
+    result: T;
+    index: number;
+};

dist/Array.js

 "use strict";
+var __read = (this && this.__read) || function (o, n) {
+    var m = typeof Symbol === "function" && o[Symbol.iterator];
+    if (!m) return o;
+    var i = m.call(o), r, ar = [], e;
+    try {
+        while ((n === void 0 || n-- > 0) && !(r = i.next()).done) ar.push(r.value);
+    }
+    catch (error) { e = { error: error }; }
+    finally {
+        try {
+            if (r && !r.done && (m = i["return"])) m.call(i);
+        }
+        finally { if (e) throw e.error; }
+    }
+    return ar;
+};
+var __spread = (this && this.__spread) || function () {
+    for (var ar = [], i = 0; i < arguments.length; i++) ar = ar.concat(__read(arguments[i]));
+    return ar;
+};
 Object.defineProperty(exports, "__esModule", { value: true });
+var Object_1 = require("./Object");
 /**
@@ -22,3 +43,3 @@ * Remove all elements from the given array that return false from the given
     if (result === void 0) { result = []; }
-    arr.forEach(function (ea) { return (Array.isArray(ea) ? result.push.apply(result, ea) : result.push(ea)); });
+    arr.forEach(function (ea) { return (Array.isArray(ea) ? result.push.apply(result, __spread(ea)) : result.push(ea)); });
     return result;
@@ -31,2 +52,19 @@ }
 exports.sortNumeric = sortNumeric;
+/**
+ * Treat an array as a round-robin list, starting from `startIndex`.
+ */
+function rrFind(arr, startIndex, predicate) {
+    return Object_1.map(rrFindResult(arr, startIndex, predicate), function (ea) { return ea.result; });
+}
+exports.rrFind = rrFind;
+function rrFindResult(arr, startIndex, predicate) {
+    for (var iter = 0; iter < arr.length; iter++) {
+        var arrIdx = (iter + startIndex) % arr.length;
+        var t = arr[arrIdx];
+        if (predicate(t, arrIdx, iter))
+            return { result: t, index: arrIdx };
+    }
+    return;
+}
+exports.rrFindResult = rrFindResult;
 //# sourceMappingURL=Array.js.map

dist/BatchCluster.d.ts

@@ -40,2 +40,4 @@ /// <reference types="node" />
     private readonly _procs;
+    private _lastUsedProcsIdx;
+    private _lastSpawnedProcTime;
     private readonly tasks;
@@ -45,3 +47,3 @@ private onIdleInterval?;
     private _spawnedProcs;
-    private _ended;
+    private endprocs?;
     private _internalErrorCount;
@@ -73,8 +75,14 @@ constructor(opts: Partial<BatchClusterOptions> & BatchProcessOptions & ChildProcessFactory);
     readonly meanTasksPerProc: number;
+    /**
+     * @return the total number of child processes created by this instance
+     */
     readonly spawnedProcs: number;
     /**
+     * @return the current number of child processes currently servicing tasks
+     */
+    readonly busyProcs: number;
+    /**
      * For integration tests:
      */
     readonly internalErrorCount: number;
-    private endPromise;
     private onInternalError;
@@ -88,3 +96,6 @@ private onStartError;
     pids(): Promise<number[]>;
-    private readonly onIdle;
+    private onIdle;
+    private vacuumProcs;
+    private execNextTask;
+    private readonly maybeLaunchNewChild;
 }

dist/BatchCluster.js

@@ -50,2 +50,12 @@ "use strict";
 };
+var __values = (this && this.__values) || function (o) {
+    var m = typeof Symbol === "function" && o[Symbol.iterator], i = 0;
+    if (m) return m.call(o);
+    return {
+        next: function () {
+            if (o && i >= o.length) o = void 0;
+            return { value: o && o[i++], done: !o };
+        }
+    };
+};
 function __export(m) {
@@ -96,11 +106,12 @@ for (var p in m) if (!exports.hasOwnProperty(p)) exports[p] = m[p];
         _this._procs = [];
+        _this._lastUsedProcsIdx = 0;
+        _this._lastSpawnedProcTime = 0;
         _this.tasks = [];
         _this.startErrorRate = new Rate_1.Rate();
         _this._spawnedProcs = 0;
-        _this._ended = false;
         _this._internalErrorCount = 0;
         _this.beforeExitListener = function () { return _this.end(true); };
         _this.exitListener = function () { return _this.end(false); };
-        _this.onIdle = Async_1.atMostOne(function () { return __awaiter(_this, void 0, void 0, function () {
-            var minStart, execNextTask, child_1, proc_1, err_1;
+        _this.maybeLaunchNewChild = Async_1.atMostOne(function () { return __awaiter(_this, void 0, void 0, function () {
+            var child_1, proc_1, err_1;
             var _this = this;
@@ -110,39 +121,14 @@ return __generator(this, function (_a) {
                     case 0:
-                        if (this._ended)
+                        // Minimize start time system load. Only launch one new proc at a time
+                        if (this.ended ||
+                            this.tasks.length === 0 ||
+                            this._procs.length >= this.opts.maxProcs ||
+                            this._lastSpawnedProcTime >
+                                Date.now() - this.opts.minDelayBetweenSpawnMillis) {
                             return [2 /*return*/];
-                        minStart = Date.now() - this.opts.maxProcAgeMillis;
-                        Array_1.filterInPlace(this._procs, function (proc) {
-                            var old = proc.start < minStart && _this.opts.maxProcAgeMillis > 0;
-                            var worn = proc.taskCount >= _this.opts.maxTasksPerProcess;
-                            var broken = proc.exited;
-                            if (proc.idle && (old || worn || broken)) {
-                                // tslint:disable-next-line: no-floating-promises
-                                proc.end(true, old ? "old" : worn ? "worn" : "broken");
-                                return false;
-                            }
-                            else {
-                                return true;
-                            }
-                        });
-                        execNextTask = function () {
-                            var proc = _this._procs.find(function (ea) { return ea.ready; });
-                            if (proc == null)
-                                return false;
-                            var task = _this.tasks.shift();
-                            if (task == null)
-                                return false;
-                            var submitted = proc.execTask(task);
-                            if (!submitted) {
-                                // tslint:disable-next-line: no-floating-promises
-                                _this.enqueueTask(task);
-                            }
-                            return false;
-                        };
-                        while (!this._ended && execNextTask()) { }
-                        if (!(!this._ended &&
-                            this.tasks.length > 0 &&
-                            this._procs.length < this.opts.maxProcs)) return [3 /*break*/, 4];
+                        }
                         _a.label = 1;
                     case 1:
                         _a.trys.push([1, 3, , 4]);
+                        this._lastSpawnedProcTime = Date.now();
                         return [4 /*yield*/, this.opts.processFactory()];
@@ -152,2 +138,3 @@ case 2:
                         proc_1 = new BatchProcess_1.BatchProcess(child_1, this.opts, this.observer);
+                        this._procs.push(proc_1);
                         this.emitter.emit("childStart", child_1);
@@ -159,9 +146,8 @@ // tslint:disable-next-line: no-floating-promises
                         });
-                        this._procs.push(proc_1);
                         this._spawnedProcs++;
-                        return [3 /*break*/, 4];
+                        return [2 /*return*/, proc_1];
                     case 3:
                         err_1 = _a.sent();
                         this.emitter.emit("startError", err_1);
-                        return [3 /*break*/, 4];
+                        return [2 /*return*/];
                     case 4: return [2 /*return*/];
@@ -191,3 +177,3 @@ }
         get: function () {
-            return this._ended;
+            return this.endprocs != null;
         },
@@ -205,31 +191,18 @@ enumerable: true,
         return __awaiter(this, void 0, void 0, function () {
-            var alreadyEnded;
             var _this = this;
             return __generator(this, function (_a) {
-                switch (_a.label) {
-                    case 0:
-                        alreadyEnded = this._ended;
-                        this._ended = true;
-                        if (!alreadyEnded) {
-                            this.emitter.emit("beforeEnd");
-                            Object_1.map(this.onIdleInterval, timers_1.clearInterval);
-                            this.onIdleInterval = undefined;
-                            _p.removeListener("beforeExit", this.beforeExitListener);
-                            _p.removeListener("exit", this.exitListener);
-                        }
-                        if (!(!gracefully || !alreadyEnded)) return [3 /*break*/, 2];
-                        return [4 /*yield*/, Promise.all(this._procs.map(function (p) {
-                                return p
-                                    .end(gracefully, "BatchCluster.end()")
-                                    .catch(function (err) { return _this.emitter.emit("endError", err); });
-                            }))];
-                    case 1:
-                        _a.sent();
-                        this._procs.length = 0;
-                        _a.label = 2;
-                    case 2: return [2 /*return*/, this.endPromise().then(function () {
-                            if (!alreadyEnded)
-                                _this.emitter.emit("end");
-                        })];
+                if (this.endprocs == null) {
+                    this.emitter.emit("beforeEnd");
+                    Object_1.map(this.onIdleInterval, timers_1.clearInterval);
+                    this.onIdleInterval = undefined;
+                    _p.removeListener("beforeExit", this.beforeExitListener);
+                    _p.removeListener("exit", this.exitListener);
+                    this.endprocs = Promise.all(this._procs.map(function (p) {
+                        return p
+                            .end(gracefully, "BatchCluster.end()")
+                            .catch(function (err) { return _this.emitter.emit("endError", err); });
+                    })).then(function () { return _this.emitter.emit("end"); });
+                    this._procs.length = 0;
                 }
+                return [2 /*return*/, this.endprocs];
             });
@@ -246,3 +219,3 @@ });
         var _this = this;
-        if (this._ended) {
+        if (this.ended) {
             task.reject(new Error("BatchCluster has ended"));
@@ -276,2 +249,5 @@ throw new Error("Cannot enqueue task " + task.command);
     Object.defineProperty(BatchCluster.prototype, "spawnedProcs", {
+        /**
+         * @return the total number of child processes created by this instance
+         */
         get: function () {
@@ -283,2 +259,13 @@ return this._spawnedProcs;
     });
+    Object.defineProperty(BatchCluster.prototype, "busyProcs", {
+        /**
+         * @return the current number of child processes currently servicing tasks
+         */
+        get: function () {
+            return this._procs.filter(function (ea) { return ea.taskCount > 0 && !ea.exited && !ea.idle; })
+                .length;
+        },
+        enumerable: true,
+        configurable: true
+    });
     Object.defineProperty(BatchCluster.prototype, "internalErrorCount", {
@@ -294,10 +281,2 @@ /**
     });
-    BatchCluster.prototype.endPromise = function () {
-        var _this = this;
-        return Promise.all(this._procs.map(function (p) { return p.exitedPromise; }))
-            .then(function () { })
-            .catch(function (err) {
-            _this.emitter.emit("endError", err);
-        });
-    };
     BatchCluster.prototype.onInternalError = function (error) {
@@ -329,21 +308,35 @@ this.emitter.emit("internalError", error);
         return __awaiter(this, void 0, void 0, function () {
-            var arr, _i, _a, pid;
-            return __generator(this, function (_b) {
-                switch (_b.label) {
+            var e_1, _a, arr, _b, _c, pid, e_1_1;
+            return __generator(this, function (_d) {
+                switch (_d.label) {
                     case 0:
                         arr = [];
-                        _i = 0, _a = this._procs.map(function (p) { return p.pid; });
-                        _b.label = 1;
+                        _d.label = 1;
                     case 1:
-                        if (!(_i < _a.length)) return [3 /*break*/, 4];
-                        pid = _a[_i];
+                        _d.trys.push([1, 6, 7, 8]);
+                        _b = __values(this._procs.map(function (p) { return p.pid; })), _c = _b.next();
+                        _d.label = 2;
+                    case 2:
+                        if (!!_c.done) return [3 /*break*/, 5];
+                        pid = _c.value;
                         return [4 /*yield*/, Pids_1.pidExists(pid)];
-                    case 2:
-                        if (_b.sent())
+                    case 3:
+                        if (_d.sent())
                             arr.push(pid);
-                        _b.label = 3;
-                    case 3:
-                        _i++;
-                        return [3 /*break*/, 1];
-                    case 4: return [2 /*return*/, arr];
+                        _d.label = 4;
+                    case 4:
+                        _c = _b.next();
+                        return [3 /*break*/, 2];
+                    case 5: return [3 /*break*/, 8];
+                    case 6:
+                        e_1_1 = _d.sent();
+                        e_1 = { error: e_1_1 };
+                        return [3 /*break*/, 8];
+                    case 7:
+                        try {
+                            if (_c && !_c.done && (_a = _b.return)) _a.call(_b);
+                        }
+                        finally { if (e_1) throw e_1.error; }
+                        return [7 /*endfinally*/];
+                    case 8: return [2 /*return*/, arr];
                 }
@@ -353,2 +346,47 @@ });
     };
+    BatchCluster.prototype.onIdle = function () {
+        if (this.ended)
+            return;
+        this.vacuumProcs();
+        while (this.execNextTask()) { }
+        // tslint:disable-next-line: no-floating-promises
+        this.maybeLaunchNewChild();
+        return;
+    };
+    BatchCluster.prototype.vacuumProcs = function () {
+        var _this = this;
+        Array_1.filterInPlace(this._procs, function (proc) {
+            // Only idle procs are eligible for deletion:
+            if (!proc.idle)
+                return true;
+            var old = _this.opts.maxProcAgeMillis > 0 &&
+                proc.start + _this.opts.maxProcAgeMillis < Date.now();
+            var wornOut = _this.opts.maxTasksPerProcess > 0 &&
+                proc.taskCount >= _this.opts.maxTasksPerProcess;
+            var broken = proc.exited;
+            var reap = old || wornOut || broken; // # me
+            if (reap) {
+                // tslint:disable-next-line: no-floating-promises
+                proc.end(true, old ? "old" : wornOut ? "worn" : "broken");
+            }
+            return !reap;
+        });
+    };
+    BatchCluster.prototype.execNextTask = function () {
+        if (this.ended)
+            return false;
+        var readyProc = Array_1.rrFindResult(this._procs, this._lastUsedProcsIdx + 1, function (ea) { return ea.ready; });
+        if (readyProc == null)
+            return false;
+        var task = this.tasks.shift();
+        if (task == null)
+            return false;
+        this._lastUsedProcsIdx = readyProc.index;
+        var submitted = readyProc.result.execTask(task);
+        if (!submitted) {
+            // tslint:disable-next-line: no-floating-promises
+            this.enqueueTask(task);
+        }
+        return submitted;
+    };
     return BatchCluster;
@@ -355,0 +393,0 @@ }(BatchClusterEmitter_1.BatchClusterEmitter));

dist/BatchClusterOptions.d.ts

@@ -56,2 +56,9 @@ import { ChildProcessFactory } from "./BatchCluster";
     /**
+     * If maxProcs > 1, spawning new child processes to process tasks can slow
+     * down initial processing, and create unnecessary processes.
+     *
+     * Must be >= 0ms. Defaults to 1.5 seconds.
+     */
+    readonly minDelayBetweenSpawnMillis: number;
+    /**
      * If commands take longer than this, presume the underlying process is dead
@@ -58,0 +65,0 @@ * and we should fail the task.

dist/BatchClusterOptions.js

@@ -68,2 +68,9 @@ "use strict";
         /**
+         * If maxProcs > 1, spawning new child processes to process tasks can slow
+         * down initial processing, and create unnecessary processes.
+         *
+         * Must be >= 0ms. Defaults to 1.5 seconds.
+         */
+        this.minDelayBetweenSpawnMillis = 1500;
+        /**
          * If commands take longer than this, presume the underlying process is dead
@@ -141,3 +148,6 @@ * and we should fail the task.
     gte("maxProcs", 1);
-    gte("maxProcAgeMillis", Math.max(result.spawnTimeoutMillis, result.taskTimeoutMillis));
+    if (opts.maxProcAgeMillis !== 0) {
+        gte("maxProcAgeMillis", Math.max(result.spawnTimeoutMillis, result.taskTimeoutMillis));
+    }
+    gte("minDelayBetweenSpawnMillis", 0);
     gte("onIdleIntervalMillis", 0);
@@ -144,0 +154,0 @@ gte("endGracefulWaitTimeMillis", 0);

dist/Error.js

 "use strict";
+var __values = (this && this.__values) || function (o) {
+    var m = typeof Symbol === "function" && o[Symbol.iterator], i = 0;
+    if (m) return m.call(o);
+    return {
+        next: function () {
+            if (o && i >= o.length) o = void 0;
+            return { value: o && o[i++], done: !o };
+        }
+    };
+};
 Object.defineProperty(exports, "__esModule", { value: true });
@@ -8,8 +18,18 @@ /**
 function tryEach(arr) {
-    for (var _i = 0, arr_1 = arr; _i < arr_1.length; _i++) {
-        var f = arr_1[_i];
+    var e_1, _a;
+    try {
+        for (var arr_1 = __values(arr), arr_1_1 = arr_1.next(); !arr_1_1.done; arr_1_1 = arr_1.next()) {
+            var f = arr_1_1.value;
+            try {
+                f();
+            }
+            catch (_) { }
+        }
+    }
+    catch (e_1_1) { e_1 = { error: e_1_1 }; }
+    finally {
         try {
-            f();
+            if (arr_1_1 && !arr_1_1.done && (_a = arr_1.return)) _a.call(arr_1);
         }
-        catch (_) { }
+        finally { if (e_1) throw e_1.error; }
     }
@@ -16,0 +36,0 @@ }

dist/Logger.js

 "use strict";
+var __read = (this && this.__read) || function (o, n) {
+    var m = typeof Symbol === "function" && o[Symbol.iterator];
+    if (!m) return o;
+    var i = m.call(o), r, ar = [], e;
+    try {
+        while ((n === void 0 || n-- > 0) && !(r = i.next()).done) ar.push(r.value);
+    }
+    catch (error) { e = { error: error }; }
+    finally {
+        try {
+            if (r && !r.done && (m = i["return"])) m.call(i);
+        }
+        finally { if (e) throw e.error; }
+    }
+    return ar;
+};
+var __spread = (this && this.__spread) || function () {
+    for (var ar = [], i = 0; i < arguments.length; i++) ar = ar.concat(__read(arguments[i]));
+    return ar;
+};
 Object.defineProperty(exports, "__esModule", { value: true });
@@ -78,3 +98,3 @@ var util_1 = require("util");
                 if (String_1.notBlank(message)) {
-                    delegate[ea].apply(delegate, [prefix + message].concat(optionalParams));
+                    delegate[ea].apply(delegate, __spread([prefix + message], optionalParams));
                 }
@@ -95,3 +115,3 @@ };
                 return Object_1.map(message, function (ea) {
-                    return delegate[level].apply(delegate, [new Date().toISOString() + " | " + ea].concat(optionalParams));
+                    return delegate[level].apply(delegate, __spread([new Date().toISOString() + " | " + ea], optionalParams));
                 });
@@ -98,0 +118,0 @@ });

package.json

 {
   "name": "batch-cluster",
-  "version": "5.3.1",
+  "version": "5.4.0",
   "description": "Manage a cluster of child processes",
@@ -40,3 +40,3 @@ "main": "dist/BatchCluster.js",
     "@types/mocha": "^5.2.6",
-    "@types/node": "^11.10.4",
+    "@types/node": "^11.11.3",
     "chai": "^4.2.0",
@@ -51,5 +51,5 @@ "chai-as-promised": "^7.1.1",
     "serve": "^10.1.2",
-    "source-map-support": "^0.5.10",
+    "source-map-support": "^0.5.11",
     "split2": "^3.1.0",
-    "timekeeper": "^2.1.2",
+    "timekeeper": "^2.2.0",
     "tslint": "^5.13.1",
@@ -56,0 +56,0 @@ "typedoc": "^0.14.2",

No alert changes

Improved metrics

Total package byte prevSize

219269

increased by93.67%

Number of package files

47

increased by2.17%

Lines of code

2546

increased by6.35%

No dependency changes