X-Git-Url: https://git.piment-noir.org/?a=blobdiff_plain;f=src%2Fpools%2Fabstract-pool.ts;h=c56beaa6d801843c043f091963cea385b72eccee;hb=refs%2Ftags%2Fv2.7.4;hp=062df07791af8666fd94282eac229aef466c7a1b;hpb=15872edc9845cab9ad28f47c669fb95fe15ae0a3;p=poolifier.git diff --git a/src/pools/abstract-pool.ts b/src/pools/abstract-pool.ts index 062df077..c56beaa6 100644 --- a/src/pools/abstract-pool.ts +++ b/src/pools/abstract-pool.ts @@ -1,12 +1,11 @@ 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 { type EventEmitter, EventEmitterAsyncResource } from 'node:events' import type { MessageValue, PromiseResponseWrapper, - Task, - Writable + Task } from '../utility-types' import { DEFAULT_TASK_NAME, @@ -15,14 +14,15 @@ import { average, isKillBehavior, isPlainObject, + max, median, - round, - updateMeasurementStatistics + min, + 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, @@ -47,6 +47,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. @@ -64,11 +70,10 @@ export abstract class AbstractPool< public readonly workerNodes: Array> = [] /** @inheritDoc */ - public readonly emitter?: PoolEmitter + public emitter?: EventEmitter | EventEmitterAsyncResource /** - * The task execution response promise map. - * + * The task execution response promise map: * - `key`: The message id of each submitted task. * - `value`: An object that contains the worker, the execution response promise resolve and reject callbacks. * @@ -92,13 +97,20 @@ export abstract class AbstractPool< protected readonly max?: number /** - * Whether the pool is starting or not. + * The task functions added at runtime map: + * - `key`: The task function name. + * - `value`: The task function itself. */ - private readonly starting: boolean + private readonly taskFunctions: Map> + /** * Whether the pool is started or not. */ private started: boolean + /** + * Whether the pool is starting or not. + */ + private starting: boolean /** * The start timestamp of the pool. */ @@ -121,8 +133,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) @@ -130,7 +142,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, @@ -144,27 +156,17 @@ export abstract class AbstractPool< this.setupHook() - this.starting = true - this.startPool() + this.taskFunctions = new Map>() + + this.started = false this.starting = false - this.started = true + if (this.opts.startWorkers === true) { + this.start() + } 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( @@ -183,51 +185,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 ) @@ -237,42 +214,35 @@ 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.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 ( - workerChoiceStrategyOptions.weights != null && + workerChoiceStrategyOptions?.weights != null && Object.keys(workerChoiceStrategyOptions.weights).length !== this.maxSize ) { throw new Error( @@ -280,7 +250,7 @@ export abstract class AbstractPool< ) } if ( - workerChoiceStrategyOptions.measurement != null && + workerChoiceStrategyOptions?.measurement != null && !Object.values(Measurements).includes( workerChoiceStrategyOptions.measurement ) @@ -291,64 +261,10 @@ export abstract class AbstractPool< } } - private checkValidTasksQueueOptions ( - tasksQueueOptions: Writable - ): 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?.queueMaxSize != null && - tasksQueueOptions?.size != null - ) { - throw new Error( - 'Invalid tasks queue options: cannot specify both queueMaxSize and size' - ) - } - 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' - ) - } - 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` - ) - } - } - - private startPool (): void { - while ( - this.workerNodes.reduce( - (accumulator, workerNode) => - !workerNode.info.dynamic ? accumulator + 1 : accumulator, - 0 - ) < this.numberOfWorkers - ) { - this.createAndSetupWorkerNode() - } + private initializeEventEmitter (): void { + this.emitter = new EventEmitterAsyncResource({ + name: `poolifier:${this.type}-${this.worker}-pool` + }) } /** @inheritDoc */ @@ -357,6 +273,7 @@ export abstract class AbstractPool< version, type: this.type, worker: this.worker, + started: this.started, ready: this.ready, strategy: this.opts.workerChoiceStrategy as WorkerChoiceStrategy, minSize: this.minSize, @@ -405,6 +322,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, @@ -414,16 +338,16 @@ export abstract class AbstractPool< .runTime.aggregate && { runTime: { minimum: round( - Math.min( + min( ...this.workerNodes.map( - (workerNode) => workerNode.usage.runTime?.minimum ?? Infinity + workerNode => workerNode.usage.runTime?.minimum ?? Infinity ) ) ), maximum: round( - Math.max( + max( ...this.workerNodes.map( - (workerNode) => workerNode.usage.runTime?.maximum ?? -Infinity + workerNode => workerNode.usage.runTime?.maximum ?? -Infinity ) ) ), @@ -457,16 +381,16 @@ export abstract class AbstractPool< .waitTime.aggregate && { waitTime: { minimum: round( - Math.min( + min( ...this.workerNodes.map( - (workerNode) => workerNode.usage.waitTime?.minimum ?? Infinity + workerNode => workerNode.usage.waitTime?.minimum ?? Infinity ) ) ), maximum: round( - Math.max( + max( ...this.workerNodes.map( - (workerNode) => workerNode.usage.waitTime?.maximum ?? -Infinity + workerNode => workerNode.usage.waitTime?.maximum ?? -Infinity ) ) ), @@ -567,7 +491,7 @@ 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 ( @@ -588,7 +512,7 @@ export abstract class AbstractPool< */ private getWorkerNodeKeyByWorker (worker: Worker): number { return this.workerNodes.findIndex( - (workerNode) => workerNode.worker === worker + workerNode => workerNode.worker === worker ) } @@ -598,9 +522,9 @@ export abstract class AbstractPool< * @param workerId - The worker id. * @returns The worker node key if the worker id is found in the pool worker nodes, `-1` otherwise. */ - private getWorkerNodeKeyByWorkerId (workerId: number): number { + private getWorkerNodeKeyByWorkerId (workerId: number | undefined): number { return this.workerNodes.findIndex( - (workerNode) => workerNode.info.id === workerId + workerNode => workerNode.info.id === workerId ) } @@ -609,7 +533,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 @@ -643,6 +567,8 @@ export abstract class AbstractPool< tasksQueueOptions?: TasksQueueOptions ): void { if (this.opts.enableTasksQueue === true && !enable) { + this.unsetTaskStealing() + this.unsetTasksStealingOnBackPressure() this.flushTasksQueues() } this.opts.enableTasksQueue = enable @@ -652,33 +578,71 @@ 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.setTasksQueueMaxSize(this.opts.tasksQueueOptions.size as number) + 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 setTasksQueueMaxSize (size: number): void { - for (const workerNode of this.workerNodes) { - workerNode.tasksQueueBackPressureSize = size - } - } - private buildTasksQueueOptions ( tasksQueueOptions: TasksQueueOptions ): TasksQueueOptions { return { ...{ size: Math.pow(this.maxSize, 2), - concurrency: 1 + concurrency: 1, + taskStealing: true, + tasksStealingOnBackPressure: true }, ...tasksQueueOptions } } + 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].onEmptyQueue = + this.taskStealingOnEmptyQueue.bind(this) + } + } + + private unsetTaskStealing (): void { + for (const [workerNodeKey] of this.workerNodes.entries()) { + delete this.workerNodes[workerNodeKey].onEmptyQueue + } + } + + private setTasksStealingOnBackPressure (): void { + for (const [workerNodeKey] of this.workerNodes.entries()) { + this.workerNodes[workerNodeKey].onBackPressure = + this.tasksStealingOnBackPressure.bind(this) + } + } + + private unsetTasksStealingOnBackPressure (): void { + for (const [workerNodeKey] of this.workerNodes.entries()) { + delete this.workerNodes[workerNodeKey].onBackPressure + } + } + /** * Whether the pool is full or not. * @@ -704,35 +668,204 @@ export abstract class AbstractPool< if (this.opts.enableTasksQueue === true) { return ( this.workerNodes.findIndex( - (workerNode) => + workerNode => workerNode.info.ready && workerNode.usage.tasks.executing < (this.opts.tasksQueueOptions?.concurrency as number) ) === -1 ) - } else { - return ( - this.workerNodes.findIndex( - (workerNode) => - workerNode.info.ready && workerNode.usage.tasks.executing === 0 - ) === -1 + } + return ( + this.workerNodes.findIndex( + workerNode => + workerNode.info.ready && workerNode.usage.tasks.executing === 0 + ) === -1 + ) + } + + private async sendTaskFunctionOperationToWorker ( + workerNodeKey: number, + message: MessageValue + ): Promise { + return await new Promise((resolve, reject) => { + const taskFunctionOperationListener = ( + message: MessageValue + ): void => { + this.checkMessageWorkerId(message) + const workerId = this.getWorkerInfo(workerNodeKey).id as number + if ( + message.taskFunctionOperationStatus != null && + message.workerId === 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: MessageValue + ): Promise { + return await new Promise((resolve, reject) => { + 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.every( + message => message.taskFunctionOperationStatus === true + ) + ) { + resolve(true) + } else if ( + responsesReceived.some( + message => message.taskFunctionOperationStatus === false + ) + ) { + const errorResponse = responsesReceived.find( + response => response.taskFunctionOperationStatus === false + ) + reject( + new Error( + `Task function operation '${ + message.taskFunctionOperation as string + }' 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) + } + }) + } + + /** @inheritDoc */ + public hasTaskFunction (name: string): boolean { + for (const workerNode of this.workerNodes) { + if ( + Array.isArray(workerNode.info.taskFunctionNames) && + workerNode.info.taskFunctionNames.includes(name) + ) { + return true + } + } + return false + } + + /** @inheritDoc */ + public async addTaskFunction ( + name: string, + fn: TaskFunction + ): Promise { + 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: fn.toString() + }) + this.taskFunctions.set(name, fn) + return opResult + } + + /** @inheritDoc */ + public async removeTaskFunction (name: string): Promise { + 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 */ - public listTaskFunctions (): string[] { + public listTaskFunctionNames (): string[] { for (const workerNode of this.workerNodes) { if ( - Array.isArray(workerNode.info.taskFunctions) && - workerNode.info.taskFunctions.length > 0 + Array.isArray(workerNode.info.taskFunctionNames) && + workerNode.info.taskFunctionNames.length > 0 ) { - return workerNode.info.taskFunctions + return workerNode.info.taskFunctionNames } } return [] } + /** @inheritDoc */ + public async setDefaultTaskFunction (name: string): Promise { + return await this.sendTaskFunctionOperationToWorkers({ + taskFunctionOperation: 'default', + taskFunctionName: name + }) + } + + private deleteTaskFunctionWorkerUsages (name: string): void { + for (const workerNode of this.workerNodes) { + workerNode.deleteTaskFunctionWorkerUsage(name) + } + } + + 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, @@ -741,10 +874,12 @@ export abstract class AbstractPool< ): Promise { return await new Promise((resolve, reject) => { if (!this.started) { - reject(new Error('Cannot execute a task on destroyed pool')) + reject(new Error('Cannot execute a task on not started pool')) + return } if (name != null && typeof name !== 'string') { reject(new TypeError('name argument must be a string')) + return } if ( name != null && @@ -752,29 +887,20 @@ 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 data: data ?? ({} as Data), transferList, timestamp, - workerId: workerInfo.id as number, taskId: randomUUID() } this.promiseResponseMap.set(task.taskId as string, { @@ -785,9 +911,7 @@ export abstract class AbstractPool< if ( this.opts.enableTasksQueue === false || (this.opts.enableTasksQueue === true && - this.tasksQueueSize(workerNodeKey) === 0 && - this.workerNodes[workerNodeKey].usage.tasks.executing < - (this.opts.tasksQueueOptions?.concurrency as number)) + this.shallExecuteTask(workerNodeKey)) ) { this.executeTask(workerNodeKey, task) } else { @@ -796,6 +920,22 @@ export abstract class AbstractPool< }) } + /** @inheritdoc */ + public start (): void { + this.starting = true + while ( + this.workerNodes.reduce( + (accumulator, workerNode) => + !workerNode.info.dynamic ? accumulator + 1 : accumulator, + 0 + ) < this.numberOfWorkers + ) { + this.createAndSetupWorkerNode() + } + this.starting = false + this.started = true + } + /** @inheritDoc */ public async destroy (): Promise { await Promise.all( @@ -804,22 +944,32 @@ export abstract class AbstractPool< }) ) this.emitter?.emit(PoolEvents.destroy, this.info) + if (this.emitter instanceof EventEmitterAsyncResource) { + this.emitter?.emitDestroy() + } this.started = false } protected async sendKillMessageToWorker ( - workerNodeKey: number, - workerId: number + 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 ${workerId} kill message handling failed`)) + reject( + new Error( + `Kill message handling failed on worker ${ + message.workerId as number + }` + ) + ) } - }) - this.sendToWorker(workerNodeKey, { kill: true, workerId }) + } + this.registerWorkerMessageListener(workerNodeKey, killMessageListener) + this.sendToWorker(workerNodeKey, { kill: true }) }) } @@ -837,7 +987,7 @@ export abstract class AbstractPool< * @virtual */ protected setupHook (): void { - /** Intentionally empty */ + /* Intentionally empty */ } /** @@ -919,8 +1069,8 @@ export abstract class AbstractPool< const workerInfo = this.getWorkerInfo(workerNodeKey) return ( workerInfo != null && - Array.isArray(workerInfo.taskFunctions) && - workerInfo.taskFunctions.length > 2 + Array.isArray(workerInfo.taskFunctionNames) && + workerInfo.taskFunctionNames.length > 2 ) } @@ -935,7 +1085,7 @@ export abstract class AbstractPool< ) { --workerTaskStatistics.executing } - if (message.taskError == null) { + if (message.workerError == null) { ++workerTaskStatistics.executed } else { ++workerTaskStatistics.failed @@ -946,7 +1096,7 @@ export abstract class AbstractPool< workerUsage: WorkerUsage, message: MessageValue ): void { - if (message.taskError != null) { + if (message.workerError != null) { return } updateMeasurementStatistics( @@ -973,7 +1123,7 @@ export abstract class AbstractPool< workerUsage: WorkerUsage, message: MessageValue ): void { - if (message.taskError != null) { + if (message.workerError != null) { return } const eluTaskStatisticsRequirements: MeasurementStatisticsRequirements = @@ -1061,16 +1211,16 @@ export abstract class AbstractPool< worker.on('online', this.opts.onlineHandler ?? EMPTY_FUNCTION) worker.on('message', this.opts.messageHandler ?? EMPTY_FUNCTION) worker.on('error', this.opts.errorHandler ?? EMPTY_FUNCTION) - worker.on('error', (error) => { + worker.on('error', error => { const workerNodeKey = this.getWorkerNodeKeyByWorker(worker) - const workerInfo = this.getWorkerInfo(workerNodeKey) as WorkerInfo + const workerInfo = this.getWorkerInfo(workerNodeKey) workerInfo.ready = false this.workerNodes[workerNodeKey].closeChannel() this.emitter?.emit(PoolEvents.error, error) if ( - this.opts.restartWorkerOnError === true && + this.started && !this.starting && - this.started + this.opts.restartWorkerOnError === true ) { if (workerInfo.dynamic) { this.createAndSetupDynamicWorkerNode() @@ -1078,7 +1228,7 @@ export abstract class AbstractPool< this.createAndSetupWorkerNode() } } - if (this.opts.enableTasksQueue === true) { + if (this.started && this.opts.enableTasksQueue === true) { this.redistributeQueuedTasks(workerNodeKey) } }) @@ -1101,7 +1251,8 @@ export abstract class AbstractPool< */ protected createAndSetupDynamicWorkerNode (): number { const workerNodeKey = this.createAndSetupWorkerNode() - this.registerWorkerMessageListener(workerNodeKey, (message) => { + this.registerWorkerMessageListener(workerNodeKey, message => { + this.checkMessageWorkerId(message) const localWorkerNodeKey = this.getWorkerNodeKeyByWorkerId( message.workerId ) @@ -1116,16 +1267,26 @@ export abstract class AbstractPool< workerUsage.tasks.executing === 0 && this.tasksQueueSize(localWorkerNodeKey) === 0))) ) { - this.destroyWorkerNode(localWorkerNodeKey).catch((error) => { + this.destroyWorkerNode(localWorkerNodeKey).catch(error => { this.emitter?.emit(PoolEvents.error, error) }) } }) - const workerInfo = this.getWorkerInfo(workerNodeKey) as WorkerInfo + const workerInfo = this.getWorkerInfo(workerNodeKey) this.sendToWorker(workerNodeKey, { - checkActive: true, - workerId: workerInfo.id as number + checkActive: true }) + if (this.taskFunctions.size > 0) { + for (const [taskFunctionName, taskFunction] of this.taskFunctions) { + this.sendTaskFunctionOperationToWorker(workerNodeKey, { + taskFunctionOperation: 'add', + taskFunctionName, + taskFunction: taskFunction.toString() + }).catch(error => { + this.emitter?.emit(PoolEvents.error, error) + }) + } + } workerInfo.dynamic = true if ( this.workerChoiceStrategyContext.getStrategyPolicy().dynamicWorkerReady || @@ -1150,6 +1311,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. @@ -1164,10 +1351,14 @@ 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) + if (this.opts.tasksQueueOptions?.taskStealing === true) { + this.workerNodes[workerNodeKey].onEmptyQueue = + this.taskStealingOnEmptyQueue.bind(this) + } + if (this.opts.tasksQueueOptions?.tasksStealingOnBackPressure === true) { + this.workerNodes[workerNodeKey].onBackPressure = + this.tasksStealingOnBackPressure.bind(this) + } } } @@ -1191,83 +1382,83 @@ export abstract class AbstractPool< .runTime.aggregate, elu: this.workerChoiceStrategyContext.getTaskStatisticsRequirements() .elu.aggregate - }, - workerId: (this.getWorkerInfo(workerNodeKey) as WorkerInfo).id as number + } }) } private redistributeQueuedTasks (workerNodeKey: number): void { while (this.tasksQueueSize(workerNodeKey) > 0) { - let destinationWorkerNodeKey!: number - let minQueuedTasks = Infinity - 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 (destinationWorkerNodeKey != null) { - const destinationWorkerNode = this.workerNodes[destinationWorkerNodeKey] - const task = { - ...(this.dequeueTask(workerNodeKey) as Task), - workerId: destinationWorkerNode.info.id as number - } - if ( - this.tasksQueueSize(destinationWorkerNodeKey) === 0 && - destinationWorkerNode.usage.tasks.executing < - (this.opts.tasksQueueOptions?.concurrency as number) - ) { - this.executeTask(destinationWorkerNodeKey, task) - } else { - this.enqueueTask(destinationWorkerNodeKey, task) - } + const destinationWorkerNodeKey = this.workerNodes.reduce( + (minWorkerNodeKey, workerNode, workerNodeKey, workerNodes) => { + return workerNode.info.ready && + workerNode.usage.tasks.queued < + workerNodes[minWorkerNodeKey].usage.tasks.queued + ? workerNodeKey + : minWorkerNodeKey + }, + 0 + ) + const task = this.dequeueTask(workerNodeKey) as Task + if (this.shallExecuteTask(destinationWorkerNodeKey)) { + this.executeTask(destinationWorkerNodeKey, task) + } else { + this.enqueueTask(destinationWorkerNodeKey, 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.tasksQueueSize(destinationWorkerNodeKey) === 0 && - destinationWorkerNode.usage.tasks.executing < - (this.opts.tasksQueueOptions?.concurrency as number) - ) { - this.executeTask(destinationWorkerNodeKey, task) - } else { - this.enqueueTask(destinationWorkerNodeKey, task) - } - break + const sourceWorkerNode = workerNodes.find( + workerNode => + workerNode.info.ready && + workerNode.info.id !== workerId && + workerNode.usage.tasks.queued > 0 + ) + if (sourceWorkerNode != null) { + const task = sourceWorkerNode.popTask() as Task + if (this.shallExecuteTask(destinationWorkerNodeKey)) { + this.executeTask(destinationWorkerNodeKey, task) + } else { + this.enqueueTask(destinationWorkerNodeKey, task) } + this.updateTaskStolenStatisticsWorkerUsage( + destinationWorkerNodeKey, + task.name as string + ) } } private tasksStealingOnBackPressure (workerId: number): void { + const sizeOffset = 1 + if ((this.opts.tasksQueueOptions?.size as number) <= sizeOffset) { + return + } const sourceWorkerNode = this.workerNodes[this.getWorkerNodeKeyByWorkerId(workerId)] const workerNodes = this.workerNodes @@ -1282,21 +1473,18 @@ export abstract class AbstractPool< workerNode.info.ready && workerNode.info.id !== workerId && workerNode.usage.tasks.queued < - (this.opts.tasksQueueOptions?.size as number) - 1 + (this.opts.tasksQueueOptions?.size as number) - sizeOffset ) { - const task = { - ...(sourceWorkerNode.popTask() as Task), - workerId: workerNode.info.id as number - } - if ( - this.tasksQueueSize(workerNodeKey) === 0 && - workerNode.usage.tasks.executing < - (this.opts.tasksQueueOptions?.concurrency as number) - ) { + const task = sourceWorkerNode.popTask() as Task + if (this.shallExecuteTask(workerNodeKey)) { this.executeTask(workerNodeKey, task) } else { this.enqueueTask(workerNodeKey, task) } + this.updateTaskStolenStatisticsWorkerUsage( + workerNodeKey, + task.name as string + ) } } } @@ -1307,51 +1495,52 @@ export abstract class AbstractPool< * @returns The listener function to execute when a message is received from a worker. */ protected workerListener (): (message: MessageValue) => void { - return (message) => { + return message => { this.checkMessageWorkerId(message) - if (message.ready != null && message.taskFunctions != null) { + 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.taskFunctions != null) { - // Task functions message received from worker - ( - this.getWorkerInfo( - this.getWorkerNodeKeyByWorkerId(message.workerId) - ) as WorkerInfo - ).taskFunctions = message.taskFunctions + } else if (message.taskFunctionNames != null) { + // Task function names message received from worker + this.getWorkerInfo( + this.getWorkerNodeKeyByWorkerId(message.workerId) + ).taskFunctionNames = message.taskFunctionNames } } } private handleWorkerReadyResponse (message: MessageValue): void { if (message.ready === false) { - throw new Error(`Worker ${message.workerId} failed to initialize`) + throw new Error( + `Worker ${message.workerId as number} failed to initialize` + ) } const workerInfo = this.getWorkerInfo( this.getWorkerNodeKeyByWorkerId(message.workerId) - ) as WorkerInfo + ) workerInfo.ready = message.ready as boolean - workerInfo.taskFunctions = message.taskFunctions - if (this.emitter != null && this.ready) { - this.emitter.emit(PoolEvents.ready, this.info) + workerInfo.taskFunctionNames = message.taskFunctionNames + if (this.ready) { + this.emitter?.emit(PoolEvents.ready, this.info) } } private handleTaskExecutionResponse (message: MessageValue): void { - const { taskId, taskError, data } = message + const { taskId, workerError, data } = message const promiseResponse = this.promiseResponseMap.get(taskId as string) if (promiseResponse != null) { - if (taskError != null) { - this.emitter?.emit(PoolEvents.taskError, taskError) - promiseResponse.reject(taskError.message) + if (workerError != null) { + this.emitter?.emit(PoolEvents.taskError, workerError) + promiseResponse.reject(workerError.message) } else { promiseResponse.resolve(data as Response) } const workerNodeKey = promiseResponse.workerNodeKey this.afterTaskExecutionHook(workerNodeKey, message) + this.workerChoiceStrategyContext.update(workerNodeKey) this.promiseResponseMap.delete(taskId as string) if ( this.opts.enableTasksQueue === true && @@ -1364,7 +1553,6 @@ export abstract class AbstractPool< this.dequeueTask(workerNodeKey) as Task ) } - this.workerChoiceStrategyContext.update(workerNodeKey) } } @@ -1394,8 +1582,8 @@ export abstract class AbstractPool< * @param workerNodeKey - The worker node key. * @returns The worker information. */ - protected getWorkerInfo (workerNodeKey: number): WorkerInfo | undefined { - return this.workerNodes[workerNodeKey]?.info + protected getWorkerInfo (workerNodeKey: number): WorkerInfo { + return this.workerNodes[workerNodeKey].info } /** @@ -1408,7 +1596,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. @@ -1418,7 +1605,7 @@ export abstract class AbstractPool< this.workerNodes.push(workerNode) const workerNodeKey = this.getWorkerNodeKeyByWorker(worker) if (workerNodeKey === -1) { - throw new Error('Worker node added not found') + throw new Error('Worker added not found in worker nodes') } return workerNodeKey } @@ -1448,7 +1635,7 @@ export abstract class AbstractPool< return ( this.opts.enableTasksQueue === true && this.workerNodes.findIndex( - (workerNode) => !workerNode.hasBackPressure() + workerNode => !workerNode.hasBackPressure() ) === -1 ) }