batch-cluster - npm Package Compare versions

Comparing version 10.1.0 to 10.1.1

CHANGELOG.md

@@ -21,2 +21,14 @@ # Changelog
 
+## v10.1.1
+
+- 🐞 Fixed [issue #15](https://github.com/photostructure/batch-cluster.js/issues/15) by restoring the call to `#onIdleLater` when tasks settle.
+
+- 🐞 Fixed issue with `setMaxProcs` which resulted in all idle processes being reaped
+
+- 📦 The `idle` event was removed. You weren't using it, though, so I'm not bumping major.
+
+- 📦 Process shutdown is handled more gracefully with new `thenOrTimeout` (rather than the prior `Promise.race` call which resulting in a dangling timeout)
+
+- 📦 Updated development dependencies and rebuild docs
+
 ## v10.1.0
@@ -23,0 +35,0 @@

dist/BatchCluster.d.ts

@@ -117,3 +117,3 @@ /// <reference types="node" />
         childEndCounts: {
-            idle: number;
+            timeout: number;
             broken: number;
@@ -123,2 +123,3 @@ closed: number;
             ended: number;
+            idle: number;
             old: number;
@@ -133,3 +134,2 @@ "proc.close": number;
             "stdout.error": number;
-            timeout: number;
             tooMany: number;
@@ -136,0 +136,0 @@ unhealthy: number;

dist/BatchCluster.js

@@ -26,3 +26,3 @@ "use strict";
 };
-var _BatchCluster_instances, _BatchCluster_tasksPerProc, _BatchCluster_logger, _BatchCluster_procs, _BatchCluster_lastSpawnedProcTime, _BatchCluster_lastPidsCheckTime, _BatchCluster_tasks, _BatchCluster_onIdleInterval, _BatchCluster_startErrorRate, _BatchCluster_spawnedProcs, _BatchCluster_endPromise, _BatchCluster_internalErrorCount, _BatchCluster_childEndCounts, _BatchCluster_beforeExitListener, _BatchCluster_exitListener, _BatchCluster_onIdleLater, _BatchCluster_onIdle, _BatchCluster_maybeCheckPids, _BatchCluster_canRetainProc, _BatchCluster_execNextTask, _BatchCluster_maybeLaunchNewChild, _BatchCluster_spawnChild;
+var _BatchCluster_instances, _BatchCluster_tasksPerProc, _BatchCluster_logger, _BatchCluster_procs, _BatchCluster_lastSpawnedProcTime, _BatchCluster_lastPidsCheckTime, _BatchCluster_tasks, _BatchCluster_onIdleInterval, _BatchCluster_startErrorRate, _BatchCluster_spawnedProcs, _BatchCluster_endPromise, _BatchCluster_internalErrorCount, _BatchCluster_childEndCounts, _BatchCluster_beforeExitListener, _BatchCluster_exitListener, _BatchCluster_onIdleLater, _BatchCluster_onIdle, _BatchCluster_maybeCheckPids, _BatchCluster_execNextTask, _BatchCluster_maybeLaunchNewChild, _BatchCluster_spawnChild;
 Object.defineProperty(exports, "__esModule", { value: true });
@@ -96,22 +96,2 @@ exports.BatchCluster = exports.Task = exports.pids = exports.pidExists = exports.kill = exports.SimpleParser = exports.Deferred = exports.BatchProcess = exports.BatchClusterOptions = void 0;
         );
-        _BatchCluster_canRetainProc.set(this, (proc) => {
-            var _a;
-            // don't bother busy procs:
-            if (!proc.idle)
-                return true;
-            const why = (_a = proc.whyNotHealthy) !== null && _a !== void 0 ? _a : (__classPrivateFieldGet(this, _BatchCluster_procs, "f").length > this.options.maxProcs ? "tooMany" : null);
-            if (why != null) {
-                void proc.end(true, why);
-                return false;
-            }
-            if (proc.idle)
-                proc.maybeRunHealthcheck();
-            return true;
-        }
-        /**
-         * Run maintenance on currently spawned child processes. This method is
-         * normally invoked automatically as tasks are enqueued and processed.
-         */
-        // NOT ASYNC: updates internal state. only exported for tests.
-        );
         this.options = (0, BatchClusterOptions_1.verifyOptions)({ ...opts, observer: this.emitter });
