X-Git-Url: https://git.piment-noir.org/?a=blobdiff_plain;f=src%2Fpools%2Fabstract-pool.ts;h=66b1ea1da3f3ba47d93cda9c94d2b2a19cc63ac3;hb=bf90656cacf88d2cfdd5b3262086ba55b2ff9818;hp=b3ebc19483d95e7c5f9863870e85823f976de157;hpb=f4ff1ce25e8ec840112e33e306e492b063738e6d;p=poolifier.git diff --git a/src/pools/abstract-pool.ts b/src/pools/abstract-pool.ts index b3ebc194..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' @@ -29,31 +26,23 @@ export abstract class AbstractPool< Response = unknown > implements IPoolInternal { /** {@inheritDoc} */ - public readonly workers: Map> = new Map< - number, - WorkerType - >() + public readonly workers: Array> = [] /** {@inheritDoc} */ public readonly emitter?: PoolEmitter /** - * Id of the next worker. - */ - protected nextWorkerId: number = 0 - - /** - * 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. @@ -96,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.getWorkerRunningTasks(workerCreated) === 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 ) @@ -150,32 +139,17 @@ 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 | undefined { - return [...this.workers].find(([, value]) => value.worker === worker)?.[0] - } - - /** {@inheritDoc} */ - public getWorkerRunningTasks (worker: Worker): number | undefined { - return this.getWorkerTasksUsage(worker)?.running - } - - /** {@inheritDoc} */ - public getWorkerRunTasks (worker: Worker): number | undefined { - return this.getWorkerTasksUsage(worker)?.run - } - - /** {@inheritDoc} */ - public getWorkerAverageTasksRunTime (worker: Worker): number | undefined { - return this.getWorkerTasksUsage(worker)?.avgRunTime + private getWorkerKey (worker: Worker): number { + return this.workers.findIndex(workerItem => workerItem.worker === worker) } /** {@inheritDoc} */ @@ -183,12 +157,13 @@ export abstract class AbstractPool< workerChoiceStrategy: WorkerChoiceStrategy ): void { this.opts.workerChoiceStrategy = workerChoiceStrategy - for (const [key, value] of this.workers) { - this.setWorker(key, value.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( @@ -202,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 value of this.workers.values()) { - if (value.tasksUsage.running === 0) { - // A worker is free, return the matching worker - return value.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 @@ -235,8 +206,8 @@ export abstract class AbstractPool< /** {@inheritDoc} */ public async destroy (): Promise { await Promise.all( - [...this.workers].map(async ([, value]) => { - await this.destroyWorker(value.worker) + this.workers.map(async workerItem => { + await this.destroyWorker(workerItem.worker) }) ) } @@ -265,26 +236,36 @@ 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.increaseWorkerRunningTasks(worker) + 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 { - this.decreaseWorkerRunningTasks(promise.worker) - this.stepWorkerRunTasks(promise.worker, 1) - this.updateWorkerTasksRunTime(promise.worker, message.taskRunTime) + const workerTasksUsage = this.getWorkerTasksUsage(worker) as TasksUsage + --workerTasksUsage.running + ++workerTasksUsage.run + if (message.error != null) { + ++workerTasksUsage.error + } + if (this.workerChoiceStrategyContext.getRequiredStatistics().runTime) { + workerTasksUsage.runTime += message.taskRunTime ?? 0 + if (workerTasksUsage.run !== 0) { + workerTasksUsage.avgRunTime = + workerTasksUsage.runTime / workerTasksUsage.run + } + } } /** @@ -293,8 +274,9 @@ export abstract class AbstractPool< * @param worker - The worker that will be removed. */ protected removeWorker (worker: Worker): void { - this.workers.delete(this.getWorkerKey(worker) as number) - --this.nextWorkerId + const workerKey = this.getWorkerKey(worker) + this.workers.splice(workerKey, 1) + this.workerChoiceStrategyContext.remove(workerKey) } /** @@ -302,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] } /** @@ -359,13 +342,13 @@ export abstract class AbstractPool< this.removeWorker(worker) }) - this.setWorker(this.nextWorkerId, worker, { + this.pushWorker(worker, { run: 0, running: 0, runTime: 0, - avgRunTime: 0 + avgRunTime: 0, + error: 0 }) - ++this.nextWorkerId this.afterWorkerSetup(worker) @@ -380,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 }) }) } @@ -411,79 +395,30 @@ export abstract class AbstractPool< } /** - * Increases the number of tasks that the given worker has applied. + * Gets worker tasks usage. * - * @param worker - Worker which running tasks is increased. - */ - private increaseWorkerRunningTasks (worker: Worker): void { - this.stepWorkerRunningTasks(worker, 1) - } - - /** - * Decreases the number of tasks that the given worker has applied. - * - * @param worker - Worker which running tasks is decreased. - */ - private decreaseWorkerRunningTasks (worker: Worker): void { - this.stepWorkerRunningTasks(worker, -1) - } - - /** - * Gets tasks usage of the given worker. - * - * @param worker - Worker which tasks usage is returned. + * @param worker - The worker. + * @returns The worker tasks usage. */ private getWorkerTasksUsage (worker: Worker): TasksUsage | undefined { - if (this.checkWorker(worker)) { - const workerKey = this.getWorkerKey(worker) as number - const workerEntry = this.workers.get(workerKey) as WorkerType - return workerEntry.tasksUsage + const workerKey = this.getWorkerKey(worker) + if (workerKey !== -1) { + return this.workers[workerKey].tasksUsage } + throw new Error('Worker could not be found in the pool') } /** - * Steps the number of tasks that the given worker has applied. + * Pushes the given worker. * - * @param worker - Worker which running tasks are stepped. - * @param step - Number of running tasks step. - */ - private stepWorkerRunningTasks (worker: Worker, step: number): void { - // prettier-ignore - (this.getWorkerTasksUsage(worker) as TasksUsage).running += step - } - - /** - * Steps the number of tasks that the given worker has run. - * - * @param worker - Worker which has run tasks. - * @param step - Number of run tasks step. - */ - private stepWorkerRunTasks (worker: Worker, step: number): void { - // prettier-ignore - (this.getWorkerTasksUsage(worker) as TasksUsage).run += step - } - - /** - * Updates tasks runtime for the given worker. - * - * @param worker - Worker which run the task. - * @param taskRunTime - Worker task runtime. + * @param worker - The worker. + * @param tasksUsage - The worker tasks usage. */ - private updateWorkerTasksRunTime ( - worker: Worker, - taskRunTime: number | undefined - ): void { - if ( - this.workerChoiceStrategyContext.getWorkerChoiceStrategy() - .requiredStatistics.runTime - ) { - const workerTasksUsage = this.getWorkerTasksUsage(worker) as TasksUsage - workerTasksUsage.runTime += taskRunTime ?? 0 - if (workerTasksUsage.run !== 0) { - workerTasksUsage.avgRunTime = - workerTasksUsage.runTime / workerTasksUsage.run - } - } + private pushWorker (worker: Worker, tasksUsage: TasksUsage): void { + this.workers.push({ + worker, + tasksUsage + }) } /** @@ -498,22 +433,9 @@ export abstract class AbstractPool< worker: Worker, tasksUsage: TasksUsage ): void { - this.workers.set(workerKey, { + this.workers[workerKey] = { worker, tasksUsage - }) - } - - /** - * Checks if the given worker is registered in the pool. - * - * @param worker - Worker to check. - * @returns `true` if the worker is registered in the pool. - */ - private checkWorker (worker: Worker): boolean { - if (this.getWorkerKey(worker) == null) { - throw new Error('Worker could not be found in the pool') } - return true } }