X-Git-Url: https://git.piment-noir.org/?a=blobdiff_plain;f=src%2Fpools%2Fabstract-pool.ts;h=cbc24e7d84c65c3313f0e27aea9f927ea7bfe543;hb=660689f017615c233020793967822c31975e5aa0;hp=aae083953c526c16f7bcaa92bddd3bf042423d3e;hpb=22c85455d2f30345148aa3e1304d3b0840102f72;p=poolifier.git diff --git a/src/pools/abstract-pool.ts b/src/pools/abstract-pool.ts index aae08395..cbc24e7d 100644 --- a/src/pools/abstract-pool.ts +++ b/src/pools/abstract-pool.ts @@ -68,8 +68,7 @@ export abstract class AbstractPool< public readonly emitter?: PoolEmitter /** - * 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,14 +91,14 @@ export abstract class AbstractPool< */ protected readonly max?: number - /** - * Whether the pool is starting or not. - */ - private readonly starting: boolean /** * 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. */ @@ -145,10 +144,11 @@ export abstract class AbstractPool< this.setupHook() - this.starting = true - this.startPool() + this.started = false this.starting = false - this.started = true + if (this.opts.startWorkers === true) { + this.start() + } this.startTimestamp = performance.now() } @@ -212,16 +212,19 @@ export abstract class AbstractPool< private checkPoolOptions (opts: PoolOptions): void { if (isPlainObject(opts)) { + this.opts.startWorkers = opts.startWorkers ?? true + this.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 @@ -241,7 +244,10 @@ export abstract class AbstractPool< private checkValidWorkerChoiceStrategy ( workerChoiceStrategy: WorkerChoiceStrategy ): void { - if (!Object.values(WorkerChoiceStrategies).includes(workerChoiceStrategy)) { + if ( + workerChoiceStrategy != null && + !Object.values(WorkerChoiceStrategies).includes(workerChoiceStrategy) + ) { throw new Error( `Invalid worker choice strategy '${workerChoiceStrategy}'` ) @@ -251,13 +257,16 @@ export abstract class AbstractPool< 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.retries != null && + workerChoiceStrategyOptions?.retries != null && !Number.isSafeInteger(workerChoiceStrategyOptions.retries) ) { throw new TypeError( @@ -265,7 +274,7 @@ export abstract class AbstractPool< ) } if ( - workerChoiceStrategyOptions.retries != null && + workerChoiceStrategyOptions?.retries != null && workerChoiceStrategyOptions.retries < 0 ) { throw new RangeError( @@ -273,7 +282,7 @@ export abstract class AbstractPool< ) } if ( - workerChoiceStrategyOptions.weights != null && + workerChoiceStrategyOptions?.weights != null && Object.keys(workerChoiceStrategyOptions.weights).length !== this.maxSize ) { throw new Error( @@ -281,7 +290,7 @@ export abstract class AbstractPool< ) } if ( - workerChoiceStrategyOptions.measurement != null && + workerChoiceStrategyOptions?.measurement != null && !Object.values(Measurements).includes( workerChoiceStrategyOptions.measurement ) @@ -300,7 +309,7 @@ export abstract class AbstractPool< } if ( tasksQueueOptions?.concurrency != null && - !Number.isSafeInteger(tasksQueueOptions?.concurrency) + !Number.isSafeInteger(tasksQueueOptions.concurrency) ) { throw new TypeError( 'Invalid worker node tasks concurrency: must be an integer' @@ -308,50 +317,34 @@ export abstract class AbstractPool< } if ( tasksQueueOptions?.concurrency != null && - tasksQueueOptions?.concurrency <= 0 + tasksQueueOptions.concurrency <= 0 ) { throw new RangeError( - `Invalid worker node tasks concurrency: ${tasksQueueOptions?.concurrency} is a negative integer or zero` - ) - } - if (tasksQueueOptions?.queueMaxSize != null) { - throw new Error( - 'Invalid tasks queue options: queueMaxSize is deprecated, please use size instead' + `Invalid worker node tasks concurrency: ${tasksQueueOptions.concurrency} is a negative integer or zero` ) } if ( tasksQueueOptions?.size != null && - !Number.isSafeInteger(tasksQueueOptions?.size) + !Number.isSafeInteger(tasksQueueOptions.size) ) { throw new TypeError( 'Invalid worker node tasks queue size: must be an integer' ) } - if (tasksQueueOptions?.size != null && tasksQueueOptions?.size <= 0) { + if (tasksQueueOptions?.size != null && tasksQueueOptions.size <= 0) { throw new RangeError( - `Invalid worker node tasks queue size: ${tasksQueueOptions?.size} is a negative integer or zero` + `Invalid worker node tasks queue 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() - } - } - /** @inheritDoc */ public get info (): PoolInfo { return { version, type: this.type, worker: this.worker, + started: this.started, ready: this.ready, strategy: this.opts.workerChoiceStrategy as WorkerChoiceStrategy, minSize: this.minSize, @@ -418,14 +411,14 @@ export abstract class AbstractPool< minimum: round( min( ...this.workerNodes.map( - (workerNode) => workerNode.usage.runTime?.minimum ?? Infinity + workerNode => workerNode.usage.runTime?.minimum ?? Infinity ) ) ), maximum: round( max( ...this.workerNodes.map( - (workerNode) => workerNode.usage.runTime?.maximum ?? -Infinity + workerNode => workerNode.usage.runTime?.maximum ?? -Infinity ) ) ), @@ -461,14 +454,14 @@ export abstract class AbstractPool< minimum: round( min( ...this.workerNodes.map( - (workerNode) => workerNode.usage.waitTime?.minimum ?? Infinity + workerNode => workerNode.usage.waitTime?.minimum ?? Infinity ) ) ), maximum: round( max( ...this.workerNodes.map( - (workerNode) => workerNode.usage.waitTime?.maximum ?? -Infinity + workerNode => workerNode.usage.waitTime?.maximum ?? -Infinity ) ) ), @@ -590,7 +583,7 @@ export abstract class AbstractPool< */ private getWorkerNodeKeyByWorker (worker: Worker): number { return this.workerNodes.findIndex( - (workerNode) => workerNode.worker === worker + workerNode => workerNode.worker === worker ) } @@ -602,7 +595,7 @@ export abstract class AbstractPool< */ private getWorkerNodeKeyByWorkerId (workerId: number): number { return this.workerNodes.findIndex( - (workerNode) => workerNode.info.id === workerId + workerNode => workerNode.info.id === workerId ) } @@ -645,6 +638,8 @@ export abstract class AbstractPool< tasksQueueOptions?: TasksQueueOptions ): void { if (this.opts.enableTasksQueue === true && !enable) { + this.unsetTaskStealing() + this.unsetTasksStealingOnBackPressure() this.flushTasksQueues() } this.opts.enableTasksQueue = enable @@ -658,29 +653,67 @@ export abstract class AbstractPool< this.opts.tasksQueueOptions = this.buildTasksQueueOptions(tasksQueueOptions) 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 setTasksQueueSize (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. * @@ -706,20 +739,19 @@ 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 + ) } /** @inheritDoc */ @@ -751,7 +783,7 @@ 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') { @@ -772,14 +804,13 @@ export abstract class AbstractPool< } const timestamp = performance.now() const workerNodeKey = this.chooseWorkerNode() - const workerInfo = this.getWorkerInfo(workerNodeKey) 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, + workerId: this.getWorkerInfo(workerNodeKey).id as number, taskId: randomUUID() } this.promiseResponseMap.set(task.taskId as string, { @@ -799,6 +830,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( @@ -815,7 +862,7 @@ export abstract class AbstractPool< workerId: number ): Promise { await new Promise((resolve, reject) => { - this.registerWorkerMessageListener(workerNodeKey, (message) => { + this.registerWorkerMessageListener(workerNodeKey, message => { if (message.kill === 'success') { resolve() } else if (message.kill === 'failure') { @@ -1064,16 +1111,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) 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() @@ -1081,7 +1128,7 @@ export abstract class AbstractPool< this.createAndSetupWorkerNode() } } - if (this.opts.enableTasksQueue === true) { + if (this.started && this.opts.enableTasksQueue === true) { this.redistributeQueuedTasks(workerNodeKey) } }) @@ -1104,7 +1151,7 @@ export abstract class AbstractPool< */ protected createAndSetupDynamicWorkerNode (): number { const workerNodeKey = this.createAndSetupWorkerNode() - this.registerWorkerMessageListener(workerNodeKey, (message) => { + this.registerWorkerMessageListener(workerNodeKey, message => { const localWorkerNodeKey = this.getWorkerNodeKeyByWorkerId( message.workerId ) @@ -1119,7 +1166,7 @@ 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) }) } @@ -1167,10 +1214,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) + } } } @@ -1203,27 +1254,23 @@ export abstract class AbstractPool< while (this.tasksQueueSize(workerNodeKey) > 0) { const destinationWorkerNodeKey = this.workerNodes.reduce( (minWorkerNodeKey, workerNode, workerNodeKey, workerNodes) => { - if (!workerNode.info.ready) { - return minWorkerNodeKey - } - return workerNode.usage.tasks.queued < - workerNodes[minWorkerNodeKey].usage.tasks.queued + return workerNode.info.ready && + workerNode.usage.tasks.queued < + workerNodes[minWorkerNodeKey].usage.tasks.queued ? workerNodeKey : minWorkerNodeKey }, 0 ) - if (destinationWorkerNodeKey != null) { - const destinationWorkerNode = this.workerNodes[destinationWorkerNodeKey] - const task = { - ...(this.dequeueTask(workerNodeKey) as Task), - workerId: destinationWorkerNode.info.id as number - } - if (this.shallExecuteTask(destinationWorkerNodeKey)) { - this.executeTask(destinationWorkerNodeKey, task) - } else { - this.enqueueTask(destinationWorkerNodeKey, task) - } + const destinationWorkerNode = this.workerNodes[destinationWorkerNodeKey] + const task = { + ...(this.dequeueTask(workerNodeKey) as Task), + workerId: destinationWorkerNode.info.id as number + } + if (this.shallExecuteTask(destinationWorkerNodeKey)) { + this.executeTask(destinationWorkerNodeKey, task) + } else { + this.enqueueTask(destinationWorkerNodeKey, task) } } } @@ -1257,7 +1304,7 @@ export abstract class AbstractPool< workerNodeB.usage.tasks.queued - workerNodeA.usage.tasks.queued ) const sourceWorkerNode = workerNodes.find( - (workerNode) => + workerNode => workerNode.info.ready && workerNode.info.id !== workerId && workerNode.usage.tasks.queued > 0 @@ -1323,7 +1370,7 @@ 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) { // Worker ready response received from worker @@ -1349,8 +1396,8 @@ export abstract class AbstractPool< ) workerInfo.ready = message.ready as boolean workerInfo.taskFunctions = message.taskFunctions - if (this.emitter != null && this.ready) { - this.emitter.emit(PoolEvents.ready, this.info) + if (this.ready) { + this.emitter?.emit(PoolEvents.ready, this.info) } } @@ -1461,7 +1508,7 @@ export abstract class AbstractPool< return ( this.opts.enableTasksQueue === true && this.workerNodes.findIndex( - (workerNode) => !workerNode.hasBackPressure() + workerNode => !workerNode.hasBackPressure() ) === -1 ) }