@@ -185,5 +165,7 @@ this.on("childEnd", (bp, why) => {
         __classPrivateFieldGet(this, _BatchCluster_tasks, "f").push(task);
-        // Run #onIdle now (not later), to make sure the task gets enqueued if
-        // possible, asap
+        // Run #onIdle now (not later), to make sure the task gets enqueued asap if
+        // possible
         __classPrivateFieldGet(this, _BatchCluster_instances, "m", _BatchCluster_onIdle).call(this);
+        // (BatchProcess will call our #onIdleLater when tasks settle or when they
+        // exit)
         return task.promise;
@@ -318,7 +300,24 @@ }
         __classPrivateFieldGet(this, _BatchCluster_instances, "m", _BatchCluster_maybeCheckPids).call(this);
-        (0, Array_1.filterInPlace)(__classPrivateFieldGet(this, _BatchCluster_procs, "f"), __classPrivateFieldGet(this, _BatchCluster_canRetainProc, "f"));
+        let pidsToReap = Math.max(0, __classPrivateFieldGet(this, _BatchCluster_procs, "f").length - this.options.maxProcs);
+        (0, Array_1.filterInPlace)(__classPrivateFieldGet(this, _BatchCluster_procs, "f"), (proc) => {
+            var _a;
+            // don't bother busy procs:
+            if (proc.idle) {
+                // don't reap more than pidsToReap pids. We can't use #procs.length
+                // within filterInPlace because #procs.length only changes at iteration
+                // completion: the prior impl resulted in all idle pids getting reaped
+                // when maxProcs was reduced.
+                const why = (_a = proc.whyNotHealthy) !== null && _a !== void 0 ? _a : (--pidsToReap >= 0 ? "tooMany" : null);
+                if (why != null) {
+                    void proc.end(true, why);
+                    return false;
+                }
+                proc.maybeRunHealthcheck();
+            }
+            return true;
+        });
     }
 }
 exports.BatchCluster = BatchCluster;
