X-Git-Url: https://git.piment-noir.org/?a=blobdiff_plain;f=src%2Fpools%2Fabstract-pool.ts;h=97225246cde2e8baab84cd2a60968ca7d09aa079;hb=d99ba5a8a292772965b5027272a8dc677a6ad008;hp=66b1ea1da3f3ba47d93cda9c94d2b2a19cc63ac3;hpb=bf90656cacf88d2cfdd5b3262086ba55b2ff9818;p=poolifier.git diff --git a/src/pools/abstract-pool.ts b/src/pools/abstract-pool.ts index 66b1ea1d..97225246 100644 --- a/src/pools/abstract-pool.ts +++ b/src/pools/abstract-pool.ts @@ -25,10 +25,10 @@ export abstract class AbstractPool< Data = unknown, Response = unknown > implements IPoolInternal { - /** {@inheritDoc} */ + /** @inheritDoc */ public readonly workers: Array> = [] - /** {@inheritDoc} */ + /** @inheritDoc */ public readonly emitter?: PoolEmitter /** @@ -45,9 +45,9 @@ export abstract class AbstractPool< > = new Map>() /** - * Worker choice strategy instance implementing the worker choice algorithm. + * Worker choice strategy context referencing a worker choice algorithm implementation. * - * Default to a strategy implementing a round robin algorithm. + * Default to a round robin algorithm. */ protected workerChoiceStrategyContext: WorkerChoiceStrategyContext< Worker, @@ -73,6 +73,13 @@ export abstract class AbstractPool< this.checkNumberOfWorkers(this.numberOfWorkers) this.checkFilePath(this.filePath) this.checkPoolOptions(this.opts) + + this.chooseWorker.bind(this) + this.internalExecute.bind(this) + this.checkAndEmitFull.bind(this) + this.checkAndEmitBusy.bind(this) + this.sendToWorker.bind(this) + this.setupHook() for (let i = 1; i <= this.numberOfWorkers; i++) { @@ -82,7 +89,11 @@ export abstract class AbstractPool< if (this.opts.enableEvents === true) { this.emitter = new PoolEmitter() } - this.workerChoiceStrategyContext = new WorkerChoiceStrategyContext( + this.workerChoiceStrategyContext = new WorkerChoiceStrategyContext< + Worker, + Data, + Response + >( this, () => { const createdWorker = this.createAndSetupWorker() @@ -131,14 +142,25 @@ export abstract class AbstractPool< private checkPoolOptions (opts: PoolOptions): void { this.opts.workerChoiceStrategy = opts.workerChoiceStrategy ?? WorkerChoiceStrategies.ROUND_ROBIN + if ( + !Object.values(WorkerChoiceStrategies).includes( + this.opts.workerChoiceStrategy + ) + ) { + throw new Error( + `Invalid worker choice strategy '${this.opts.workerChoiceStrategy}'` + ) + } this.opts.enableEvents = opts.enableEvents ?? true } - /** {@inheritDoc} */ + /** @inheritDoc */ public abstract get type (): PoolType - /** {@inheritDoc} */ - public get numberOfRunningTasks (): number { + /** + * Number of tasks concurrently running in the pool. + */ + private get numberOfRunningTasks (): number { return this.promiseResponseMap.size } @@ -152,7 +174,7 @@ export abstract class AbstractPool< return this.workers.findIndex(workerItem => workerItem.worker === worker) } - /** {@inheritDoc} */ + /** @inheritDoc */ public setWorkerChoiceStrategy ( workerChoiceStrategy: WorkerChoiceStrategy ): void { @@ -171,28 +193,32 @@ export abstract class AbstractPool< ) } - /** {@inheritDoc} */ + /** @inheritDoc */ + public abstract get full (): boolean + + /** @inheritDoc */ public abstract get busy (): boolean - protected internalGetBusyStatus (): boolean { + protected internalBusy (): boolean { return ( this.numberOfRunningTasks >= this.numberOfWorkers && this.findFreeWorkerKey() === -1 ) } - /** {@inheritDoc} */ + /** @inheritDoc */ public findFreeWorkerKey (): number { return this.workers.findIndex(workerItem => { return workerItem.tasksUsage.running === 0 }) } - /** {@inheritDoc} */ + /** @inheritDoc */ public async execute (data: Data): Promise { const [workerKey, worker] = this.chooseWorker() const messageId = crypto.randomUUID() const res = this.internalExecute(workerKey, worker, messageId) + this.checkAndEmitFull() this.checkAndEmitBusy() this.sendToWorker(worker, { // eslint-disable-next-line @typescript-eslint/consistent-type-assertions @@ -203,7 +229,7 @@ export abstract class AbstractPool< return res } - /** {@inheritDoc} */ + /** @inheritDoc */ public async destroy (): Promise { await Promise.all( this.workers.map(async workerItem => { @@ -213,7 +239,7 @@ export abstract class AbstractPool< } /** - * Shutdowns given worker. + * Shutdowns given worker in the pool. * * @param worker - A worker within `workers`. */ @@ -222,9 +248,12 @@ export abstract class AbstractPool< /** * Setup hook that can be overridden by a Poolifier pool implementation * to run code before workers are created in the abstract constructor. + * Can be overridden + * + * @virtual */ protected setupHook (): void { - // Can be overridden + // Intentionally empty } /** @@ -261,28 +290,20 @@ export abstract class AbstractPool< } if (this.workerChoiceStrategyContext.getRequiredStatistics().runTime) { workerTasksUsage.runTime += message.taskRunTime ?? 0 - if (workerTasksUsage.run !== 0) { + if ( + this.workerChoiceStrategyContext.getRequiredStatistics().avgRunTime && + workerTasksUsage.run !== 0 + ) { workerTasksUsage.avgRunTime = workerTasksUsage.runTime / workerTasksUsage.run } } } - /** - * Removes the given worker from the pool. - * - * @param worker - The worker that will be removed. - */ - protected removeWorker (worker: Worker): void { - const workerKey = this.getWorkerKey(worker) - this.workers.splice(workerKey, 1) - this.workerChoiceStrategyContext.remove(workerKey) - } - /** * Chooses a worker for the next task. * - * The default implementation uses a round robin algorithm to distribute the load. + * The default uses a round robin algorithm to distribute the load. * * @returns [worker key, worker]. */ @@ -323,6 +344,7 @@ export abstract class AbstractPool< * Can be used to update the `maxListeners` or binding the `main-worker`\<-\>`worker` connection if not bind by default. * * @param worker - The newly created worker. + * @virtual */ protected abstract afterWorkerSetup (worker: Worker): void @@ -362,9 +384,9 @@ export abstract class AbstractPool< */ protected workerListener (): (message: MessageValue) => void { return message => { - if (message.id !== undefined) { + if (message.id != null) { const promiseResponse = this.promiseResponseMap.get(message.id) - if (promiseResponse !== undefined) { + if (promiseResponse != null) { if (message.error != null) { promiseResponse.reject(message.error) } else { @@ -394,8 +416,18 @@ export abstract class AbstractPool< } } + private checkAndEmitFull (): void { + if ( + this.type === PoolType.DYNAMIC && + this.opts.enableEvents === true && + this.full + ) { + this.emitter?.emit('full') + } + } + /** - * Gets worker tasks usage. + * Gets the given worker tasks usage in the pool. * * @param worker - The worker. * @returns The worker tasks usage. @@ -409,7 +441,7 @@ export abstract class AbstractPool< } /** - * Pushes the given worker. + * Pushes the given worker in the pool. * * @param worker - The worker. * @param tasksUsage - The worker tasks usage. @@ -422,7 +454,7 @@ export abstract class AbstractPool< } /** - * Sets the given worker. + * Sets the given worker in the pool. * * @param workerKey - The worker key. * @param worker - The worker. @@ -438,4 +470,15 @@ export abstract class AbstractPool< tasksUsage } } + + /** + * Removes the given worker from the pool. + * + * @param worker - The worker that will be removed. + */ + protected removeWorker (worker: Worker): void { + const workerKey = this.getWorkerKey(worker) + this.workers.splice(workerKey, 1) + this.workerChoiceStrategyContext.remove(workerKey) + } }