node-apparatus - npm Package Compare versions

Comparing version 0.1.9 to 0.1.10

dist/src/index.d.ts

@@ -6,2 +6,4 @@ import { InjectableConstructor } from "./injectable-constructor/injectable-constructor.js";
 import { SortedMap } from "./sorted-map/sorted-map.js";
-export { kWayMerge, StatefulRecipient, StatefulProxyManager, InjectableConstructor, SortedMap };
+import { SequentialInvocationQueue } from "./sequential-invocation-queue/sequential-invocation-queue";
+import { SpinWaitLock } from "./spin-wait-lock/spin-wait-lock";
+export { kWayMerge, StatefulRecipient, StatefulProxyManager, InjectableConstructor, SortedMap, SpinWaitLock, SequentialInvocationQueue };

dist/src/index.js

 "use strict";
 Object.defineProperty(exports, "__esModule", { value: true });
-exports.SortedMap = exports.InjectableConstructor = exports.StatefulProxyManager = exports.StatefulRecipient = exports.kWayMerge = void 0;
+exports.SequentialInvocationQueue = exports.SpinWaitLock = exports.SortedMap = exports.InjectableConstructor = exports.StatefulProxyManager = exports.StatefulRecipient = exports.kWayMerge = void 0;
 const injectable_constructor_js_1 = require("./injectable-constructor/injectable-constructor.js");
@@ -14,2 +14,6 @@ Object.defineProperty(exports, "InjectableConstructor", { enumerable: true, get: function () { return injectable_constructor_js_1.InjectableConstructor; } });
 Object.defineProperty(exports, "SortedMap", { enumerable: true, get: function () { return sorted_map_js_1.SortedMap; } });
+const sequential_invocation_queue_1 = require("./sequential-invocation-queue/sequential-invocation-queue");
+Object.defineProperty(exports, "SequentialInvocationQueue", { enumerable: true, get: function () { return sequential_invocation_queue_1.SequentialInvocationQueue; } });
+const spin_wait_lock_1 = require("./spin-wait-lock/spin-wait-lock");
+Object.defineProperty(exports, "SpinWaitLock", { enumerable: true, get: function () { return spin_wait_lock_1.SpinWaitLock; } });
 //# sourceMappingURL=index.js.map

dist/src/stateful-threads/stateful-proxy-manager.d.ts

@@ -12,3 +12,5 @@ /**
     private readonly workerFilePath;
-    private workersExistingWork;
+    private readonly invocationQueueSizePerWorker;
+    private readonly spinWaitTime;
+    private workerSequentialQueue;
     private readonly workers;
@@ -22,4 +24,6 @@ WorkerCount: number;
      * @param {string} workerFilePath - The path to the worker script file which has exported instance of StatefulRecipient.
+     * @param {number} [invocationQueueSizePerWorker=Number.MAX_SAFE_INTEGER] - The maximum number of method invocations to queue per worker.
+     * @param {number} [spinWaitTime=100] - The time to wait between attempts to acquire the lock in milliseconds.
      */
-    constructor(workerCount: number, workerFilePath: string);
+    constructor(workerCount: number, workerFilePath: string, invocationQueueSizePerWorker?: number, spinWaitTime?: number);
     /**
@@ -40,5 +44,5 @@ * Initializes the worker threads, must be called before invoking any methods.
      */
-    invokeMethod<T>(methodName: string, methodArguments?: any[], workerIndex?: number, methodInvocationId?: number): Promise<T>;
+    invokeMethod<T>(methodName: string, methodArguments?: any[], workerIndex?: number, methodInvocationId?: number): Promise<T | undefined>;
     private invokeRemoteMethod;
     [Symbol.asyncDispose](): Promise<void>;
 }

dist/src/stateful-threads/stateful-proxy-manager.js

@@ -7,2 +7,4 @@ "use strict";
 const node_worker_threads_1 = require("node:worker_threads");
