worker-swarmer - npm Package Compare versions

Comparing version 1.0.0 to 2.0.0

dist/esm/handler.d.ts

@@ -7,2 +7,4 @@ import type { Scheduler } from './scheduler';
     private _id;
+    private working;
+    private retireRequested;
     constructor(scheduler: Scheduler<I, O>, worker: Worker, _id: string);
@@ -13,2 +15,3 @@ handle(req: IQueueRequest<I, O>['details']): Promise<TInterruptableReq<O>>;
     handleRequest({ details, report }: IQueueRequest<I, O>): void;
+    retire(): boolean;
 }

dist/esm/handler.js

@@ -6,2 +6,4 @@ export class Handler {
         this._id = _id;
+        this.working = false;
+        this.retireRequested = false;
     }
@@ -20,6 +22,12 @@ handle(req) {
     handleRequest({ details, report }) {
+        this.working = true;
         const msgHandler = ({ data }) => {
             this.worker.removeEventListener("message", msgHandler);
             report(data);
-            this.scheduler.handleQueuedRequest(this);
+            if (this.retireRequested) {
+                this.destroy();
+            }
+            else {
+                this.scheduler.handleQueuedRequest(this);
+            }
         };
@@ -29,2 +37,10 @@ this.worker.addEventListener("message", msgHandler);
     }
+    retire() {
+        if (this.working) {
+            this.retireRequested = true;
+            return false;
+        }
+        this.destroy();
+        return true;
+    }
 }

dist/esm/index.d.ts

@@ -1,3 +0,20 @@
-import { TWorkerMaker } from './types';
-export { REQ_EARLY_TERMINATION_TOKEN } from './constants';
-export declare function swarm<R1 = any, R2 = any>(workerMaker: TWorkerMaker, maxCount: number): (req: R1) => Promise<import("./types").TInterruptableReq<R2>>;
+import { Scheduler } from "./scheduler";
+import { TWorkerMaker } from "./types";
+export { REQ_EARLY_TERMINATION_TOKEN } from "./constants";
+export { TWorkerMaker } from "./types";
+type SchedulerInstance<I, O> = InstanceType<typeof Scheduler<I, O>>;
+type ISwarmed<I, O> = SchedulerInstance<I, O>["doRequest"] & {
+    terminate: SchedulerInstance<I, O>["kill"];
+};
+type ISwarmedWithResourceSaving<I, O> = ISwarmed<I, O> & {
+    disableRecycling: () => void;
+    enableRecycling: () => void;
+};
+export declare function swarm<I, O>(workerMaker: TWorkerMaker, args: {
+    maxCount?: number;
+    recyclable?: true;
+}): ISwarmedWithResourceSaving<I, O>;
+export declare function swarm<I, O>(workerMaker: TWorkerMaker, args: {
+    maxCount?: number;
+    recyclable: false;
+}): ISwarmed<I, O>;

dist/esm/index.js

