feat(DynamicWorkerPool): idleTimeout for workers

Author: hellivan

src/worker-pool/dynamic-worker-pool.spec.ts

@@ -22,12 +22,12 @@ describe('DynamicWorkerPool', () => {
     });
 
     test('size of new pool should be 0', () => {
-        const testPool = new DynamicWorkerPool<string, string, TestWorker>(workerFactory, 0, 5);
+        const testPool = new DynamicWorkerPool<string, string, TestWorker>(workerFactory, 0, 5, 0);
         expect(testPool.size).toEqual(0);
     });
 
     test('execute task should aquire new workers and execute tasks on them', async () => {
-        const testPool = new DynamicWorkerPool<string, string, TestWorker>(workerFactory, 0, 2);
+        const testPool = new DynamicWorkerPool<string, string, TestWorker>(workerFactory, 0, 2, 0);
 
         const workerDisposeSpy = jest.spyOn(testWorkerInstance, 'dispose');
         const createWorkerSpy = jest.spyOn(workerFactory, 'createWorker');
@@ -47,7 +47,7 @@ describe('DynamicWorkerPool', () => {
     });
 
     test('worker pool should hould minSize workers', async () => {
-        const testPool = new DynamicWorkerPool<string, string, TestWorker>(workerFactory, 3, 5);
+        const testPool = new DynamicWorkerPool<string, string, TestWorker>(workerFactory, 3, 5, 0);
 
         const workerDisposeSpy = jest.spyOn(testWorkerInstance, 'dispose');
         const createWorkerSpy = jest.spyOn(workerFactory, 'createWorker');
@@ -72,7 +72,7 @@ describe('DynamicWorkerPool', () => {
     });
 
     test('executeTask should reject if worker pool max size is reached', async () => {
-        const testPool = new DynamicWorkerPool<string, string, TestWorker>(workerFactory, 0, 1);
+        const testPool = new DynamicWorkerPool<string, string, TestWorker>(workerFactory, 0, 1, 0);
 
         const workerDisposeSpy = jest.spyOn(testWorkerInstance, 'dispose');
         const createWorkerSpy = jest.spyOn(workerFactory, 'createWorker');
@@ -91,7 +91,7 @@ describe('DynamicWorkerPool', () => {
     });
 
     test('worker pool should re-use existing workers in pool', async () => {
-        const testPool = new DynamicWorkerPool<string, string, TestWorker>(workerFactory, 1, 1);
+        const testPool = new DynamicWorkerPool<string, string, TestWorker>(workerFactory, 1, 1, 0);
 
         const workerDisposeSpy = jest.spyOn(testWorkerInstance, 'dispose');
         const createWorkerSpy = jest.spyOn(workerFactory, 'createWorker');
@@ -110,7 +110,7 @@ describe('DynamicWorkerPool', () => {
     });
 
     test('worker pool should emit an error on error$ if worker disposal fails', async () => {
-        const testPool = new DynamicWorkerPool<string, string, TestWorker>(workerFactory, 0, 1);
+        const testPool = new DynamicWorkerPool<string, string, TestWorker>(workerFactory, 0, 1, 0);
 
         jest.spyOn(testWorkerInstance, 'dispose').mockImplementationOnce(async () => {
             throw new Error('DisposeError');
@@ -126,13 +126,13 @@ describe('DynamicWorkerPool', () => {
     });
 
     test('executeTask on a stopped DynamicWorkerPool should reject', async () => {
-        const testPool = new DynamicWorkerPool<string, string, TestWorker>(workerFactory, 2, 5);
+        const testPool = new DynamicWorkerPool<string, string, TestWorker>(workerFactory, 2, 5, 0);
         testPool.stop();
         await expect(testPool.executeTask('1')).rejects.toThrowError('Cannot aquire worker from stopped worker pool!');
     });
 
     test('stop should terminate all current workers', async () => {
-        const testPool = new DynamicWorkerPool<string, string, TestWorker>(workerFactory, 2, 5);
+        const testPool = new DynamicWorkerPool<string, string, TestWorker>(workerFactory, 2, 5, 0);
         const workerDisposeSpy = jest.spyOn(testWorkerInstance, 'dispose');
 
         await Promise.all([
@@ -149,7 +149,7 @@ describe('DynamicWorkerPool', () => {
     });
 
     test('busyWorkers$ should emit number of workers in use', async () => {
-        const testPool = new DynamicWorkerPool<string, string, TestWorker>(workerFactory, 2, 5);
+        const testPool = new DynamicWorkerPool<string, string, TestWorker>(workerFactory, 2, 5, 0);
 
         const busyWorkersHistoryPromise = testPool.busyWorkers$.pipe(bufferCount(13), first()).toPromise();
 
@@ -167,7 +167,7 @@ describe('DynamicWorkerPool', () => {
     });
 
     test('idleWorkers$ should emit number of idle workers', async () => {
-        const testPool = new DynamicWorkerPool<string, string, TestWorker>(workerFactory, 2, 5);
+        const testPool = new DynamicWorkerPool<string, string, TestWorker>(workerFactory, 2, 5, 0);
 
         const idleWorkersHistoryPromise = testPool.idleWorkers$.pipe(bufferCount(5), first()).toPromise();
 
@@ -185,7 +185,7 @@ describe('DynamicWorkerPool', () => {
     });
 
     test('availableWorkers$ should emit number of available workers', async () => {
-        const testPool = new DynamicWorkerPool<string, string, TestWorker>(workerFactory, 2, 5);
+        const testPool = new DynamicWorkerPool<string, string, TestWorker>(workerFactory, 2, 5, 0);
 
         const availableWorkersHistoryPromise = testPool.availableWorkers$.pipe(bufferCount(13), first()).toPromise();
 
@@ -203,7 +203,7 @@ describe('DynamicWorkerPool', () => {
     });
 
     test('availableWorkers$ should emit 0 for a stopped pool', async () => {
-        const testPool = new DynamicWorkerPool<string, string, TestWorker>(workerFactory, 2, 5);
+        const testPool = new DynamicWorkerPool<string, string, TestWorker>(workerFactory, 2, 5, 0);
 
         const availableWorkersHistoryPromise = testPool.availableWorkers$.pipe(bufferCount(4), first()).toPromise();
 
@@ -215,13 +215,13 @@ describe('DynamicWorkerPool', () => {
     });
 
     test('availableWorkers should return 0 for a stopped pool', async () => {
-        const testPool = new DynamicWorkerPool<string, string, TestWorker>(workerFactory, 2, 5);
+        const testPool = new DynamicWorkerPool<string, string, TestWorker>(workerFactory, 2, 5, 0);
         await testPool.stop();
         expect(testPool.availableWorkers).toEqual(0);
     });
 
     test('availableWorkers should return number of possible free workers', async () => {
-        const testPool = new DynamicWorkerPool<string, string, TestWorker>(workerFactory, 2, 5);
+        const testPool = new DynamicWorkerPool<string, string, TestWorker>(workerFactory, 2, 5, 0);
         expect(testPool.availableWorkers).toEqual(5);
         const taskResultPromise = testPool.executeTask('1');
         expect(testPool.availableWorkers).toEqual(4);
@@ -230,13 +230,13 @@ describe('DynamicWorkerPool', () => {
     });
 
     test('idleWorkers should return 0 for a stopped pool', async () => {
-        const testPool = new DynamicWorkerPool<string, string, TestWorker>(workerFactory, 2, 5);
+        const testPool = new DynamicWorkerPool<string, string, TestWorker>(workerFactory, 2, 5, 0);
         await testPool.stop();
         expect(testPool.idleWorkers).toEqual(0);
     });
 
     test('idleWorkers should return the number of created workers in idle state', async () => {
-        const testPool = new DynamicWorkerPool<string, string, TestWorker>(workerFactory, 2, 5);
+        const testPool = new DynamicWorkerPool<string, string, TestWorker>(workerFactory, 2, 5, 0);
         expect(testPool.idleWorkers).toEqual(0);
         const taskResultPromise = testPool.executeTask('1');
         expect(testPool.idleWorkers).toEqual(0);
@@ -245,17 +245,44 @@ describe('DynamicWorkerPool', () => {
     });
 
     test('busyWorkers should return 0 for a stopped pool', async () => {
-        const testPool = new DynamicWorkerPool<string, string, TestWorker>(workerFactory, 2, 5);
+        const testPool = new DynamicWorkerPool<string, string, TestWorker>(workerFactory, 2, 5, 0);
         await testPool.stop();
         expect(testPool.busyWorkers).toEqual(0);
     });
 
     test('busyWorkers should return the number of working workers', async () => {
-        const testPool = new DynamicWorkerPool<string, string, TestWorker>(workerFactory, 2, 5);
+        const testPool = new DynamicWorkerPool<string, string, TestWorker>(workerFactory, 2, 5, 0);
         expect(testPool.busyWorkers).toEqual(0);
         const taskResultPromise = testPool.executeTask('1');
         expect(testPool.busyWorkers).toEqual(1);
         await taskResultPromise;
         expect(testPool.busyWorkers).toEqual(0);
     });
+
+    test('DynamicWorkerPool with idleTimeout should keep idle workers for specified time', async () => {
+        const testPool = new DynamicWorkerPool<string, string, TestWorker>(workerFactory, 2, 5, 100);
+        const workerDisposeSpy = jest.spyOn(testWorkerInstance, 'dispose');
+
+        await Promise.all([
+            testPool.executeTask('1'),
+            testPool.executeTask('2'),
+            testPool.executeTask('3'),
+            testPool.executeTask('4'),
+            testPool.executeTask('5')
+        ]);
+
+        await new Promise((resolve) => setTimeout(resolve, 10));
+
+        expect(workerDisposeSpy).toHaveBeenCalledTimes(0);
+        expect(testPool.size).toEqual(5);
+
+        await new Promise((resolve) => setTimeout(resolve, 100));
+
+        expect(testPool.size).toEqual(2);
+        expect(workerDisposeSpy).toHaveBeenCalledTimes(3);
+
+        await testPool.stop();
+        expect(testPool.size).toEqual(0);
+        expect(workerDisposeSpy).toHaveBeenCalledTimes(5);
+    });
 });

src/worker-pool/dynamic-worker-pool.ts

@@ -4,6 +4,7 @@ import { distinctUntilChanged, filter, first } from 'rxjs/operators';
 import { ObservableQueue } from '../data-structures';
 import { WorkerFactory } from '../worker-factory';
 import { AbstractWorkerPool, AbstractWorkerPoolWorker } from './abstract-worker-pool';
+import { IdleWorkerDescription } from './idle-worker-description';
 
 export interface DynamicWorkerPoolWorker<TTask, TResult> extends AbstractWorkerPoolWorker<TTask, TResult> {
     dispose(): Promise<void>;
@@ -27,13 +28,14 @@ export class DynamicWorkerPool<
     private readonly availableWorkersSubject: BehaviorSubject<number>;
     private readonly errorSubject = new Subject<Error>();
     private readonly busyWorkersQueue = new ObservableQueue<TWorker>();
-    private readonly idleWorkersQueue = new ObservableQueue<TWorker>();
+    private readonly idleWorkersQueue = new ObservableQueue<IdleWorkerDescription<TWorker>>();
     private stopped = false;
 
     constructor(
         private readonly workerFactory: WorkerFactory<TWorker>,
         private readonly minSize: number,
-        private readonly maxSize: number
+        private readonly maxSize: number,
+        private readonly workersIdleTimeout: number
     ) {
         super();
 
@@ -66,26 +68,30 @@ export class DynamicWorkerPool<
 
         await this.busyWorkers$
             .pipe(
-                filter((x) => x === 0),
+                filter((count) => count === 0),
                 first()
             )
             .toPromise();
 
-        const existingWorkers = this.idleWorkersQueue.clear();
+        const existingWorkerDescriptions = this.idleWorkersQueue.clear();
+        existingWorkerDescriptions.forEach((d) => d.cancelIdleTimeout());
+
+        const existingWorkers = existingWorkerDescriptions.map(({ worker }) => worker);
         this.updatePoolSize();
         this.updateAvailableWorkers();
-        await Promise.all(existingWorkers.map((w) => this.safeDispose(w)));
+        await Promise.all(existingWorkers.map((worker) => this.safeDispose(worker)));
     }
 
     protected aquireWorker(): TWorker {
         if (this.stopped) {
             throw new Error('Cannot aquire worker from stopped worker pool!');
         } else if (this.idleWorkersQueue.size > 0) {
-            const worker = this.idleWorkersQueue.dequeue();
-            this.busyWorkersQueue.enqueue(worker);
+            const workerDescription = this.idleWorkersQueue.dequeue();
+            workerDescription.cancelIdleTimeout();
+            this.busyWorkersQueue.enqueue(workerDescription.worker);
             // NOTE: poolsize did not change since we reused an existing worker -> no need to update poolsize
             this.updateAvailableWorkers();
-            return worker;
+            return workerDescription.worker;
         } else if (this.size < this.maxSize) {
             const worker = this.workerFactory.createWorker();
             this.busyWorkersQueue.enqueue(worker);
@@ -98,16 +104,34 @@ export class DynamicWorkerPool<
     }
 
     protected releaseWorker(worker: TWorker): void {
-        const currentWorkersCount = this.size;
-        this.busyWorkersQueue.drop(worker);
-        if (currentWorkersCount > this.minSize) {
-            // NOTE: safeDispose will never reject -> leave promise uncatched is safe
-            this.safeDispose(worker);
-            this.updatePoolSize();
+        if (this.workersIdleTimeout < 1) {
+            // Workers do not have an idle timeout -> remove or enqueue them immediately
+            const currentWorkersCount = this.size;
+            this.busyWorkersQueue.drop(worker);
+
+            if (currentWorkersCount > this.minSize) {
+                // NOTE: safeDispose will never reject -> leave promise uncatched is safe
+                this.safeDispose(worker);
+                this.updatePoolSize();
+            } else {
+                this.idleWorkersQueue.enqueue(new IdleWorkerDescription<TWorker>(worker));
+                // NOTE: poolsize did not change since we standbyed an existing worker -> no need to update poolsize
+            }
         } else {
-            this.idleWorkersQueue.enqueue(worker);
-            // NOTE: poolsize did not change since we standbyed an existing worker -> no need to update poolsize
+            // Workers do have an idle timeout -> enqueue them as idle worker and remove them if possible after timeout
+            this.busyWorkersQueue.drop(worker);
+            const idleWorkerDescription = new IdleWorkerDescription<TWorker>(worker, this.workersIdleTimeout);
+            idleWorkerDescription.addTeardownLogic(() => {
+                if (this.size > this.minSize) {
+                    this.idleWorkersQueue.drop(idleWorkerDescription);
+                    this.safeDispose(worker);
+                    this.updatePoolSize();
+                    this.updateAvailableWorkers();
+                }
+            });
+            this.idleWorkersQueue.enqueue(idleWorkerDescription);
         }
+
         this.updateAvailableWorkers();
     }
 

src/worker-pool/idle-worker-description.spec.ts

@@ -0,0 +1,47 @@
+import { IdleWorkerDescription } from './idle-worker-description';
+
+describe('IdleWorkerDescription', () => {
+    jest.useFakeTimers();
+
+    beforeEach(() => {
+        jest.clearAllMocks();
+    });
+
+    test('creating a new IdleWorkerDescription should start idleTimout if specified', () => {
+        const idleWorkerDescription = new IdleWorkerDescription<unknown>(undefined, 3000);
+        // required to avoid unused local variable warning
+        expect(idleWorkerDescription).toBeDefined();
+        expect(setTimeout).toHaveBeenCalledWith(expect.any(Function), 3000);
+    });
+
+    test('cancelIdleTimeout should clear idle-teardown-logic timeout', () => {
+        const idleWorkerDescription = new IdleWorkerDescription<unknown>(undefined, 3000);
+        idleWorkerDescription.cancelIdleTimeout();
+        expect(clearTimeout).toHaveBeenCalledTimes(1);
+    });
+
+    test('cancelIdleTimeout should not have any effect if no timeout was specified', () => {
+        const idleWorkerDescription = new IdleWorkerDescription<unknown>(undefined);
+        idleWorkerDescription.cancelIdleTimeout();
+        expect(clearTimeout).toHaveBeenCalledTimes(0);
+    });
+
+    test('creating a new IdleWorkerDescription should not start idleTimout if not specified', () => {
+        const idleWorkerDescription = new IdleWorkerDescription<unknown>(undefined);
+        // required to avoid unused local variable warning
+        expect(idleWorkerDescription).toBeDefined();
+        expect(setTimeout).toHaveBeenCalledTimes(0);
+    });
+
+    test('IdleWorkerDescription should execute teardown logics if idle timeout times out', () => {
+        const idleWorkerDescription = new IdleWorkerDescription<unknown>(undefined, 3000);
+        const tearDownLogics = [jest.fn(), jest.fn()];
+        for (const tearDownLogic of tearDownLogics) {
+            idleWorkerDescription.addTeardownLogic(tearDownLogic);
+        }
+        jest.runAllTimers();
+        for (const tearDownLogic of tearDownLogics) {
+            expect(tearDownLogic).toHaveBeenCalledTimes(1);
+        }
+    });
+});

src/worker-pool/idle-worker-description.ts

@@ -0,0 +1,26 @@
+export type TeardownLogic = () => void;
+
+// TODO: test
+export class IdleWorkerDescription<TWorker> {
+    private readonly timeoutTimer: NodeJS.Timeout | undefined;
+    private readonly teardownLogics: Array<TeardownLogic> = [];
+
+    constructor(public readonly worker: TWorker, idleTimeout?: number) {
+        if (idleTimeout != null) {
+            this.timeoutTimer = setTimeout(
+                () => this.teardownLogics.forEach((teardownLogic) => teardownLogic()),
+                idleTimeout
+            );
+        }
+    }
+
+    public addTeardownLogic(teardownLogic: () => void): void {
+        this.teardownLogics.push(teardownLogic);
+    }
+
+    public cancelIdleTimeout(): void {
+        if (this.timeoutTimer != null) {
+            clearTimeout(this.timeoutTimer);
+        }
+    }
+}