X-Git-Url: https://git.piment-noir.org/?a=blobdiff_plain;f=src%2Fpools%2Fabstract-pool.ts;h=b3ebc19483d95e7c5f9863870e85823f976de157;hb=f4ff1ce25e8ec840112e33e306e492b063738e6d;hp=4d572d0ab5ba351df0d936f88901d4f25335cacf;hpb=4f3c3d894171421375559b43ce469bd5ccb475da;p=poolifier.git diff --git a/src/pools/abstract-pool.ts b/src/pools/abstract-pool.ts index 4d572d0a..b3ebc194 100644 --- a/src/pools/abstract-pool.ts +++ b/src/pools/abstract-pool.ts @@ -1,12 +1,13 @@ +import crypto from 'node:crypto' import type { MessageValue, PromiseWorkerResponseWrapper } from '../utility-types' -import { EMPTY_FUNCTION, EMPTY_OBJECT_LITERAL } from '../utils' -import { isKillBehavior, KillBehaviors } from '../worker/worker-options' +import { EMPTY_FUNCTION } from '../utils' +import { KillBehaviors, isKillBehavior } from '../worker/worker-options' import type { PoolOptions } from './pool' import { PoolEmitter } from './pool' -import type { IPoolInternal, TasksUsage } from './pool-internal' +import type { IPoolInternal, TasksUsage, WorkerType } from './pool-internal' import { PoolType } from './pool-internal' import type { IPoolWorker } from './pool-worker' import { @@ -28,34 +29,31 @@ export abstract class AbstractPool< Response = unknown > implements IPoolInternal { /** {@inheritDoc} */ - public readonly workers: Worker[] = [] - - /** {@inheritDoc} */ - public readonly workersTasksUsage: Map = new Map< - Worker, - TasksUsage + public readonly workers: Map> = new Map< + number, + WorkerType >() /** {@inheritDoc} */ public readonly emitter?: PoolEmitter + /** + * Id of the next worker. + */ + protected nextWorkerId: number = 0 + /** * The promise map. * - * - `key`: This is the message Id of each submitted task. + * - `key`: This is the message id of each submitted task. * - `value`: An object that contains the worker, the resolve function and the reject function. * * When we receive a message from the worker we get a map entry and resolve/reject the promise based on the message. */ protected promiseMap: Map< - number, + string, PromiseWorkerResponseWrapper - > = new Map>() - - /** - * Id of the next message. - */ - protected nextMessageId: number = 0 + > = new Map>() /** * Worker choice strategy instance implementing the worker choice algorithm. @@ -105,7 +103,7 @@ export abstract class AbstractPool< this.getWorkerRunningTasks(workerCreated) === 0 ) { // Kill received from the worker, means that no new tasks are submitted to that worker for a while ( > maxInactiveTime) - void (this.destroyWorker(workerCreated) as Promise) + void this.destroyWorker(workerCreated) } }) return workerCreated @@ -115,7 +113,10 @@ export abstract class AbstractPool< } private checkFilePath (filePath: string): void { - if (filePath == null || filePath.length === 0) { + if ( + filePath == null || + (typeof filePath === 'string' && filePath.trim().length === 0) + ) { throw new Error('Please specify a file with a worker implementation') } } @@ -152,19 +153,29 @@ export abstract class AbstractPool< return this.promiseMap.size } - /** {@inheritDoc} */ - public getWorkerIndex (worker: Worker): number { - return this.workers.indexOf(worker) + /** + * Gets the given worker key. + * + * @param worker - The worker. + * @returns The worker key. + */ + private getWorkerKey (worker: Worker): number | undefined { + return [...this.workers].find(([, value]) => value.worker === worker)?.[0] } /** {@inheritDoc} */ public getWorkerRunningTasks (worker: Worker): number | undefined { - return this.workersTasksUsage.get(worker)?.running + 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.workersTasksUsage.get(worker)?.avgRunTime + return this.getWorkerTasksUsage(worker)?.avgRunTime } /** {@inheritDoc} */ @@ -172,8 +183,13 @@ export abstract class AbstractPool< workerChoiceStrategy: WorkerChoiceStrategy ): void { this.opts.workerChoiceStrategy = workerChoiceStrategy - for (const worker of this.workers) { - this.resetWorkerTasksUsage(worker) + for (const [key, value] of this.workers) { + this.setWorker(key, value.worker, { + run: 0, + running: 0, + runTime: 0, + avgRunTime: 0 + }) } this.workerChoiceStrategyContext.setWorkerChoiceStrategy( workerChoiceStrategy @@ -192,10 +208,10 @@ export abstract class AbstractPool< /** {@inheritDoc} */ public findFreeWorker (): Worker | false { - for (const worker of this.workers) { - if (this.getWorkerRunningTasks(worker) === 0) { + for (const value of this.workers.values()) { + if (value.tasksUsage.running === 0) { // A worker is free, return the matching worker - return worker + return value.worker } } return false @@ -203,22 +219,26 @@ export abstract class AbstractPool< /** {@inheritDoc} */ public async execute (data: Data): Promise { - // Configure worker to handle message with the specified task const worker = this.chooseWorker() - const res = this.internalExecute(worker, this.nextMessageId) + const messageId = crypto.randomUUID() + const res = this.internalExecute(worker, messageId) this.checkAndEmitBusy() this.sendToWorker(worker, { - data: data ?? (EMPTY_OBJECT_LITERAL as Data), - id: this.nextMessageId + // eslint-disable-next-line @typescript-eslint/consistent-type-assertions + data: data ?? ({} as Data), + id: messageId }) - ++this.nextMessageId // eslint-disable-next-line @typescript-eslint/return-await return res } /** {@inheritDoc} */ public async destroy (): Promise { - await Promise.all(this.workers.map(worker => this.destroyWorker(worker))) + await Promise.all( + [...this.workers].map(async ([, value]) => { + await this.destroyWorker(value.worker) + }) + ) } /** @@ -273,9 +293,8 @@ export abstract class AbstractPool< * @param worker - The worker that will be removed. */ protected removeWorker (worker: Worker): void { - // Clean worker from data structure - this.workers.splice(this.getWorkerIndex(worker), 1) - this.removeWorkerTasksUsage(worker) + this.workers.delete(this.getWorkerKey(worker) as number) + --this.nextWorkerId } /** @@ -340,10 +359,13 @@ export abstract class AbstractPool< this.removeWorker(worker) }) - this.workers.push(worker) - - // Init worker tasks usage map - this.initWorkerTasksUsage(worker) + this.setWorker(this.nextWorkerId, worker, { + run: 0, + running: 0, + runTime: 0, + avgRunTime: 0 + }) + ++this.nextWorkerId this.afterWorkerSetup(worker) @@ -374,7 +396,7 @@ export abstract class AbstractPool< private async internalExecute ( worker: Worker, - messageId: number + messageId: string ): Promise { this.beforePromiseWorkerResponseHook(worker) return await new Promise((resolve, reject) => { @@ -406,6 +428,19 @@ export abstract class AbstractPool< this.stepWorkerRunningTasks(worker, -1) } + /** + * Gets tasks usage of the given worker. + * + * @param worker - Worker which tasks usage is returned. + */ + 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 + } + } + /** * Steps the number of tasks that the given worker has applied. * @@ -413,11 +448,8 @@ export abstract class AbstractPool< * @param step - Number of running tasks step. */ private stepWorkerRunningTasks (worker: Worker, step: number): void { - if (this.checkWorkerTasksUsage(worker)) { - const tasksUsage = this.workersTasksUsage.get(worker) as TasksUsage - tasksUsage.running = tasksUsage.running + step - this.workersTasksUsage.set(worker, tasksUsage) - } + // prettier-ignore + (this.getWorkerTasksUsage(worker) as TasksUsage).running += step } /** @@ -427,11 +459,8 @@ export abstract class AbstractPool< * @param step - Number of run tasks step. */ private stepWorkerRunTasks (worker: Worker, step: number): void { - if (this.checkWorkerTasksUsage(worker)) { - const tasksUsage = this.workersTasksUsage.get(worker) as TasksUsage - tasksUsage.run = tasksUsage.run + step - this.workersTasksUsage.set(worker, tasksUsage) - } + // prettier-ignore + (this.getWorkerTasksUsage(worker) as TasksUsage).run += step } /** @@ -446,62 +475,45 @@ export abstract class AbstractPool< ): void { if ( this.workerChoiceStrategyContext.getWorkerChoiceStrategy() - .requiredStatistics.runTime && - this.checkWorkerTasksUsage(worker) + .requiredStatistics.runTime ) { - const tasksUsage = this.workersTasksUsage.get(worker) as TasksUsage - tasksUsage.runTime += taskRunTime ?? 0 - if (tasksUsage.run !== 0) { - tasksUsage.avgRunTime = tasksUsage.runTime / tasksUsage.run + const workerTasksUsage = this.getWorkerTasksUsage(worker) as TasksUsage + workerTasksUsage.runTime += taskRunTime ?? 0 + if (workerTasksUsage.run !== 0) { + workerTasksUsage.avgRunTime = + workerTasksUsage.runTime / workerTasksUsage.run } - this.workersTasksUsage.set(worker, tasksUsage) } } /** - * Checks if the given worker is registered in the workers tasks usage map. - * - * @param worker - Worker to check. - * @returns `true` if the worker is registered in the workers tasks usage map. `false` otherwise. - */ - private checkWorkerTasksUsage (worker: Worker): boolean { - const hasTasksUsage = this.workersTasksUsage.has(worker) - if (!hasTasksUsage) { - throw new Error('Worker could not be found in workers tasks usage map') - } - return hasTasksUsage - } - - /** - * Initializes tasks usage statistics. + * Sets the given worker. * + * @param workerKey - The worker key. * @param worker - The worker. + * @param tasksUsage - The worker tasks usage. */ - private initWorkerTasksUsage (worker: Worker): void { - this.workersTasksUsage.set(worker, { - run: 0, - running: 0, - runTime: 0, - avgRunTime: 0 + private setWorker ( + workerKey: number, + worker: Worker, + tasksUsage: TasksUsage + ): void { + this.workers.set(workerKey, { + worker, + tasksUsage }) } /** - * Removes worker tasks usage statistics. - * - * @param worker - The worker. - */ - private removeWorkerTasksUsage (worker: Worker): void { - this.workersTasksUsage.delete(worker) - } - - /** - * Resets worker tasks usage statistics. + * Checks if the given worker is registered in the pool. * - * @param worker - The worker. + * @param worker - Worker to check. + * @returns `true` if the worker is registered in the pool. */ - private resetWorkerTasksUsage (worker: Worker): void { - this.removeWorkerTasksUsage(worker) - this.initWorkerTasksUsage(worker) + private checkWorker (worker: Worker): boolean { + if (this.getWorkerKey(worker) == null) { + throw new Error('Worker could not be found in the pool') + } + return true } }