X-Git-Url: https://git.piment-noir.org/?a=blobdiff_plain;f=src%2Fpools%2Fabstract-pool.ts;h=91cb5facd01f736d43703d4d0530ec7c01a3bf41;hb=3d76750ac8dacd95003ca89cf9542c3b2a014b8c;hp=0b359d93b6006c124c692fda70f9fb21926691f4;hpb=10ecf8fd5f751ffb5477284ae1b6935f3c81ec2d;p=poolifier.git diff --git a/src/pools/abstract-pool.ts b/src/pools/abstract-pool.ts index 0b359d93..91cb5fac 100644 --- a/src/pools/abstract-pool.ts +++ b/src/pools/abstract-pool.ts @@ -10,7 +10,6 @@ import { DEFAULT_TASK_NAME, DEFAULT_WORKER_CHOICE_STRATEGY_OPTIONS, EMPTY_FUNCTION, - isAsyncFunction, isKillBehavior, isPlainObject, median, @@ -173,13 +172,21 @@ export abstract class AbstractPool< protected checkDynamicPoolSize (min: number, max: number): void { if (this.type === PoolTypes.dynamic) { - if (min > max) { + if (max == null) { + throw new Error( + '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 pool size equal to zero' + 'Cannot instantiate a dynamic pool with a maximum pool size equal to zero' ) } else if (min === max) { throw new RangeError( @@ -327,16 +334,20 @@ export abstract class AbstractPool< accumulator + workerNode.usage.tasks.executing, 0 ), - queuedTasks: this.workerNodes.reduce( - (accumulator, workerNode) => - accumulator + workerNode.usage.tasks.queued, - 0 - ), - maxQueuedTasks: this.workerNodes.reduce( - (accumulator, workerNode) => - accumulator + (workerNode.usage.tasks?.maxQueued ?? 0), - 0 - ), + ...(this.opts.enableTasksQueue === true && { + queuedTasks: this.workerNodes.reduce( + (accumulator, workerNode) => + accumulator + workerNode.usage.tasks.queued, + 0 + ) + }), + ...(this.opts.enableTasksQueue === true && { + maxQueuedTasks: this.workerNodes.reduce( + (accumulator, workerNode) => + accumulator + (workerNode.usage.tasks?.maxQueued ?? 0), + 0 + ) + }), failedTasks: this.workerNodes.reduce( (accumulator, workerNode) => accumulator + workerNode.usage.tasks.failed, @@ -494,7 +505,7 @@ export abstract class AbstractPool< private checkMessageWorkerId (message: MessageValue): void { if ( message.workerId != null && - this.getWorkerNodeKeyByWorkerId(message.workerId) == null + this.getWorkerNodeKeyByWorkerId(message.workerId) === -1 ) { throw new Error( `Worker message received from unknown worker '${message.workerId}'` @@ -508,7 +519,7 @@ export abstract class AbstractPool< * @param worker - The worker. * @returns The worker node key if found in the pool worker nodes, `-1` otherwise. */ - private getWorkerNodeKey (worker: Worker): number { + private getWorkerNodeKeyByWorker (worker: Worker): number { return this.workerNodes.findIndex( workerNode => workerNode.worker === worker ) @@ -518,14 +529,12 @@ export abstract class AbstractPool< * Gets the worker node key given its worker id. * * @param workerId - The worker id. - * @returns The worker node key if the worker id is found in the pool worker nodes, `undefined` otherwise. + * @returns The worker node key if the worker id is found in the pool worker nodes, `-1` otherwise. */ - private getWorkerNodeKeyByWorkerId (workerId: number): number | undefined { - for (const [workerNodeKey, workerNode] of this.workerNodes.entries()) { - if (workerNode.info.id === workerId) { - return workerNodeKey - } - } + private getWorkerNodeKeyByWorkerId (workerId: number): number { + return this.workerNodes.findIndex( + workerNode => workerNode.info.id === workerId + ) } /** @inheritDoc */ @@ -606,16 +615,28 @@ export abstract class AbstractPool< protected abstract get busy (): boolean /** - * Whether worker nodes are executing at least one task. + * Whether worker nodes are executing concurrently their tasks quota or not. * * @returns Worker nodes busyness boolean status. */ protected internalBusy (): boolean { - return ( - this.workerNodes.findIndex(workerNode => { - return workerNode.usage.tasks.executing === 0 - }) === -1 - ) + if (this.opts.enableTasksQueue === true) { + return ( + this.workerNodes.findIndex( + 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 + ) + } } /** @inheritDoc */ @@ -629,23 +650,22 @@ export abstract class AbstractPool< data: data ?? ({} as Data), timestamp, workerId: this.getWorkerInfo(workerNodeKey).id as number, - id: randomUUID() + taskId: randomUUID() } - this.promiseResponseMap.set(task.id as string, { + this.promiseResponseMap.set(task.taskId as string, { resolve, reject, workerNodeKey }) if ( - this.opts.enableTasksQueue === true && - (this.busy || - this.workerNodes[workerNodeKey].usage.tasks.executing >= - ((this.opts.tasksQueueOptions as TasksQueueOptions) - .concurrency as number)) + this.opts.enableTasksQueue === false || + (this.opts.enableTasksQueue === true && + this.workerNodes[workerNodeKey].usage.tasks.executing < + (this.opts.tasksQueueOptions?.concurrency as number)) ) { - this.enqueueTask(workerNodeKey, task) - } else { this.executeTask(workerNodeKey, task) + } else { + this.enqueueTask(workerNodeKey, task) } this.checkAndEmitEvents() }) @@ -654,16 +674,8 @@ export abstract class AbstractPool< /** @inheritDoc */ public async destroy (): Promise { await Promise.all( - this.workerNodes.map(async (workerNode, workerNodeKey) => { - this.flushTasksQueue(workerNodeKey) - // FIXME: wait for tasks to be finished - const workerExitPromise = new Promise(resolve => { - workerNode.worker.on('exit', () => { - resolve() - }) - }) + this.workerNodes.map(async (_, workerNodeKey) => { await this.destroyWorkerNode(workerNodeKey) - await workerExitPromise }) ) } @@ -673,9 +685,7 @@ export abstract class AbstractPool< * * @param workerNodeKey - The worker node key. */ - protected abstract destroyWorkerNode ( - workerNodeKey: number - ): void | Promise + protected abstract destroyWorkerNode (workerNodeKey: number): Promise /** * Setup hook to execute code before worker nodes are created in the abstract constructor. @@ -864,7 +874,7 @@ export abstract class AbstractPool< worker.on('message', this.opts.messageHandler ?? EMPTY_FUNCTION) worker.on('error', this.opts.errorHandler ?? EMPTY_FUNCTION) worker.on('error', error => { - const workerNodeKey = this.getWorkerNodeKey(worker) + const workerNodeKey = this.getWorkerNodeKeyByWorker(worker) const workerInfo = this.getWorkerInfo(workerNodeKey) workerInfo.ready = false this.workerNodes[workerNodeKey].closeChannel() @@ -903,8 +913,9 @@ export abstract class AbstractPool< this.registerWorkerMessageListener(workerNodeKey, message => { const localWorkerNodeKey = this.getWorkerNodeKeyByWorkerId( message.workerId - ) as number + ) const workerUsage = this.workerNodes[localWorkerNodeKey].usage + // Kill message received from worker if ( isKillBehavior(KillBehaviors.HARD, message.kill) || (message.kill != null && @@ -914,28 +925,18 @@ export abstract class AbstractPool< workerUsage.tasks.executing === 0 && this.tasksQueueSize(localWorkerNodeKey) === 0))) ) { - // Kill message received from the worker: no new tasks are submitted to that worker for a while ( > maxInactiveTime) - const destroyWorkerNodeBounded = this.destroyWorkerNode.bind(this) - if (isAsyncFunction(destroyWorkerNodeBounded)) { - ( - destroyWorkerNodeBounded as (workerNodeKey: number) => Promise - )(localWorkerNodeKey).catch(EMPTY_FUNCTION) - } else { - (destroyWorkerNodeBounded as (workerNodeKey: number) => void)( - localWorkerNodeKey - ) - } + this.destroyWorkerNode(localWorkerNodeKey).catch(EMPTY_FUNCTION) } }) const workerInfo = this.getWorkerInfo(workerNodeKey) - workerInfo.dynamic = true - if (this.workerChoiceStrategyContext.getStrategyPolicy().useDynamicWorker) { - workerInfo.ready = true - } this.sendToWorker(workerNodeKey, { checkActive: true, workerId: workerInfo.id as number }) + workerInfo.dynamic = true + if (this.workerChoiceStrategyContext.getStrategyPolicy().useDynamicWorker) { + workerInfo.ready = true + } return workerNodeKey } @@ -1004,7 +1005,10 @@ export abstract class AbstractPool< workerInfo.ready && workerNode.usage.tasks.queued === 0 ) { - if (workerNode.usage.tasks.executing === 0) { + if ( + this.workerNodes[workerNodeId].usage.tasks.executing < + (this.opts.tasksQueueOptions?.concurrency as number) + ) { executeTask = true } targetWorkerNodeKey = workerNodeId @@ -1042,10 +1046,10 @@ export abstract class AbstractPool< return message => { this.checkMessageWorkerId(message) if (message.ready != null) { - // Worker ready response received + // Worker ready response received from worker this.handleWorkerReadyResponse(message) - } else if (message.id != null) { - // Task execution response received + } else if (message.taskId != null) { + // Task execution response received from worker this.handleTaskExecutionResponse(message) } } @@ -1053,7 +1057,7 @@ export abstract class AbstractPool< private handleWorkerReadyResponse (message: MessageValue): void { this.getWorkerInfo( - this.getWorkerNodeKeyByWorkerId(message.workerId) as number + this.getWorkerNodeKeyByWorkerId(message.workerId) ).ready = message.ready as boolean if (this.emitter != null && this.ready) { this.emitter.emit(PoolEvents.ready, this.info) @@ -1061,7 +1065,9 @@ export abstract class AbstractPool< } private handleTaskExecutionResponse (message: MessageValue): void { - const promiseResponse = this.promiseResponseMap.get(message.id as string) + const promiseResponse = this.promiseResponseMap.get( + message.taskId as string + ) if (promiseResponse != null) { if (message.taskError != null) { this.emitter?.emit(PoolEvents.taskError, message.taskError) @@ -1071,10 +1077,12 @@ export abstract class AbstractPool< } const workerNodeKey = promiseResponse.workerNodeKey this.afterTaskExecutionHook(workerNodeKey, message) - this.promiseResponseMap.delete(message.id as string) + this.promiseResponseMap.delete(message.taskId as string) if ( this.opts.enableTasksQueue === true && - this.tasksQueueSize(workerNodeKey) > 0 + this.tasksQueueSize(workerNodeKey) > 0 && + this.workerNodes[workerNodeKey].usage.tasks.executing < + (this.opts.tasksQueueOptions?.concurrency as number) ) { this.executeTask( workerNodeKey, @@ -1120,7 +1128,7 @@ export abstract class AbstractPool< workerNode.info.ready = true } this.workerNodes.push(workerNode) - const workerNodeKey = this.getWorkerNodeKey(worker) + const workerNodeKey = this.getWorkerNodeKeyByWorker(worker) if (workerNodeKey === -1) { throw new Error('Worker node not found') } @@ -1133,7 +1141,7 @@ export abstract class AbstractPool< * @param worker - The worker. */ private removeWorkerNode (worker: Worker): void { - const workerNodeKey = this.getWorkerNodeKey(worker) + const workerNodeKey = this.getWorkerNodeKeyByWorker(worker) if (workerNodeKey !== -1) { this.workerNodes.splice(workerNodeKey, 1) this.workerChoiceStrategyContext.remove(workerNodeKey) @@ -1163,7 +1171,7 @@ export abstract class AbstractPool< return this.workerNodes[workerNodeKey].tasksQueueSize() } - private flushTasksQueue (workerNodeKey: number): void { + protected flushTasksQueue (workerNodeKey: number): void { while (this.tasksQueueSize(workerNodeKey) > 0) { this.executeTask( workerNodeKey,