X-Git-Url: https://git.piment-noir.org/?a=blobdiff_plain;f=src%2Fpools%2Fabstract-pool.ts;h=66b1ea1da3f3ba47d93cda9c94d2b2a19cc63ac3;hb=bf90656cacf88d2cfdd5b3262086ba55b2ff9818;hp=5ed353f602d4f0420e6fca66044122c749098074;hpb=e65c6cd9a3d6ed2e5b8af95120a5aa070101e945;p=poolifier.git diff --git a/src/pools/abstract-pool.ts b/src/pools/abstract-pool.ts index 5ed353f6..66b1ea1d 100644 --- a/src/pools/abstract-pool.ts +++ b/src/pools/abstract-pool.ts @@ -1,8 +1,5 @@ import crypto from 'node:crypto' -import type { - MessageValue, - PromiseWorkerResponseWrapper -} from '../utility-types' +import type { MessageValue, PromiseResponseWrapper } from '../utility-types' import { EMPTY_FUNCTION } from '../utils' import { KillBehaviors, isKillBehavior } from '../worker/worker-options' import type { PoolOptions } from './pool' @@ -35,17 +32,17 @@ export abstract class AbstractPool< public readonly emitter?: PoolEmitter /** - * The promise map. + * The promise response map. * - * - `key`: This is the message id of each submitted task. - * - `value`: An object that contains the worker, the resolve function and the reject function. + * - `key`: The message id of each submitted task. + * - `value`: An object that contains the worker, the promise resolve and reject callbacks. * - * When we receive a message from the worker we get a map entry and resolve/reject the promise based on the message. + * When we receive a message from the worker we get a map entry with the promise resolve/reject bound to the message. */ - protected promiseMap: Map< + protected promiseResponseMap: Map< string, - PromiseWorkerResponseWrapper - > = new Map>() + PromiseResponseWrapper + > = new Map>() /** * Worker choice strategy instance implementing the worker choice algorithm. @@ -88,17 +85,17 @@ export abstract class AbstractPool< this.workerChoiceStrategyContext = new WorkerChoiceStrategyContext( this, () => { - const workerCreated = this.createAndSetupWorker() - this.registerWorkerMessageListener(workerCreated, message => { + const createdWorker = this.createAndSetupWorker() + this.registerWorkerMessageListener(createdWorker, message => { if ( isKillBehavior(KillBehaviors.HARD, message.kill) || - this.getWorkerTasksUsage(workerCreated)?.running === 0 + this.getWorkerTasksUsage(createdWorker)?.running === 0 ) { // Kill received from the worker, means that no new tasks are submitted to that worker for a while ( > maxInactiveTime) - void this.destroyWorker(workerCreated) + void this.destroyWorker(createdWorker) } }) - return workerCreated + return this.getWorkerKey(createdWorker) }, this.opts.workerChoiceStrategy ) @@ -142,14 +139,14 @@ export abstract class AbstractPool< /** {@inheritDoc} */ public get numberOfRunningTasks (): number { - return this.promiseMap.size + return this.promiseResponseMap.size } /** * Gets the given worker key. * * @param worker - The worker. - * @returns The worker key. + * @returns The worker key if the worker is found in the pool, `-1` otherwise. */ private getWorkerKey (worker: Worker): number { return this.workers.findIndex(workerItem => workerItem.worker === worker) @@ -160,12 +157,13 @@ export abstract class AbstractPool< workerChoiceStrategy: WorkerChoiceStrategy ): void { this.opts.workerChoiceStrategy = workerChoiceStrategy - for (const workerItem of this.workers) { - this.setWorker(workerItem.worker, { + for (const [index, workerItem] of this.workers.entries()) { + this.setWorker(index, workerItem.worker, { run: 0, running: 0, runTime: 0, - avgRunTime: 0 + avgRunTime: 0, + error: 0 }) } this.workerChoiceStrategyContext.setWorkerChoiceStrategy( @@ -179,26 +177,22 @@ export abstract class AbstractPool< protected internalGetBusyStatus (): boolean { return ( this.numberOfRunningTasks >= this.numberOfWorkers && - this.findFreeWorker() === false + this.findFreeWorkerKey() === -1 ) } /** {@inheritDoc} */ - public findFreeWorker (): Worker | false { - for (const workerItem of this.workers) { - if (workerItem.tasksUsage.running === 0) { - // A worker is free, return the matching worker - return workerItem.worker - } - } - return false + public findFreeWorkerKey (): number { + return this.workers.findIndex(workerItem => { + return workerItem.tasksUsage.running === 0 + }) } /** {@inheritDoc} */ public async execute (data: Data): Promise { - const worker = this.chooseWorker() + const [workerKey, worker] = this.chooseWorker() const messageId = crypto.randomUUID() - const res = this.internalExecute(worker, messageId) + const res = this.internalExecute(workerKey, worker, messageId) this.checkAndEmitBusy() this.sendToWorker(worker, { // eslint-disable-next-line @typescript-eslint/consistent-type-assertions @@ -242,32 +236,30 @@ export abstract class AbstractPool< * Hook executed before the worker task promise resolution. * Can be overridden. * - * @param worker - The worker. + * @param workerKey - The worker key. */ - protected beforePromiseWorkerResponseHook (worker: Worker): void { - ++(this.getWorkerTasksUsage(worker) as TasksUsage).running + protected beforePromiseResponseHook (workerKey: number): void { + ++this.workers[workerKey].tasksUsage.running } /** * Hook executed after the worker task promise resolution. * Can be overridden. * + * @param worker - The worker. * @param message - The received message. - * @param promise - The Promise response. */ - protected afterPromiseWorkerResponseHook ( - message: MessageValue, - promise: PromiseWorkerResponseWrapper + protected afterPromiseResponseHook ( + worker: Worker, + message: MessageValue ): void { - const workerTasksUsage = this.getWorkerTasksUsage( - promise.worker - ) as TasksUsage + const workerTasksUsage = this.getWorkerTasksUsage(worker) as TasksUsage --workerTasksUsage.running ++workerTasksUsage.run - if ( - this.workerChoiceStrategyContext.getWorkerChoiceStrategy() - .requiredStatistics.runTime - ) { + if (message.error != null) { + ++workerTasksUsage.error + } + if (this.workerChoiceStrategyContext.getRequiredStatistics().runTime) { workerTasksUsage.runTime += message.taskRunTime ?? 0 if (workerTasksUsage.run !== 0) { workerTasksUsage.avgRunTime = @@ -282,7 +274,9 @@ export abstract class AbstractPool< * @param worker - The worker that will be removed. */ protected removeWorker (worker: Worker): void { - this.workers.splice(this.getWorkerKey(worker), 1) + const workerKey = this.getWorkerKey(worker) + this.workers.splice(workerKey, 1) + this.workerChoiceStrategyContext.remove(workerKey) } /** @@ -290,10 +284,11 @@ export abstract class AbstractPool< * * The default implementation uses a round robin algorithm to distribute the load. * - * @returns Worker. + * @returns [worker key, worker]. */ - protected chooseWorker (): Worker { - return this.workerChoiceStrategyContext.execute() + protected chooseWorker (): [number, Worker] { + const workerKey = this.workerChoiceStrategyContext.execute() + return [workerKey, this.workers[workerKey].worker] } /** @@ -347,11 +342,12 @@ export abstract class AbstractPool< this.removeWorker(worker) }) - this.setWorker(worker, { + this.pushWorker(worker, { run: 0, running: 0, runTime: 0, - avgRunTime: 0 + avgRunTime: 0, + error: 0 }) this.afterWorkerSetup(worker) @@ -367,27 +363,28 @@ export abstract class AbstractPool< protected workerListener (): (message: MessageValue) => void { return message => { if (message.id !== undefined) { - const promise = this.promiseMap.get(message.id) - if (promise !== undefined) { + const promiseResponse = this.promiseResponseMap.get(message.id) + if (promiseResponse !== undefined) { if (message.error != null) { - promise.reject(message.error) + promiseResponse.reject(message.error) } else { - promise.resolve(message.data as Response) + promiseResponse.resolve(message.data as Response) } - this.afterPromiseWorkerResponseHook(message, promise) - this.promiseMap.delete(message.id) + this.afterPromiseResponseHook(promiseResponse.worker, message) + this.promiseResponseMap.delete(message.id) } } } } private async internalExecute ( + workerKey: number, worker: Worker, messageId: string ): Promise { - this.beforePromiseWorkerResponseHook(worker) + this.beforePromiseResponseHook(workerKey) return await new Promise((resolve, reject) => { - this.promiseMap.set(messageId, { resolve, reject, worker }) + this.promiseResponseMap.set(messageId, { resolve, reject, worker }) }) } @@ -397,8 +394,13 @@ export abstract class AbstractPool< } } - /** {@inheritDoc} */ - public getWorkerTasksUsage (worker: Worker): TasksUsage | undefined { + /** + * Gets worker tasks usage. + * + * @param worker - The worker. + * @returns The worker tasks usage. + */ + private getWorkerTasksUsage (worker: Worker): TasksUsage | undefined { const workerKey = this.getWorkerKey(worker) if (workerKey !== -1) { return this.workers[workerKey].tasksUsage @@ -407,15 +409,33 @@ export abstract class AbstractPool< } /** - * Sets the given worker. + * Pushes the given worker. * * @param worker - The worker. * @param tasksUsage - The worker tasks usage. */ - private setWorker (worker: Worker, tasksUsage: TasksUsage): void { + private pushWorker (worker: Worker, tasksUsage: TasksUsage): void { this.workers.push({ worker, tasksUsage }) } + + /** + * Sets the given worker. + * + * @param workerKey - The worker key. + * @param worker - The worker. + * @param tasksUsage - The worker tasks usage. + */ + private setWorker ( + workerKey: number, + worker: Worker, + tasksUsage: TasksUsage + ): void { + this.workers[workerKey] = { + worker, + tasksUsage + } + } }