X-Git-Url: https://git.piment-noir.org/?a=blobdiff_plain;f=src%2Fpools%2Fabstract-pool.ts;h=da0eb335b26f41092fb57560d255b5b0b624ad6e;hb=d8163b08686d39aca2a2c9896176ceede981672e;hp=52f9c0e3fc25aeb6a2169cf49243cfacb03d2153;hpb=bfc75ccaf49d915d2b7e73c92360787b3245321a;p=poolifier.git diff --git a/src/pools/abstract-pool.ts b/src/pools/abstract-pool.ts index 52f9c0e3..da0eb335 100644 --- a/src/pools/abstract-pool.ts +++ b/src/pools/abstract-pool.ts @@ -1,6 +1,8 @@ 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 { AsyncResource } from 'node:async_hooks' import type { MessageValue, PromiseResponseWrapper, @@ -11,18 +13,19 @@ import { DEFAULT_WORKER_CHOICE_STRATEGY_OPTIONS, EMPTY_FUNCTION, average, + exponentialDelay, isKillBehavior, isPlainObject, max, median, min, - round + round, + sleep } from '../utils' import { KillBehaviors } from '../worker/worker-options' import type { TaskFunction } from '../worker/task-functions' import { type IPool, - PoolEmitter, PoolEvents, type PoolInfo, type PoolOptions, @@ -33,7 +36,9 @@ import { import type { IWorker, IWorkerNode, + TaskStatistics, WorkerInfo, + WorkerNodeEventDetail, WorkerType, WorkerUsage } from './worker' @@ -70,7 +75,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 +101,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. @@ -111,6 +116,14 @@ export abstract class AbstractPool< * Whether the pool is starting or not. */ private starting: boolean + /** + * Whether the pool is destroying or not. + */ + private destroying: boolean + /** + * Whether the pool ready event has been emitted or not. + */ + private readyEventEmitted: boolean /** * The start timestamp of the pool. */ @@ -133,8 +146,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 +155,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, @@ -160,6 +173,8 @@ export abstract class AbstractPool< this.started = false this.starting = false + this.destroying = false + this.readyEventEmitted = false if (this.opts.startWorkers === true) { this.start() } @@ -261,6 +276,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,13 +506,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}'` ) @@ -577,11 +595,13 @@ export abstract class AbstractPool< this.buildTasksQueueOptions(tasksQueueOptions) this.setTasksQueueSize(this.opts.tasksQueueOptions.size as number) if (this.opts.tasksQueueOptions.taskStealing === true) { + this.unsetTaskStealing() this.setTaskStealing() } else { this.unsetTaskStealing() } if (this.opts.tasksQueueOptions.tasksStealingOnBackPressure === true) { + this.unsetTasksStealingOnBackPressure() this.setTasksStealingOnBackPressure() } else { this.unsetTasksStealingOnBackPressure() @@ -613,27 +633,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].on( + 'idleWorkerNode', + this.handleIdleWorkerNodeEvent + ) } } private unsetTaskStealing (): void { for (const [workerNodeKey] of this.workerNodes.entries()) { - delete this.workerNodes[workerNodeKey].onEmptyQueue + this.workerNodes[workerNodeKey].off( + 'idleWorkerNode', + this.handleIdleWorkerNodeEvent + ) } } private setTasksStealingOnBackPressure (): void { for (const [workerNodeKey] of this.workerNodes.entries()) { - this.workerNodes[workerNodeKey].onBackPressure = - this.tasksStealingOnBackPressure.bind(this) + this.workerNodes[workerNodeKey].on( + 'backPressure', + this.handleBackPressureEvent + ) } } private unsetTasksStealingOnBackPressure (): void { for (const [workerNodeKey] of this.workerNodes.entries()) { - delete this.workerNodes[workerNodeKey].onBackPressure + this.workerNodes[workerNodeKey].off( + 'backPressure', + this.handleBackPressureEvent + ) } } @@ -682,28 +712,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 +752,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 +786,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) } }) @@ -850,6 +901,10 @@ export abstract class AbstractPool< reject(new Error('Cannot execute a task on not started pool')) return } + if (this.destroying) { + reject(new Error('Cannot execute a task on destroying pool')) + return + } if (name != null && typeof name !== 'string') { reject(new TypeError('name argument must be a string')) return @@ -879,7 +934,13 @@ export abstract class AbstractPool< this.promiseResponseMap.set(task.taskId as string, { resolve, reject, - workerNodeKey + workerNodeKey, + ...(this.emitter != null && { + asyncResource: new AsyncResource('poolifier:task', { + triggerAsyncId: this.emitter.asyncId, + requireManualDestroy: true + }) + }) }) if ( this.opts.enableTasksQueue === false || @@ -895,6 +956,15 @@ export abstract class AbstractPool< /** @inheritdoc */ public start (): void { + if (this.started) { + throw new Error('Cannot start an already started pool') + } + if (this.starting) { + throw new Error('Cannot start an already starting pool') + } + if (this.destroying) { + throw new Error('Cannot start a destroying pool') + } this.starting = true while ( this.workerNodes.reduce( @@ -911,12 +981,26 @@ export abstract class AbstractPool< /** @inheritDoc */ public async destroy (): Promise { + if (!this.started) { + throw new Error('Cannot destroy an already destroyed pool') + } + if (this.starting) { + throw new Error('Cannot destroy an starting pool') + } + if (this.destroying) { + throw new Error('Cannot destroy an already destroying pool') + } + this.destroying = true await Promise.all( - this.workerNodes.map(async (_, workerNodeKey) => { + this.workerNodes.map(async (_workerNode, workerNodeKey) => { await this.destroyWorkerNode(workerNodeKey) }) ) this.emitter?.emit(PoolEvents.destroy, this.info) + this.emitter?.emitDestroy() + this.emitter?.removeAllListeners() + this.readyEventEmitted = false + this.destroying = false this.started = false } @@ -924,19 +1008,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 }) }) } @@ -1181,13 +1268,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.started && !this.starting && + !this.destroying && this.opts.restartWorkerOnError === true ) { if (workerInfo.dynamic) { @@ -1220,6 +1308,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 ) @@ -1234,6 +1323,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) }) @@ -1278,6 +1369,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 +1403,26 @@ export abstract class AbstractPool< */ protected afterWorkerNodeSetup (workerNodeKey: number): void { // Listen to worker messages. - this.registerWorkerMessageListener(workerNodeKey, this.workerListener()) + this.registerWorkerMessageListener( + workerNodeKey, + this.workerMessageListener + ) // 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].on( + 'idleWorkerNode', + this.handleIdleWorkerNodeEvent + ) } if (this.opts.tasksQueueOptions?.tasksStealingOnBackPressure === true) { - this.workerNodes[workerNodeKey].onBackPressure = - this.tasksStealingOnBackPressure.bind(this) + this.workerNodes[workerNodeKey].on( + 'backPressure', + this.handleBackPressureEvent + ) } } } @@ -1328,6 +1452,9 @@ export abstract class AbstractPool< } private redistributeQueuedTasks (workerNodeKey: number): void { + if (this.workerNodes.length <= 1) { + return + } while (this.tasksQueueSize(workerNodeKey) > 0) { const destinationWorkerNodeKey = this.workerNodes.reduce( (minWorkerNodeKey, workerNode, workerNodeKey, workerNodes) => { @@ -1367,8 +1494,123 @@ export abstract class AbstractPool< } } - private taskStealingOnEmptyQueue (workerId: number): void { - const destinationWorkerNodeKey = this.getWorkerNodeKeyByWorkerId(workerId) + private updateTaskSequentiallyStolenStatisticsWorkerUsage ( + workerNodeKey: number + ): void { + const workerNode = this.workerNodes[workerNodeKey] + if (workerNode?.usage != null) { + ++workerNode.usage.tasks.sequentiallyStolen + } + } + + private updateTaskSequentiallyStolenStatisticsTaskFunctionWorkerUsage ( + workerNodeKey: number, + taskName: string + ): void { + const workerNode = this.workerNodes[workerNodeKey] + if ( + this.shallUpdateTaskFunctionWorkerUsage(workerNodeKey) && + workerNode.getTaskFunctionWorkerUsage(taskName) != null + ) { + const taskFunctionWorkerUsage = workerNode.getTaskFunctionWorkerUsage( + taskName + ) as WorkerUsage + ++taskFunctionWorkerUsage.tasks.sequentiallyStolen + } + } + + private resetTaskSequentiallyStolenStatisticsWorkerUsage ( + workerNodeKey: number + ): void { + const workerNode = this.workerNodes[workerNodeKey] + if (workerNode?.usage != null) { + workerNode.usage.tasks.sequentiallyStolen = 0 + } + } + + private resetTaskSequentiallyStolenStatisticsTaskFunctionWorkerUsage ( + workerNodeKey: number, + taskName: string + ): void { + const workerNode = this.workerNodes[workerNodeKey] + if ( + this.shallUpdateTaskFunctionWorkerUsage(workerNodeKey) && + workerNode.getTaskFunctionWorkerUsage(taskName) != null + ) { + const taskFunctionWorkerUsage = workerNode.getTaskFunctionWorkerUsage( + taskName + ) as WorkerUsage + taskFunctionWorkerUsage.tasks.sequentiallyStolen = 0 + } + } + + private readonly handleIdleWorkerNodeEvent = ( + eventDetail: WorkerNodeEventDetail, + previousStolenTask?: Task + ): void => { + if (this.workerNodes.length <= 1) { + return + } + const { workerNodeKey } = eventDetail + if (workerNodeKey == null) { + throw new Error( + 'WorkerNode event detail workerNodeKey attribute must be defined' + ) + } + const workerNodeTasksUsage = this.workerNodes[workerNodeKey].usage.tasks + if ( + previousStolenTask != null && + workerNodeTasksUsage.sequentiallyStolen > 0 && + (workerNodeTasksUsage.executing > 0 || + this.tasksQueueSize(workerNodeKey) > 0) + ) { + for (const taskName of this.workerNodes[workerNodeKey].info + .taskFunctionNames as string[]) { + this.resetTaskSequentiallyStolenStatisticsTaskFunctionWorkerUsage( + workerNodeKey, + taskName + ) + } + this.resetTaskSequentiallyStolenStatisticsWorkerUsage(workerNodeKey) + return + } + const stolenTask = this.workerNodeStealTask(workerNodeKey) + if ( + this.shallUpdateTaskFunctionWorkerUsage(workerNodeKey) && + stolenTask != null + ) { + const taskFunctionTasksWorkerUsage = this.workerNodes[ + workerNodeKey + ].getTaskFunctionWorkerUsage(stolenTask.name as string) + ?.tasks as TaskStatistics + if ( + taskFunctionTasksWorkerUsage.sequentiallyStolen === 0 || + (previousStolenTask != null && + previousStolenTask.name === stolenTask.name && + taskFunctionTasksWorkerUsage.sequentiallyStolen > 0) + ) { + this.updateTaskSequentiallyStolenStatisticsTaskFunctionWorkerUsage( + workerNodeKey, + stolenTask.name as string + ) + } else { + this.resetTaskSequentiallyStolenStatisticsTaskFunctionWorkerUsage( + workerNodeKey, + stolenTask.name as string + ) + } + } + sleep(exponentialDelay(workerNodeTasksUsage.sequentiallyStolen)) + .then(() => { + this.handleIdleWorkerNodeEvent(eventDetail, stolenTask) + return undefined + }) + .catch(EMPTY_FUNCTION) + } + + private readonly workerNodeStealTask = ( + workerNodeKey: number + ): Task | undefined => { const workerNodes = this.workerNodes .slice() .sort( @@ -1376,26 +1618,34 @@ export abstract class AbstractPool< workerNodeB.usage.tasks.queued - workerNodeA.usage.tasks.queued ) const sourceWorkerNode = workerNodes.find( - workerNode => - workerNode.info.ready && - workerNode.info.id !== workerId && - workerNode.usage.tasks.queued > 0 + (sourceWorkerNode, sourceWorkerNodeKey) => + sourceWorkerNode.info.ready && + sourceWorkerNodeKey !== workerNodeKey && + sourceWorkerNode.usage.tasks.queued > 0 ) if (sourceWorkerNode != null) { const task = sourceWorkerNode.popTask() as Task - if (this.shallExecuteTask(destinationWorkerNodeKey)) { - this.executeTask(destinationWorkerNodeKey, task) + if (this.shallExecuteTask(workerNodeKey)) { + this.executeTask(workerNodeKey, task) } else { - this.enqueueTask(destinationWorkerNodeKey, task) + this.enqueueTask(workerNodeKey, task) } + this.updateTaskSequentiallyStolenStatisticsWorkerUsage(workerNodeKey) this.updateTaskStolenStatisticsWorkerUsage( - destinationWorkerNodeKey, + workerNodeKey, task.name as string ) + return task } } - private tasksStealingOnBackPressure (workerId: number): void { + private readonly handleBackPressureEvent = ( + eventDetail: WorkerNodeEventDetail + ): void => { + if (this.workerNodes.length <= 1) { + return + } + const { workerId } = eventDetail const sizeOffset = 1 if ((this.opts.tasksQueueOptions?.size as number) <= sizeOffset) { return @@ -1431,68 +1681,88 @@ 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 readonly workerMessageListener = ( + message: MessageValue + ): void => { + this.checkMessageWorkerId(message) + const { workerId, ready, taskId, taskFunctionNames } = message + if (ready != null && taskFunctionNames != null) { + // Worker ready response received from worker + this.handleWorkerReadyResponse(message) + } else if (taskId != null) { + // Task execution response received from worker + this.handleTaskExecutionResponse(message) + } else if (taskFunctionNames != null) { + // Task function names message received from worker + this.getWorkerInfo( + this.getWorkerNodeKeyByWorkerId(workerId) + ).taskFunctionNames = taskFunctionNames } } private handleWorkerReadyResponse (message: MessageValue): void { - if (message.ready === false) { - throw new Error( - `Worker ${message.workerId as number} failed to initialize` - ) + const { workerId, ready, taskFunctionNames } = message + if (ready === false) { + throw new Error(`Worker ${workerId as number} failed to initialize`) } const workerInfo = this.getWorkerInfo( - this.getWorkerNodeKeyByWorkerId(message.workerId) + this.getWorkerNodeKeyByWorkerId(workerId) ) - workerInfo.ready = message.ready as boolean - workerInfo.taskFunctionNames = message.taskFunctionNames - if (this.ready) { + workerInfo.ready = ready as boolean + workerInfo.taskFunctionNames = taskFunctionNames + if (!this.readyEventEmitted && this.ready) { + this.readyEventEmitted = true this.emitter?.emit(PoolEvents.ready, this.info) } } private handleTaskExecutionResponse (message: MessageValue): void { - const { taskId, workerError, data } = message + const { workerId, taskId, workerError, data } = message const promiseResponse = this.promiseResponseMap.get(taskId as string) if (promiseResponse != null) { + const { resolve, reject, workerNodeKey, asyncResource } = promiseResponse if (workerError != null) { this.emitter?.emit(PoolEvents.taskError, workerError) - promiseResponse.reject(workerError.message) + asyncResource != null + ? asyncResource.runInAsyncScope( + reject, + this.emitter, + workerError.message + ) + : reject(workerError.message) } else { - promiseResponse.resolve(data as Response) + asyncResource != null + ? asyncResource.runInAsyncScope(resolve, this.emitter, data) + : resolve(data as Response) } - const workerNodeKey = promiseResponse.workerNodeKey + asyncResource?.emitDestroy() this.afterTaskExecutionHook(workerNodeKey, message) this.workerChoiceStrategyContext.update(workerNodeKey) this.promiseResponseMap.delete(taskId as string) - if ( - this.opts.enableTasksQueue === true && - this.tasksQueueSize(workerNodeKey) > 0 && - this.workerNodes[workerNodeKey].usage.tasks.executing < - (this.opts.tasksQueueOptions?.concurrency as number) - ) { - this.executeTask( - workerNodeKey, - this.dequeueTask(workerNodeKey) as Task - ) + if (this.opts.enableTasksQueue === true) { + const workerNodeTasksUsage = this.workerNodes[workerNodeKey].usage.tasks + if ( + this.tasksQueueSize(workerNodeKey) > 0 && + workerNodeTasksUsage.executing < + (this.opts.tasksQueueOptions?.concurrency as number) + ) { + this.executeTask( + workerNodeKey, + this.dequeueTask(workerNodeKey) as Task + ) + } + if ( + workerNodeTasksUsage.executing === 0 && + this.tasksQueueSize(workerNodeKey) === 0 && + workerNodeTasksUsage.sequentiallyStolen === 0 + ) { + this.workerNodes[workerNodeKey].emit('idleWorkerNode', { + workerId: workerId as number, + workerNodeKey + }) + } } } } @@ -1524,7 +1794,7 @@ export abstract class AbstractPool< * @returns The worker information. */ protected getWorkerInfo (workerNodeKey: number): WorkerInfo { - return this.workerNodes[workerNodeKey].info + return this.workerNodes[workerNodeKey]?.info } /** @@ -1564,6 +1834,10 @@ export abstract class AbstractPool< } } + protected flagWorkerNodeAsNotReady (workerNodeKey: number): void { + this.getWorkerInfo(workerNodeKey).ready = false + } + /** @inheritDoc */ public hasWorkerNodeBackPressure (workerNodeKey: number): boolean { return (