-_BatchCluster_tasksPerProc = new WeakMap(), _BatchCluster_logger = new WeakMap(), _BatchCluster_procs = new WeakMap(), _BatchCluster_lastSpawnedProcTime = new WeakMap(), _BatchCluster_lastPidsCheckTime = new WeakMap(), _BatchCluster_tasks = new WeakMap(), _BatchCluster_onIdleInterval = new WeakMap(), _BatchCluster_startErrorRate = new WeakMap(), _BatchCluster_spawnedProcs = new WeakMap(), _BatchCluster_endPromise = new WeakMap(), _BatchCluster_internalErrorCount = new WeakMap(), _BatchCluster_childEndCounts = new WeakMap(), _BatchCluster_beforeExitListener = new WeakMap(), _BatchCluster_exitListener = new WeakMap(), _BatchCluster_onIdleLater = new WeakMap(), _BatchCluster_canRetainProc = new WeakMap(), _BatchCluster_instances = new WeakSet(), _BatchCluster_onIdle = function _BatchCluster_onIdle() {
+_BatchCluster_tasksPerProc = new WeakMap(), _BatchCluster_logger = new WeakMap(), _BatchCluster_procs = new WeakMap(), _BatchCluster_lastSpawnedProcTime = new WeakMap(), _BatchCluster_lastPidsCheckTime = new WeakMap(), _BatchCluster_tasks = new WeakMap(), _BatchCluster_onIdleInterval = new WeakMap(), _BatchCluster_startErrorRate = new WeakMap(), _BatchCluster_spawnedProcs = new WeakMap(), _BatchCluster_endPromise = new WeakMap(), _BatchCluster_internalErrorCount = new WeakMap(), _BatchCluster_childEndCounts = new WeakMap(), _BatchCluster_beforeExitListener = new WeakMap(), _BatchCluster_exitListener = new WeakMap(), _BatchCluster_onIdleLater = new WeakMap(), _BatchCluster_instances = new WeakSet(), _BatchCluster_onIdle = function _BatchCluster_onIdle() {
     this.vacuumProcs();
@@ -328,2 +327,3 @@ while (__classPrivateFieldGet(this, _BatchCluster_instances, "m", _BatchCluster_execNextTask).call(this)) {
     }
+    // this will be a no-op if we're ending or there are no pending tasks or ...
     __classPrivateFieldGet(this, _BatchCluster_instances, "m", _BatchCluster_maybeLaunchNewChild).call(this);
@@ -378,6 +378,4 @@ }, _BatchCluster_maybeCheckPids = function _BatchCluster_maybeCheckPids() {
         const child = await this.options.processFactory();
-        const proc = new BatchProcess_1.BatchProcess(child, this.options);
+        const proc = new BatchProcess_1.BatchProcess(child, this.options, __classPrivateFieldGet(this, _BatchCluster_onIdleLater, "f"));
         __classPrivateFieldGet(this, _BatchCluster_procs, "f").push(proc);
-        // As soon as this is ready, run onIdle
-        proc.startupPromise.then(__classPrivateFieldGet(this, _BatchCluster_onIdleLater, "f"), __classPrivateFieldGet(this, _BatchCluster_onIdleLater, "f"));
         return proc;
@@ -384,0 +382,0 @@ }

dist/BatchClusterEmitter.d.ts

@@ -67,7 +67,2 @@ /// <reference types="node" />
     /**
-     * Emitted when a task is completed, asking for more work to be scheduled, if
-     * possible.
-     */
-    idle: () => void;
-    /**
      * Emitted when this instance has ended. No child processes should remain at
@@ -74,0 +69,0 @@ * this point.

dist/BatchClusterOptions.d.ts

@@ -105,4 +105,9 @@ import { ChildProcessFactory } from "./BatchCluster";
      *
-     * Defaults to 10ms on Linux and 100ms on macOS and Windows due to slow stream
-     * handling.
+     * Defaults to 15ms on Linux and 100ms on macOS and Windows due to slower
+     * stream handling on those platforms. Your system may support a smaller value
+     * than the default: these are somewhat pessimistic.
+     *
+     * Setting this to 0 makes whatever flushes first--stdout and stderr--and will
+     * most likely result in internal errors (due to stream buffers not being able
+     * to be associated to tasks that were just settled)
      */
@@ -109,0 +114,0 @@ streamFlushMillis: number;

dist/BatchClusterOptions.js

@@ -109,4 +109,9 @@ "use strict";
          *
-         * Defaults to 10ms on Linux and 100ms on macOS and Windows due to slow stream
-         * handling.
+         * Defaults to 15ms on Linux and 100ms on macOS and Windows due to slower
+         * stream handling on those platforms. Your system may support a smaller value
+         * than the default: these are somewhat pessimistic.
+         *
+         * Setting this to 0 makes whatever flushes first--stdout and stderr--and will
+         * most likely result in internal errors (due to stream buffers not being able
+         * to be associated to tasks that were just settled)
          */
@@ -113,0 +118,0 @@ this.streamFlushMillis = Platform_1.isLinux ? 15 : 100;

dist/BatchProcess.d.ts

@@ -14,2 +14,3 @@ /// <reference types="node" />
     readonly opts: InternalBatchProcessOptions;
+    private readonly onIdle;
     readonly name: string;
@@ -19,3 +20,7 @@ readonly pid: number;
     failedTaskCount: number;
-    constructor(proc: _cp.ChildProcess, opts: InternalBatchProcessOptions);
+    /**
+     * @param onIdle to be called when internal state changes (like the current
+     * task is resolved, or the process exits)
+     */
+    constructor(proc: _cp.ChildProcess, opts: InternalBatchProcessOptions, onIdle: () => void);
     get startupPromise(): Promise<void>;
@@ -22,0 +27,0 @@ get currentTask(): Task | undefined;

dist/BatchProcess.js

@@ -25,2 +25,3 @@ "use strict";
 const Task_1 = require("./Task");
+const Timeout_1 = require("./Timeout");
 /**
@@ -30,5 +31,10 @@ * BatchProcess manages the care and feeding of a single child process.
 class BatchProcess {
-    constructor(proc, opts) {
+    /**
+     * @param onIdle to be called when internal state changes (like the current
+     * task is resolved, or the process exits)
+     */
+    constructor(proc, opts, onIdle) {
         this.proc = proc;
         this.opts = opts;
+        this.onIdle = onIdle;
         _BatchProcess_instances.add(this);
@@ -293,7 +299,9 @@ this.start = Date.now();
     const isStartupTask = task.taskId === __classPrivateFieldGet(this, _BatchProcess_startupTask, "f").taskId;
-    const timeoutMs = isStartupTask
+    const taskTimeoutMs = isStartupTask
         ? this.opts.spawnTimeoutMillis
         : this.opts.taskTimeoutMillis;
-    if (timeoutMs > 0) {
-        __classPrivateFieldSet(this, _BatchProcess_currentTaskTimeout, setTimeout(() => __classPrivateFieldGet(this, _BatchProcess_instances, "m", _BatchProcess_onTimeout).call(this, task, timeoutMs), timeoutMs), "f");
+    if (taskTimeoutMs > 0) {
+        // add the stream flush millis to the taskTimeoutMs, because that time
+        // should not be counted against the task.
+        __classPrivateFieldSet(this, _BatchProcess_currentTaskTimeout, setTimeout(() => __classPrivateFieldGet(this, _BatchProcess_instances, "m", _BatchProcess_onTimeout).call(this, task, taskTimeoutMs), taskTimeoutMs + this.opts.streamFlushMillis), "f");
     }
@@ -307,2 +315,4 @@ // CAREFUL! If you add a .catch or .finally, the pipeline can emit unhandled
         }
+        // Call _after_ we've cleared the current task:
+        this.onIdle();
     }, (err) => {
@@ -316,2 +326,4 @@ __classPrivateFieldGet(this, _BatchProcess_instances, "m", _BatchProcess_clearCurrentTask).call(this, task);
         }
+        // Call _after_ we've cleared the current task:
+        this.onIdle();
     });
