X-Git-Url: https://git.piment-noir.org/?a=blobdiff_plain;f=src%2Fpools%2Fabstract-pool.ts;h=b5e528ec643bb7d58325ab46d49f82c64f0f442b;hb=e334515bf28581c234d6b9f5239425c9a0c6a7ec;hp=2b834158914a93a49aca61cd0ed7f41b2f1e2451;hpb=fb43035cdeaef2c68f9d3c9f8a31071fd6b71e05;p=poolifier.git diff --git a/src/pools/abstract-pool.ts b/src/pools/abstract-pool.ts index 2b834158..b5e528ec 100644 --- a/src/pools/abstract-pool.ts +++ b/src/pools/abstract-pool.ts @@ -1,17 +1,18 @@ +import { AsyncResource } from 'node:async_hooks' import { randomUUID } from 'node:crypto' +import { EventEmitterAsyncResource } from 'node:events' import { performance } from 'node:perf_hooks' import type { TransferListItem } from 'node:worker_threads' -import { EventEmitterAsyncResource } from 'node:events' -import { AsyncResource } from 'node:async_hooks' + import type { MessageValue, PromiseResponseWrapper, Task } from '../utility-types.js' import { + average, DEFAULT_TASK_NAME, EMPTY_FUNCTION, - average, exponentialDelay, isKillBehavior, isPlainObject, @@ -21,8 +22,8 @@ import { round, sleep } from '../utils.js' -import { KillBehaviors } from '../worker/worker-options.js' import type { TaskFunction } from '../worker/task-functions.js' +import { KillBehaviors } from '../worker/worker-options.js' import { type IPool, PoolEvents, @@ -32,13 +33,6 @@ import { PoolTypes, type TasksQueueOptions } from './pool.js' -import type { - IWorker, - IWorkerNode, - WorkerInfo, - WorkerNodeEventDetail, - WorkerType -} from './worker.js' import { Measurements, WorkerChoiceStrategies, @@ -46,8 +40,6 @@ import { type WorkerChoiceStrategyOptions } from './selection-strategies/selection-strategies-types.js' import { WorkerChoiceStrategyContext } from './selection-strategies/worker-choice-strategy-context.js' -import { version } from './version.js' -import { WorkerNode } from './worker-node.js' import { checkFilePath, checkValidTasksQueueOptions, @@ -59,6 +51,15 @@ import { updateWaitTimeWorkerUsage, waitWorkerNodeEvents } from './utils.js' +import { version } from './version.js' +import type { + IWorker, + IWorkerNode, + WorkerInfo, + WorkerNodeEventDetail, + WorkerType +} from './worker.js' +import { WorkerNode } from './worker-node.js' /** * Base class that implements some shared logic for all poolifier pools. @@ -116,6 +117,10 @@ export abstract class AbstractPool< * Whether the pool is destroying or not. */ private destroying: boolean + /** + * Whether the minimum number of workers is starting or not. + */ + private startingMinimumNumberOfWorkers: boolean /** * Whether the pool ready event has been emitted or not. */ @@ -174,6 +179,7 @@ export abstract class AbstractPool< this.starting = false this.destroying = false this.readyEventEmitted = false + this.startingMinimumNumberOfWorkers = false if (this.opts.startWorkers === true) { this.start() } @@ -283,10 +289,11 @@ export abstract class AbstractPool< ready: this.ready, // eslint-disable-next-line @typescript-eslint/no-non-null-assertion strategy: this.opts.workerChoiceStrategy!, + strategyRetries: this.workerChoiceStrategyContext?.retriesCount ?? 0, minSize: this.minimumNumberOfWorkers, maxSize: this.maximumNumberOfWorkers ?? this.minimumNumberOfWorkers, ...(this.workerChoiceStrategyContext?.getTaskStatisticsRequirements() - ?.runTime.aggregate === true && + .runTime.aggregate === true && this.workerChoiceStrategyContext.getTaskStatisticsRequirements() .waitTime.aggregate && { utilization: round(this.utilization) @@ -351,7 +358,7 @@ export abstract class AbstractPool< 0 ), ...(this.workerChoiceStrategyContext?.getTaskStatisticsRequirements() - ?.runTime.aggregate === true && { + .runTime.aggregate === true && { runTime: { minimum: round( min( @@ -394,7 +401,7 @@ export abstract class AbstractPool< } }), ...(this.workerChoiceStrategyContext?.getTaskStatisticsRequirements() - ?.waitTime.aggregate === true && { + .waitTime.aggregate === true && { waitTime: { minimum: round( min( @@ -947,6 +954,23 @@ export abstract class AbstractPool< }) } + /** + * Starts the minimum number of workers. + */ + private startMinimumNumberOfWorkers (): void { + this.startingMinimumNumberOfWorkers = true + while ( + this.workerNodes.reduce( + (accumulator, workerNode) => + !workerNode.info.dynamic ? accumulator + 1 : accumulator, + 0 + ) < this.minimumNumberOfWorkers + ) { + this.createAndSetupWorkerNode() + } + this.startingMinimumNumberOfWorkers = false + } + /** @inheritdoc */ public start (): void { if (this.started) { @@ -959,15 +983,7 @@ export abstract class AbstractPool< throw new Error('Cannot start a destroying pool') } this.starting = true - while ( - this.workerNodes.reduce( - (accumulator, workerNode) => - !workerNode.info.dynamic ? accumulator + 1 : accumulator, - 0 - ) < this.minimumNumberOfWorkers - ) { - this.createAndSetupWorkerNode() - } + this.startMinimumNumberOfWorkers() this.starting = false this.started = true } @@ -991,7 +1007,6 @@ export abstract class AbstractPool< ) this.emitter?.emit(PoolEvents.destroy, this.info) this.emitter?.emitDestroy() - this.emitter?.removeAllListeners() this.readyEventEmitted = false this.destroying = false this.started = false @@ -1055,7 +1070,9 @@ export abstract class AbstractPool< } /** - * Should return whether the worker is the main worker or not. + * Returns whether the worker is the main worker or not. + * + * @returns `true` if the worker is the main worker, `false` otherwise. */ protected abstract isMain (): boolean @@ -1233,7 +1250,7 @@ export abstract class AbstractPool< 'error', this.opts.errorHandler ?? EMPTY_FUNCTION ) - workerNode.registerWorkerEventHandler('error', (error: Error) => { + workerNode.registerOnceWorkerEventHandler('error', (error: Error) => { workerNode.info.ready = false this.emitter?.emit(PoolEvents.error, error) if ( @@ -1243,8 +1260,8 @@ export abstract class AbstractPool< ) { if (workerNode.info.dynamic) { this.createAndSetupDynamicWorkerNode() - } else { - this.createAndSetupWorkerNode() + } else if (!this.startingMinimumNumberOfWorkers) { + this.startMinimumNumberOfWorkers() } } if ( @@ -1255,7 +1272,7 @@ export abstract class AbstractPool< this.redistributeQueuedTasks(this.workerNodes.indexOf(workerNode)) } // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition - workerNode?.terminate().catch(error => { + workerNode?.terminate().catch((error: unknown) => { this.emitter?.emit(PoolEvents.error, error) }) }) @@ -1265,6 +1282,13 @@ export abstract class AbstractPool< ) workerNode.registerOnceWorkerEventHandler('exit', () => { this.removeWorkerNode(workerNode) + if ( + this.started && + !this.startingMinimumNumberOfWorkers && + !this.destroying + ) { + this.startMinimumNumberOfWorkers() + } }) const workerNodeKey = this.addWorkerNode(workerNode) this.afterWorkerNodeSetup(workerNodeKey) @@ -1296,7 +1320,7 @@ export abstract class AbstractPool< ) { // Flag the worker node as not ready immediately this.flagWorkerNodeAsNotReady(localWorkerNodeKey) - this.destroyWorkerNode(localWorkerNodeKey).catch(error => { + this.destroyWorkerNode(localWorkerNodeKey).catch((error: unknown) => { this.emitter?.emit(PoolEvents.error, error) }) } @@ -1310,7 +1334,7 @@ export abstract class AbstractPool< taskFunctionOperation: 'add', taskFunctionName, taskFunction: taskFunction.toString() - }).catch(error => { + }).catch((error: unknown) => { this.emitter?.emit(PoolEvents.error, error) }) } @@ -1417,9 +1441,9 @@ export abstract class AbstractPool< statistics: { runTime: this.workerChoiceStrategyContext?.getTaskStatisticsRequirements() - ?.runTime.aggregate ?? false, + .runTime.aggregate ?? false, elu: - this.workerChoiceStrategyContext?.getTaskStatisticsRequirements()?.elu + this.workerChoiceStrategyContext?.getTaskStatisticsRequirements().elu .aggregate ?? false } }) @@ -1563,8 +1587,7 @@ export abstract class AbstractPool< ) { workerInfo.stealing = false // eslint-disable-next-line @typescript-eslint/no-non-null-assertion - for (const taskName of this.workerNodes[workerNodeKey].info - .taskFunctionNames!) { + for (const taskName of workerInfo.taskFunctionNames!) { this.resetTaskSequentiallyStolenStatisticsTaskFunctionWorkerUsage( workerNodeKey, taskName @@ -1613,7 +1636,7 @@ export abstract class AbstractPool< this.handleWorkerNodeIdleEvent(eventDetail, stolenTask) return undefined }) - .catch(error => { + .catch((error: unknown) => { this.emitter?.emit(PoolEvents.error, error) }) } @@ -1767,7 +1790,12 @@ export abstract class AbstractPool< this.promiseResponseMap.delete(taskId!) // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition workerNode?.emit('taskFinished', taskId) - if (this.opts.enableTasksQueue === true && !this.destroying) { + if ( + this.opts.enableTasksQueue === true && + !this.destroying && + // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition + workerNode != null + ) { const workerNodeTasksUsage = workerNode.usage.tasks if ( this.tasksQueueSize(workerNodeKey) > 0 &&