+const sequential_invocation_queue_js_1 = require("../sequential-invocation-queue/sequential-invocation-queue.js");
+const spin_wait_lock_js_1 = require("../spin-wait-lock/spin-wait-lock.js");
 /**
@@ -22,7 +24,11 @@ * Manages a pool of stateful worker threads to handle method invocations.
      * @param {string} workerFilePath - The path to the worker script file which has exported instance of StatefulRecipient.
+     * @param {number} [invocationQueueSizePerWorker=Number.MAX_SAFE_INTEGER] - The maximum number of method invocations to queue per worker.
+     * @param {number} [spinWaitTime=100] - The time to wait between attempts to acquire the lock in milliseconds.
      */
-    constructor(workerCount, workerFilePath) {
+    constructor(workerCount, workerFilePath, invocationQueueSizePerWorker = Number.MAX_SAFE_INTEGER, spinWaitTime = 100) {
         this.workerCount = workerCount;
         this.workerFilePath = workerFilePath;
-        this.workersExistingWork = new Array();
+        this.invocationQueueSizePerWorker = invocationQueueSizePerWorker;
+        this.spinWaitTime = spinWaitTime;
+        this.workerSequentialQueue = new Map();
         this.workers = new Array();
@@ -36,7 +42,8 @@ }
         for (let index = 0; index < parsedWorkerCount; index++) {
-            //TODO:Have Handshake with threads that they are ready to accept commands.
+            //TODO:Have Handshake with threads that they are ready to accept commands Ping-Pong.
             this.workers.push(new node_worker_threads_1.Worker(this.workerFilePath, { workerData: null }));
+            this.workerSequentialQueue.set(index, new sequential_invocation_queue_js_1.SequentialInvocationQueue(new spin_wait_lock_js_1.SpinWaitLock(), this.invokeRemoteMethod.bind(this), this.invocationQueueSizePerWorker));
         }
         if (this.workers.length === 0) {
-            const module = await import(this.workerFilePath);
+            const module = await import(`file://${this.workerFilePath}`);
             this.selfWorker = module.default;
@@ -48,3 +55,2 @@ this.WorkerCount = 1;
         }
-        this.workersExistingWork = new Array(this.workers.length);
     }
@@ -68,10 +74,18 @@ /**
         else {
-            return this.invokeRemoteMethod(methodName, methodArguments, workerIdx, methodInvocationId);
+            const q = this.workerSequentialQueue.get(workerIdx);
+            const result = await q.invoke([methodName, methodArguments, workerIdx, methodInvocationId], this.spinWaitTime);
+            if (result.state === "success") {
+                return result.result;
+            }
+            else if (result.state === "error:queue-full") {
+                throw new Error(`Worker ${workerIdx} reported:${result.state}`);
+            }
+            else {
+                return undefined;
+            }
         }
     }
-    async invokeRemoteMethod(methodName, methodArguments, workerIndex, methodInvocationId = Number.NaN) {
-        if (this.workersExistingWork[workerIndex] !== undefined) {
-            await this.workersExistingWork[workerIndex];
-        }
-        this.workersExistingWork[workerIndex] = new Promise((resolve, reject) => {
+    async invokeRemoteMethod(methodParameters) {
+        const [methodName, methodArguments, workerIndex, methodInvocationId = Number.NaN] = methodParameters;
+        return await new Promise((resolve, reject) => {
             const worker = this.workers[workerIndex];
@@ -85,6 +99,2 @@ const workerErrorHandler = (error) => {
                 const returnValue = (0, i_proxy_method_js_1.deserialize)(message);
-                if (Number.isNaN(returnValue.workerId) === false) {
-                    const workerIdx = Math.max(0, Math.min(returnValue.workerId, this.workers.length - 1));
-                    this.workersExistingWork[workerIdx] = undefined;
-                }
                 if (returnValue.error !== undefined) {
@@ -101,3 +111,2 @@ reject(new Error(returnValue.error));
         });
-        return this.workersExistingWork[workerIndex];
     }
@@ -108,3 +117,5 @@ async [Symbol.asyncDispose]() {
         }
-        await Promise.allSettled(this.workersExistingWork);
+        for (const [workerIndex, q] of this.workerSequentialQueue.entries()) {
+            await q[Symbol.asyncDispose]();
+        }
         for (const worker of this.workers) {
@@ -114,3 +125,3 @@ worker.postMessage((0, i_proxy_method_js_1.serialize)(i_proxy_method_js_1.DisposeMethodPayload));
         this.workers.length = 0;
-        this.workersExistingWork.length = 0;
+        this.workerSequentialQueue.clear();
     }
@@ -117,0 +128,0 @@ }

package.json

 {
   "name": "node-apparatus",
-  "version": "0.1.9",
+  "version": "0.1.10",
   "description": "A mix of common components needed for awesome node experience",
@@ -12,4 +12,3 @@ "main": "dist/src/index.js",
     "build-test": "npm run clean-build && npx tsc && npm run copy-files",
-    "start": "npm run build && node -r dotenv/config dist/src/server.js ./.env",
-    "test": "npm run build-test && node --test ./dist/test",
+    "test": "npm run build-test && node --test ./dist/test/**/**.js",
     "docker-local": "docker build -f ./Dockerfile -t notification-hub-svc:local .",
