X-Git-Url: https://git.piment-noir.org/?a=blobdiff_plain;f=src%2Fpools%2Fabstract-pool.ts;h=3da4c70fa41e0cf5168dfccbe517d2ae626f8542;hb=1df5c5ee5a2115a50bdb3ef380e0d74d1d1d5c1e;hp=8c2dcc31d2e849c0fdbf4e0ed9d8bfa297c585de;hpb=b4e93467358d93166716dc6e5075d58c18c961c7;p=poolifier.git diff --git a/src/pools/abstract-pool.ts b/src/pools/abstract-pool.ts index 8c2dcc31..3da4c70f 100644 --- a/src/pools/abstract-pool.ts +++ b/src/pools/abstract-pool.ts @@ -12,12 +12,14 @@ 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' @@ -33,6 +35,7 @@ import { import type { IWorker, IWorkerNode, + TaskStatistics, WorkerInfo, WorkerNodeEventDetail, WorkerType, @@ -591,11 +594,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() @@ -627,36 +632,36 @@ export abstract class AbstractPool< private setTaskStealing (): void { for (const [workerNodeKey] of this.workerNodes.entries()) { - this.workerNodes[workerNodeKey].addEventListener( + this.workerNodes[workerNodeKey].on( 'idleWorkerNode', - this.handleIdleWorkerNodeEvent as EventListener + this.handleIdleWorkerNodeEvent ) } } private unsetTaskStealing (): void { for (const [workerNodeKey] of this.workerNodes.entries()) { - this.workerNodes[workerNodeKey].removeEventListener( + this.workerNodes[workerNodeKey].off( 'idleWorkerNode', - this.handleIdleWorkerNodeEvent as EventListener + this.handleIdleWorkerNodeEvent ) } } private setTasksStealingOnBackPressure (): void { for (const [workerNodeKey] of this.workerNodes.entries()) { - this.workerNodes[workerNodeKey].addEventListener( + this.workerNodes[workerNodeKey].on( 'backPressure', - this.handleBackPressureEvent as EventListener + this.handleBackPressureEvent ) } } private unsetTasksStealingOnBackPressure (): void { for (const [workerNodeKey] of this.workerNodes.entries()) { - this.workerNodes[workerNodeKey].removeEventListener( + this.workerNodes[workerNodeKey].off( 'backPressure', - this.handleBackPressureEvent as EventListener + this.handleBackPressureEvent ) } } @@ -980,12 +985,13 @@ export abstract class AbstractPool< } 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 @@ -1392,7 +1398,7 @@ export abstract class AbstractPool< // Listen to worker messages. this.registerWorkerMessageListener( workerNodeKey, - this.workerMessageListener.bind(this) + this.workerMessageListener ) // Send the startup message to worker. this.sendStartupMessageToWorker(workerNodeKey) @@ -1400,15 +1406,15 @@ export abstract class AbstractPool< this.sendStatisticsMessageToWorker(workerNodeKey) if (this.opts.enableTasksQueue === true) { if (this.opts.tasksQueueOptions?.taskStealing === true) { - this.workerNodes[workerNodeKey].addEventListener( + this.workerNodes[workerNodeKey].on( 'idleWorkerNode', - this.handleIdleWorkerNodeEvent as EventListener + this.handleIdleWorkerNodeEvent ) } if (this.opts.tasksQueueOptions?.tasksStealingOnBackPressure === true) { - this.workerNodes[workerNodeKey].addEventListener( + this.workerNodes[workerNodeKey].on( 'backPressure', - this.handleBackPressureEvent as EventListener + this.handleBackPressureEvent ) } } @@ -1478,11 +1484,120 @@ export abstract class AbstractPool< } } + 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 = ( - event: CustomEvent + eventDetail: WorkerNodeEventDetail, + previousStolenTask?: Task ): void => { - const { workerId } = event.detail - const destinationWorkerNodeKey = this.getWorkerNodeKeyByWorkerId(workerId) + 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( @@ -1490,29 +1605,31 @@ 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 readonly handleBackPressureEvent = ( - event: CustomEvent + eventDetail: WorkerNodeEventDetail ): void => { - const { workerId } = event.detail + const { workerId } = eventDetail const sizeOffset = 1 if ((this.opts.tasksQueueOptions?.size as number) <= sizeOffset) { return @@ -1550,33 +1667,35 @@ export abstract class AbstractPool< /** * This method is the message listener registered on each worker. */ - protected workerMessageListener (message: MessageValue): void { + protected readonly workerMessageListener = ( + message: MessageValue + ): void => { this.checkMessageWorkerId(message) - if (message.ready != null && message.taskFunctionNames != null) { + const { workerId, ready, taskId, taskFunctionNames } = message + if (ready != null && taskFunctionNames != null) { // Worker ready response received from worker this.handleWorkerReadyResponse(message) - } else if (message.taskId != null) { + } else if (taskId != null) { // Task execution response received from worker this.handleTaskExecutionResponse(message) - } else if (message.taskFunctionNames != null) { + } else if (taskFunctionNames != null) { // Task function names message received from worker this.getWorkerInfo( - this.getWorkerNodeKeyByWorkerId(message.workerId) - ).taskFunctionNames = message.taskFunctionNames + 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 + workerInfo.ready = ready as boolean + workerInfo.taskFunctionNames = taskFunctionNames if (!this.readyEventEmitted && this.ready) { this.readyEventEmitted = true this.emitter?.emit(PoolEvents.ready, this.info) @@ -1584,7 +1703,7 @@ export abstract class AbstractPool< } 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 } = promiseResponse @@ -1597,16 +1716,28 @@ export abstract class AbstractPool< 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 + }) + } } } }