X-Git-Url: https://git.piment-noir.org/?a=blobdiff_plain;f=src%2Fpools%2Fabstract-pool.ts;h=a41996ff918d527d2ef81ac29f431cdd601b91c7;hb=1a880eca683648d5228e2424db74553f58ecee7d;hp=3a1a87807cc065df871bb1ca6d2195619672a64d;hpb=30c799e98a2d088cbca774d347983a04ebc5f5a5;p=poolifier.git diff --git a/src/pools/abstract-pool.ts b/src/pools/abstract-pool.ts index 3a1a8780..a41996ff 100644 --- a/src/pools/abstract-pool.ts +++ b/src/pools/abstract-pool.ts @@ -5,13 +5,13 @@ import { type TransferListItem } from 'node:worker_threads' import type { MessageValue, PromiseResponseWrapper, - Task, - Writable + Task } from '../utility-types' import { DEFAULT_TASK_NAME, DEFAULT_WORKER_CHOICE_STRATEGY_OPTIONS, EMPTY_FUNCTION, + average, isKillBehavior, isPlainObject, median, @@ -255,19 +255,19 @@ export abstract class AbstractPool< ) } if ( - workerChoiceStrategyOptions.choiceRetries != null && - !Number.isSafeInteger(workerChoiceStrategyOptions.choiceRetries) + workerChoiceStrategyOptions.retries != null && + !Number.isSafeInteger(workerChoiceStrategyOptions.retries) ) { throw new TypeError( - 'Invalid worker choice strategy options: choice retries must be an integer' + 'Invalid worker choice strategy options: retries must be an integer' ) } if ( - workerChoiceStrategyOptions.choiceRetries != null && - workerChoiceStrategyOptions.choiceRetries <= 0 + workerChoiceStrategyOptions.retries != null && + workerChoiceStrategyOptions.retries < 0 ) { throw new RangeError( - `Invalid worker choice strategy options: choice retries '${workerChoiceStrategyOptions.choiceRetries}' must be greater than zero` + `Invalid worker choice strategy options: retries '${workerChoiceStrategyOptions.retries}' must be greater or equal than zero` ) } if ( @@ -291,7 +291,7 @@ export abstract class AbstractPool< } private checkValidTasksQueueOptions ( - tasksQueueOptions: Writable + tasksQueueOptions: TasksQueueOptions ): void { if (tasksQueueOptions != null && !isPlainObject(tasksQueueOptions)) { throw new TypeError('Invalid tasks queue options: must be a plain object') @@ -312,28 +312,22 @@ export abstract class AbstractPool< `Invalid worker node tasks concurrency: ${tasksQueueOptions.concurrency} is a negative integer or zero` ) } - if ( - tasksQueueOptions?.queueMaxSize != null && - tasksQueueOptions?.size != null - ) { + if (tasksQueueOptions?.queueMaxSize != null) { throw new Error( - 'Invalid tasks queue options: cannot specify both queueMaxSize and size' + 'Invalid tasks queue options: queueMaxSize is deprecated, please use size instead' ) } - if (tasksQueueOptions?.queueMaxSize != null) { - tasksQueueOptions.size = tasksQueueOptions.queueMaxSize - } if ( tasksQueueOptions?.size != null && !Number.isSafeInteger(tasksQueueOptions.size) ) { throw new TypeError( - 'Invalid worker node tasks queue max size: must be an integer' + '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 max size: ${tasksQueueOptions.size} is a negative integer or zero` + `Invalid worker node tasks queue size: ${tasksQueueOptions.size} is a negative integer or zero` ) } } @@ -404,6 +398,13 @@ export abstract class AbstractPool< ...(this.opts.enableTasksQueue === true && { backPressure: this.hasBackPressure() }), + ...(this.opts.enableTasksQueue === true && { + stolenTasks: this.workerNodes.reduce( + (accumulator, workerNode) => + accumulator + workerNode.usage.tasks.stolen, + 0 + ) + }), failedTasks: this.workerNodes.reduce( (accumulator, workerNode) => accumulator + workerNode.usage.tasks.failed, @@ -426,24 +427,26 @@ export abstract class AbstractPool< ) ) ), - average: round( - this.workerNodes.reduce( - (accumulator, workerNode) => - accumulator + (workerNode.usage.runTime?.aggregate ?? 0), - 0 - ) / - this.workerNodes.reduce( - (accumulator, workerNode) => - accumulator + (workerNode.usage.tasks?.executed ?? 0), - 0 + ...(this.workerChoiceStrategyContext.getTaskStatisticsRequirements() + .runTime.average && { + average: round( + average( + this.workerNodes.reduce( + (accumulator, workerNode) => + accumulator.concat(workerNode.usage.runTime.history), + [] + ) ) - ), + ) + }), ...(this.workerChoiceStrategyContext.getTaskStatisticsRequirements() .runTime.median && { median: round( median( - this.workerNodes.map( - (workerNode) => workerNode.usage.runTime?.median ?? 0 + this.workerNodes.reduce( + (accumulator, workerNode) => + accumulator.concat(workerNode.usage.runTime.history), + [] ) ) ) @@ -467,24 +470,26 @@ export abstract class AbstractPool< ) ) ), - average: round( - this.workerNodes.reduce( - (accumulator, workerNode) => - accumulator + (workerNode.usage.waitTime?.aggregate ?? 0), - 0 - ) / - this.workerNodes.reduce( - (accumulator, workerNode) => - accumulator + (workerNode.usage.tasks?.executed ?? 0), - 0 + ...(this.workerChoiceStrategyContext.getTaskStatisticsRequirements() + .waitTime.average && { + average: round( + average( + this.workerNodes.reduce( + (accumulator, workerNode) => + accumulator.concat(workerNode.usage.waitTime.history), + [] + ) ) - ), + ) + }), ...(this.workerChoiceStrategyContext.getTaskStatisticsRequirements() .waitTime.median && { median: round( median( - this.workerNodes.map( - (workerNode) => workerNode.usage.waitTime?.median ?? 0 + this.workerNodes.reduce( + (accumulator, workerNode) => + accumulator.concat(workerNode.usage.waitTime.history), + [] ) ) ) @@ -728,6 +733,14 @@ export abstract class AbstractPool< return [] } + private shallExecuteTask (workerNodeKey: number): boolean { + return ( + this.tasksQueueSize(workerNodeKey) === 0 && + this.workerNodes[workerNodeKey].usage.tasks.executing < + (this.opts.tasksQueueOptions?.concurrency as number) + ) + } + /** @inheritDoc */ public async execute ( data?: Data, @@ -737,9 +750,11 @@ export abstract class AbstractPool< return await new Promise((resolve, reject) => { if (!this.started) { reject(new Error('Cannot execute a task on destroyed pool')) + return } if (name != null && typeof name !== 'string') { reject(new TypeError('name argument must be a string')) + return } if ( name != null && @@ -747,22 +762,15 @@ export abstract class AbstractPool< name.trim().length === 0 ) { reject(new TypeError('name argument must not be an empty string')) + return } if (transferList != null && !Array.isArray(transferList)) { reject(new TypeError('transferList argument must be an array')) + return } const timestamp = performance.now() const workerNodeKey = this.chooseWorkerNode() const workerInfo = this.getWorkerInfo(workerNodeKey) as WorkerInfo - if ( - name != null && - Array.isArray(workerInfo.taskFunctions) && - !workerInfo.taskFunctions.includes(name) - ) { - reject( - new Error(`Task function '${name}' is not registered in the pool`) - ) - } const task: Task = { name: name ?? DEFAULT_TASK_NAME, // eslint-disable-next-line @typescript-eslint/consistent-type-assertions @@ -780,8 +788,7 @@ export abstract class AbstractPool< if ( this.opts.enableTasksQueue === false || (this.opts.enableTasksQueue === true && - this.workerNodes[workerNodeKey].usage.tasks.executing < - (this.opts.tasksQueueOptions?.concurrency as number)) + this.shallExecuteTask(workerNodeKey)) ) { this.executeTask(workerNodeKey, task) } else { @@ -831,7 +838,7 @@ export abstract class AbstractPool< * @virtual */ protected setupHook (): void { - // Intentionally empty + /** Intentionally empty */ } /** @@ -940,11 +947,13 @@ export abstract class AbstractPool< workerUsage: WorkerUsage, message: MessageValue ): void { + if (message.taskError != null) { + return + } updateMeasurementStatistics( workerUsage.runTime, this.workerChoiceStrategyContext.getTaskStatisticsRequirements().runTime, - message.taskPerformance?.runTime ?? 0, - workerUsage.tasks.executed + message.taskPerformance?.runTime ?? 0 ) } @@ -957,8 +966,7 @@ export abstract class AbstractPool< updateMeasurementStatistics( workerUsage.waitTime, this.workerChoiceStrategyContext.getTaskStatisticsRequirements().waitTime, - taskWaitTime, - workerUsage.tasks.executed + taskWaitTime ) } @@ -966,19 +974,20 @@ export abstract class AbstractPool< workerUsage: WorkerUsage, message: MessageValue ): void { + if (message.taskError != null) { + return + } const eluTaskStatisticsRequirements: MeasurementStatisticsRequirements = this.workerChoiceStrategyContext.getTaskStatisticsRequirements().elu updateMeasurementStatistics( workerUsage.elu.active, eluTaskStatisticsRequirements, - message.taskPerformance?.elu?.active ?? 0, - workerUsage.tasks.executed + message.taskPerformance?.elu?.active ?? 0 ) updateMeasurementStatistics( workerUsage.elu.idle, eluTaskStatisticsRequirements, - message.taskPerformance?.elu?.idle ?? 0, - workerUsage.tasks.executed + message.taskPerformance?.elu?.idle ?? 0 ) if (eluTaskStatisticsRequirements.aggregate) { if (message.taskPerformance?.elu != null) { @@ -1156,6 +1165,8 @@ export abstract class AbstractPool< // Send the statistics message to worker. this.sendStatisticsMessageToWorker(workerNodeKey) if (this.opts.enableTasksQueue === true) { + this.workerNodes[workerNodeKey].onEmptyQueue = + this.taskStealingOnEmptyQueue.bind(this) this.workerNodes[workerNodeKey].onBackPressure = this.tasksStealingOnBackPressure.bind(this) } @@ -1187,75 +1198,124 @@ export abstract class AbstractPool< } private redistributeQueuedTasks (workerNodeKey: number): void { - const workerNodes = this.workerNodes.filter( - (_, workerNodeId) => workerNodeId !== workerNodeKey - ) while (this.tasksQueueSize(workerNodeKey) > 0) { - let targetWorkerNodeKey: number = workerNodeKey + let destinationWorkerNodeKey!: number let minQueuedTasks = Infinity - let executeTask = false - for (const [workerNodeId, workerNode] of workerNodes.entries()) { - if ( - workerNode.usage.tasks.executing < - (this.opts.tasksQueueOptions?.concurrency as number) - ) { - executeTask = true + for (const [workerNodeId, workerNode] of this.workerNodes.entries()) { + if (workerNode.info.ready && workerNodeId !== workerNodeKey) { + if (workerNode.usage.tasks.queued === 0) { + destinationWorkerNodeKey = workerNodeId + break + } + if (workerNode.usage.tasks.queued < minQueuedTasks) { + minQueuedTasks = workerNode.usage.tasks.queued + destinationWorkerNodeKey = workerNodeId + } } - if (workerNode.info.ready && workerNode.usage.tasks.queued === 0) { - targetWorkerNodeKey = workerNodeId - break + } + if (destinationWorkerNodeKey != null) { + const destinationWorkerNode = this.workerNodes[destinationWorkerNodeKey] + const task = { + ...(this.dequeueTask(workerNodeKey) as Task), + workerId: destinationWorkerNode.info.id as number } - if ( - workerNode.info.ready && - workerNode.usage.tasks.queued < minQueuedTasks - ) { - minQueuedTasks = workerNode.usage.tasks.queued - targetWorkerNodeKey = workerNodeId + if (this.shallExecuteTask(destinationWorkerNodeKey)) { + this.executeTask(destinationWorkerNodeKey, task) + } else { + this.enqueueTask(destinationWorkerNodeKey, task) } } - if (executeTask) { - this.executeTask( - targetWorkerNodeKey, - this.dequeueTask(workerNodeKey) as Task - ) - } else { - this.enqueueTask( - targetWorkerNodeKey, - this.dequeueTask(workerNodeKey) as Task + } + } + + private updateTaskStolenStatisticsWorkerUsage ( + workerNodeKey: number, + taskName: string + ): void { + const workerNode = this.workerNodes[workerNodeKey] + if (workerNode?.usage != null) { + ++workerNode.usage.tasks.stolen + } + if ( + this.shallUpdateTaskFunctionWorkerUsage(workerNodeKey) && + workerNode.getTaskFunctionWorkerUsage(taskName) != null + ) { + const taskFunctionWorkerUsage = workerNode.getTaskFunctionWorkerUsage( + taskName + ) as WorkerUsage + ++taskFunctionWorkerUsage.tasks.stolen + } + } + + private taskStealingOnEmptyQueue (workerId: number): void { + const destinationWorkerNodeKey = this.getWorkerNodeKeyByWorkerId(workerId) + const destinationWorkerNode = this.workerNodes[destinationWorkerNodeKey] + const workerNodes = this.workerNodes + .slice() + .sort( + (workerNodeA, workerNodeB) => + workerNodeB.usage.tasks.queued - workerNodeA.usage.tasks.queued + ) + for (const sourceWorkerNode of workerNodes) { + if (sourceWorkerNode.usage.tasks.queued === 0) { + break + } + if ( + sourceWorkerNode.info.ready && + sourceWorkerNode.info.id !== workerId && + sourceWorkerNode.usage.tasks.queued > 0 + ) { + const task = { + ...(sourceWorkerNode.popTask() as Task), + workerId: destinationWorkerNode.info.id as number + } + if (this.shallExecuteTask(destinationWorkerNodeKey)) { + this.executeTask(destinationWorkerNodeKey, task) + } else { + this.enqueueTask(destinationWorkerNodeKey, task) + } + this.updateTaskStolenStatisticsWorkerUsage( + destinationWorkerNodeKey, + task.name as string ) + break } } } private tasksStealingOnBackPressure (workerId: number): void { + if ((this.opts.tasksQueueOptions?.size as number) <= 1) { + return + } const sourceWorkerNode = this.workerNodes[this.getWorkerNodeKeyByWorkerId(workerId)] const workerNodes = this.workerNodes - .filter((workerNode) => workerNode.info.id !== workerId) + .slice() .sort( (workerNodeA, workerNodeB) => workerNodeA.usage.tasks.queued - workerNodeB.usage.tasks.queued ) for (const [workerNodeKey, workerNode] of workerNodes.entries()) { if ( - workerNode.info.ready && sourceWorkerNode.usage.tasks.queued > 0 && - !workerNode.hasBackPressure() + workerNode.info.ready && + workerNode.info.id !== workerId && + workerNode.usage.tasks.queued < + (this.opts.tasksQueueOptions?.size as number) - 1 ) { - if ( - workerNode.usage.tasks.executing < - (this.opts.tasksQueueOptions?.concurrency as number) - ) { - this.executeTask( - workerNodeKey, - sourceWorkerNode.popTask() as Task - ) + const task = { + ...(sourceWorkerNode.popTask() as Task), + workerId: workerNode.info.id as number + } + if (this.shallExecuteTask(workerNodeKey)) { + this.executeTask(workerNodeKey, task) } else { - this.enqueueTask( - workerNodeKey, - sourceWorkerNode.popTask() as Task - ) + this.enqueueTask(workerNodeKey, task) } + this.updateTaskStolenStatisticsWorkerUsage( + workerNodeKey, + task.name as string + ) } } } @@ -1367,7 +1427,6 @@ export abstract class AbstractPool< private addWorkerNode (worker: Worker): number { const workerNode = new WorkerNode( worker, - this.worker, this.opts.tasksQueueOptions?.size ?? Math.pow(this.maxSize, 2) ) // Flag the worker node as ready at pool startup.