@@ -351,5 +363,6 @@ try {
         try {
-            // Let's wait for streams to flush, as that may actually allow the task
-            // to complete successfully. Let's not wait forever, though.
-            await Promise.race([lastTask.promise, (0, Async_1.delay)(gracefully ? 2000 : 250)]);
+            // Let's wait for the process to complete and the streams to flush, as
+            // that may actually allow the task to complete successfully. Let's not
+            // wait forever, though.
+            await (0, Timeout_1.thenOrTimeout)(lastTask.promise, gracefully ? 2000 : 250);
         }
@@ -392,2 +405,4 @@ catch {
     }
+    // OK, we're all cleaned up: parent may be able to spin up another worker:
+    this.onIdle();
     return __classPrivateFieldGet(this, _BatchProcess_resolvedOnExit, "f");
@@ -481,3 +496,2 @@ }, _BatchProcess_awaitNotRunning = function _BatchProcess_awaitNotRunning(timeout) {
     var _a;
-    setImmediate(() => this.opts.observer.emit("idle"));
     if (task != null && task.taskId !== ((_a = __classPrivateFieldGet(this, _BatchProcess_currentTask, "f")) === null || _a === void 0 ? void 0 : _a.taskId))
@@ -484,0 +498,0 @@ return;

dist/Task.js

@@ -17,2 +17,3 @@ "use strict";
 const Async_1 = require("./Async");
+const BatchClusterOptions_1 = require("./BatchClusterOptions");
 const Deferred_1 = require("./Deferred");
@@ -136,3 +137,6 @@ let _taskId = 1;
     // wait for stderr and stdout to flush:
-    await (0, Async_1.delay)((_b = (_a = __classPrivateFieldGet(this, _Task_opts, "f")) === null || _a === void 0 ? void 0 : _a.streamFlushMillis) !== null && _b !== void 0 ? _b : 10, true);
+    const flushMs = (_b = (_a = __classPrivateFieldGet(this, _Task_opts, "f")) === null || _a === void 0 ? void 0 : _a.streamFlushMillis) !== null && _b !== void 0 ? _b : new BatchClusterOptions_1.BatchClusterOptions().streamFlushMillis;
+    if (flushMs > 0) {
+        await (0, Async_1.delay)(flushMs);
+    }
     // we're expecting this method may be called concurrently (if there are both
@@ -139,0 +143,0 @@ // pass and fail tokens found in stderr and stdout), but we only want to run

package.json

 {
   "name": "batch-cluster",
-  "version": "10.1.0",
+  "version": "10.1.1",
   "description": "Manage a cluster of child processes",
@@ -33,2 +33,13 @@ "main": "dist/BatchCluster.js",
   },
+  "release-it": {
+    "hooks": {
+      "before:init": [
+        "yarn install",
+        "yarn test"
+      ]
+    },
+    "github": {
+      "release": true
+    }
+  },
   "author": "Matthew McEachen <[e-mail]>",
@@ -41,3 +52,3 @@ "license": "MIT",
     "@types/mocha": "^9.1.0",
-    "@types/node": "^17.0.19",
+    "@types/node": "^17.0.21",
     "@typescript-eslint/eslint-plugin": "^5.12.1",
@@ -44,0 +55,0 @@ "@typescript-eslint/parser": "^5.12.1",

No alert changes

Improved metrics

Total package byte prevSize

198202

increased by2.21%

Number of package files

73

increased by4.29%

Lines of code

2807

increased by2.11%

No dependency changes