X-Git-Url: https://git.piment-noir.org/?a=blobdiff_plain;f=src%2Fpools%2Fabstract-pool.ts;h=6668361f9491a7cf927fba8734a11c37781a3b87;hb=refs%2Ftags%2Fv2.6.41;hp=be8ec1a0125644bbcb00b3a5ec48e3056e16ed11;hpb=375f7504cdd1e27b29c47f6a4ffc0669cab855f3;p=poolifier.git diff --git a/src/pools/abstract-pool.ts b/src/pools/abstract-pool.ts index be8ec1a0..6668361f 100644 --- a/src/pools/abstract-pool.ts +++ b/src/pools/abstract-pool.ts @@ -14,7 +14,9 @@ import { average, isKillBehavior, isPlainObject, + max, median, + min, round, updateMeasurementStatistics } from '../utils' @@ -255,19 +257,19 @@ export abstract class AbstractPool< ) } if ( - workerChoiceStrategyOptions.choiceRetries != null && - !Number.isSafeInteger(workerChoiceStrategyOptions.choiceRetries) + workerChoiceStrategyOptions.retries != null && + !Number.isSafeInteger(workerChoiceStrategyOptions.retries) ) { throw new TypeError( - 'Invalid worker choice strategy options: choice retries must be an integer' + 'Invalid worker choice strategy options: retries must be an integer' ) } if ( - workerChoiceStrategyOptions.choiceRetries != null && - workerChoiceStrategyOptions.choiceRetries < 0 + workerChoiceStrategyOptions.retries != null && + workerChoiceStrategyOptions.retries < 0 ) { throw new RangeError( - `Invalid worker choice strategy options: choice retries '${workerChoiceStrategyOptions.choiceRetries}' must be greater or equal than zero` + `Invalid worker choice strategy options: retries '${workerChoiceStrategyOptions.retries}' must be greater or equal than zero` ) } if ( @@ -298,7 +300,7 @@ export abstract class AbstractPool< } if ( tasksQueueOptions?.concurrency != null && - !Number.isSafeInteger(tasksQueueOptions.concurrency) + !Number.isSafeInteger(tasksQueueOptions?.concurrency) ) { throw new TypeError( 'Invalid worker node tasks concurrency: must be an integer' @@ -306,10 +308,10 @@ export abstract class AbstractPool< } if ( tasksQueueOptions?.concurrency != null && - tasksQueueOptions.concurrency <= 0 + tasksQueueOptions?.concurrency <= 0 ) { throw new RangeError( - `Invalid worker node tasks concurrency: ${tasksQueueOptions.concurrency} is a negative integer or zero` + `Invalid worker node tasks concurrency: ${tasksQueueOptions?.concurrency} is a negative integer or zero` ) } if (tasksQueueOptions?.queueMaxSize != null) { @@ -319,15 +321,15 @@ export abstract class AbstractPool< } if ( tasksQueueOptions?.size != null && - !Number.isSafeInteger(tasksQueueOptions.size) + !Number.isSafeInteger(tasksQueueOptions?.size) ) { throw new TypeError( 'Invalid worker node tasks queue size: must be an integer' ) } - if (tasksQueueOptions?.size != null && tasksQueueOptions.size <= 0) { + if (tasksQueueOptions?.size != null && tasksQueueOptions?.size <= 0) { throw new RangeError( - `Invalid worker node tasks queue size: ${tasksQueueOptions.size} is a negative integer or zero` + `Invalid worker node tasks queue size: ${tasksQueueOptions?.size} is a negative integer or zero` ) } } @@ -414,14 +416,14 @@ export abstract class AbstractPool< .runTime.aggregate && { runTime: { minimum: round( - Math.min( + min( ...this.workerNodes.map( (workerNode) => workerNode.usage.runTime?.minimum ?? Infinity ) ) ), maximum: round( - Math.max( + max( ...this.workerNodes.map( (workerNode) => workerNode.usage.runTime?.maximum ?? -Infinity ) @@ -457,14 +459,14 @@ export abstract class AbstractPool< .waitTime.aggregate && { waitTime: { minimum: round( - Math.min( + min( ...this.workerNodes.map( (workerNode) => workerNode.usage.waitTime?.minimum ?? Infinity ) ) ), maximum: round( - Math.max( + max( ...this.workerNodes.map( (workerNode) => workerNode.usage.waitTime?.maximum ?? -Infinity ) @@ -655,13 +657,13 @@ export abstract class AbstractPool< this.checkValidTasksQueueOptions(tasksQueueOptions) this.opts.tasksQueueOptions = this.buildTasksQueueOptions(tasksQueueOptions) - this.setTasksQueueMaxSize(this.opts.tasksQueueOptions.size as number) + this.setTasksQueueSize(this.opts.tasksQueueOptions.size as number) } else if (this.opts.tasksQueueOptions != null) { delete this.opts.tasksQueueOptions } } - private setTasksQueueMaxSize (size: number): void { + private setTasksQueueSize (size: number): void { for (const workerNode of this.workerNodes) { workerNode.tasksQueueBackPressureSize = size } @@ -770,7 +772,7 @@ export abstract class AbstractPool< } const timestamp = performance.now() const workerNodeKey = this.chooseWorkerNode() - const workerInfo = this.getWorkerInfo(workerNodeKey) as WorkerInfo + const workerInfo = this.getWorkerInfo(workerNodeKey) const task: Task = { name: name ?? DEFAULT_TASK_NAME, // eslint-disable-next-line @typescript-eslint/consistent-type-assertions @@ -838,7 +840,7 @@ export abstract class AbstractPool< * @virtual */ protected setupHook (): void { - /** Intentionally empty */ + /* Intentionally empty */ } /** @@ -1064,7 +1066,7 @@ export abstract class AbstractPool< worker.on('error', this.opts.errorHandler ?? EMPTY_FUNCTION) worker.on('error', (error) => { const workerNodeKey = this.getWorkerNodeKeyByWorker(worker) - const workerInfo = this.getWorkerInfo(workerNodeKey) as WorkerInfo + const workerInfo = this.getWorkerInfo(workerNodeKey) workerInfo.ready = false this.workerNodes[workerNodeKey].closeChannel() this.emitter?.emit(PoolEvents.error, error) @@ -1122,7 +1124,7 @@ export abstract class AbstractPool< }) } }) - const workerInfo = this.getWorkerInfo(workerNodeKey) as WorkerInfo + const workerInfo = this.getWorkerInfo(workerNodeKey) this.sendToWorker(workerNodeKey, { checkActive: true, workerId: workerInfo.id as number @@ -1193,7 +1195,7 @@ export abstract class AbstractPool< elu: this.workerChoiceStrategyContext.getTaskStatisticsRequirements() .elu.aggregate }, - workerId: (this.getWorkerInfo(workerNodeKey) as WorkerInfo).id as number + workerId: this.getWorkerInfo(workerNodeKey).id as number }) } @@ -1228,6 +1230,25 @@ export abstract class AbstractPool< } } + private updateTaskStolenStatisticsWorkerUsage ( + workerNodeKey: number, + taskName: string + ): void { + const workerNode = this.workerNodes[workerNodeKey] + if (workerNode?.usage != null) { + ++workerNode.usage.tasks.stolen + } + if ( + this.shallUpdateTaskFunctionWorkerUsage(workerNodeKey) && + workerNode.getTaskFunctionWorkerUsage(taskName) != null + ) { + const taskFunctionWorkerUsage = workerNode.getTaskFunctionWorkerUsage( + taskName + ) as WorkerUsage + ++taskFunctionWorkerUsage.tasks.stolen + } + } + private taskStealingOnEmptyQueue (workerId: number): void { const destinationWorkerNodeKey = this.getWorkerNodeKeyByWorkerId(workerId) const destinationWorkerNode = this.workerNodes[destinationWorkerNodeKey] @@ -1255,28 +1276,18 @@ export abstract class AbstractPool< } else { this.enqueueTask(destinationWorkerNodeKey, task) } - if (destinationWorkerNode?.usage != null) { - ++destinationWorkerNode.usage.tasks.stolen - } - if ( - this.shallUpdateTaskFunctionWorkerUsage(destinationWorkerNodeKey) && - destinationWorkerNode.getTaskFunctionWorkerUsage( - task.name as string - ) != null - ) { - const taskFunctionWorkerUsage = - destinationWorkerNode.getTaskFunctionWorkerUsage( - task.name as string - ) as WorkerUsage - ++taskFunctionWorkerUsage.tasks.stolen - } + this.updateTaskStolenStatisticsWorkerUsage( + destinationWorkerNodeKey, + task.name as string + ) break } } } private tasksStealingOnBackPressure (workerId: number): void { - if ((this.opts.tasksQueueOptions?.size as number) <= 1) { + const sizeOffset = 1 + if ((this.opts.tasksQueueOptions?.size as number) <= sizeOffset) { return } const sourceWorkerNode = @@ -1293,7 +1304,7 @@ export abstract class AbstractPool< workerNode.info.ready && workerNode.info.id !== workerId && workerNode.usage.tasks.queued < - (this.opts.tasksQueueOptions?.size as number) - 1 + (this.opts.tasksQueueOptions?.size as number) - sizeOffset ) { const task = { ...(sourceWorkerNode.popTask() as Task), @@ -1304,18 +1315,10 @@ export abstract class AbstractPool< } else { this.enqueueTask(workerNodeKey, task) } - if (workerNode?.usage != null) { - ++workerNode.usage.tasks.stolen - } - if ( - this.shallUpdateTaskFunctionWorkerUsage(workerNodeKey) && - workerNode.getTaskFunctionWorkerUsage(task.name as string) != null - ) { - const taskFunctionWorkerUsage = workerNode.getTaskFunctionWorkerUsage( - task.name as string - ) as WorkerUsage - ++taskFunctionWorkerUsage.tasks.stolen - } + this.updateTaskStolenStatisticsWorkerUsage( + workerNodeKey, + task.name as string + ) } } } @@ -1336,10 +1339,8 @@ export abstract class AbstractPool< this.handleTaskExecutionResponse(message) } else if (message.taskFunctions != null) { // Task functions message received from worker - ( - this.getWorkerInfo( - this.getWorkerNodeKeyByWorkerId(message.workerId) - ) as WorkerInfo + this.getWorkerInfo( + this.getWorkerNodeKeyByWorkerId(message.workerId) ).taskFunctions = message.taskFunctions } } @@ -1351,7 +1352,7 @@ export abstract class AbstractPool< } const workerInfo = this.getWorkerInfo( this.getWorkerNodeKeyByWorkerId(message.workerId) - ) as WorkerInfo + ) workerInfo.ready = message.ready as boolean workerInfo.taskFunctions = message.taskFunctions if (this.emitter != null && this.ready) { @@ -1371,6 +1372,7 @@ export abstract class AbstractPool< } const workerNodeKey = promiseResponse.workerNodeKey this.afterTaskExecutionHook(workerNodeKey, message) + this.workerChoiceStrategyContext.update(workerNodeKey) this.promiseResponseMap.delete(taskId as string) if ( this.opts.enableTasksQueue === true && @@ -1383,7 +1385,6 @@ export abstract class AbstractPool< this.dequeueTask(workerNodeKey) as Task ) } - this.workerChoiceStrategyContext.update(workerNodeKey) } } @@ -1413,8 +1414,8 @@ export abstract class AbstractPool< * @param workerNodeKey - The worker node key. * @returns The worker information. */ - protected getWorkerInfo (workerNodeKey: number): WorkerInfo | undefined { - return this.workerNodes[workerNodeKey]?.info + protected getWorkerInfo (workerNodeKey: number): WorkerInfo { + return this.workerNodes[workerNodeKey].info } /** @@ -1427,7 +1428,6 @@ export abstract class AbstractPool< private addWorkerNode (worker: Worker): number { const workerNode = new WorkerNode( worker, - this.worker, this.opts.tasksQueueOptions?.size ?? Math.pow(this.maxSize, 2) ) // Flag the worker node as ready at pool startup. @@ -1437,7 +1437,7 @@ export abstract class AbstractPool< this.workerNodes.push(workerNode) const workerNodeKey = this.getWorkerNodeKeyByWorker(worker) if (workerNodeKey === -1) { - throw new Error('Worker node added not found') + throw new Error('Worker added not found in worker nodes') } return workerNodeKey }