X-Git-Url: https://git.piment-noir.org/?a=blobdiff_plain;f=src%2Fpools%2Fabstract-pool.ts;h=fb1bf8458ad7a970f27ae7a4dcbb300faf5bf848;hb=dbd73092cca6cabb2b41e18b944656fc43f8757b;hp=482a20db4fc044d0f118b6d7ae2cc9f5af962de9;hpb=75de9f41ce00bec38febd6d82653d3d82f1bb884;p=poolifier.git diff --git a/src/pools/abstract-pool.ts b/src/pools/abstract-pool.ts index 482a20db..fb1bf845 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' @@ -90,14 +92,14 @@ export abstract class AbstractPool< */ protected readonly max?: number - /** - * Whether the pool is starting or not. - */ - private readonly starting: boolean /** * Whether the pool is started or not. */ private started: boolean + /** + * Whether the pool is starting or not. + */ + private starting: boolean /** * The start timestamp of the pool. */ @@ -143,10 +145,11 @@ export abstract class AbstractPool< this.setupHook() - this.starting = true - this.startPool() + this.started = false this.starting = false - this.started = true + if (this.opts.startWorkers === true) { + this.start() + } this.startTimestamp = performance.now() } @@ -210,6 +213,7 @@ export abstract class AbstractPool< private checkPoolOptions (opts: PoolOptions): void { if (isPlainObject(opts)) { + this.opts.startWorkers = opts.startWorkers ?? true this.opts.workerChoiceStrategy = opts.workerChoiceStrategy ?? WorkerChoiceStrategies.ROUND_ROBIN this.checkValidWorkerChoiceStrategy(this.opts.workerChoiceStrategy) @@ -255,19 +259,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 +302,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,50 +310,34 @@ 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` - ) - } - if (tasksQueueOptions?.queueMaxSize != null) { - throw new Error( - 'Invalid tasks queue options: queueMaxSize is deprecated, please use size instead' + `Invalid worker node tasks concurrency: ${tasksQueueOptions?.concurrency} is a negative integer or zero` ) } 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` ) } } - private startPool (): void { - while ( - this.workerNodes.reduce( - (accumulator, workerNode) => - !workerNode.info.dynamic ? accumulator + 1 : accumulator, - 0 - ) < this.numberOfWorkers - ) { - this.createAndSetupWorkerNode() - } - } - /** @inheritDoc */ public get info (): PoolInfo { return { version, type: this.type, worker: this.worker, + started: this.started, ready: this.ready, strategy: this.opts.workerChoiceStrategy as WorkerChoiceStrategy, minSize: this.minSize, @@ -414,16 +402,16 @@ export abstract class AbstractPool< .runTime.aggregate && { runTime: { minimum: round( - Math.min( + min( ...this.workerNodes.map( - (workerNode) => workerNode.usage.runTime?.minimum ?? Infinity + workerNode => workerNode.usage.runTime?.minimum ?? Infinity ) ) ), maximum: round( - Math.max( + max( ...this.workerNodes.map( - (workerNode) => workerNode.usage.runTime?.maximum ?? -Infinity + workerNode => workerNode.usage.runTime?.maximum ?? -Infinity ) ) ), @@ -457,16 +445,16 @@ export abstract class AbstractPool< .waitTime.aggregate && { waitTime: { minimum: round( - Math.min( + min( ...this.workerNodes.map( - (workerNode) => workerNode.usage.waitTime?.minimum ?? Infinity + workerNode => workerNode.usage.waitTime?.minimum ?? Infinity ) ) ), maximum: round( - Math.max( + max( ...this.workerNodes.map( - (workerNode) => workerNode.usage.waitTime?.maximum ?? -Infinity + workerNode => workerNode.usage.waitTime?.maximum ?? -Infinity ) ) ), @@ -588,7 +576,7 @@ export abstract class AbstractPool< */ private getWorkerNodeKeyByWorker (worker: Worker): number { return this.workerNodes.findIndex( - (workerNode) => workerNode.worker === worker + workerNode => workerNode.worker === worker ) } @@ -600,7 +588,7 @@ export abstract class AbstractPool< */ private getWorkerNodeKeyByWorkerId (workerId: number): number { return this.workerNodes.findIndex( - (workerNode) => workerNode.info.id === workerId + workerNode => workerNode.info.id === workerId ) } @@ -655,13 +643,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 } @@ -673,7 +661,9 @@ export abstract class AbstractPool< return { ...{ size: Math.pow(this.maxSize, 2), - concurrency: 1 + concurrency: 1, + taskStealing: true, + tasksStealingOnBackPressure: true }, ...tasksQueueOptions } @@ -704,7 +694,7 @@ export abstract class AbstractPool< if (this.opts.enableTasksQueue === true) { return ( this.workerNodes.findIndex( - (workerNode) => + workerNode => workerNode.info.ready && workerNode.usage.tasks.executing < (this.opts.tasksQueueOptions?.concurrency as number) @@ -713,7 +703,7 @@ export abstract class AbstractPool< } else { return ( this.workerNodes.findIndex( - (workerNode) => + workerNode => workerNode.info.ready && workerNode.usage.tasks.executing === 0 ) === -1 ) @@ -749,7 +739,7 @@ export abstract class AbstractPool< ): Promise { return await new Promise((resolve, reject) => { if (!this.started) { - reject(new Error('Cannot execute a task on destroyed pool')) + reject(new Error('Cannot execute a task on not started pool')) return } if (name != null && typeof name !== 'string') { @@ -770,14 +760,13 @@ export abstract class AbstractPool< } const timestamp = performance.now() const workerNodeKey = this.chooseWorkerNode() - const workerInfo = this.getWorkerInfo(workerNodeKey) as WorkerInfo const task: Task = { name: name ?? DEFAULT_TASK_NAME, // eslint-disable-next-line @typescript-eslint/consistent-type-assertions data: data ?? ({} as Data), transferList, timestamp, - workerId: workerInfo.id as number, + workerId: this.getWorkerInfo(workerNodeKey).id as number, taskId: randomUUID() } this.promiseResponseMap.set(task.taskId as string, { @@ -797,6 +786,22 @@ export abstract class AbstractPool< }) } + /** @inheritdoc */ + public start (): void { + this.starting = true + while ( + this.workerNodes.reduce( + (accumulator, workerNode) => + !workerNode.info.dynamic ? accumulator + 1 : accumulator, + 0 + ) < this.numberOfWorkers + ) { + this.createAndSetupWorkerNode() + } + this.starting = false + this.started = true + } + /** @inheritDoc */ public async destroy (): Promise { await Promise.all( @@ -813,7 +818,7 @@ export abstract class AbstractPool< workerId: number ): Promise { await new Promise((resolve, reject) => { - this.registerWorkerMessageListener(workerNodeKey, (message) => { + this.registerWorkerMessageListener(workerNodeKey, message => { if (message.kill === 'success') { resolve() } else if (message.kill === 'failure') { @@ -838,7 +843,7 @@ export abstract class AbstractPool< * @virtual */ protected setupHook (): void { - /** Intentionally empty */ + /* Intentionally empty */ } /** @@ -1062,16 +1067,16 @@ export abstract class AbstractPool< worker.on('online', this.opts.onlineHandler ?? EMPTY_FUNCTION) worker.on('message', this.opts.messageHandler ?? EMPTY_FUNCTION) worker.on('error', this.opts.errorHandler ?? EMPTY_FUNCTION) - worker.on('error', (error) => { + 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) if ( this.opts.restartWorkerOnError === true && - !this.starting && - this.started + this.started && + !this.starting ) { if (workerInfo.dynamic) { this.createAndSetupDynamicWorkerNode() @@ -1102,7 +1107,7 @@ export abstract class AbstractPool< */ protected createAndSetupDynamicWorkerNode (): number { const workerNodeKey = this.createAndSetupWorkerNode() - this.registerWorkerMessageListener(workerNodeKey, (message) => { + this.registerWorkerMessageListener(workerNodeKey, message => { const localWorkerNodeKey = this.getWorkerNodeKeyByWorkerId( message.workerId ) @@ -1117,12 +1122,12 @@ export abstract class AbstractPool< workerUsage.tasks.executing === 0 && this.tasksQueueSize(localWorkerNodeKey) === 0))) ) { - this.destroyWorkerNode(localWorkerNodeKey).catch((error) => { + this.destroyWorkerNode(localWorkerNodeKey).catch(error => { this.emitter?.emit(PoolEvents.error, error) }) } }) - const workerInfo = this.getWorkerInfo(workerNodeKey) as WorkerInfo + const workerInfo = this.getWorkerInfo(workerNodeKey) this.sendToWorker(workerNodeKey, { checkActive: true, workerId: workerInfo.id as number @@ -1165,10 +1170,14 @@ export abstract class AbstractPool< // Send the statistics message to worker. this.sendStatisticsMessageToWorker(workerNodeKey) if (this.opts.enableTasksQueue === true) { - this.workerNodes[workerNodeKey].onEmptyQueue = - this.taskStealingOnEmptyQueue.bind(this) - this.workerNodes[workerNodeKey].onBackPressure = - this.tasksStealingOnBackPressure.bind(this) + if (this.opts.tasksQueueOptions?.taskStealing === true) { + this.workerNodes[workerNodeKey].onEmptyQueue = + this.taskStealingOnEmptyQueue.bind(this) + } + if (this.opts.tasksQueueOptions?.tasksStealingOnBackPressure === true) { + this.workerNodes[workerNodeKey].onBackPressure = + this.tasksStealingOnBackPressure.bind(this) + } } } @@ -1193,41 +1202,54 @@ 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 }) } private redistributeQueuedTasks (workerNodeKey: number): void { while (this.tasksQueueSize(workerNodeKey) > 0) { - let destinationWorkerNodeKey!: number - let minQueuedTasks = Infinity - for (const [workerNodeId, workerNode] of this.workerNodes.entries()) { - if (workerNode.info.ready && workerNodeId !== workerNodeKey) { - if (workerNode.usage.tasks.queued === 0) { - destinationWorkerNodeKey = workerNodeId - break - } - if (workerNode.usage.tasks.queued < minQueuedTasks) { - minQueuedTasks = workerNode.usage.tasks.queued - destinationWorkerNodeKey = workerNodeId - } - } + const destinationWorkerNodeKey = this.workerNodes.reduce( + (minWorkerNodeKey, workerNode, workerNodeKey, workerNodes) => { + return workerNode.info.ready && + workerNode.usage.tasks.queued < + workerNodes[minWorkerNodeKey].usage.tasks.queued + ? workerNodeKey + : minWorkerNodeKey + }, + 0 + ) + const destinationWorkerNode = this.workerNodes[destinationWorkerNodeKey] + const task = { + ...(this.dequeueTask(workerNodeKey) as Task), + workerId: destinationWorkerNode.info.id as number } - if (destinationWorkerNodeKey != null) { - const destinationWorkerNode = this.workerNodes[destinationWorkerNodeKey] - const task = { - ...(this.dequeueTask(workerNodeKey) as Task), - workerId: destinationWorkerNode.info.id as number - } - if (this.shallExecuteTask(destinationWorkerNodeKey)) { - this.executeTask(destinationWorkerNodeKey, task) - } else { - this.enqueueTask(destinationWorkerNodeKey, task) - } + if (this.shallExecuteTask(destinationWorkerNodeKey)) { + this.executeTask(destinationWorkerNodeKey, task) + } else { + this.enqueueTask(destinationWorkerNodeKey, task) } } } + 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] @@ -1237,46 +1259,32 @@ export abstract class AbstractPool< (workerNodeA, workerNodeB) => workerNodeB.usage.tasks.queued - workerNodeA.usage.tasks.queued ) - for (const sourceWorkerNode of workerNodes) { - if (sourceWorkerNode.usage.tasks.queued === 0) { - break + const sourceWorkerNode = workerNodes.find( + workerNode => + workerNode.info.ready && + workerNode.info.id !== workerId && + workerNode.usage.tasks.queued > 0 + ) + if (sourceWorkerNode != null) { + const task = { + ...(sourceWorkerNode.popTask() as Task), + workerId: destinationWorkerNode.info.id as number } - if ( - sourceWorkerNode.info.ready && - sourceWorkerNode.info.id !== workerId && - sourceWorkerNode.usage.tasks.queued > 0 - ) { - const task = { - ...(sourceWorkerNode.popTask() as Task), - workerId: destinationWorkerNode.info.id 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 - } - break + if (this.shallExecuteTask(destinationWorkerNodeKey)) { + this.executeTask(destinationWorkerNodeKey, task) + } else { + this.enqueueTask(destinationWorkerNodeKey, task) } + this.updateTaskStolenStatisticsWorkerUsage( + destinationWorkerNodeKey, + task.name as string + ) } } 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 +1301,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 +1312,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 + ) } } } @@ -1326,7 +1326,7 @@ export abstract class AbstractPool< * @returns The listener function to execute when a message is received from a worker. */ protected workerListener (): (message: MessageValue) => void { - return (message) => { + return message => { this.checkMessageWorkerId(message) if (message.ready != null && message.taskFunctions != null) { // Worker ready response received from worker @@ -1336,10 +1336,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 +1349,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 +1369,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 +1382,6 @@ export abstract class AbstractPool< this.dequeueTask(workerNodeKey) as Task ) } - this.workerChoiceStrategyContext.update(workerNodeKey) } } @@ -1413,8 +1411,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 } /** @@ -1436,7 +1434,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 } @@ -1466,7 +1464,7 @@ export abstract class AbstractPool< return ( this.opts.enableTasksQueue === true && this.workerNodes.findIndex( - (workerNode) => !workerNode.hasBackPressure() + workerNode => !workerNode.hasBackPressure() ) === -1 ) }