@@ -1,6 +0,17 @@
-import { Scheduler } from './scheduler';
-export { REQ_EARLY_TERMINATION_TOKEN } from './constants';
-export function swarm(workerMaker, maxCount) {
-    const scheduler = new Scheduler(workerMaker, maxCount);
-    return scheduler.doRequest;
+import { Scheduler } from "./scheduler";
+export { REQ_EARLY_TERMINATION_TOKEN } from "./constants";
+export function swarm(workerMaker, { maxCount = 3, recyclable = true, } = {}) {
+    const scheduler = new Scheduler(workerMaker, maxCount, recyclable);
+    if (recyclable) {
+        const swarmed = (req) => scheduler.doRequest(req);
+        swarmed["disableRecycling"] = () => scheduler.pauseResourceSaving();
+        swarmed["enableRecycling"] = () => scheduler.restartResourceSaving();
+        swarmed["terminate"] = () => scheduler.kill();
+        return swarmed;
+    }
+    else {
+        const swarmed = (req) => scheduler.doRequest(req);
+        swarmed["terminate"] = () => scheduler.kill();
+        return swarmed;
+    }
 }

dist/esm/scheduler.d.ts

@@ -1,3 +0,3 @@
-import { Handler } from './handler';
-import { TInterruptableReq, TWorkerMaker } from './types';
+import { Handler } from "./handler";
+import { TInterruptableReq, TWorkerMaker } from "./types";
 export declare class Scheduler<I, O> {
@@ -12,3 +12,5 @@ private workerMaker;
     private ownHandlerIds;
-    constructor(workerMaker: TWorkerMaker, maxTotal?: number);
+    private dischargeChecker;
+    get idleCount(): number;
+    constructor(workerMaker: TWorkerMaker, maxTotal?: number, recycleIdleWorkers?: boolean);
     private init;
@@ -20,2 +22,5 @@ doRequest: (req: I) => Promise<TInterruptableReq<O>>;
     kill(): void;
+    discharge(numToDischarge: number): void;
+    pauseResourceSaving(): void;
+    restartResourceSaving(): void;
 }

dist/esm/scheduler.js

@@ -1,6 +0,10 @@
-import { nanoid } from 'nanoid';
-import { REQ_EARLY_TERMINATION_TOKEN } from './constants';
-import { Handler } from './handler';
+import { nanoid } from "nanoid";
+import { REQ_EARLY_TERMINATION_TOKEN } from "./constants";
+import { Handler } from "./handler";
+import { DischargeChecker } from "./discharge-checker";
 export class Scheduler {
-    constructor(workerMaker, maxTotal = 3) {
+    get idleCount() {
+        return this.idleHandlers.length;
+    }
+    constructor(workerMaker, maxTotal = 3, recycleIdleWorkers = true) {
         this.workerMaker = workerMaker;
@@ -28,2 +32,6 @@ this.maxTotal = maxTotal;
         this.init();
+        if (recycleIdleWorkers) {
+            this.dischargeChecker = new DischargeChecker(this);
+            this.dischargeChecker.start();
+        }
     }
@@ -73,6 +81,6 @@ init() {
     kill() {
-        this.handlers.forEach(h => {
+        this.handlers.forEach((h) => {
             h.destroy();
         });
-        this.requestQueue.forEach(r => {
+        this.requestQueue.forEach((r) => {
             r.report(REQ_EARLY_TERMINATION_TOKEN);
@@ -82,2 +90,21 @@ });
     }
+    discharge(numToDischarge) {
+        while (this.idleCount > 0 && numToDischarge--) {
+            const handlerIdToDischarge = this.idleHandlers.pop();
+            const handlerToDischarge = this.handlers.get(handlerIdToDischarge);
+            if (handlerToDischarge.retire()) {
+                this.handlers.delete(handlerIdToDischarge);
+                this.ownHandlerIds.delete(handlerIdToDischarge);
+                this.spawned--;
+            }
+        }
+    }
+    pauseResourceSaving() {
+        var _a;
+        (_a = this.dischargeChecker) === null || _a === void 0 ? void 0 : _a.stop();
+    }
+    restartResourceSaving() {
+        var _a;
+        (_a = this.dischargeChecker) === null || _a === void 0 ? void 0 : _a.start();
+    }
 }

dist/esm/types.d.ts

@@ -8,5 +8,1 @@ import { REQ_EARLY_TERMINATION_TOKEN } from './constants';
 export type TWorkerMaker = () => Worker;
-export interface IReport<T> {
-    reqTs: number;
-    data: T;
-}

dist/umd/index.js

@@ -1,2 +0,2 @@
-!function(e,t){"object"==typeof exports&&"object"==typeof module?module.exports=t():"function"==typeof define&&define.amd?define([],t):"object"==typeof exports?exports.WorkerSwarmer=t():e.WorkerSwarmer=t()}(globalThis,(()=>(()=>{"use strict";var e={601:(e,t)=>{Object.defineProperty(t,"__esModule",{value:!0}),t.REQ_EARLY_TERMINATION_TOKEN=void 0,t.REQ_EARLY_TERMINATION_TOKEN=Symbol("request processing is interrupted")},102:(e,t)=>{Object.defineProperty(t,"__esModule",{value:!0}),t.Handler=void 0,t.Handler=class{constructor(e,t,r){this.scheduler=e,this.worker=t,this._id=r}handle(e){return new Promise((t=>{this.handleRequest({details:e,report:t})}))}get id(){return this._id}destroy(){this.worker.terminate()}handleRequest({details:e,report:t}){const r=({data:e})=>{this.worker.removeEventListener("message",r),t(e),this.scheduler.handleQueuedRequest(this)};this.worker.addEventListener("message",r),this.worker.postMessage(e)}}},399:(e,t,r)=>{Object.defineProperty(t,"__esModule",{value:!0}),t.Scheduler=void 0;const s=r(296),n=r(601),i=r(102);t.Scheduler=class{constructor(e,t=3){this.workerMaker=e,this.maxTotal=t,this.busyHandlers=new Set,this.idleHandlers=[],this.handlers=new Map,this.spawned=0,this.requestQueue=[],this.ownHandlerIds=new Set,this.doRequest=e=>{const t=this.getHandler();let r;const s=Object.assign({},e);return t?(this.busyHandlers.add(t.id),r=t.handle(s)):r=this.arrangeRequest(s),r},this.init()}init(){this.busyHandlers=new Set,this.idleHandlers=[],this.handlers=new Map,this.spawned=0,this.requestQueue=[],this.ownHandlerIds=new Set}getHandler(){let e=this.idleHandlers.pop();return!e&&this.spawned<this.maxTotal&&(e=(0,s.nanoid)(),this.handlers.set(e,new i.Handler(this,this.workerMaker(),e)),this.ownHandlerIds.add(e),this.spawned++),this.handlers.get(e)}arrangeRequest(e){let t;const r=new Promise((e=>{t=e}));return this.requestQueue.push({details:e,report:t}),r}handleQueuedRequest(e){if(this.ownHandlerIds.has(e.id)){const t=this.requestQueue.shift();t?(e.handleRequest(t),this.busyHandlers.add(e.id)):(this.busyHandlers.delete(e.id),this.idleHandlers.push(e.id))}}kill(){this.handlers.forEach((e=>{e.destroy()})),this.requestQueue.forEach((e=>{e.report(n.REQ_EARLY_TERMINATION_TOKEN)})),this.init()}}},296:(e,t,r)=>{r.r(t),r.d(t,{customAlphabet:()=>o,customRandom:()=>i,nanoid:()=>d,random:()=>n,urlAlphabet:()=>s});const s="useandom-26T198340PX75pxJACKVERYMINDBUSHWOLF_GQZbfghjklqvwyzrict";let n=e=>crypto.getRandomValues(new Uint8Array(e)),i=(e,t,r)=>{let s=(2<<Math.log(e.length-1)/Math.LN2)-1,n=-~(1.6*s*t/e.length);return(i=t)=>{let o="";for(;;){let t=r(n),d=n;for(;d--;)if(o+=e[t[d]&s]||"",o.length===i)return o}}},o=(e,t=21)=>i(e,t,n),d=(e=21)=>crypto.getRandomValues(new Uint8Array(e)).reduce(((e,t)=>e+((t&=63)<36?t.toString(36):t<62?(t-26).toString(36).toUpperCase():t>62?"-":"_")),"")}},t={};function r(s){var n=t[s];if(void 0!==n)return n.exports;var i=t[s]={exports:{}};return e[s](i,i.exports,r),i.exports}r.d=(e,t)=>{for(var s in t)r.o(t,s)&&!r.o(e,s)&&Object.defineProperty(e,s,{enumerable:!0,get:t[s]})},r.o=(e,t)=>Object.prototype.hasOwnProperty.call(e,t),r.r=e=>{"undefined"!=typeof Symbol&&Symbol.toStringTag&&Object.defineProperty(e,Symbol.toStringTag,{value:"Module"}),Object.defineProperty(e,"__esModule",{value:!0})};var s={};return(()=>{var e=s;Object.defineProperty(e,"__esModule",{value:!0}),e.swarm=e.REQ_EARLY_TERMINATION_TOKEN=void 0;const t=r(399);var n=r(601);Object.defineProperty(e,"REQ_EARLY_TERMINATION_TOKEN",{enumerable:!0,get:function(){return n.REQ_EARLY_TERMINATION_TOKEN}}),e.swarm=function(e,r){return new t.Scheduler(e,r).doRequest}})(),s})()));
+!function(e,t){"object"==typeof exports&&"object"==typeof module?module.exports=t():"function"==typeof define&&define.amd?define([],t):"object"==typeof exports?exports.WorkerSwarmer=t():e.WorkerSwarmer=t()}(globalThis,(()=>(()=>{"use strict";var e={601:(e,t)=>{Object.defineProperty(t,"__esModule",{value:!0}),t.REQ_EARLY_TERMINATION_TOKEN=void 0,t.REQ_EARLY_TERMINATION_TOKEN=Symbol("request processing is interrupted")},900:(e,t)=>{Object.defineProperty(t,"__esModule",{value:!0}),t.DischargeChecker=void 0,t.DischargeChecker=class{constructor(e){this.scheduler=e,this.checkerCycleStartTs=0,this.accumulatedIdleCount=0,this.totalChecksOfCycle=0}start(){this.doCheck(0)}doCheck(e){if(e-this.checkerCycleStartTs>=168){const t=Math.ceil(this.scheduler.idleCount-this.accumulatedIdleCount/this.totalChecksOfCycle);t>0&&this.scheduler.discharge(t),this.checkerCycleStartTs=e,this.accumulatedIdleCount=0,this.totalChecksOfCycle=0}else e>0&&(this.accumulatedIdleCount+=this.scheduler.idleCount,this.totalChecksOfCycle++);this._handle=requestAnimationFrame((e=>this.doCheck(e)))}stop(){cancelAnimationFrame(this._handle),this.checkerCycleStartTs=0,this.accumulatedIdleCount=0,this.totalChecksOfCycle=0}}},102:(e,t)=>{Object.defineProperty(t,"__esModule",{value:!0}),t.Handler=void 0,t.Handler=class{constructor(e,t,s){this.scheduler=e,this.worker=t,this._id=s,this.working=!1,this.retireRequested=!1}handle(e){return new Promise((t=>{this.handleRequest({details:e,report:t})}))}get id(){return this._id}destroy(){this.worker.terminate()}handleRequest({details:e,report:t}){this.working=!0;const s=({data:e})=>{this.worker.removeEventListener("message",s),t(e),this.retireRequested?this.destroy():this.scheduler.handleQueuedRequest(this)};this.worker.addEventListener("message",s),this.worker.postMessage(e)}retire(){return this.working?(this.retireRequested=!0,!1):(this.destroy(),!0)}}},399:(e,t,s)=>{Object.defineProperty(t,"__esModule",{value:!0}),t.Scheduler=void 0;const r=s(296),i=s(601),n=s(102),a=s(900);t.Scheduler=class{get idleCount(){return this.idleHandlers.length}constructor(e,t=3,s=!0){this.workerMaker=e,this.maxTotal=t,this.busyHandlers=new Set,this.idleHandlers=[],this.handlers=new Map,this.spawned=0,this.requestQueue=[],this.ownHandlerIds=new Set,this.doRequest=e=>{const t=this.getHandler();let s;const r=Object.assign({},e);return t?(this.busyHandlers.add(t.id),s=t.handle(r)):s=this.arrangeRequest(r),s},this.init(),s&&(this.dischargeChecker=new a.DischargeChecker(this),this.dischargeChecker.start())}init(){this.busyHandlers=new Set,this.idleHandlers=[],this.handlers=new Map,this.spawned=0,this.requestQueue=[],this.ownHandlerIds=new Set}getHandler(){let e=this.idleHandlers.pop();return!e&&this.spawned<this.maxTotal&&(e=(0,r.nanoid)(),this.handlers.set(e,new n.Handler(this,this.workerMaker(),e)),this.ownHandlerIds.add(e),this.spawned++),this.handlers.get(e)}arrangeRequest(e){let t;const s=new Promise((e=>{t=e}));return this.requestQueue.push({details:e,report:t}),s}handleQueuedRequest(e){if(this.ownHandlerIds.has(e.id)){const t=this.requestQueue.shift();t?(e.handleRequest(t),this.busyHandlers.add(e.id)):(this.busyHandlers.delete(e.id),this.idleHandlers.push(e.id))}}kill(){this.handlers.forEach((e=>{e.destroy()})),this.requestQueue.forEach((e=>{e.report(i.REQ_EARLY_TERMINATION_TOKEN)})),this.init()}discharge(e){for(;this.idleCount>0&&e--;){const e=this.idleHandlers.pop();this.handlers.get(e).retire()&&(this.handlers.delete(e),this.ownHandlerIds.delete(e),this.spawned--)}}pauseResourceSaving(){var e;null===(e=this.dischargeChecker)||void 0===e||e.stop()}restartResourceSaving(){var e;null===(e=this.dischargeChecker)||void 0===e||e.start()}}},296:(e,t,s)=>{s.r(t),s.d(t,{customAlphabet:()=>a,customRandom:()=>n,nanoid:()=>d,random:()=>i,urlAlphabet:()=>r});const r="useandom-26T198340PX75pxJACKVERYMINDBUSHWOLF_GQZbfghjklqvwyzrict";let i=e=>crypto.getRandomValues(new Uint8Array(e)),n=(e,t,s)=>{let r=(2<<Math.log(e.length-1)/Math.LN2)-1,i=-~(1.6*r*t/e.length);return(n=t)=>{let a="";for(;;){let t=s(i),d=i;for(;d--;)if(a+=e[t[d]&r]||"",a.length===n)return a}}},a=(e,t=21)=>n(e,t,i),d=(e=21)=>crypto.getRandomValues(new Uint8Array(e)).reduce(((e,t)=>e+((t&=63)<36?t.toString(36):t<62?(t-26).toString(36).toUpperCase():t>62?"-":"_")),"")}},t={};function s(r){var i=t[r];if(void 0!==i)return i.exports;var n=t[r]={exports:{}};return e[r](n,n.exports,s),n.exports}s.d=(e,t)=>{for(var r in t)s.o(t,r)&&!s.o(e,r)&&Object.defineProperty(e,r,{enumerable:!0,get:t[r]})},s.o=(e,t)=>Object.prototype.hasOwnProperty.call(e,t),s.r=e=>{"undefined"!=typeof Symbol&&Symbol.toStringTag&&Object.defineProperty(e,Symbol.toStringTag,{value:"Module"}),Object.defineProperty(e,"__esModule",{value:!0})};var r={};return(()=>{var e=r;Object.defineProperty(e,"__esModule",{value:!0}),e.swarm=e.REQ_EARLY_TERMINATION_TOKEN=void 0;const t=s(399);var i=s(601);Object.defineProperty(e,"REQ_EARLY_TERMINATION_TOKEN",{enumerable:!0,get:function(){return i.REQ_EARLY_TERMINATION_TOKEN}}),e.swarm=function(e,{maxCount:s=3,recyclable:r=!0}={}){const i=new t.Scheduler(e,s,r);if(r){const e=e=>i.doRequest(e);return e.disableRecycling=()=>i.pauseResourceSaving(),e.enableRecycling=()=>i.restartResourceSaving(),e.terminate=()=>i.kill(),e}{const e=e=>i.doRequest(e);return e.terminate=()=>i.kill(),e}}})(),r})()));
 //# sourceMappingURL=index.js.map

package.json

 {
   "name": "worker-swarmer",
-  "version": "1.0.0",
+  "version": "2.0.0",
   "description": "A lib that allows spawning a worker implementation into many. This is essentially a worker pooler with simple API surface.",
@@ -5,0 +5,0 @@ "main": "./dist/umd/index.js",

README.md

 # worker-swarmer
 A lib that allows spawning a worker implementation into many. This is essentially a worker pooler with simple API surface.
+
+## Usage
+
+### 1. Basic
+
+Wrap your worker implementation in a worker factory function and pass this function to the `swarm`. Along with this factory function, you can also specify the maximum total (default to 3) of web workers can spawn.
+
+```ts
+import { swarm, REQ_EARLY_TERMINATION_TOKEN } from "worker-swarmer";
+
+// webpack@4 or below + worker-loader
+import SuperWorker from "path/to/my/superworker.ts"; 
+const workerMaker = () => new SuperWorker();
+
+/**
+ * for webpack@5+ and other bundler tools supporting native web worker instantiation, simply do
+ * 
+ * const workerMaker = () => new Worker(new URL("path/to/my/suprerworker.ts", import.meta.url));
+ */
+
+interface ISuperWorkerInput {
+  taskName: string;
+  timestamp: number;
+}
+
+interface ISuperWorkerOutput {
+  taskResult: any;
+  timestamp: number;
+}
+
+const maxCount = 5; // at most, 5 web workers of SuperWorker will exist
+const swarmedSuperWorker = swarm<ISuperWorkerInput, ISuperWorkerOutput>(
+  () => new SuperWorker(),
+  { maxCount },
+);
+
+swarmedSuperworker({
+  taskName: "meaning-of-life",
+}).then((output) => {
+  if (output === REQ_EARLY_TERMINATION_TOKEN) {
+    throw new Error("task processing was interrupted.");
+  }
+
+  const {
+    taskResult,
+  } = output;
+
+  if (taskResult === 42) {
+    console.log("found the meaning of life");
+
+    return true;
+  }
+
+  return false;
+});
+
+// let's destroy all spawned super workers.
+// This might cause some running ones throw `REQ_EARLY_TERMINATION_TOKEN`.
+// If you want to avoid this, do the termination in the Promise's
+// lifecycle (e.g., `then`, `catch` or `finally`).
+swarmedSuperWorker.terminate();
+```
+
+### 2. Control idle web worker's eligibility for recycling.
+
+By default, a swarmed instance will try to recycle some web workers after they are idle for certain period of time. This can potentially reserve some resource consumption. But, it's inteneded to avoid the overhead of spinning up a web worker. The recycling can be disabled completely or paused for as long as needed.
+
+```ts
+// disable the recycling completely
+import { swarm } from "worker-swarmer";
+
+...
+
+const swarmed = swarm(() => new Worker(
+  new URL("path/to/worker.ts", import.meta.url)),
+  { recyclable: false /* no recycling at all */ },
+);
+
+
+
+// pause&resume recycling behavior
+const swarmedWithRecycling = swarm(() => new Worker(
+  new URL("path/to/worker.ts", import.meta.url)),
+);
+
+...
+
+swarmedWithRecycling.disableRecycling(); // Let's pause recycling idle workers
+
+...
+
+// mission-critical work has been finished, let's re-enable recycling.
+swarmedWithRecycling.enableRecycling();
+```

No alert changes

Improved metrics

Total package byte prevSize

37882

increased by58.63%

Number of package files

19

increased by11.76%

Lines of code

302

increased by72.57%

Number of lines in readme file

96

increased by3100%

No dependency changes