X-Git-Url: https://git.piment-noir.org/?a=blobdiff_plain;f=src%2Fpools%2Fabstract-pool.ts;h=604c7af8ad32dde3959dcb75166427458999c9f7;hb=4ef0b45781a9a2a1416c4bb62d700a0c70e71145;hp=6761882589bb14bd9d205c33684ac7e7b3ab5128;hpb=440042a64e5beb9e3e97f513198f4bd2d4a896a6;p=poolifier.git diff --git a/src/pools/abstract-pool.ts b/src/pools/abstract-pool.ts index 67618825..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 { existsSync } from 'node:fs' -import { type TransferListItem } from 'node:worker_threads' +import type { TransferListItem } from 'node:worker_threads' +import { EventEmitterAsyncResource } from 'node:events' import type { MessageValue, PromiseResponseWrapper, @@ -17,14 +17,13 @@ import { max, median, min, - round, - updateMeasurementStatistics + once, + round } from '../utils' import { KillBehaviors } from '../worker/worker-options' import type { TaskFunction } from '../worker/task-functions' import { type IPool, - PoolEmitter, PoolEvents, type PoolInfo, type PoolOptions, @@ -36,6 +35,7 @@ import type { IWorker, IWorkerNode, WorkerInfo, + WorkerNodeEventDetail, WorkerType, WorkerUsage } from './worker' @@ -49,6 +49,12 @@ import { import { WorkerChoiceStrategyContext } from './selection-strategies/worker-choice-strategy-context' import { version } from './version' import { WorkerNode } from './worker-node' +import { + checkFilePath, + checkValidTasksQueueOptions, + checkValidWorkerChoiceStrategy, + updateMeasurementStatistics +} from './utils' /** * Base class that implements some shared logic for all poolifier pools. @@ -66,7 +72,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: @@ -87,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. @@ -129,8 +135,8 @@ export abstract class AbstractPool< 'Cannot start a pool from a worker with the same type as the pool' ) } + checkFilePath(this.filePath) this.checkNumberOfWorkers(this.numberOfWorkers) - this.checkFilePath(this.filePath) this.checkPoolOptions(this.opts) this.chooseWorkerNode = this.chooseWorkerNode.bind(this) @@ -138,7 +144,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, @@ -163,19 +169,6 @@ export abstract class AbstractPool< this.startTimestamp = performance.now() } - private checkFilePath (filePath: string): void { - if ( - filePath == null || - typeof filePath !== 'string' || - (typeof filePath === 'string' && filePath.trim().length === 0) - ) { - throw new Error('Please specify a file with a worker implementation') - } - if (!existsSync(filePath)) { - throw new Error(`Cannot find the worker file '${filePath}'`) - } - } - private checkNumberOfWorkers (numberOfWorkers: number): void { if (numberOfWorkers == null) { throw new Error( @@ -194,52 +187,26 @@ export abstract class AbstractPool< } } - protected checkDynamicPoolSize (min: number, max: number): void { - if (this.type === PoolTypes.dynamic) { - if (max == null) { - throw new TypeError( - 'Cannot instantiate a dynamic pool without specifying the maximum pool size' - ) - } else if (!Number.isSafeInteger(max)) { - throw new TypeError( - 'Cannot instantiate a dynamic pool with a non safe integer maximum pool size' - ) - } else if (min > max) { - throw new RangeError( - 'Cannot instantiate a dynamic pool with a maximum pool size inferior to the minimum pool size' - ) - } else if (max === 0) { - throw new RangeError( - 'Cannot instantiate a dynamic pool with a maximum pool size equal to zero' - ) - } else if (min === max) { - throw new RangeError( - 'Cannot instantiate a dynamic pool with a minimum pool size equal to the maximum pool size. Use a fixed pool instead' - ) - } - } - } - private checkPoolOptions (opts: PoolOptions): void { if (isPlainObject(opts)) { this.opts.startWorkers = opts.startWorkers ?? true + checkValidWorkerChoiceStrategy( + opts.workerChoiceStrategy as WorkerChoiceStrategy + ) this.opts.workerChoiceStrategy = opts.workerChoiceStrategy ?? WorkerChoiceStrategies.ROUND_ROBIN - this.checkValidWorkerChoiceStrategy(this.opts.workerChoiceStrategy) + this.checkValidWorkerChoiceStrategyOptions( + opts.workerChoiceStrategyOptions as WorkerChoiceStrategyOptions + ) this.opts.workerChoiceStrategyOptions = { ...DEFAULT_WORKER_CHOICE_STRATEGY_OPTIONS, ...opts.workerChoiceStrategyOptions } - this.checkValidWorkerChoiceStrategyOptions( - this.opts.workerChoiceStrategyOptions - ) this.opts.restartWorkerOnError = opts.restartWorkerOnError ?? true this.opts.enableEvents = opts.enableEvents ?? true this.opts.enableTasksQueue = opts.enableTasksQueue ?? false if (this.opts.enableTasksQueue) { - this.checkValidTasksQueueOptions( - opts.tasksQueueOptions as TasksQueueOptions - ) + checkValidTasksQueueOptions(opts.tasksQueueOptions as TasksQueueOptions) this.opts.tasksQueueOptions = this.buildTasksQueueOptions( opts.tasksQueueOptions as TasksQueueOptions ) @@ -249,26 +216,19 @@ export abstract class AbstractPool< } } - private checkValidWorkerChoiceStrategy ( - workerChoiceStrategy: WorkerChoiceStrategy - ): void { - if (!Object.values(WorkerChoiceStrategies).includes(workerChoiceStrategy)) { - throw new Error( - `Invalid worker choice strategy '${workerChoiceStrategy}'` - ) - } - } - private checkValidWorkerChoiceStrategyOptions ( workerChoiceStrategyOptions: WorkerChoiceStrategyOptions ): void { - if (!isPlainObject(workerChoiceStrategyOptions)) { + if ( + workerChoiceStrategyOptions != null && + !isPlainObject(workerChoiceStrategyOptions) + ) { throw new TypeError( 'Invalid worker choice strategy options: must be a plain object' ) } if ( - workerChoiceStrategyOptions.retries != null && + workerChoiceStrategyOptions?.retries != null && !Number.isSafeInteger(workerChoiceStrategyOptions.retries) ) { throw new TypeError( @@ -276,7 +236,7 @@ export abstract class AbstractPool< ) } if ( - workerChoiceStrategyOptions.retries != null && + workerChoiceStrategyOptions?.retries != null && workerChoiceStrategyOptions.retries < 0 ) { throw new RangeError( @@ -284,7 +244,7 @@ export abstract class AbstractPool< ) } if ( - workerChoiceStrategyOptions.weights != null && + workerChoiceStrategyOptions?.weights != null && Object.keys(workerChoiceStrategyOptions.weights).length !== this.maxSize ) { throw new Error( @@ -292,7 +252,7 @@ export abstract class AbstractPool< ) } if ( - workerChoiceStrategyOptions.measurement != null && + workerChoiceStrategyOptions?.measurement != null && !Object.values(Measurements).includes( workerChoiceStrategyOptions.measurement ) @@ -303,41 +263,10 @@ export abstract class AbstractPool< } } - private checkValidTasksQueueOptions ( - tasksQueueOptions: TasksQueueOptions - ): void { - if (tasksQueueOptions != null && !isPlainObject(tasksQueueOptions)) { - throw new TypeError('Invalid tasks queue options: must be a plain object') - } - if ( - tasksQueueOptions?.concurrency != null && - !Number.isSafeInteger(tasksQueueOptions?.concurrency) - ) { - throw new TypeError( - 'Invalid worker node tasks concurrency: must be an integer' - ) - } - if ( - tasksQueueOptions?.concurrency != null && - tasksQueueOptions?.concurrency <= 0 - ) { - throw new RangeError( - `Invalid worker node tasks concurrency: ${tasksQueueOptions?.concurrency} is a negative integer or zero` - ) - } - if ( - tasksQueueOptions?.size != null && - !Number.isSafeInteger(tasksQueueOptions?.size) - ) { - throw new TypeError( - 'Invalid worker node tasks queue size: must be an integer' - ) - } - if (tasksQueueOptions?.size != null && tasksQueueOptions?.size <= 0) { - throw new RangeError( - `Invalid worker node tasks queue size: ${tasksQueueOptions?.size} is a negative integer or zero` - ) - } + private initializeEventEmitter (): void { + this.emitter = new EventEmitterAsyncResource({ + name: `poolifier:${this.type}-${this.worker}-pool` + }) } /** @inheritDoc */ @@ -564,13 +493,10 @@ 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 ( - 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}'` ) @@ -606,7 +532,7 @@ export abstract class AbstractPool< workerChoiceStrategy: WorkerChoiceStrategy, workerChoiceStrategyOptions?: WorkerChoiceStrategyOptions ): void { - this.checkValidWorkerChoiceStrategy(workerChoiceStrategy) + checkValidWorkerChoiceStrategy(workerChoiceStrategy) this.opts.workerChoiceStrategy = workerChoiceStrategy this.workerChoiceStrategyContext.setWorkerChoiceStrategy( this.opts.workerChoiceStrategy @@ -640,6 +566,8 @@ export abstract class AbstractPool< tasksQueueOptions?: TasksQueueOptions ): void { if (this.opts.enableTasksQueue === true && !enable) { + this.unsetTaskStealing() + this.unsetTasksStealingOnBackPressure() this.flushTasksQueues() } this.opts.enableTasksQueue = enable @@ -649,21 +577,25 @@ export abstract class AbstractPool< /** @inheritDoc */ public setTasksQueueOptions (tasksQueueOptions: TasksQueueOptions): void { if (this.opts.enableTasksQueue === true) { - this.checkValidTasksQueueOptions(tasksQueueOptions) + checkValidTasksQueueOptions(tasksQueueOptions) this.opts.tasksQueueOptions = this.buildTasksQueueOptions(tasksQueueOptions) this.setTasksQueueSize(this.opts.tasksQueueOptions.size as number) + if (this.opts.tasksQueueOptions.taskStealing === true) { + this.setTaskStealing() + } else { + this.unsetTaskStealing() + } + if (this.opts.tasksQueueOptions.tasksStealingOnBackPressure === true) { + this.setTasksStealingOnBackPressure() + } else { + this.unsetTasksStealingOnBackPressure() + } } else if (this.opts.tasksQueueOptions != null) { delete this.opts.tasksQueueOptions } } - private setTasksQueueSize (size: number): void { - for (const workerNode of this.workerNodes) { - workerNode.tasksQueueBackPressureSize = size - } - } - private buildTasksQueueOptions ( tasksQueueOptions: TasksQueueOptions ): TasksQueueOptions { @@ -678,6 +610,48 @@ export abstract class AbstractPool< } } + private setTasksQueueSize (size: number): void { + for (const workerNode of this.workerNodes) { + workerNode.tasksQueueBackPressureSize = size + } + } + + private setTaskStealing (): void { + for (const [workerNodeKey] of this.workerNodes.entries()) { + this.workerNodes[workerNodeKey].addEventListener( + 'emptyqueue', + this.handleEmptyQueueEvent as EventListener + ) + } + } + + private unsetTaskStealing (): void { + for (const [workerNodeKey] of this.workerNodes.entries()) { + this.workerNodes[workerNodeKey].removeEventListener( + 'emptyqueue', + this.handleEmptyQueueEvent as EventListener + ) + } + } + + private setTasksStealingOnBackPressure (): void { + for (const [workerNodeKey] of this.workerNodes.entries()) { + this.workerNodes[workerNodeKey].addEventListener( + 'backpressure', + this.handleBackPressureEvent as EventListener + ) + } + } + + private unsetTasksStealingOnBackPressure (): void { + for (const [workerNodeKey] of this.workerNodes.entries()) { + this.workerNodes[workerNodeKey].removeEventListener( + 'backpressure', + this.handleBackPressureEvent as EventListener + ) + } + } + /** * Whether the pool is full or not. * @@ -722,63 +696,93 @@ export abstract class AbstractPool< workerNodeKey: number, message: MessageValue ): Promise { - const workerId = this.getWorkerInfo(workerNodeKey).id as number return await new Promise((resolve, reject) => { - 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}` + 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) }) } private async sendTaskFunctionOperationToWorkers ( - message: Omit, 'workerId'> + 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 ) ) { + const errorResponse = responsesReceived.find( + response => response.taskFunctionOperationStatus === false + ) reject( new Error( - `Task function operation ${ + `Task function operation '${ message.taskFunctionOperation as string - } failed on worker ${message.workerId as number}` + }' failed on worker ${ + errorResponse?.workerId as number + } with error: '${ + errorResponse?.workerError?.message as string + }'` ) ) } + this.deregisterWorkerMessageListener( + this.getWorkerNodeKeyByWorkerId(message.workerId), + taskFunctionOperationsListener + ) } - }) + } + } + for (const [workerNodeKey] of this.workerNodes.entries()) { + this.registerWorkerMessageListener( + workerNodeKey, + taskFunctionOperationsListener + ) this.sendToWorker(workerNodeKey, message) } }) @@ -800,23 +804,40 @@ export abstract class AbstractPool< /** @inheritDoc */ public async addTaskFunction ( name: string, - taskFunction: TaskFunction + fn: TaskFunction ): Promise { - this.taskFunctions.set(name, taskFunction) - return await this.sendTaskFunctionOperationToWorkers({ + if (typeof name !== 'string') { + throw new TypeError('name argument must be a string') + } + if (typeof name === 'string' && name.trim().length === 0) { + throw new TypeError('name argument must not be an empty string') + } + if (typeof fn !== 'function') { + throw new TypeError('fn argument must be a function') + } + const opResult = await this.sendTaskFunctionOperationToWorkers({ taskFunctionOperation: 'add', taskFunctionName: name, - taskFunction: taskFunction.toString() + taskFunction: fn.toString() }) + this.taskFunctions.set(name, fn) + return opResult } /** @inheritDoc */ public async removeTaskFunction (name: string): Promise { - this.taskFunctions.delete(name) - return await this.sendTaskFunctionOperationToWorkers({ + if (!this.taskFunctions.has(name)) { + throw new Error( + 'Cannot remove a task function not handled on the pool side' + ) + } + const opResult = await this.sendTaskFunctionOperationToWorkers({ taskFunctionOperation: 'remove', taskFunctionName: name }) + this.deleteTaskFunctionWorkerUsages(name) + this.taskFunctions.delete(name) + return opResult } /** @inheritDoc */ @@ -840,6 +861,12 @@ export abstract class AbstractPool< }) } + private deleteTaskFunctionWorkerUsages (name: string): void { + for (const workerNode of this.workerNodes) { + workerNode.deleteTaskFunctionWorkerUsage(name) + } + } + private shallExecuteTask (workerNodeKey: number): boolean { return ( this.tasksQueueSize(workerNodeKey) === 0 && @@ -926,6 +953,7 @@ export abstract class AbstractPool< }) ) this.emitter?.emit(PoolEvents.destroy, this.info) + this.emitter?.emitDestroy() this.started = false } @@ -933,19 +961,22 @@ 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` + }` ) ) } - }) + } + // FIXME: should be registered only once + this.registerWorkerMessageListener(workerNodeKey, killMessageListener) this.sendToWorker(workerNodeKey, { kill: true }) }) } @@ -1190,14 +1221,14 @@ 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.opts.restartWorkerOnError === true && this.started && - !this.starting + !this.starting && + this.opts.restartWorkerOnError === true ) { if (workerInfo.dynamic) { this.createAndSetupDynamicWorkerNode() @@ -1205,7 +1236,7 @@ export abstract class AbstractPool< this.createAndSetupWorkerNode() } } - if (this.opts.enableTasksQueue === true) { + if (this.started && this.opts.enableTasksQueue === true) { this.redistributeQueuedTasks(workerNodeKey) } }) @@ -1229,6 +1260,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 ) @@ -1243,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) }) @@ -1287,6 +1321,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. @@ -1295,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 + ) } } } @@ -1376,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( @@ -1387,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) { @@ -1404,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( @@ -1421,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 ) { @@ -1440,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 } } @@ -1473,8 +1542,12 @@ export abstract class AbstractPool< ) workerInfo.ready = message.ready as boolean workerInfo.taskFunctionNames = message.taskFunctionNames - if (this.emitter != null && this.ready) { - this.emitter.emit(PoolEvents.ready, this.info) + if (this.ready) { + const emitPoolReadyEventOnce = once( + () => this.emitter?.emit(PoolEvents.ready, this.info), + this + ) + emitPoolReadyEventOnce() } } @@ -1533,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 } /** @@ -1573,6 +1646,10 @@ export abstract class AbstractPool< } } + protected flagWorkerNodeAsNotReady (workerNodeKey: number): void { + this.getWorkerInfo(workerNodeKey).ready = false + } + /** @inheritDoc */ public hasWorkerNodeBackPressure (workerNodeKey: number): boolean { return (