X-Git-Url: https://git.piment-noir.org/?a=blobdiff_plain;f=src%2Fpools%2Fabstract-pool.ts;h=12ad699511efdc9e812d893e66c6233091fc1c9d;hb=1df04dd8d2010ba57ec11f96f7d9995bba9a9dfb;hp=63f1f8595c928a6db6e8daf2fe8180b9685b459a;hpb=bcfb06ce041a682baf396a099c633a848d6a4045;p=poolifier.git diff --git a/src/pools/abstract-pool.ts b/src/pools/abstract-pool.ts index 63f1f859..12ad6995 100644 --- a/src/pools/abstract-pool.ts +++ b/src/pools/abstract-pool.ts @@ -47,6 +47,7 @@ import { import { WorkerChoiceStrategiesContext } from './selection-strategies/worker-choice-strategies-context.js' import { checkFilePath, + checkValidPriority, checkValidTasksQueueOptions, checkValidWorkerChoiceStrategy, getDefaultTasksQueueOptions, @@ -91,8 +92,13 @@ export abstract class AbstractPool< * * When we receive a message from the worker, we get a map entry with the promise resolve/reject bound to the message id. */ - protected promiseResponseMap: Map> = - new Map>() + protected promiseResponseMap: Map< + `${string}-${string}-${string}-${string}-${string}`, + PromiseResponseWrapper + > = new Map< + `${string}-${string}-${string}-${string}-${string}`, + PromiseResponseWrapper + >() /** * Worker choice strategies context referencing worker choice algorithms implementation. @@ -136,7 +142,7 @@ export abstract class AbstractPool< /** * The start timestamp of the pool. */ - private readonly startTimestamp + private startTimestamp?: number /** * Constructs a new poolifier pool. @@ -167,7 +173,7 @@ export abstract class AbstractPool< this.enqueueTask = this.enqueueTask.bind(this) if (this.opts.enableEvents === true) { - this.initializeEventEmitter() + this.initEventEmitter() } this.workerChoiceStrategiesContext = new WorkerChoiceStrategiesContext< Worker, @@ -192,8 +198,6 @@ export abstract class AbstractPool< if (this.opts.startWorkers === true) { this.start() } - - this.startTimestamp = performance.now() } private checkPoolType (): void { @@ -282,7 +286,7 @@ export abstract class AbstractPool< } } - private initializeEventEmitter (): void { + private initEventEmitter (): void { this.emitter = new EventEmitterAsyncResource({ name: `poolifier:${this.type}-${this.worker}-pool` }) @@ -323,7 +327,7 @@ export abstract class AbstractPool< ) }), busyWorkerNodes: this.workerNodes.reduce( - (accumulator, _workerNode, workerNodeKey) => + (accumulator, _, workerNodeKey) => this.isWorkerNodeBusy(workerNodeKey) ? accumulator + 1 : accumulator, 0 ), @@ -451,6 +455,91 @@ export abstract class AbstractPool< ) }) } + }), + ...(this.workerChoiceStrategiesContext?.getTaskStatisticsRequirements() + .elu.aggregate === true && { + elu: { + idle: { + minimum: round( + min( + ...this.workerNodes.map( + workerNode => workerNode.usage.elu.idle.minimum ?? Infinity + ) + ) + ), + maximum: round( + max( + ...this.workerNodes.map( + workerNode => workerNode.usage.elu.idle.maximum ?? -Infinity + ) + ) + ), + ...(this.workerChoiceStrategiesContext.getTaskStatisticsRequirements() + .elu.average && { + average: round( + average( + this.workerNodes.reduce( + (accumulator, workerNode) => + accumulator.concat(workerNode.usage.elu.idle.history), + [] + ) + ) + ) + }), + ...(this.workerChoiceStrategiesContext.getTaskStatisticsRequirements() + .elu.median && { + median: round( + median( + this.workerNodes.reduce( + (accumulator, workerNode) => + accumulator.concat(workerNode.usage.elu.idle.history), + [] + ) + ) + ) + }) + }, + active: { + minimum: round( + min( + ...this.workerNodes.map( + workerNode => workerNode.usage.elu.active.minimum ?? Infinity + ) + ) + ), + maximum: round( + max( + ...this.workerNodes.map( + workerNode => workerNode.usage.elu.active.maximum ?? -Infinity + ) + ) + ), + ...(this.workerChoiceStrategiesContext.getTaskStatisticsRequirements() + .elu.average && { + average: round( + average( + this.workerNodes.reduce( + (accumulator, workerNode) => + accumulator.concat(workerNode.usage.elu.active.history), + [] + ) + ) + ) + }), + ...(this.workerChoiceStrategiesContext.getTaskStatisticsRequirements() + .elu.median && { + median: round( + median( + this.workerNodes.reduce( + (accumulator, workerNode) => + accumulator.concat(workerNode.usage.elu.active.history), + [] + ) + ) + ) + }) + } + } }) } } @@ -486,6 +575,9 @@ export abstract class AbstractPool< * @returns The pool utilization. */ private get utilization (): number { + if (this.startTimestamp == null) { + return 0 + } const poolTimeCapacity = (performance.now() - this.startTimestamp) * (this.maximumNumberOfWorkers ?? this.minimumNumberOfWorkers) @@ -547,28 +639,52 @@ export abstract class AbstractPool< workerChoiceStrategy: WorkerChoiceStrategy, workerChoiceStrategyOptions?: WorkerChoiceStrategyOptions ): void { + let requireSync = false checkValidWorkerChoiceStrategy(workerChoiceStrategy) - this.opts.workerChoiceStrategy = workerChoiceStrategy - this.workerChoiceStrategiesContext?.setDefaultWorkerChoiceStrategy( - this.opts.workerChoiceStrategy, - workerChoiceStrategyOptions - ) - for (const [workerNodeKey] of this.workerNodes.entries()) { - this.sendStatisticsMessageToWorker(workerNodeKey) + if (workerChoiceStrategyOptions != null) { + requireSync = !this.setWorkerChoiceStrategyOptions( + workerChoiceStrategyOptions + ) + } + if (workerChoiceStrategy !== this.opts.workerChoiceStrategy) { + this.opts.workerChoiceStrategy = workerChoiceStrategy + this.workerChoiceStrategiesContext?.setDefaultWorkerChoiceStrategy( + this.opts.workerChoiceStrategy, + this.opts.workerChoiceStrategyOptions + ) + requireSync = true + } + if (requireSync) { + this.workerChoiceStrategiesContext?.syncWorkerChoiceStrategies( + this.getWorkerChoiceStrategies(), + this.opts.workerChoiceStrategyOptions + ) + for (const workerNodeKey of this.workerNodes.keys()) { + this.sendStatisticsMessageToWorker(workerNodeKey) + } } } /** @inheritDoc */ public setWorkerChoiceStrategyOptions ( workerChoiceStrategyOptions: WorkerChoiceStrategyOptions | undefined - ): void { + ): boolean { this.checkValidWorkerChoiceStrategyOptions(workerChoiceStrategyOptions) if (workerChoiceStrategyOptions != null) { this.opts.workerChoiceStrategyOptions = workerChoiceStrategyOptions + this.workerChoiceStrategiesContext?.setOptions( + this.opts.workerChoiceStrategyOptions + ) + this.workerChoiceStrategiesContext?.syncWorkerChoiceStrategies( + this.getWorkerChoiceStrategies(), + this.opts.workerChoiceStrategyOptions + ) + for (const workerNodeKey of this.workerNodes.keys()) { + this.sendStatisticsMessageToWorker(workerNodeKey) + } + return true } - this.workerChoiceStrategiesContext?.setOptions( - this.opts.workerChoiceStrategyOptions - ) + return false } /** @inheritDoc */ @@ -630,13 +746,13 @@ export abstract class AbstractPool< } private setTaskStealing (): void { - for (const [workerNodeKey] of this.workerNodes.entries()) { + for (const workerNodeKey of this.workerNodes.keys()) { this.workerNodes[workerNodeKey].on('idle', this.handleWorkerNodeIdleEvent) } } private unsetTaskStealing (): void { - for (const [workerNodeKey] of this.workerNodes.entries()) { + for (const workerNodeKey of this.workerNodes.keys()) { this.workerNodes[workerNodeKey].off( 'idle', this.handleWorkerNodeIdleEvent @@ -645,7 +761,7 @@ export abstract class AbstractPool< } private setTasksStealingOnBackPressure (): void { - for (const [workerNodeKey] of this.workerNodes.entries()) { + for (const workerNodeKey of this.workerNodes.keys()) { this.workerNodes[workerNodeKey].on( 'backPressure', this.handleWorkerNodeBackPressureEvent @@ -654,7 +770,7 @@ export abstract class AbstractPool< } private unsetTasksStealingOnBackPressure (): void { - for (const [workerNodeKey] of this.workerNodes.entries()) { + for (const workerNodeKey of this.workerNodes.keys()) { this.workerNodes[workerNodeKey].off( 'backPressure', this.handleWorkerNodeBackPressureEvent @@ -797,7 +913,7 @@ export abstract class AbstractPool< } } } - for (const [workerNodeKey] of this.workerNodes.entries()) { + for (const workerNodeKey of this.workerNodes.keys()) { this.registerWorkerMessageListener( workerNodeKey, taskFunctionOperationsListener @@ -809,17 +925,9 @@ export abstract class AbstractPool< /** @inheritDoc */ public hasTaskFunction (name: string): boolean { - for (const workerNode of this.workerNodes) { - if ( - Array.isArray(workerNode.info.taskFunctionsProperties) && - workerNode.info.taskFunctionsProperties.some( - taskFunctionProperties => taskFunctionProperties.name === name - ) - ) { - return true - } - } - return false + return this.listTaskFunctionsProperties().some( + taskFunctionProperties => taskFunctionProperties.name === name + ) } /** @inheritDoc */ @@ -839,12 +947,20 @@ export abstract class AbstractPool< if (typeof fn.taskFunction !== 'function') { throw new TypeError('taskFunction property must be a function') } + checkValidPriority(fn.priority) + checkValidWorkerChoiceStrategy(fn.strategy) const opResult = await this.sendTaskFunctionOperationToWorkers({ taskFunctionOperation: 'add', taskFunctionProperties: buildTaskFunctionProperties(name, fn), taskFunction: fn.taskFunction.toString() }) this.taskFunctions.set(name, fn) + this.workerChoiceStrategiesContext?.syncWorkerChoiceStrategies( + this.getWorkerChoiceStrategies() + ) + for (const workerNodeKey of this.workerNodes.keys()) { + this.sendStatisticsMessageToWorker(workerNodeKey) + } return opResult } @@ -862,8 +978,16 @@ export abstract class AbstractPool< this.taskFunctions.get(name) ) }) - this.deleteTaskFunctionWorkerUsages(name) + for (const workerNode of this.workerNodes) { + workerNode.deleteTaskFunctionWorkerUsage(name) + } this.taskFunctions.delete(name) + this.workerChoiceStrategiesContext?.syncWorkerChoiceStrategies( + this.getWorkerChoiceStrategies() + ) + for (const workerNodeKey of this.workerNodes.keys()) { + this.sendStatisticsMessageToWorker(workerNodeKey) + } return opResult } @@ -881,22 +1005,96 @@ export abstract class AbstractPool< } /** - * Gets task function strategy, if any. + * Gets task function worker choice strategy, if any. * * @param name - The task function name. * @returns The task function worker choice strategy if the task function worker choice strategy is defined, `undefined` otherwise. */ - private readonly getTaskFunctionWorkerWorkerChoiceStrategy = ( + private readonly getTaskFunctionWorkerChoiceStrategy = ( name?: string ): WorkerChoiceStrategy | undefined => { - if (name != null) { - return this.listTaskFunctionsProperties().find( - (taskFunctionProperties: TaskFunctionProperties) => - taskFunctionProperties.name === name - )?.strategy + name = name ?? DEFAULT_TASK_NAME + const taskFunctionsProperties = this.listTaskFunctionsProperties() + if (name === DEFAULT_TASK_NAME) { + name = taskFunctionsProperties[1]?.name } + return taskFunctionsProperties.find( + (taskFunctionProperties: TaskFunctionProperties) => + taskFunctionProperties.name === name + )?.strategy } + /** + * Gets worker node task function worker choice strategy, if any. + * + * @param workerNodeKey - The worker node key. + * @param name - The task function name. + * @returns The worker node task function worker choice strategy if the worker node task function worker choice strategy is defined, `undefined` otherwise. + */ + private readonly getWorkerNodeTaskFunctionWorkerChoiceStrategy = ( + workerNodeKey: number, + name?: string + ): WorkerChoiceStrategy | undefined => { + const workerInfo = this.getWorkerInfo(workerNodeKey) + if (workerInfo == null) { + return + } + name = name ?? DEFAULT_TASK_NAME + if (name === DEFAULT_TASK_NAME) { + name = workerInfo.taskFunctionsProperties?.[1]?.name + } + return workerInfo.taskFunctionsProperties?.find( + (taskFunctionProperties: TaskFunctionProperties) => + taskFunctionProperties.name === name + )?.strategy + } + + /** + * Gets worker node task function priority, if any. + * + * @param workerNodeKey - The worker node key. + * @param name - The task function name. + * @returns The worker node task function priority if the worker node task function priority is defined, `undefined` otherwise. + */ + private readonly getWorkerNodeTaskFunctionPriority = ( + workerNodeKey: number, + name?: string + ): number | undefined => { + const workerInfo = this.getWorkerInfo(workerNodeKey) + if (workerInfo == null) { + return + } + name = name ?? DEFAULT_TASK_NAME + if (name === DEFAULT_TASK_NAME) { + name = workerInfo.taskFunctionsProperties?.[1]?.name + } + return workerInfo.taskFunctionsProperties?.find( + (taskFunctionProperties: TaskFunctionProperties) => + taskFunctionProperties.name === name + )?.priority + } + + /** + * Gets the worker choice strategies registered in this pool. + * + * @returns The worker choice strategies. + */ + private readonly getWorkerChoiceStrategies = + (): Set => { + return new Set([ + // eslint-disable-next-line @typescript-eslint/no-non-null-assertion + this.opts.workerChoiceStrategy!, + ...(this.listTaskFunctionsProperties() + .map( + (taskFunctionProperties: TaskFunctionProperties) => + taskFunctionProperties.strategy + ) + .filter( + (strategy: WorkerChoiceStrategy | undefined) => strategy != null + ) as WorkerChoiceStrategy[]) + ]) + } + /** @inheritDoc */ public async setDefaultTaskFunction (name: string): Promise { return await this.sendTaskFunctionOperationToWorkers({ @@ -908,12 +1106,6 @@ 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 && @@ -955,13 +1147,16 @@ export abstract class AbstractPool< return } const timestamp = performance.now() - const workerNodeKey = this.chooseWorkerNode( - this.getTaskFunctionWorkerWorkerChoiceStrategy(name) - ) + const workerNodeKey = this.chooseWorkerNode(name) const task: Task = { name: name ?? DEFAULT_TASK_NAME, // eslint-disable-next-line @typescript-eslint/consistent-type-assertions data: data ?? ({} as Data), + priority: this.getWorkerNodeTaskFunctionPriority(workerNodeKey, name), + strategy: this.getWorkerNodeTaskFunctionWorkerChoiceStrategy( + workerNodeKey, + name + ), transferList, timestamp, taskId: randomUUID() @@ -993,7 +1188,7 @@ export abstract class AbstractPool< /** * Starts the minimum number of workers. */ - private startMinimumNumberOfWorkers (): void { + private startMinimumNumberOfWorkers (initWorkerNodeUsage = false): void { this.startingMinimumNumberOfWorkers = true while ( this.workerNodes.reduce( @@ -1002,7 +1197,9 @@ export abstract class AbstractPool< 0 ) < this.minimumNumberOfWorkers ) { - this.createAndSetupWorkerNode() + const workerNodeKey = this.createAndSetupWorkerNode() + initWorkerNodeUsage && + this.initWorkerNodeUsage(this.workerNodes[workerNodeKey]) } this.startingMinimumNumberOfWorkers = false } @@ -1020,6 +1217,7 @@ export abstract class AbstractPool< } this.starting = true this.startMinimumNumberOfWorkers() + this.startTimestamp = performance.now() this.starting = false this.started = true } @@ -1037,13 +1235,14 @@ export abstract class AbstractPool< } this.destroying = true await Promise.all( - this.workerNodes.map(async (_workerNode, workerNodeKey) => { + this.workerNodes.map(async (_, workerNodeKey) => { await this.destroyWorkerNode(workerNodeKey) }) ) this.emitter?.emit(PoolEvents.destroy, this.info) this.emitter?.emitDestroy() this.readyEventEmitted = false + delete this.startTimestamp this.destroying = false this.started = false } @@ -1164,7 +1363,7 @@ export abstract class AbstractPool< workerNodeKey: number, message: MessageValue ): void { - let needWorkerChoiceStrategyUpdate = false + let needWorkerChoiceStrategiesUpdate = false // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition if (this.workerNodes[workerNodeKey]?.usage != null) { const workerUsage = this.workerNodes[workerNodeKey].usage @@ -1179,7 +1378,7 @@ export abstract class AbstractPool< workerUsage, message ) - needWorkerChoiceStrategyUpdate = true + needWorkerChoiceStrategiesUpdate = true } if ( this.shallUpdateTaskFunctionWorkerUsage(workerNodeKey) && @@ -1204,9 +1403,9 @@ export abstract class AbstractPool< taskFunctionWorkerUsage, message ) - needWorkerChoiceStrategyUpdate = true + needWorkerChoiceStrategiesUpdate = true } - if (needWorkerChoiceStrategyUpdate) { + if (needWorkerChoiceStrategiesUpdate) { this.workerChoiceStrategiesContext?.update(workerNodeKey) } } @@ -1229,12 +1428,10 @@ export abstract class AbstractPool< /** * Chooses a worker node for the next task. * - * @param workerChoiceStrategy - The worker choice strategy. + * @param name - The task function name. * @returns The chosen worker node key */ - private chooseWorkerNode ( - workerChoiceStrategy?: WorkerChoiceStrategy - ): number { + private chooseWorkerNode (name?: string): number { if (this.shallCreateDynamicWorker()) { const workerNodeKey = this.createAndSetupDynamicWorkerNode() if ( @@ -1245,7 +1442,9 @@ export abstract class AbstractPool< } } // eslint-disable-next-line @typescript-eslint/no-non-null-assertion - return this.workerChoiceStrategiesContext!.execute(workerChoiceStrategy) + return this.workerChoiceStrategiesContext!.execute( + this.getTaskFunctionWorkerChoiceStrategy(name) + ) } /** @@ -1268,6 +1467,44 @@ export abstract class AbstractPool< transferList?: readonly TransferListItem[] ): void + /** + * Initializes the worker node usage with sensible default values gathered during runtime. + * + * @param workerNode - The worker node. + */ + private initWorkerNodeUsage (workerNode: IWorkerNode): void { + if ( + this.workerChoiceStrategiesContext?.getTaskStatisticsRequirements() + .runTime.aggregate === true + ) { + workerNode.usage.runTime.aggregate = min( + ...this.workerNodes.map( + workerNode => workerNode.usage.runTime.aggregate ?? Infinity + ) + ) + } + if ( + this.workerChoiceStrategiesContext?.getTaskStatisticsRequirements() + .waitTime.aggregate === true + ) { + workerNode.usage.waitTime.aggregate = min( + ...this.workerNodes.map( + workerNode => workerNode.usage.waitTime.aggregate ?? Infinity + ) + ) + } + if ( + this.workerChoiceStrategiesContext?.getTaskStatisticsRequirements().elu + .aggregate === true + ) { + workerNode.usage.elu.active.aggregate = min( + ...this.workerNodes.map( + workerNode => workerNode.usage.elu.active.aggregate ?? Infinity + ) + ) + } + } + /** * Creates a new, completely set up worker node. * @@ -1298,7 +1535,7 @@ export abstract class AbstractPool< if (workerNode.info.dynamic) { this.createAndSetupDynamicWorkerNode() } else if (!this.startingMinimumNumberOfWorkers) { - this.startMinimumNumberOfWorkers() + this.startMinimumNumberOfWorkers(true) } } if ( @@ -1324,7 +1561,7 @@ export abstract class AbstractPool< !this.startingMinimumNumberOfWorkers && !this.destroying ) { - this.startMinimumNumberOfWorkers() + this.startMinimumNumberOfWorkers(true) } }) const workerNodeKey = this.addWorkerNode(workerNode) @@ -1389,6 +1626,7 @@ export abstract class AbstractPool< ) { workerNode.info.ready = true } + this.initWorkerNodeUsage(workerNode) this.checkAndEmitDynamicWorkerCreationEvents() return workerNodeKey } @@ -1501,14 +1739,15 @@ export abstract class AbstractPool< } } - private redistributeQueuedTasks (workerNodeKey: number): void { - if (workerNodeKey === -1 || this.cannotStealTask()) { + private redistributeQueuedTasks (sourceWorkerNodeKey: number): void { + if (sourceWorkerNodeKey === -1 || this.cannotStealTask()) { return } - while (this.tasksQueueSize(workerNodeKey) > 0) { + while (this.tasksQueueSize(sourceWorkerNodeKey) > 0) { const destinationWorkerNodeKey = this.workerNodes.reduce( (minWorkerNodeKey, workerNode, workerNodeKey, workerNodes) => { - return workerNode.info.ready && + return sourceWorkerNodeKey !== workerNodeKey && + workerNode.info.ready && workerNode.usage.tasks.queued < workerNodes[minWorkerNodeKey].usage.tasks.queued ? workerNodeKey @@ -1519,7 +1758,7 @@ export abstract class AbstractPool< this.handleTask( destinationWorkerNodeKey, // eslint-disable-next-line @typescript-eslint/no-non-null-assertion - this.dequeueTask(workerNodeKey)! + this.dequeueTask(sourceWorkerNodeKey)! ) } } @@ -1537,28 +1776,21 @@ export abstract class AbstractPool< this.shallUpdateTaskFunctionWorkerUsage(workerNodeKey) && workerNode.getTaskFunctionWorkerUsage(taskName) != null ) { - const taskFunctionWorkerUsage = - // eslint-disable-next-line @typescript-eslint/no-non-null-assertion - workerNode.getTaskFunctionWorkerUsage(taskName)! - ++taskFunctionWorkerUsage.tasks.stolen + // eslint-disable-next-line @typescript-eslint/no-non-null-assertion + ++workerNode.getTaskFunctionWorkerUsage(taskName)!.tasks.stolen } } private updateTaskSequentiallyStolenStatisticsWorkerUsage ( - workerNodeKey: number + workerNodeKey: number, + taskName: string, + previousTaskName?: string ): void { const workerNode = this.workerNodes[workerNodeKey] // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition 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 @@ -1566,33 +1798,36 @@ export abstract class AbstractPool< const taskFunctionWorkerUsage = // eslint-disable-next-line @typescript-eslint/no-non-null-assertion workerNode.getTaskFunctionWorkerUsage(taskName)! - ++taskFunctionWorkerUsage.tasks.sequentiallyStolen + if ( + taskFunctionWorkerUsage.tasks.sequentiallyStolen === 0 || + (previousTaskName != null && + previousTaskName === taskName && + taskFunctionWorkerUsage.tasks.sequentiallyStolen > 0) + ) { + ++taskFunctionWorkerUsage.tasks.sequentiallyStolen + } else if (taskFunctionWorkerUsage.tasks.sequentiallyStolen > 0) { + taskFunctionWorkerUsage.tasks.sequentiallyStolen = 0 + } } } private resetTaskSequentiallyStolenStatisticsWorkerUsage ( - workerNodeKey: number + workerNodeKey: number, + taskName: string ): void { const workerNode = this.workerNodes[workerNodeKey] // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition 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 = - // eslint-disable-next-line @typescript-eslint/no-non-null-assertion - workerNode.getTaskFunctionWorkerUsage(taskName)! - taskFunctionWorkerUsage.tasks.sequentiallyStolen = 0 + // eslint-disable-next-line @typescript-eslint/no-non-null-assertion + workerNode.getTaskFunctionWorkerUsage( + taskName + )!.tasks.sequentiallyStolen = 0 } } @@ -1607,69 +1842,49 @@ export abstract class AbstractPool< ) } const workerInfo = this.getWorkerInfo(workerNodeKey) + if (workerInfo == null) { + throw new Error( + `Worker node with key '${workerNodeKey}' not found in pool` + ) + } if ( this.cannotStealTask() || (this.info.stealingWorkerNodes ?? 0) > Math.floor(this.workerNodes.length / 2) ) { - if (workerInfo != null && previousStolenTask != null) { + if (previousStolenTask != null) { workerInfo.stealing = false + this.resetTaskSequentiallyStolenStatisticsWorkerUsage( + workerNodeKey, + // eslint-disable-next-line @typescript-eslint/no-non-null-assertion + previousStolenTask.name! + ) } return } const workerNodeTasksUsage = this.workerNodes[workerNodeKey].usage.tasks if ( - workerInfo != null && previousStolenTask != null && - workerNodeTasksUsage.sequentiallyStolen > 0 && (workerNodeTasksUsage.executing > 0 || this.tasksQueueSize(workerNodeKey) > 0) ) { workerInfo.stealing = false - // eslint-disable-next-line @typescript-eslint/no-non-null-assertion - for (const taskFunctionProperties of workerInfo.taskFunctionsProperties!) { - this.resetTaskSequentiallyStolenStatisticsTaskFunctionWorkerUsage( - workerNodeKey, - taskFunctionProperties.name - ) - } - this.resetTaskSequentiallyStolenStatisticsWorkerUsage(workerNodeKey) - return - } - if (workerInfo == null) { - throw new Error( - `Worker node with key '${workerNodeKey}' not found in pool` + this.resetTaskSequentiallyStolenStatisticsWorkerUsage( + workerNodeKey, + // eslint-disable-next-line @typescript-eslint/no-non-null-assertion + previousStolenTask.name! ) + return } workerInfo.stealing = true const stolenTask = this.workerNodeStealTask(workerNodeKey) - if ( - this.shallUpdateTaskFunctionWorkerUsage(workerNodeKey) && - stolenTask != null - ) { - // eslint-disable-next-line @typescript-eslint/no-non-null-assertion - const taskFunctionTasksWorkerUsage = this.workerNodes[ - workerNodeKey + if (stolenTask != null) { + this.updateTaskSequentiallyStolenStatisticsWorkerUsage( + workerNodeKey, // eslint-disable-next-line @typescript-eslint/no-non-null-assertion - ].getTaskFunctionWorkerUsage(stolenTask.name!)!.tasks - if ( - taskFunctionTasksWorkerUsage.sequentiallyStolen === 0 || - (previousStolenTask != null && - previousStolenTask.name === stolenTask.name && - taskFunctionTasksWorkerUsage.sequentiallyStolen > 0) - ) { - this.updateTaskSequentiallyStolenStatisticsTaskFunctionWorkerUsage( - workerNodeKey, - // eslint-disable-next-line @typescript-eslint/no-non-null-assertion - stolenTask.name! - ) - } else { - this.resetTaskSequentiallyStolenStatisticsTaskFunctionWorkerUsage( - workerNodeKey, - // eslint-disable-next-line @typescript-eslint/no-non-null-assertion - stolenTask.name! - ) - } + stolenTask.name!, + previousStolenTask?.name + ) } sleep(exponentialDelay(workerNodeTasksUsage.sequentiallyStolen)) .then(() => { @@ -1699,9 +1914,8 @@ export abstract class AbstractPool< ) if (sourceWorkerNode != null) { // eslint-disable-next-line @typescript-eslint/no-non-null-assertion - const task = sourceWorkerNode.popTask()! + const task = sourceWorkerNode.dequeueLastPrioritizedTask()! this.handleTask(workerNodeKey, task) - this.updateTaskSequentiallyStolenStatisticsWorkerUsage(workerNodeKey) // eslint-disable-next-line @typescript-eslint/no-non-null-assertion this.updateTaskStolenStatisticsWorkerUsage(workerNodeKey, task.name!) return task @@ -1713,17 +1927,18 @@ export abstract class AbstractPool< ): void => { if ( this.cannotStealTask() || + this.hasBackPressure() || (this.info.stealingWorkerNodes ?? 0) > Math.floor(this.workerNodes.length / 2) ) { return } - const { workerId } = eventDetail const sizeOffset = 1 // eslint-disable-next-line @typescript-eslint/no-non-null-assertion if (this.opts.tasksQueueOptions!.size! <= sizeOffset) { return } + const { workerId } = eventDetail const sourceWorkerNode = this.workerNodes[this.getWorkerNodeKeyByWorkerId(workerId)] const workerNodes = this.workerNodes @@ -1750,7 +1965,7 @@ export abstract class AbstractPool< } workerInfo.stealing = true // eslint-disable-next-line @typescript-eslint/no-non-null-assertion - const task = sourceWorkerNode.popTask()! + const task = sourceWorkerNode.dequeueLastPrioritizedTask()! this.handleTask(workerNodeKey, task) // eslint-disable-next-line @typescript-eslint/no-non-null-assertion this.updateTaskStolenStatisticsWorkerUsage(workerNodeKey, task.name!) @@ -1772,11 +1987,11 @@ export abstract class AbstractPool< this.handleWorkerReadyResponse(message) } else if (taskFunctionsProperties != null) { // Task function properties message received from worker - const workerInfo = this.getWorkerInfo( - this.getWorkerNodeKeyByWorkerId(workerId) - ) + const workerNodeKey = this.getWorkerNodeKeyByWorkerId(workerId) + const workerInfo = this.getWorkerInfo(workerNodeKey) if (workerInfo != null) { workerInfo.taskFunctionsProperties = taskFunctionsProperties + this.sendStatisticsMessageToWorker(workerNodeKey) } } else if (taskId != null) { // Task execution response received from worker @@ -1796,10 +2011,11 @@ export abstract class AbstractPool< if (ready == null || !ready) { throw new Error(`Worker ${workerId} failed to initialize`) } - const workerNode = - this.workerNodes[this.getWorkerNodeKeyByWorkerId(workerId)] + const workerNodeKey = this.getWorkerNodeKeyByWorkerId(workerId) + const workerNode = this.workerNodes[workerNodeKey] workerNode.info.ready = ready workerNode.info.taskFunctionsProperties = taskFunctionsProperties + this.sendStatisticsMessageToWorker(workerNodeKey) this.checkAndEmitReadyEvent() } @@ -1903,7 +2119,9 @@ export abstract class AbstractPool< this.opts.tasksQueueOptions?.size ?? getDefaultTasksQueueOptions( this.maximumNumberOfWorkers ?? this.minimumNumberOfWorkers - ).size + ).size, + tasksQueueBucketSize: + (this.maximumNumberOfWorkers ?? this.minimumNumberOfWorkers) * 2 } ) // Flag the worker node as ready at pool startup. @@ -2004,7 +2222,7 @@ export abstract class AbstractPool< } private flushTasksQueues (): void { - for (const [workerNodeKey] of this.workerNodes.entries()) { + for (const workerNodeKey of this.workerNodes.keys()) { this.flushTasksQueue(workerNodeKey) } }