X-Git-Url: https://git.piment-noir.org/?a=blobdiff_plain;f=src%2Fpools%2Fabstract-pool.ts;h=604c7af8ad32dde3959dcb75166427458999c9f7;hb=16ce27f3d5276072eb9f4633dd70c1e46ba6a868;hp=c56beaa6d801843c043f091963cea385b72eccee;hpb=5b7d00b4d0f276c521db9f2145104a64f441d2fb;p=poolifier.git diff --git a/src/pools/abstract-pool.ts b/src/pools/abstract-pool.ts index c56beaa6..604c7af8 100644 --- a/src/pools/abstract-pool.ts +++ b/src/pools/abstract-pool.ts @@ -1,7 +1,7 @@ import { randomUUID } from 'node:crypto' import { performance } from 'node:perf_hooks' import type { TransferListItem } from 'node:worker_threads' -import { type EventEmitter, EventEmitterAsyncResource } from 'node:events' +import { EventEmitterAsyncResource } from 'node:events' import type { MessageValue, PromiseResponseWrapper, @@ -17,6 +17,7 @@ import { max, median, min, + once, round } from '../utils' import { KillBehaviors } from '../worker/worker-options' @@ -34,6 +35,7 @@ import type { IWorker, IWorkerNode, WorkerInfo, + WorkerNodeEventDetail, WorkerType, WorkerUsage } from './worker' @@ -70,7 +72,12 @@ export abstract class AbstractPool< public readonly workerNodes: Array> = [] /** @inheritDoc */ - public emitter?: EventEmitter | EventEmitterAsyncResource + public emitter?: EventEmitterAsyncResource + + /** + * Dynamic pool maximum size property placeholder. + */ + protected readonly max?: number /** * The task execution response promise map: @@ -91,11 +98,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. @@ -494,10 +496,7 @@ export abstract class AbstractPool< private checkMessageWorkerId (message: MessageValue): void { if (message.workerId == null) { throw new Error('Worker message received without worker id') - } else if ( - message.workerId != null && - this.getWorkerNodeKeyByWorkerId(message.workerId) === -1 - ) { + } else if (this.getWorkerNodeKeyByWorkerId(message.workerId) === -1) { throw new Error( `Worker message received from unknown worker '${message.workerId}'` ) @@ -619,27 +618,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 as EventListener + ) } } private unsetTaskStealing (): void { for (const [workerNodeKey] of this.workerNodes.entries()) { - delete this.workerNodes[workerNodeKey].onEmptyQueue + this.workerNodes[workerNodeKey].removeEventListener( + 'emptyqueue', + this.handleEmptyQueueEvent as EventListener + ) } } 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 as EventListener + ) } } private unsetTasksStealingOnBackPressure (): void { for (const [workerNodeKey] of this.workerNodes.entries()) { - delete this.workerNodes[workerNodeKey].onBackPressure + this.workerNodes[workerNodeKey].removeEventListener( + 'backpressure', + this.handleBackPressureEvent as EventListener + ) } } @@ -944,9 +953,7 @@ export abstract class AbstractPool< }) ) this.emitter?.emit(PoolEvents.destroy, this.info) - if (this.emitter instanceof EventEmitterAsyncResource) { - this.emitter?.emitDestroy() - } + this.emitter?.emitDestroy() this.started = false } @@ -968,6 +975,7 @@ export abstract class AbstractPool< ) } } + // FIXME: should be registered only once this.registerWorkerMessageListener(workerNodeKey, killMessageListener) this.sendToWorker(workerNodeKey, { kill: true }) }) @@ -1213,10 +1221,10 @@ export abstract class AbstractPool< worker.on('error', this.opts.errorHandler ?? EMPTY_FUNCTION) worker.on('error', error => { const workerNodeKey = this.getWorkerNodeKeyByWorker(worker) + this.flagWorkerNodeAsNotReady(workerNodeKey) 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 && @@ -1267,6 +1275,8 @@ export abstract class AbstractPool< workerUsage.tasks.executing === 0 && this.tasksQueueSize(localWorkerNodeKey) === 0))) ) { + // Flag the worker node as not ready immediately + this.flagWorkerNodeAsNotReady(localWorkerNodeKey) this.destroyWorkerNode(localWorkerNodeKey).catch(error => { this.emitter?.emit(PoolEvents.error, error) }) @@ -1345,19 +1355,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 as EventListener + ) } if (this.opts.tasksQueueOptions?.tasksStealingOnBackPressure === true) { - this.workerNodes[workerNodeKey].onBackPressure = - this.tasksStealingOnBackPressure.bind(this) + this.workerNodes[workerNodeKey].addEventListener( + 'backpressure', + this.handleBackPressureEvent as EventListener + ) } } } @@ -1426,8 +1443,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( @@ -1437,7 +1458,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) { @@ -1454,13 +1475,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( @@ -1471,7 +1494,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 ) { @@ -1490,25 +1513,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 } } @@ -1524,7 +1543,11 @@ export abstract class AbstractPool< workerInfo.ready = message.ready as boolean workerInfo.taskFunctionNames = message.taskFunctionNames if (this.ready) { - this.emitter?.emit(PoolEvents.ready, this.info) + const emitPoolReadyEventOnce = once( + () => this.emitter?.emit(PoolEvents.ready, this.info), + this + ) + emitPoolReadyEventOnce() } } @@ -1583,7 +1606,7 @@ export abstract class AbstractPool< * @returns The worker information. */ protected getWorkerInfo (workerNodeKey: number): WorkerInfo { - return this.workerNodes[workerNodeKey].info + return this.workerNodes[workerNodeKey]?.info } /** @@ -1623,6 +1646,10 @@ export abstract class AbstractPool< } } + protected flagWorkerNodeAsNotReady (workerNodeKey: number): void { + this.getWorkerInfo(workerNodeKey).ready = false + } + /** @inheritDoc */ public hasWorkerNodeBackPressure (workerNodeKey: number): boolean { return (