From 4a6952ffec2bc4dba2d73bd747c003d0fe59fe7c Mon Sep 17 00:00:00 2001 From: =?utf8?q?J=C3=A9r=C3=B4me=20Benoit?= Date: Wed, 24 Feb 2021 17:47:05 +0100 Subject: [PATCH] Untangle pool abstract class from worker strategy selection (#234) Co-authored-by: Shinigami --- README.md | 2 +- benchmarks/versus-external-pools/threadjs.js | 9 +-- .../threadjs/function-to-bench-worker.js | 2 +- src/pools/abstract-pool.ts | 61 +++++++++++++------ src/pools/pool-internal.ts | 26 -------- src/pools/selection-strategies.ts | 33 ++++------ 6 files changed, 63 insertions(+), 70 deletions(-) diff --git a/README.md b/README.md index ce026b14..fe74335d 100644 --- a/README.md +++ b/README.md @@ -203,7 +203,7 @@ We already have a bench folder where you can find some comparisons. Thread pools are built on top of Node.js [worker-threads](https://nodejs.org/api/worker_threads.html#worker_threads_worker_threads) module. **Cluster pools** (FixedClusterPool and DynamicClusterPool) are suggested to run I/O intensive tasks, again you can still run CPU intensive tasks into cluster pools, but performance enhancement is expected to be minimal. -Cluster pools are built on top of Node.js [cluster](https://nodejs.org/api/cluster.html) module. +Cluster pools are built on top of Node.js [cluster](https://nodejs.org/api/cluster.html) module. **Remember** that some Node.js tasks are execute by Node.js into the libuv worker pool at process level as explained [here](https://nodejs.org/en/docs/guides/dont-block-the-event-loop/#what-code-runs-on-the-worker-pool). diff --git a/benchmarks/versus-external-pools/threadjs.js b/benchmarks/versus-external-pools/threadjs.js index a70692bb..df5c087f 100644 --- a/benchmarks/versus-external-pools/threadjs.js +++ b/benchmarks/versus-external-pools/threadjs.js @@ -11,18 +11,19 @@ const data = { // Threads.js is not really a pool so we need to write few additional code const workers = [] async function poolify () { - for (let i = 0; i < size ; i++ ){ - const worker = await spawn(new Worker("./workers/threadjs/function-to-bench-worker.js")) + for (let i = 0; i < size; i++) { + const worker = await spawn( + new Worker('./workers/threadjs/function-to-bench-worker.js') + ) workers.push(worker) } } - async function run () { await poolify() const promises = [] for (let i = 0; i < iterations; i++) { - const worker = workers[(i % size)] + const worker = workers[i % size] promises.push(worker.exposedFunction(data)) } await Promise.all(promises) diff --git a/benchmarks/versus-external-pools/workers/threadjs/function-to-bench-worker.js b/benchmarks/versus-external-pools/workers/threadjs/function-to-bench-worker.js index 81f2c58e..71e48388 100644 --- a/benchmarks/versus-external-pools/workers/threadjs/function-to-bench-worker.js +++ b/benchmarks/versus-external-pools/workers/threadjs/function-to-bench-worker.js @@ -1,5 +1,5 @@ 'use strict' -const { expose } = require("threads/worker") +const { expose } = require('threads/worker') const functionToBench = require('../../functions/function-to-bench') expose({ diff --git a/src/pools/abstract-pool.ts b/src/pools/abstract-pool.ts index 7024f099..0c4b59c9 100644 --- a/src/pools/abstract-pool.ts +++ b/src/pools/abstract-pool.ts @@ -2,6 +2,7 @@ import type { MessageValue, PromiseWorkerResponseWrapper } from '../utility-types' +import { isKillBehavior, KillBehaviors } from '../worker/worker-options' import type { IPoolInternal } from './pool-internal' import { PoolEmitter } from './pool-internal' import type { WorkerChoiceStrategy } from './selection-strategies' @@ -100,6 +101,15 @@ export abstract class AbstractPool< Data = unknown, Response = unknown > implements IPoolInternal { + /** @inheritdoc */ + public readonly workers: Worker[] = [] + + /** @inheritdoc */ + public readonly tasks: Map = new Map() + + /** @inheritdoc */ + public readonly emitter: PoolEmitter + /** * The promise map. * @@ -113,15 +123,6 @@ export abstract class AbstractPool< PromiseWorkerResponseWrapper > = new Map>() - /** @inheritdoc */ - public readonly workers: Worker[] = [] - - /** @inheritdoc */ - public readonly tasks: Map = new Map() - - /** @inheritdoc */ - public readonly emitter: PoolEmitter - /** * ID of the next message. */ @@ -164,6 +165,20 @@ export abstract class AbstractPool< this.emitter = new PoolEmitter() this.workerChoiceStrategyContext = new WorkerChoiceStrategyContext( this, + () => { + const workerCreated = this.createAndSetupWorker() + this.registerWorkerMessageListener(workerCreated, message => { + const tasksInProgress = this.tasks.get(workerCreated) + if ( + isKillBehavior(KillBehaviors.HARD, message.kill) || + tasksInProgress === 0 + ) { + // Kill received from the worker, means that no new tasks are submitted to that worker for a while ( > maxInactiveTime) + void this.destroyWorker(workerCreated) + } + }) + return workerCreated + }, opts.workerChoiceStrategy ?? WorkerChoiceStrategies.ROUND_ROBIN ) } @@ -223,8 +238,12 @@ export abstract class AbstractPool< await Promise.all(this.workers.map(worker => this.destroyWorker(worker))) } - /** @inheritdoc */ - public abstract destroyWorker (worker: Worker): void | Promise + /** + * Shut down given worker. + * + * @param worker A worker within `workers`. + */ + protected abstract destroyWorker (worker: Worker): void | Promise /** * Setup hook that can be overridden by a Poolifier pool implementation @@ -306,8 +325,13 @@ export abstract class AbstractPool< message: MessageValue ): void - /** @inheritdoc */ - public abstract registerWorkerMessageListener< + /** + * Register a listener callback on a given worker. + * + * @param worker A worker. + * @param listener A message listener callback. + */ + protected abstract registerWorkerMessageListener< Message extends Data | Response > (worker: Worker, listener: (message: MessageValue) => void): void @@ -334,8 +358,12 @@ export abstract class AbstractPool< */ protected abstract afterWorkerSetup (worker: Worker): void - /** @inheritdoc */ - public createAndSetupWorker (): Worker { + /** + * Creates a new worker for this pool and sets it up completely. + * + * @returns New, completely set up worker. + */ + protected createAndSetupWorker (): Worker { const worker: Worker = this.createWorker() worker.on('error', this.opts.errorHandler ?? EMPTY_FUNCTION) @@ -359,7 +387,7 @@ export abstract class AbstractPool< * @returns The listener function to execute when a message is sent from a worker. */ protected workerListener (): (message: MessageValue) => void { - const listener: (message: MessageValue) => void = message => { + return message => { if (message.id) { const value = this.promiseMap.get(message.id) if (value) { @@ -370,6 +398,5 @@ export abstract class AbstractPool< } } } - return listener } } diff --git a/src/pools/pool-internal.ts b/src/pools/pool-internal.ts index d641fcf9..6152e3a4 100644 --- a/src/pools/pool-internal.ts +++ b/src/pools/pool-internal.ts @@ -1,5 +1,4 @@ import EventEmitter from 'events' -import type { MessageValue } from '../utility-types' import type { IWorker } from './abstract-pool' import type { IPool } from './pool' @@ -53,29 +52,4 @@ export interface IPoolInternal< * Maximum number of workers that can be created by this pool. */ readonly max?: number - - /** - * Creates a new worker for this pool and sets it up completely. - * - * @returns New, completely set up worker. - */ - createAndSetupWorker(): Worker - - /** - * Shut down given worker. - * - * @param worker A worker within `workers`. - */ - destroyWorker(worker: Worker): void | Promise - - /** - * Register a listener callback on a given worker. - * - * @param worker A worker. - * @param listener A message listener callback. - */ - registerWorkerMessageListener( - worker: Worker, - listener: (message: MessageValue) => void - ): void } diff --git a/src/pools/selection-strategies.ts b/src/pools/selection-strategies.ts index f86442b0..53b129c9 100644 --- a/src/pools/selection-strategies.ts +++ b/src/pools/selection-strategies.ts @@ -1,4 +1,3 @@ -import { isKillBehavior, KillBehaviors } from '../worker/worker-options' import type { IWorker } from './abstract-pool' import type { IPoolInternal } from './pool-internal' @@ -121,10 +120,12 @@ class DynamicPoolWorkerChoiceStrategy * Constructs a worker choice strategy for dynamical pools. * * @param pool The pool instance. + * @param createDynamicallyWorkerCallback The worker creation callback for dynamic pool. * @param workerChoiceStrategy The worker choice strategy when the pull is full. */ public constructor ( private readonly pool: IPoolInternal, + private createDynamicallyWorkerCallback: () => Worker, workerChoiceStrategy: WorkerChoiceStrategy = WorkerChoiceStrategies.ROUND_ROBIN ) { this.workerChoiceStrategy = SelectionStrategiesUtils.getWorkerChoiceStrategy( @@ -136,7 +137,7 @@ class DynamicPoolWorkerChoiceStrategy /** @inheritdoc */ public choose (): Worker { const freeWorker = SelectionStrategiesUtils.findFreeWorkerBasedOnTasks( - this.pool + this.pool.tasks ) if (freeWorker) { return freeWorker @@ -148,18 +149,7 @@ class DynamicPoolWorkerChoiceStrategy } // All workers are busy, create a new worker - const workerCreated = this.pool.createAndSetupWorker() - this.pool.registerWorkerMessageListener(workerCreated, message => { - const tasksInProgress = this.pool.tasks.get(workerCreated) - if ( - isKillBehavior(KillBehaviors.HARD, message.kill) || - tasksInProgress === 0 - ) { - // Kill received from the worker, means that no new tasks are submitted to that worker for a while ( > maxInactiveTime) - void this.pool.destroyWorker(workerCreated) - } - }) - return workerCreated + return this.createDynamicallyWorkerCallback() } } @@ -182,10 +172,12 @@ export class WorkerChoiceStrategyContext< * Worker choice strategy context constructor. * * @param pool The pool instance. + * @param createDynamicallyWorkerCallback The worker creation callback for dynamic pool. * @param workerChoiceStrategy The worker choice strategy. */ public constructor ( private readonly pool: IPoolInternal, + private createDynamicallyWorkerCallback: () => Worker, workerChoiceStrategy: WorkerChoiceStrategy = WorkerChoiceStrategies.ROUND_ROBIN ) { this.setWorkerChoiceStrategy(workerChoiceStrategy) @@ -203,6 +195,7 @@ export class WorkerChoiceStrategyContext< if (this.pool.dynamic) { return new DynamicPoolWorkerChoiceStrategy( this.pool, + this.createDynamicallyWorkerCallback, workerChoiceStrategy ) } @@ -246,15 +239,13 @@ class SelectionStrategiesUtils { * * If no free worker was found, `null` will be returned. * - * @param pool The pool instance. + * @param workerTasksMap The pool worker tasks map. * @returns A free worker if there was one, otherwise `null`. */ - public static findFreeWorkerBasedOnTasks< - Worker extends IWorker, - Data, - Response - > (pool: IPoolInternal): Worker | null { - for (const [worker, numberOfTasks] of pool.tasks) { + public static findFreeWorkerBasedOnTasks ( + workerTasksMap: Map + ): Worker | null { + for (const [worker, numberOfTasks] of workerTasksMap) { if (numberOfTasks === 0) { // A worker is free, use it return worker -- 2.34.1