@@ -21,3 +20,3 @@ "coverage": "c8 --config ./coverage-config.json npm run test",
     "type": "git",
-    "url": "git+https://github.com/LRagji/node-toolset.git"
+    "url": "git+https://github.com/LRagji/node-apparatus.git"
   },
@@ -27,3 +26,3 @@ "author": "Laukik",
   "bugs": {
-    "url": "https://github.com/LRagji/node-toolset/issues"
+    "url": "https://github.com/LRagji/node-apparatus/issues"
   },
@@ -39,3 +38,3 @@ "keywords": [
   ],
-  "homepage": "https://github.com/LRagji/node-toolset#readme",
+  "homepage": "https://github.com/LRagji/node-apparatus#readme",
   "devDependencies": {
@@ -42,0 +41,0 @@ "@types/node": "^22.7.8",

README.md

@@ -27,6 +27,8 @@ # node-apparatus
 
-1. [k-way-merge](https://github.com/LRagji/node-toolset/blob/main/src/k-way-merge/readme.md)
-2. [stateful-threads](https://github.com/LRagji/node-toolset/blob/main/src/stateful-threads/readme.md)
-3. [injectable-constructor](https://github.com/LRagji/node-toolset/blob/main/src/injectable-constructor/readme.md)
-3. [sorted-map](https://github.com/LRagji/node-toolset/blob/main/src/sorted-map/readme.md)
+1. [k-way-merge](https://github.com/LRagji/node-apparatus/blob/main/src/k-way-merge/readme.md)
+2. [stateful-threads](https://github.com/LRagji/node-apparatus/blob/main/src/stateful-threads/readme.md)
+3. [injectable-constructor](https://github.com/LRagji/node-apparatus/blob/main/src/injectable-constructor/readme.md)
+4. [sorted-map](https://github.com/LRagji/node-apparatus/blob/main/src/sorted-map/readme.md)
+5. [sequential-invocation-queue](https://github.com/LRagji/node-apparatus/blob/main/src/sequential-invocation-queue/readme.md)
+6. [spin-wait-lock](https://github.com/LRagji/node-apparatus/blob/main/src/spin-wait-lock/readme.md)
 
@@ -33,0 +35,0 @@ Many more to come

New alerts

License Policy Violation

License

This package is not allowed per your license policy. Review the package's license to ensure compliance.

Found 1 instance in 1 package

Fixed alerts

License Policy Violation

License

This package is not allowed per your license policy. Review the package's license to ensure compliance.

Found 1 instance in 1 package

Improved metrics

Total package byte prevSize

64422

increased by2.7%

Number of package files

30

increased by15.38%

Lines of code

768

increased by49.71%

Number of lines in readme file

47

increased by4.44%

No dependency changes