X-Git-Url: https://git.piment-noir.org/?a=blobdiff_plain;f=src%2Fpools%2Fabstract-pool.ts;h=52a2aa383da0dd89acf4b3db7e518fd18fc33f21;hb=985d0e7986b2cad23bb08ae0561b2a6ff9afdf9e;hp=ba05d92959a83f362fd721b4ca3615e66c86d196;hpb=f65efa796405c785a1ab5e646e30fc27418a6151;p=poolifier.git diff --git a/src/pools/abstract-pool.ts b/src/pools/abstract-pool.ts index ba05d929..52a2aa38 100644 --- a/src/pools/abstract-pool.ts +++ b/src/pools/abstract-pool.ts @@ -5,7 +5,8 @@ import { DEFAULT_WORKER_CHOICE_STRATEGY_OPTIONS, EMPTY_FUNCTION, isPlainObject, - median + median, + round } from '../utils' import { KillBehaviors, isKillBehavior } from '../worker/worker-options' import { CircularArray } from '../circular-array' @@ -77,6 +78,11 @@ export abstract class AbstractPool< Response > + /** + * The start timestamp of the pool. + */ + private readonly startTimestamp + /** * Constructs a new poolifier pool. * @@ -116,9 +122,11 @@ export abstract class AbstractPool< this.setupHook() - for (let i = 1; i <= this.numberOfWorkers; i++) { + while (this.workerNodes.length < this.numberOfWorkers) { this.createAndSetupWorker() } + + this.startTimestamp = performance.now() } private checkFilePath (filePath: string): void { @@ -237,14 +245,6 @@ export abstract class AbstractPool< } } - private get starting (): boolean { - return this.workerNodes.some(workerNode => !workerNode.info.started) - } - - private get started (): boolean { - return this.workerNodes.some(workerNode => workerNode.info.started) - } - /** @inheritDoc */ public get info (): PoolInfo { return { @@ -252,6 +252,7 @@ export abstract class AbstractPool< worker: this.worker, minSize: this.minSize, maxSize: this.maxSize, + utilization: round(this.utilization), workerNodes: this.workerNodes.length, idleWorkerNodes: this.workerNodes.reduce( (accumulator, workerNode) => @@ -293,6 +294,35 @@ export abstract class AbstractPool< } } + /** + * Gets the pool run time. + * + * @returns The pool run time in milliseconds. + */ + private get runTime (): number { + return performance.now() - this.startTimestamp + } + + /** + * Gets the approximate pool utilization. + * + * @returns The pool utilization. + */ + private get utilization (): number { + const poolRunTimeCapacity = this.runTime * this.maxSize + const totalTasksRunTime = this.workerNodes.reduce( + (accumulator, workerNode) => + accumulator + workerNode.usage.runTime.aggregate, + 0 + ) + const totalTasksWaitTime = this.workerNodes.reduce( + (accumulator, workerNode) => + accumulator + workerNode.usage.waitTime.aggregate, + 0 + ) + return (totalTasksRunTime + totalTasksWaitTime) / poolRunTimeCapacity + } + /** * Pool type. * @@ -732,7 +762,7 @@ export abstract class AbstractPool< if (this.emitter != null) { this.emitter.emit(PoolEvents.error, error) } - if (this.opts.restartWorkerOnError === true && !this.starting) { + if (this.opts.restartWorkerOnError === true) { this.createAndSetupWorker() } }) @@ -790,7 +820,9 @@ export abstract class AbstractPool< this.workerNodes[this.getWorkerNodeKey(worker)].info.started = message.started } else { - throw new Error('Worker started message received from unknown worker') + throw new Error( + `Worker started message received from unknown worker '${message.workerId}'` + ) } } else if (message.id != null) { // Task execution response received @@ -855,7 +887,7 @@ export abstract class AbstractPool< private pushWorkerNode (worker: Worker): number { this.workerNodes.push({ worker, - info: { id: this.getWorkerId(worker), started: false }, + info: { id: this.getWorkerId(worker), started: true }, usage: this.getWorkerUsage(), tasksQueue: new Queue>() })