X-Git-Url: https://git.piment-noir.org/?a=blobdiff_plain;f=src%2Fpools%2Fabstract-pool.ts;h=a0fc76c371584d3e69855fe732d720db482cc0bf;hb=8ea697554e82d9f2854acee9fddb8d2fe2d8f9c4;hp=52f9c0e3fc25aeb6a2169cf49243cfacb03d2153;hpb=8bb271861d8632c109fb0293f2a6f43bd5e3bea6;p=poolifier.git diff --git a/src/pools/abstract-pool.ts b/src/pools/abstract-pool.ts index 52f9c0e3..a0fc76c3 100644 --- a/src/pools/abstract-pool.ts +++ b/src/pools/abstract-pool.ts @@ -1,6 +1,7 @@ import { randomUUID } from 'node:crypto' import { performance } from 'node:perf_hooks' -import { type TransferListItem } from 'node:worker_threads' +import type { TransferListItem } from 'node:worker_threads' +import { EventEmitterAsyncResource } from 'node:events' import type { MessageValue, PromiseResponseWrapper, @@ -22,7 +23,6 @@ import { KillBehaviors } from '../worker/worker-options' import type { TaskFunction } from '../worker/task-functions' import { type IPool, - PoolEmitter, PoolEvents, type PoolInfo, type PoolOptions, @@ -34,6 +34,7 @@ import type { IWorker, IWorkerNode, WorkerInfo, + WorkerNodeEventDetail, WorkerType, WorkerUsage } from './worker' @@ -70,7 +71,12 @@ export abstract class AbstractPool< public readonly workerNodes: Array> = [] /** @inheritDoc */ - public readonly emitter?: PoolEmitter + public emitter?: EventEmitterAsyncResource + + /** + * Dynamic pool maximum size property placeholder. + */ + protected readonly max?: number /** * The task execution response promise map: @@ -91,11 +97,6 @@ export abstract class AbstractPool< Response > - /** - * Dynamic pool maximum size property placeholder. - */ - protected readonly max?: number - /** * The task functions added at runtime map: * - `key`: The task function name. @@ -133,8 +134,8 @@ export abstract class AbstractPool< 'Cannot start a pool from a worker with the same type as the pool' ) } - this.checkNumberOfWorkers(this.numberOfWorkers) checkFilePath(this.filePath) + this.checkNumberOfWorkers(this.numberOfWorkers) this.checkPoolOptions(this.opts) this.chooseWorkerNode = this.chooseWorkerNode.bind(this) @@ -142,7 +143,7 @@ export abstract class AbstractPool< this.enqueueTask = this.enqueueTask.bind(this) if (this.opts.enableEvents === true) { - this.emitter = new PoolEmitter() + this.initializeEventEmitter() } this.workerChoiceStrategyContext = new WorkerChoiceStrategyContext< Worker, @@ -261,6 +262,12 @@ export abstract class AbstractPool< } } + private initializeEventEmitter (): void { + this.emitter = new EventEmitterAsyncResource({ + name: `poolifier:${this.type}-${this.worker}-pool` + }) + } + /** @inheritDoc */ public get info (): PoolInfo { return { @@ -485,7 +492,7 @@ export abstract class AbstractPool< * @param message - The received message. * @throws {@link https://nodejs.org/api/errors.html#class-error} If the worker id is invalid. */ - private checkMessageWorkerId (message: MessageValue): void { + private checkMessageWorkerId (message: MessageValue): void { if (message.workerId == null) { throw new Error('Worker message received without worker id') } else if ( @@ -613,27 +620,37 @@ export abstract class AbstractPool< private setTaskStealing (): void { for (const [workerNodeKey] of this.workerNodes.entries()) { - this.workerNodes[workerNodeKey].onEmptyQueue = - this.taskStealingOnEmptyQueue.bind(this) + this.workerNodes[workerNodeKey].addEventListener( + 'emptyqueue', + this.handleEmptyQueueEvent + ) } } private unsetTaskStealing (): void { for (const [workerNodeKey] of this.workerNodes.entries()) { - delete this.workerNodes[workerNodeKey].onEmptyQueue + this.workerNodes[workerNodeKey].removeEventListener( + 'emptyqueue', + this.handleEmptyQueueEvent + ) } } private setTasksStealingOnBackPressure (): void { for (const [workerNodeKey] of this.workerNodes.entries()) { - this.workerNodes[workerNodeKey].onBackPressure = - this.tasksStealingOnBackPressure.bind(this) + this.workerNodes[workerNodeKey].addEventListener( + 'backpressure', + this.handleBackPressureEvent + ) } } private unsetTasksStealingOnBackPressure (): void { for (const [workerNodeKey] of this.workerNodes.entries()) { - delete this.workerNodes[workerNodeKey].onBackPressure + this.workerNodes[workerNodeKey].removeEventListener( + 'backpressure', + this.handleBackPressureEvent + ) } } @@ -682,28 +699,38 @@ export abstract class AbstractPool< message: MessageValue ): Promise { return await new Promise((resolve, reject) => { - const workerId = this.getWorkerInfo(workerNodeKey).id as number - this.registerWorkerMessageListener(workerNodeKey, message => { + const taskFunctionOperationListener = ( + message: MessageValue + ): void => { + this.checkMessageWorkerId(message) + const workerId = this.getWorkerInfo(workerNodeKey).id as number if ( - message.workerId === workerId && - message.taskFunctionOperationStatus === true - ) { - resolve(true) - } else if ( - message.workerId === workerId && - message.taskFunctionOperationStatus === false + message.taskFunctionOperationStatus != null && + message.workerId === workerId ) { - reject( - new Error( - `Task function operation '${ - message.taskFunctionOperation as string - }' failed on worker ${message.workerId} with error: '${ - message.workerError?.message as string - }'` + if (message.taskFunctionOperationStatus) { + resolve(true) + } else if (!message.taskFunctionOperationStatus) { + reject( + new Error( + `Task function operation '${ + message.taskFunctionOperation as string + }' failed on worker ${message.workerId} with error: '${ + message.workerError?.message as string + }'` + ) ) + } + this.deregisterWorkerMessageListener( + this.getWorkerNodeKeyByWorkerId(message.workerId), + taskFunctionOperationListener ) } - }) + } + this.registerWorkerMessageListener( + workerNodeKey, + taskFunctionOperationListener + ) this.sendToWorker(workerNodeKey, message) }) } @@ -712,20 +739,21 @@ export abstract class AbstractPool< message: MessageValue ): Promise { return await new Promise((resolve, reject) => { - const responsesReceived = new Array>() - for (const [workerNodeKey] of this.workerNodes.entries()) { - this.registerWorkerMessageListener(workerNodeKey, message => { - if (message.taskFunctionOperationStatus != null) { - responsesReceived.push(message) + const responsesReceived = new Array>() + const taskFunctionOperationsListener = ( + message: MessageValue + ): void => { + this.checkMessageWorkerId(message) + if (message.taskFunctionOperationStatus != null) { + responsesReceived.push(message) + if (responsesReceived.length === this.workerNodes.length) { if ( - responsesReceived.length === this.workerNodes.length && responsesReceived.every( message => message.taskFunctionOperationStatus === true ) ) { resolve(true) } else if ( - responsesReceived.length === this.workerNodes.length && responsesReceived.some( message => message.taskFunctionOperationStatus === false ) @@ -745,8 +773,18 @@ export abstract class AbstractPool< ) ) } + this.deregisterWorkerMessageListener( + this.getWorkerNodeKeyByWorkerId(message.workerId), + taskFunctionOperationsListener + ) } - }) + } + } + for (const [workerNodeKey] of this.workerNodes.entries()) { + this.registerWorkerMessageListener( + workerNodeKey, + taskFunctionOperationsListener + ) this.sendToWorker(workerNodeKey, message) } }) @@ -917,6 +955,7 @@ export abstract class AbstractPool< }) ) this.emitter?.emit(PoolEvents.destroy, this.info) + this.emitter?.emitDestroy() this.started = false } @@ -924,19 +963,21 @@ export abstract class AbstractPool< workerNodeKey: number ): Promise { await new Promise((resolve, reject) => { - this.registerWorkerMessageListener(workerNodeKey, message => { + const killMessageListener = (message: MessageValue): void => { + this.checkMessageWorkerId(message) if (message.kill === 'success') { resolve() } else if (message.kill === 'failure') { reject( new Error( - `Worker ${ + `Kill message handling failed on worker ${ message.workerId as number - } kill message handling failed` + }` ) ) } - }) + } + this.registerWorkerMessageListener(workerNodeKey, killMessageListener) this.sendToWorker(workerNodeKey, { kill: true }) }) } @@ -1183,8 +1224,8 @@ export abstract class AbstractPool< const workerNodeKey = this.getWorkerNodeKeyByWorker(worker) const workerInfo = this.getWorkerInfo(workerNodeKey) workerInfo.ready = false - this.workerNodes[workerNodeKey].closeChannel() this.emitter?.emit(PoolEvents.error, error) + this.workerNodes[workerNodeKey].closeChannel() if ( this.started && !this.starting && @@ -1220,6 +1261,7 @@ export abstract class AbstractPool< protected createAndSetupDynamicWorkerNode (): number { const workerNodeKey = this.createAndSetupWorkerNode() this.registerWorkerMessageListener(workerNodeKey, message => { + this.checkMessageWorkerId(message) const localWorkerNodeKey = this.getWorkerNodeKeyByWorkerId( message.workerId ) @@ -1278,6 +1320,32 @@ export abstract class AbstractPool< listener: (message: MessageValue) => void ): void + /** + * Registers once a listener callback on the worker given its worker node key. + * + * @param workerNodeKey - The worker node key. + * @param listener - The message listener callback. + */ + protected abstract registerOnceWorkerMessageListener< + Message extends Data | Response + >( + workerNodeKey: number, + listener: (message: MessageValue) => void + ): void + + /** + * Deregisters a listener callback on the worker given its worker node key. + * + * @param workerNodeKey - The worker node key. + * @param listener - The message listener callback. + */ + protected abstract deregisterWorkerMessageListener< + Message extends Data | Response + >( + workerNodeKey: number, + listener: (message: MessageValue) => void + ): void + /** * Method hooked up after a worker node has been newly created. * Can be overridden. @@ -1286,19 +1354,26 @@ export abstract class AbstractPool< */ protected afterWorkerNodeSetup (workerNodeKey: number): void { // Listen to worker messages. - this.registerWorkerMessageListener(workerNodeKey, this.workerListener()) + this.registerWorkerMessageListener( + workerNodeKey, + this.workerMessageListener.bind(this) + ) // Send the startup message to worker. this.sendStartupMessageToWorker(workerNodeKey) // Send the statistics message to worker. this.sendStatisticsMessageToWorker(workerNodeKey) if (this.opts.enableTasksQueue === true) { if (this.opts.tasksQueueOptions?.taskStealing === true) { - this.workerNodes[workerNodeKey].onEmptyQueue = - this.taskStealingOnEmptyQueue.bind(this) + this.workerNodes[workerNodeKey].addEventListener( + 'emptyqueue', + this.handleEmptyQueueEvent + ) } if (this.opts.tasksQueueOptions?.tasksStealingOnBackPressure === true) { - this.workerNodes[workerNodeKey].onBackPressure = - this.tasksStealingOnBackPressure.bind(this) + this.workerNodes[workerNodeKey].addEventListener( + 'backpressure', + this.handleBackPressureEvent + ) } } } @@ -1367,8 +1442,12 @@ export abstract class AbstractPool< } } - private taskStealingOnEmptyQueue (workerId: number): void { - const destinationWorkerNodeKey = this.getWorkerNodeKeyByWorkerId(workerId) + private readonly handleEmptyQueueEvent = ( + event: CustomEvent + ): void => { + const destinationWorkerNodeKey = this.getWorkerNodeKeyByWorkerId( + event.detail.workerId + ) const workerNodes = this.workerNodes .slice() .sort( @@ -1378,7 +1457,7 @@ export abstract class AbstractPool< const sourceWorkerNode = workerNodes.find( workerNode => workerNode.info.ready && - workerNode.info.id !== workerId && + workerNode.info.id !== event.detail.workerId && workerNode.usage.tasks.queued > 0 ) if (sourceWorkerNode != null) { @@ -1395,13 +1474,15 @@ export abstract class AbstractPool< } } - private tasksStealingOnBackPressure (workerId: number): void { + private readonly handleBackPressureEvent = ( + event: CustomEvent + ): void => { const sizeOffset = 1 if ((this.opts.tasksQueueOptions?.size as number) <= sizeOffset) { return } const sourceWorkerNode = - this.workerNodes[this.getWorkerNodeKeyByWorkerId(workerId)] + this.workerNodes[this.getWorkerNodeKeyByWorkerId(event.detail.workerId)] const workerNodes = this.workerNodes .slice() .sort( @@ -1412,7 +1493,7 @@ export abstract class AbstractPool< if ( sourceWorkerNode.usage.tasks.queued > 0 && workerNode.info.ready && - workerNode.info.id !== workerId && + workerNode.info.id !== event.detail.workerId && workerNode.usage.tasks.queued < (this.opts.tasksQueueOptions?.size as number) - sizeOffset ) { @@ -1431,25 +1512,21 @@ export abstract class AbstractPool< } /** - * This method is the listener registered for each worker message. - * - * @returns The listener function to execute when a message is received from a worker. + * This method is the message listener registered on each worker. */ - protected workerListener (): (message: MessageValue) => void { - return message => { - this.checkMessageWorkerId(message) - if (message.ready != null && message.taskFunctionNames != null) { - // Worker ready response received from worker - this.handleWorkerReadyResponse(message) - } else if (message.taskId != null) { - // Task execution response received from worker - this.handleTaskExecutionResponse(message) - } else if (message.taskFunctionNames != null) { - // Task function names message received from worker - this.getWorkerInfo( - this.getWorkerNodeKeyByWorkerId(message.workerId) - ).taskFunctionNames = message.taskFunctionNames - } + protected workerMessageListener (message: MessageValue): void { + this.checkMessageWorkerId(message) + if (message.ready != null && message.taskFunctionNames != null) { + // Worker ready response received from worker + this.handleWorkerReadyResponse(message) + } else if (message.taskId != null) { + // Task execution response received from worker + this.handleTaskExecutionResponse(message) + } else if (message.taskFunctionNames != null) { + // Task function names message received from worker + this.getWorkerInfo( + this.getWorkerNodeKeyByWorkerId(message.workerId) + ).taskFunctionNames = message.taskFunctionNames } }