X-Git-Url: https://git.piment-noir.org/?a=blobdiff_plain;f=src%2Fpools%2Fabstract-pool.ts;h=534cb49080a7127f10a366cf9b99d6187f18fb61;hb=946b809b91118cd7442b90971344e53e15c26466;hp=6081902720554c2bcdb809d37f04d5f93ed89da7;hpb=172ed3de085bd6c59ffe877d48d4c49eeaffa8c7;p=poolifier.git diff --git a/src/pools/abstract-pool.ts b/src/pools/abstract-pool.ts index 60819027..534cb490 100644 --- a/src/pools/abstract-pool.ts +++ b/src/pools/abstract-pool.ts @@ -5,8 +5,7 @@ import { type TransferListItem } from 'node:worker_threads' import type { MessageValue, PromiseResponseWrapper, - Task, - Writable + Task } from '../utility-types' import { DEFAULT_TASK_NAME, @@ -256,19 +255,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 than zero` + `Invalid worker choice strategy options: retries '${workerChoiceStrategyOptions.retries}' must be greater or equal than zero` ) } if ( @@ -292,7 +291,7 @@ export abstract class AbstractPool< } private checkValidTasksQueueOptions ( - tasksQueueOptions: Writable + tasksQueueOptions: TasksQueueOptions ): void { if (tasksQueueOptions != null && !isPlainObject(tasksQueueOptions)) { throw new TypeError('Invalid tasks queue options: must be a plain object') @@ -313,28 +312,22 @@ export abstract class AbstractPool< `Invalid worker node tasks concurrency: ${tasksQueueOptions.concurrency} is a negative integer or zero` ) } - if ( - tasksQueueOptions?.queueMaxSize != null && - tasksQueueOptions?.size != null - ) { + if (tasksQueueOptions?.queueMaxSize != null) { throw new Error( - 'Invalid tasks queue options: cannot specify both queueMaxSize and size' + 'Invalid tasks queue options: queueMaxSize is deprecated, please use size instead' ) } - if (tasksQueueOptions?.queueMaxSize != null) { - tasksQueueOptions.size = tasksQueueOptions.queueMaxSize - } if ( tasksQueueOptions?.size != null && !Number.isSafeInteger(tasksQueueOptions.size) ) { throw new TypeError( - 'Invalid worker node tasks queue max size: must be an integer' + 'Invalid worker node tasks queue size: must be an integer' ) } if (tasksQueueOptions?.size != null && tasksQueueOptions.size <= 0) { throw new RangeError( - `Invalid worker node tasks queue max size: ${tasksQueueOptions.size} is a negative integer or zero` + `Invalid worker node tasks queue size: ${tasksQueueOptions.size} is a negative integer or zero` ) } } @@ -662,13 +655,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 } @@ -740,6 +733,14 @@ export abstract class AbstractPool< return [] } + private shallExecuteTask (workerNodeKey: number): boolean { + return ( + this.tasksQueueSize(workerNodeKey) === 0 && + this.workerNodes[workerNodeKey].usage.tasks.executing < + (this.opts.tasksQueueOptions?.concurrency as number) + ) + } + /** @inheritDoc */ public async execute ( data?: Data, @@ -749,9 +750,11 @@ export abstract class AbstractPool< return await new Promise((resolve, reject) => { if (!this.started) { reject(new Error('Cannot execute a task on destroyed pool')) + return } if (name != null && typeof name !== 'string') { reject(new TypeError('name argument must be a string')) + return } if ( name != null && @@ -759,22 +762,15 @@ export abstract class AbstractPool< name.trim().length === 0 ) { reject(new TypeError('name argument must not be an empty string')) + return } if (transferList != null && !Array.isArray(transferList)) { reject(new TypeError('transferList argument must be an array')) + return } const timestamp = performance.now() const workerNodeKey = this.chooseWorkerNode() const workerInfo = this.getWorkerInfo(workerNodeKey) as WorkerInfo - if ( - name != null && - Array.isArray(workerInfo.taskFunctions) && - !workerInfo.taskFunctions.includes(name) - ) { - reject( - new Error(`Task function '${name}' is not registered in the pool`) - ) - } const task: Task = { name: name ?? DEFAULT_TASK_NAME, // eslint-disable-next-line @typescript-eslint/consistent-type-assertions @@ -792,9 +788,7 @@ export abstract class AbstractPool< if ( this.opts.enableTasksQueue === false || (this.opts.enableTasksQueue === true && - this.tasksQueueSize(workerNodeKey) === 0 && - this.workerNodes[workerNodeKey].usage.tasks.executing < - (this.opts.tasksQueueOptions?.concurrency as number)) + this.shallExecuteTask(workerNodeKey)) ) { this.executeTask(workerNodeKey, task) } else { @@ -844,7 +838,7 @@ export abstract class AbstractPool< * @virtual */ protected setupHook (): void { - /** Intentionally empty */ + /* Intentionally empty */ } /** @@ -1225,11 +1219,7 @@ export abstract class AbstractPool< ...(this.dequeueTask(workerNodeKey) as Task), workerId: destinationWorkerNode.info.id as number } - if ( - this.tasksQueueSize(destinationWorkerNodeKey) === 0 && - destinationWorkerNode.usage.tasks.executing < - (this.opts.tasksQueueOptions?.concurrency as number) - ) { + if (this.shallExecuteTask(destinationWorkerNodeKey)) { this.executeTask(destinationWorkerNodeKey, task) } else { this.enqueueTask(destinationWorkerNodeKey, task) @@ -1238,6 +1228,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] @@ -1260,36 +1269,25 @@ export abstract class AbstractPool< ...(sourceWorkerNode.popTask() as Task), workerId: destinationWorkerNode.info.id as number } - if ( - this.tasksQueueSize(destinationWorkerNodeKey) === 0 && - destinationWorkerNode.usage.tasks.executing < - (this.opts.tasksQueueOptions?.concurrency as number) - ) { + if (this.shallExecuteTask(destinationWorkerNodeKey)) { this.executeTask(destinationWorkerNodeKey, task) } 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 { + const sizeOffset = 1 + if ((this.opts.tasksQueueOptions?.size as number) <= sizeOffset) { + return + } const sourceWorkerNode = this.workerNodes[this.getWorkerNodeKeyByWorkerId(workerId)] const workerNodes = this.workerNodes @@ -1304,33 +1302,21 @@ 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), workerId: workerNode.info.id as number } - if ( - this.tasksQueueSize(workerNodeKey) === 0 && - workerNode.usage.tasks.executing < - (this.opts.tasksQueueOptions?.concurrency as number) - ) { + if (this.shallExecuteTask(workerNodeKey)) { this.executeTask(workerNodeKey, task) } 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 + ) } } } @@ -1386,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 && @@ -1398,7 +1385,6 @@ export abstract class AbstractPool< this.dequeueTask(workerNodeKey) as Task ) } - this.workerChoiceStrategyContext.update(workerNodeKey) } } @@ -1442,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. @@ -1452,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 }