X-Git-Url: https://git.piment-noir.org/?a=blobdiff_plain;f=src%2Fpools%2Fabstract-pool.ts;h=55c380b80c5b92de72bb1c4c87e57f4614c56f37;hb=184855e69fea29f1018024a34be10de2c8e3141a;hp=a558ef504568ad95abcf91d8f413968a2f07bf8f;hpb=1f68cedebde2cf95b6345300ac52cf7ca3ecbdff;p=poolifier.git diff --git a/src/pools/abstract-pool.ts b/src/pools/abstract-pool.ts index a558ef50..55c380b8 100644 --- a/src/pools/abstract-pool.ts +++ b/src/pools/abstract-pool.ts @@ -13,9 +13,12 @@ import { type IPool, PoolEmitter, PoolEvents, + type PoolInfo, type PoolOptions, - PoolType, - type TasksQueueOptions + type PoolType, + PoolTypes, + type TasksQueueOptions, + type WorkerType } from './pool' import type { IWorker, Task, TasksUsage, WorkerNode } from './worker' import { @@ -133,7 +136,7 @@ export abstract class AbstractPool< throw new RangeError( 'Cannot instantiate a pool with a negative number of workers' ) - } else if (this.type === PoolType.FIXED && numberOfWorkers === 0) { + } else if (this.type === PoolTypes.fixed && numberOfWorkers === 0) { throw new Error('Cannot instantiate a fixed pool with no worker') } } @@ -185,7 +188,7 @@ export abstract class AbstractPool< } if ( workerChoiceStrategyOptions.weights != null && - Object.keys(workerChoiceStrategyOptions.weights).length !== this.size + Object.keys(workerChoiceStrategyOptions.weights).length !== this.maxSize ) { throw new Error( 'Invalid worker choice strategy options: must have a weight for each worker node' @@ -212,30 +215,54 @@ export abstract class AbstractPool< public abstract get type (): PoolType /** @inheritDoc */ - public abstract get size (): number + public get info (): PoolInfo { + return { + type: this.type, + worker: this.worker, + minSize: this.minSize, + maxSize: this.maxSize, + workerNodes: this.workerNodes.length, + idleWorkerNodes: this.workerNodes.reduce( + (accumulator, workerNode) => + workerNode.tasksUsage.running === 0 ? accumulator + 1 : accumulator, + 0 + ), + busyWorkerNodes: this.workerNodes.reduce( + (accumulator, workerNode) => + workerNode.tasksUsage.running > 0 ? accumulator + 1 : accumulator, + 0 + ), + runningTasks: this.workerNodes.reduce( + (accumulator, workerNode) => + accumulator + workerNode.tasksUsage.running, + 0 + ), + queuedTasks: this.workerNodes.reduce( + (accumulator, workerNode) => accumulator + workerNode.tasksQueue.size, + 0 + ), + maxQueuedTasks: this.workerNodes.reduce( + (accumulator, workerNode) => + accumulator + workerNode.tasksQueue.maxSize, + 0 + ) + } + } /** - * Number of tasks running in the pool. + * Gets the worker type. */ - private get numberOfRunningTasks (): number { - return this.workerNodes.reduce( - (accumulator, workerNode) => accumulator + workerNode.tasksUsage.running, - 0 - ) - } + protected abstract get worker (): WorkerType /** - * Number of tasks queued in the pool. + * Pool minimum size. */ - private get numberOfQueuedTasks (): number { - if (this.opts.enableTasksQueue === false) { - return 0 - } - return this.workerNodes.reduce( - (accumulator, workerNode) => accumulator + workerNode.tasksQueue.size, - 0 - ) - } + protected abstract get minSize (): number + + /** + * Pool maximum size. + */ + protected abstract get maxSize (): number /** * Gets the given worker its worker node key. @@ -497,7 +524,7 @@ export abstract class AbstractPool< */ protected chooseWorkerNode (): number { let workerNodeKey: number - if (this.type === PoolType.DYNAMIC && !this.full && this.internalBusy()) { + if (this.type === PoolTypes.dynamic && !this.full && this.internalBusy()) { const workerCreated = this.createAndSetupWorker() this.registerWorkerMessageListener(workerCreated, message => { const currentWorkerNodeKey = this.getWorkerNodeKey(workerCreated) @@ -599,6 +626,12 @@ export abstract class AbstractPool< if (promiseResponse != null) { if (message.error != null) { promiseResponse.reject(message.error) + if (this.emitter != null) { + this.emitter.emit(PoolEvents.taskError, { + error: message.error, + errorData: message.errorData + }) + } } else { promiseResponse.resolve(message.data as Response) } @@ -622,10 +655,10 @@ export abstract class AbstractPool< private checkAndEmitEvents (): void { if (this.emitter != null) { if (this.busy) { - this.emitter?.emit(PoolEvents.busy) + this.emitter?.emit(PoolEvents.busy, this.info) } - if (this.type === PoolType.DYNAMIC && this.full) { - this.emitter?.emit(PoolEvents.full) + if (this.type === PoolTypes.dynamic && this.full) { + this.emitter?.emit(PoolEvents.full, this.info) } } }