X-Git-Url: https://git.piment-noir.org/?a=blobdiff_plain;f=src%2Fpools%2Fabstract-pool.ts;h=3b79565e062aa7b1660e42eb506478ca07d79da4;hb=27469db4f589429f5dd309187b1de2aae0275cb0;hp=a7d3e944522e90d5ee5dce103be13cd5e334bdc4;hpb=f12182ad6dc553c7a5dfeee01bcde65c0177f671;p=poolifier.git diff --git a/src/pools/abstract-pool.ts b/src/pools/abstract-pool.ts index a7d3e944..3b79565e 100644 --- a/src/pools/abstract-pool.ts +++ b/src/pools/abstract-pool.ts @@ -4,11 +4,12 @@ import { EventEmitterAsyncResource } from 'node:events' import { performance } from 'node:perf_hooks' import type { TransferListItem } from 'node:worker_threads' +import { defaultBucketSize } from '../priority-queue.js' import type { MessageValue, PromiseResponseWrapper, Task, - TaskFunctionProperties + TaskFunctionProperties, } from '../utility-types.js' import { average, @@ -22,11 +23,11 @@ import { median, min, round, - sleep + sleep, } from '../utils.js' import type { TaskFunction, - TaskFunctionObject + TaskFunctionObject, } from '../worker/task-functions.js' import { KillBehaviors } from '../worker/worker-options.js' import { @@ -36,13 +37,13 @@ import { type PoolOptions, type PoolType, PoolTypes, - type TasksQueueOptions + type TasksQueueOptions, } from './pool.js' import { Measurements, WorkerChoiceStrategies, type WorkerChoiceStrategy, - type WorkerChoiceStrategyOptions + type WorkerChoiceStrategyOptions, } from './selection-strategies/selection-strategies-types.js' import { WorkerChoiceStrategiesContext } from './selection-strategies/worker-choice-strategies-context.js' import { @@ -55,7 +56,7 @@ import { updateRunTimeWorkerUsage, updateTaskStatisticsWorkerUsage, updateWaitTimeWorkerUsage, - waitWorkerNodeEvents + waitWorkerNodeEvents, } from './utils.js' import { version } from './version.js' import type { @@ -63,13 +64,12 @@ import type { IWorkerNode, WorkerInfo, WorkerNodeEventDetail, - WorkerType + WorkerType, } from './worker.js' import { WorkerNode } from './worker-node.js' /** * Base class that implements some shared logic for all poolifier pools. - * * @typeParam Worker - Type of worker which manages this pool. * @typeParam Data - Type of data sent to the worker. This can only be structured-cloneable data. * @typeParam Response - Type of execution response. This can only be structured-cloneable data. @@ -80,7 +80,7 @@ export abstract class AbstractPool< Response = unknown > implements IPool { /** @inheritDoc */ - public readonly workerNodes: Array> = [] + public readonly workerNodes: IWorkerNode[] = [] /** @inheritDoc */ public emitter?: EventEmitterAsyncResource @@ -94,19 +94,19 @@ export abstract class AbstractPool< */ protected promiseResponseMap: Map< `${string}-${string}-${string}-${string}-${string}`, - PromiseResponseWrapper + PromiseResponseWrapper > = new Map< `${string}-${string}-${string}-${string}-${string}`, PromiseResponseWrapper - >() + >() /** * Worker choice strategies context referencing worker choice algorithms implementation. */ protected workerChoiceStrategiesContext?: WorkerChoiceStrategiesContext< - Worker, - Data, - Response + Worker, + Data, + Response > /** @@ -115,8 +115,8 @@ export abstract class AbstractPool< * - `value`: The task function object. */ private readonly taskFunctions: Map< - string, - TaskFunctionObject + string, + TaskFunctionObject > /** @@ -146,7 +146,6 @@ export abstract class AbstractPool< /** * Constructs a new poolifier pool. - * * @param minimumNumberOfWorkers - Minimum number of workers that this pool manages. * @param filePath - Path to the worker file. * @param opts - Options for the pool. @@ -176,9 +175,9 @@ export abstract class AbstractPool< this.initEventEmitter() } this.workerChoiceStrategiesContext = new WorkerChoiceStrategiesContext< - Worker, - Data, - Response + Worker, + Data, + Response >( this, // eslint-disable-next-line @typescript-eslint/no-non-null-assertion @@ -288,7 +287,7 @@ export abstract class AbstractPool< private initEventEmitter (): void { this.emitter = new EventEmitterAsyncResource({ - name: `poolifier:${this.type}-${this.worker}-pool` + name: `poolifier:${this.type}-${this.worker}-pool`, }) } @@ -309,7 +308,7 @@ export abstract class AbstractPool< .runTime.aggregate === true && this.workerChoiceStrategiesContext.getTaskStatisticsRequirements() .waitTime.aggregate && { - utilization: round(this.utilization) + utilization: round(this.utilization), }), workerNodes: this.workerNodes.length, idleWorkerNodes: this.workerNodes.reduce( @@ -324,7 +323,7 @@ export abstract class AbstractPool< (accumulator, workerNode) => workerNode.info.stealing ? accumulator + 1 : accumulator, 0 - ) + ), }), busyWorkerNodes: this.workerNodes.reduce( (accumulator, _, workerNodeKey) => @@ -346,24 +345,24 @@ export abstract class AbstractPool< (accumulator, workerNode) => accumulator + workerNode.usage.tasks.queued, 0 - ) + ), }), ...(this.opts.enableTasksQueue === true && { maxQueuedTasks: this.workerNodes.reduce( (accumulator, workerNode) => accumulator + (workerNode.usage.tasks.maxQueued ?? 0), 0 - ) + ), }), ...(this.opts.enableTasksQueue === true && { - backPressure: this.hasBackPressure() + backPressure: this.hasBackPressure(), }), ...(this.opts.enableTasksQueue === true && { stolenTasks: this.workerNodes.reduce( (accumulator, workerNode) => accumulator + workerNode.usage.tasks.stolen, 0 - ) + ), }), failedTasks: this.workerNodes.reduce( (accumulator, workerNode) => @@ -401,7 +400,7 @@ export abstract class AbstractPool< [] ) ) - ) + ), }), ...(this.workerChoiceStrategiesContext.getTaskStatisticsRequirements() .runTime.median && { @@ -415,9 +414,9 @@ export abstract class AbstractPool< [] ) ) - ) - }) - } + ), + }), + }, }), ...(this.workerChoiceStrategiesContext?.getTaskStatisticsRequirements() .waitTime.aggregate === true && { @@ -450,7 +449,7 @@ export abstract class AbstractPool< [] ) ) - ) + ), }), ...(this.workerChoiceStrategiesContext.getTaskStatisticsRequirements() .waitTime.median && { @@ -464,9 +463,9 @@ export abstract class AbstractPool< [] ) ) - ) - }) - } + ), + }), + }, }), ...(this.workerChoiceStrategiesContext?.getTaskStatisticsRequirements() .elu.aggregate === true && { @@ -502,7 +501,7 @@ export abstract class AbstractPool< [] ) ) - ) + ), }), ...(this.workerChoiceStrategiesContext.getTaskStatisticsRequirements() .elu.median && { @@ -516,8 +515,8 @@ export abstract class AbstractPool< [] ) ) - ) - }) + ), + }), }, active: { minimum: round( @@ -550,7 +549,7 @@ export abstract class AbstractPool< [] ) ) - ) + ), }), ...(this.workerChoiceStrategiesContext.getTaskStatisticsRequirements() .elu.median && { @@ -564,8 +563,8 @@ export abstract class AbstractPool< [] ) ) - ) - }) + ), + }), }, utilization: { average: round( @@ -581,10 +580,10 @@ export abstract class AbstractPool< workerNode => workerNode.usage.elu.utilization ?? 0 ) ) - ) - } - } - }) + ), + }, + }, + }), } } @@ -615,7 +614,6 @@ export abstract class AbstractPool< /** * The approximate pool utilization. - * * @returns The pool utilization. */ private get utilization (): number { @@ -652,7 +650,6 @@ export abstract class AbstractPool< /** * Checks if the worker id sent in the received message from a worker is valid. - * * @param message - The received message. * @throws {@link https://nodejs.org/api/errors.html#class-error} If the worker id is invalid. */ @@ -661,14 +658,13 @@ export abstract class AbstractPool< throw new Error('Worker message received without worker id') } else if (this.getWorkerNodeKeyByWorkerId(message.workerId) === -1) { throw new Error( - `Worker message received from unknown worker '${message.workerId}'` + `Worker message received from unknown worker '${message.workerId.toString()}'` ) } } /** * Gets the worker node key given its worker id. - * * @param workerId - The worker id. * @returns The worker node key if the worker id is found in the pool worker nodes, `-1` otherwise. */ @@ -779,7 +775,7 @@ export abstract class AbstractPool< ...getDefaultTasksQueueOptions( this.maximumNumberOfWorkers ?? this.minimumNumberOfWorkers ), - ...tasksQueueOptions + ...tasksQueueOptions, } } @@ -843,7 +839,6 @@ export abstract class AbstractPool< /** * Whether worker nodes are executing concurrently their tasks quota or not. - * * @returns Worker nodes busyness boolean status. */ protected internalBusy (): boolean { @@ -896,7 +891,11 @@ export abstract class AbstractPool< } else { reject( new Error( - `Task function operation '${message.taskFunctionOperation}' failed on worker ${message.workerId} with error: '${message.workerError?.message}'` + // eslint-disable-next-line @typescript-eslint/restrict-template-expressions + `Task function operation '${message.taskFunctionOperation?.toString()}' failed on worker ${message.workerId?.toString()} with error: '${ + // eslint-disable-next-line @typescript-eslint/restrict-template-expressions + message.workerError?.message + }'` ) ) } @@ -944,7 +943,9 @@ export abstract class AbstractPool< new Error( `Task function operation '${ message.taskFunctionOperation as string - }' failed on worker ${errorResponse?.workerId} with error: '${ + // eslint-disable-next-line @typescript-eslint/restrict-template-expressions + }' failed on worker ${errorResponse?.workerId?.toString()} with error: '${ + // eslint-disable-next-line @typescript-eslint/restrict-template-expressions errorResponse?.workerError?.message }'` ) @@ -996,7 +997,7 @@ export abstract class AbstractPool< const opResult = await this.sendTaskFunctionOperationToWorkers({ taskFunctionOperation: 'add', taskFunctionProperties: buildTaskFunctionProperties(name, fn), - taskFunction: fn.taskFunction.toString() + taskFunction: fn.taskFunction.toString(), }) this.taskFunctions.set(name, fn) this.workerChoiceStrategiesContext?.syncWorkerChoiceStrategies( @@ -1020,7 +1021,7 @@ export abstract class AbstractPool< taskFunctionProperties: buildTaskFunctionProperties( name, this.taskFunctions.get(name) - ) + ), }) for (const workerNode of this.workerNodes) { workerNode.deleteTaskFunctionWorkerUsage(name) @@ -1050,7 +1051,6 @@ export abstract class AbstractPool< /** * Gets task function worker choice strategy, if any. - * * @param name - The task function name. * @returns The task function worker choice strategy if the task function worker choice strategy is defined, `undefined` otherwise. */ @@ -1070,7 +1070,6 @@ export abstract class AbstractPool< /** * Gets worker node task function worker choice strategy, if any. - * * @param workerNodeKey - The worker node key. * @param name - The task function name. * @returns The worker node task function worker choice strategy if the worker node task function worker choice strategy is defined, `undefined` otherwise. @@ -1095,7 +1094,6 @@ export abstract class AbstractPool< /** * Gets worker node task function priority, if any. - * * @param workerNodeKey - The worker node key. * @param name - The task function name. * @returns The worker node task function priority if the worker node task function priority is defined, `undefined` otherwise. @@ -1120,7 +1118,6 @@ export abstract class AbstractPool< /** * Gets the worker choice strategies registered in this pool. - * * @returns The worker choice strategies. */ private readonly getWorkerChoiceStrategies = @@ -1135,7 +1132,7 @@ export abstract class AbstractPool< ) .filter( (strategy: WorkerChoiceStrategy | undefined) => strategy != null - ) as WorkerChoiceStrategy[]) + ) as WorkerChoiceStrategy[]), ]) } @@ -1146,7 +1143,7 @@ export abstract class AbstractPool< taskFunctionProperties: buildTaskFunctionProperties( name, this.taskFunctions.get(name) - ) + ), }) } @@ -1194,7 +1191,6 @@ export abstract class AbstractPool< const workerNodeKey = this.chooseWorkerNode(name) const task: Task = { name: name ?? DEFAULT_TASK_NAME, - // eslint-disable-next-line @typescript-eslint/consistent-type-assertions data: data ?? ({} as Data), priority: this.getWorkerNodeTaskFunctionPriority(workerNodeKey, name), strategy: this.getWorkerNodeTaskFunctionWorkerChoiceStrategy( @@ -1203,7 +1199,7 @@ export abstract class AbstractPool< ), transferList, timestamp, - taskId: randomUUID() + taskId: randomUUID(), } // eslint-disable-next-line @typescript-eslint/no-non-null-assertion this.promiseResponseMap.set(task.taskId!, { @@ -1213,9 +1209,9 @@ export abstract class AbstractPool< ...(this.emitter != null && { asyncResource: new AsyncResource('poolifier:task', { triggerAsyncId: this.emitter.asyncId, - requireManualDestroy: true - }) - }) + requireManualDestroy: true, + }), + }), }) if ( this.opts.enableTasksQueue === false || @@ -1229,8 +1225,31 @@ export abstract class AbstractPool< }) } + + /** @inheritDoc */ + public mapExecute ( + data: Iterable, + name?: string, + transferList?: readonly TransferListItem[] + ): Promise { + // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition + if (data == null) { + throw new TypeError('data argument must be a defined iterable') + } + if (typeof data[Symbol.iterator] !== 'function') { + throw new TypeError('data argument must be an iterable') + } + if (!Array.isArray(data)) { + data = [...data] + } + return Promise.all( + (data as Data[]).map(data => this.execute(data, name, transferList)) + ) + } + /** * Starts the minimum number of workers. + * @param initWorkerNodeUsage - Whether to initialize the worker node usage or not. @defaultValue false */ private startMinimumNumberOfWorkers (initWorkerNodeUsage = false): void { this.startingMinimumNumberOfWorkers = true @@ -1305,7 +1324,8 @@ export abstract class AbstractPool< } else if (message.kill === 'failure') { reject( new Error( - `Kill message handling failed on worker ${message.workerId}` + // eslint-disable-next-line @typescript-eslint/restrict-template-expressions + `Kill message handling failed on worker ${message.workerId?.toString()}` ) ) } @@ -1318,7 +1338,6 @@ export abstract class AbstractPool< /** * Terminates the worker node given its worker node key. - * * @param workerNodeKey - The worker node key. */ protected async destroyWorkerNode (workerNodeKey: number): Promise { @@ -1341,8 +1360,6 @@ export abstract class AbstractPool< /** * Setup hook to execute code before worker nodes are created in the abstract constructor. * Can be overridden. - * - * @virtual */ protected setupHook (): void { /* Intentionally empty */ @@ -1350,7 +1367,6 @@ export abstract class AbstractPool< /** * 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 @@ -1358,7 +1374,6 @@ export abstract class AbstractPool< /** * Hook executed before the worker task execution. * Can be overridden. - * * @param workerNodeKey - The worker node key. * @param task - The task to execute. */ @@ -1399,7 +1414,6 @@ export abstract class AbstractPool< /** * Hook executed after the worker task execution. * Can be overridden. - * * @param workerNodeKey - The worker node key. * @param message - The received message. */ @@ -1456,7 +1470,6 @@ export abstract class AbstractPool< /** * Whether the worker node shall update its task function worker usage or not. - * * @param workerNodeKey - The worker node key. * @returns `true` if the worker node shall update its task function worker usage, `false` otherwise. */ @@ -1471,9 +1484,8 @@ export abstract class AbstractPool< /** * Chooses a worker node for the next task. - * * @param name - The task function name. - * @returns The chosen worker node key + * @returns The chosen worker node key. */ private chooseWorkerNode (name?: string): number { if (this.shallCreateDynamicWorker()) { @@ -1493,14 +1505,12 @@ export abstract class AbstractPool< /** * Conditions for dynamic worker creation. - * * @returns Whether to create a dynamic worker or not. */ protected abstract shallCreateDynamicWorker (): boolean /** * Sends a message to worker given its worker node key. - * * @param workerNodeKey - The worker node key. * @param message - The message. * @param transferList - The optional array of transferable objects. @@ -1513,7 +1523,6 @@ export abstract class AbstractPool< /** * Initializes the worker node usage with sensible default values gathered during runtime. - * * @param workerNode - The worker node. */ private initWorkerNodeUsage (workerNode: IWorkerNode): void { @@ -1554,7 +1563,6 @@ export abstract class AbstractPool< /** * Creates a new, completely set up worker node. - * * @returns New, completely set up worker node key. */ protected createAndSetupWorkerNode (): number { @@ -1618,7 +1626,6 @@ export abstract class AbstractPool< /** * Creates a new, completely set up dynamic worker node. - * * @returns New, completely set up dynamic worker node key. */ protected createAndSetupDynamicWorkerNode (): number { @@ -1650,7 +1657,7 @@ export abstract class AbstractPool< } }) this.sendToWorker(workerNodeKey, { - checkActive: true + checkActive: true, }) if (this.taskFunctions.size > 0) { for (const [taskFunctionName, taskFunctionObject] of this.taskFunctions) { @@ -1660,7 +1667,7 @@ export abstract class AbstractPool< taskFunctionName, taskFunctionObject ), - taskFunction: taskFunctionObject.taskFunction.toString() + taskFunction: taskFunctionObject.taskFunction.toString(), }).catch((error: unknown) => { this.emitter?.emit(PoolEvents.error, error) }) @@ -1683,7 +1690,6 @@ export abstract class AbstractPool< /** * Registers a listener callback on the worker given its worker node key. - * * @param workerNodeKey - The worker node key. * @param listener - The message listener callback. */ @@ -1696,7 +1702,6 @@ export abstract class AbstractPool< /** * Registers once a listener callback on the worker given its worker node key. - * * @param workerNodeKey - The worker node key. * @param listener - The message listener callback. */ @@ -1709,7 +1714,6 @@ export abstract class AbstractPool< /** * Deregisters a listener callback on the worker given its worker node key. - * * @param workerNodeKey - The worker node key. * @param listener - The message listener callback. */ @@ -1723,7 +1727,6 @@ export abstract class AbstractPool< /** * Method hooked up after a worker node has been newly created. * Can be overridden. - * * @param workerNodeKey - The newly created worker node key. */ protected afterWorkerNodeSetup (workerNodeKey: number): void { @@ -1754,14 +1757,12 @@ export abstract class AbstractPool< /** * Sends the startup message to worker given its worker node key. - * * @param workerNodeKey - The worker node key. */ protected abstract sendStartupMessageToWorker (workerNodeKey: number): void /** * Sends the statistics message to worker given its worker node key. - * * @param workerNodeKey - The worker node key. */ private sendStatisticsMessageToWorker (workerNodeKey: number): void { @@ -1772,8 +1773,8 @@ export abstract class AbstractPool< .runTime.aggregate ?? false, elu: this.workerChoiceStrategiesContext?.getTaskStatisticsRequirements() - .elu.aggregate ?? false - } + .elu.aggregate ?? false, + }, }) } @@ -1894,7 +1895,7 @@ export abstract class AbstractPool< const workerInfo = this.getWorkerInfo(workerNodeKey) if (workerInfo == null) { throw new Error( - `Worker node with key '${workerNodeKey}' not found in pool` + `Worker node with key '${workerNodeKey.toString()}' not found in pool` ) } if ( @@ -2010,7 +2011,7 @@ export abstract class AbstractPool< const workerInfo = this.getWorkerInfo(workerNodeKey) if (workerInfo == null) { throw new Error( - `Worker node with key '${workerNodeKey}' not found in pool` + `Worker node with key '${workerNodeKey.toString()}' not found in pool` ) } workerInfo.stealing = true @@ -2024,8 +2025,15 @@ export abstract class AbstractPool< } } + private setTasksQueuePriority (workerNodeKey: number): void { + this.workerNodes[workerNodeKey].setTasksQueuePriority( + this.getTasksQueuePriority() + ) + } + /** * This method is the message listener registered on each worker. + * @param message - The message received from the worker. */ protected readonly workerMessageListener = ( message: MessageValue @@ -2042,6 +2050,7 @@ export abstract class AbstractPool< if (workerInfo != null) { workerInfo.taskFunctionsProperties = taskFunctionsProperties this.sendStatisticsMessageToWorker(workerNodeKey) + this.setTasksQueuePriority(workerNodeKey) } } else if (taskId != null) { // Task execution response received from worker @@ -2059,13 +2068,15 @@ export abstract class AbstractPool< private handleWorkerReadyResponse (message: MessageValue): void { const { workerId, ready, taskFunctionsProperties } = message if (ready == null || !ready) { - throw new Error(`Worker ${workerId} failed to initialize`) + // eslint-disable-next-line @typescript-eslint/restrict-template-expressions + throw new Error(`Worker ${workerId?.toString()} failed to initialize`) } const workerNodeKey = this.getWorkerNodeKeyByWorkerId(workerId) const workerNode = this.workerNodes[workerNodeKey] workerNode.info.ready = ready workerNode.info.taskFunctionsProperties = taskFunctionsProperties this.sendStatisticsMessageToWorker(workerNodeKey) + this.setTasksQueuePriority(workerNodeKey) this.checkAndEmitReadyEvent() } @@ -2119,7 +2130,7 @@ export abstract class AbstractPool< ) { workerNode.emit('idle', { workerId, - workerNodeKey + workerNodeKey, }) } } @@ -2145,7 +2156,6 @@ export abstract class AbstractPool< /** * Gets the worker information given its worker node key. - * * @param workerNodeKey - The worker node key. * @returns The worker information. */ @@ -2153,9 +2163,14 @@ export abstract class AbstractPool< return this.workerNodes[workerNodeKey]?.info } + private getTasksQueuePriority (): boolean { + return this.listTaskFunctionsProperties().some( + taskFunctionProperties => taskFunctionProperties.priority != null + ) + } + /** * Creates a worker node. - * * @returns The created worker node. */ private createWorkerNode (): IWorkerNode { @@ -2170,8 +2185,8 @@ export abstract class AbstractPool< getDefaultTasksQueueOptions( this.maximumNumberOfWorkers ?? this.minimumNumberOfWorkers ).size, - tasksQueueBucketSize: - (this.maximumNumberOfWorkers ?? this.minimumNumberOfWorkers) * 2 + tasksQueueBucketSize: defaultBucketSize, + tasksQueuePriority: this.getTasksQueuePriority(), } ) // Flag the worker node as ready at pool startup. @@ -2183,7 +2198,6 @@ export abstract class AbstractPool< /** * Adds the given worker node in the pool worker nodes. - * * @param workerNode - The worker node. * @returns The added worker node key. * @throws {@link https://nodejs.org/api/errors.html#class-error} If the added worker node is not found. @@ -2206,7 +2220,6 @@ export abstract class AbstractPool< /** * Removes the worker node from the pool worker nodes. - * * @param workerNode - The worker node. */ private removeWorkerNode (workerNode: IWorkerNode): void { @@ -2236,7 +2249,6 @@ export abstract class AbstractPool< /** * Executes the given task on the worker given its worker node key. - * * @param workerNodeKey - The worker node key. * @param task - The task to execute. */