X-Git-Url: https://git.piment-noir.org/?a=blobdiff_plain;ds=inline;f=src%2Fpools%2Fabstract-pool.ts;h=da0f2e77358525336a5936a42841edce9016f44b;hb=70726c2810ba4768dae5ae1f40ed6ddcecf142ba;hp=6f5f3aeb1f581ea3306169cdb35d2bf9b71e0c01;hpb=4134429293801daed1762d927f66cee979de8864;p=poolifier.git diff --git a/src/pools/abstract-pool.ts b/src/pools/abstract-pool.ts index 6f5f3aeb..da0f2e77 100644 --- a/src/pools/abstract-pool.ts +++ b/src/pools/abstract-pool.ts @@ -23,12 +23,13 @@ import { } from './pool' import type { IWorker, + MessageHandler, Task, - TaskStatistics, WorkerNode, WorkerUsage } from './worker' import { + Measurements, WorkerChoiceStrategies, type WorkerChoiceStrategy, type WorkerChoiceStrategyOptions @@ -39,8 +40,8 @@ import { WorkerChoiceStrategyContext } from './selection-strategies/worker-choic * 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 serializable data. - * @typeParam Response - Type of execution response. This can only be serializable data. + * @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. */ export abstract class AbstractPool< Worker extends IWorker, @@ -199,6 +200,16 @@ export abstract class AbstractPool< 'Invalid worker choice strategy options: must have a weight for each worker node' ) } + if ( + workerChoiceStrategyOptions.measurement != null && + !Object.values(Measurements).includes( + workerChoiceStrategyOptions.measurement + ) + ) { + throw new Error( + `Invalid worker choice strategy options: invalid measurement '${workerChoiceStrategyOptions.measurement}'` + ) + } } private checkValidTasksQueueOptions ( @@ -207,11 +218,20 @@ export abstract class AbstractPool< if (tasksQueueOptions != null && !isPlainObject(tasksQueueOptions)) { throw new TypeError('Invalid tasks queue options: must be a plain object') } - if ((tasksQueueOptions?.concurrency as number) <= 0) { + if ( + tasksQueueOptions?.concurrency != null && + !Number.isSafeInteger(tasksQueueOptions.concurrency) + ) { + throw new TypeError( + 'Invalid worker tasks concurrency: must be an integer' + ) + } + if ( + tasksQueueOptions?.concurrency != null && + tasksQueueOptions.concurrency <= 0 + ) { throw new Error( - `Invalid worker tasks concurrency '${ - tasksQueueOptions.concurrency as number - }'` + `Invalid worker tasks concurrency '${tasksQueueOptions.concurrency}'` ) } } @@ -249,12 +269,13 @@ export abstract class AbstractPool< 0 ), queuedTasks: this.workerNodes.reduce( - (accumulator, workerNode) => accumulator + workerNode.tasksQueue.size, + (accumulator, workerNode) => + accumulator + workerNode.workerUsage.tasks.queued, 0 ), maxQueuedTasks: this.workerNodes.reduce( (accumulator, workerNode) => - accumulator + workerNode.tasksQueue.maxSize, + accumulator + workerNode.workerUsage.tasks.maxQueued, 0 ), failedTasks: this.workerNodes.reduce( @@ -312,10 +333,10 @@ export abstract class AbstractPool< if (workerChoiceStrategyOptions != null) { this.setWorkerChoiceStrategyOptions(workerChoiceStrategyOptions) } - for (const workerNode of this.workerNodes) { + for (const [workerNodeKey, workerNode] of this.workerNodes.entries()) { this.setWorkerNodeTasksUsage( workerNode, - this.getWorkerUsage(workerNode.worker) + this.getWorkerUsage(workerNodeKey) ) this.setWorkerStatistics(workerNode.worker) } @@ -421,7 +442,6 @@ export abstract class AbstractPool< } else { this.executeTask(workerNodeKey, submittedTask) } - this.workerChoiceStrategyContext.update(workerNodeKey) this.checkAndEmitEvents() // eslint-disable-next-line @typescript-eslint/return-await return res @@ -446,8 +466,8 @@ export abstract class AbstractPool< protected abstract destroyWorker (worker: Worker): void | Promise /** - * Setup hook to execute code before worker node are created in the abstract constructor. - * Can be overridden + * Setup hook to execute code before worker nodes are created in the abstract constructor. + * Can be overridden. * * @virtual */ @@ -656,9 +676,12 @@ export abstract class AbstractPool< * @param worker - The worker which should register a listener. * @param listener - The message listener callback. */ - protected abstract registerWorkerMessageListener< - Message extends Data | Response - >(worker: Worker, listener: (message: MessageValue) => void): void + private registerWorkerMessageListener( + worker: Worker, + listener: (message: MessageValue) => void + ): void { + worker.on('message', listener as MessageHandler) + } /** * Creates a new worker. @@ -669,12 +692,14 @@ export abstract class AbstractPool< /** * Function that can be hooked up when a worker has been newly created and moved to the pool worker nodes. - * - * Can be used to update the `maxListeners` or binding the `main-worker`\<-\>`worker` connection if not bind by default. + * Can be overridden. * * @param worker - The newly created worker. */ - protected abstract afterWorkerSetup (worker: Worker): void + protected afterWorkerSetup (worker: Worker): void { + // Listen to worker messages. + this.registerWorkerMessageListener(worker, this.workerListener()) + } /** * Creates a new worker and sets it up completely in the pool worker nodes. @@ -717,16 +742,19 @@ export abstract class AbstractPool< protected createAndSetupDynamicWorker (): Worker { const worker = this.createAndSetupWorker() this.registerWorkerMessageListener(worker, message => { - const currentWorkerNodeKey = this.getWorkerNodeKey(worker) + const workerNodeKey = this.getWorkerNodeKey(worker) if ( isKillBehavior(KillBehaviors.HARD, message.kill) || (message.kill != null && - this.workerNodes[currentWorkerNodeKey].workerUsage.tasks.executing === - 0) + ((this.opts.enableTasksQueue === false && + this.workerNodes[workerNodeKey].workerUsage.tasks.executing === + 0) || + (this.opts.enableTasksQueue === true && + this.workerNodes[workerNodeKey].workerUsage.tasks.executing === + 0 && + this.tasksQueueSize(workerNodeKey) === 0))) ) { // Kill message received from the worker: no new tasks are submitted to that worker for a while ( > maxInactiveTime) - this.flushTasksQueue(currentWorkerNodeKey) - // FIXME: wait for tasks to be finished void (this.destroyWorker(worker) as Promise) } }) @@ -745,10 +773,10 @@ export abstract class AbstractPool< const promiseResponse = this.promiseResponseMap.get(message.id) if (promiseResponse != null) { if (message.taskError != null) { - promiseResponse.reject(message.taskError.message) if (this.emitter != null) { this.emitter.emit(PoolEvents.taskError, message.taskError) } + promiseResponse.reject(message.taskError.message) } else { promiseResponse.resolve(message.data as Response) } @@ -764,6 +792,7 @@ export abstract class AbstractPool< this.dequeueTask(workerNodeKey) as Task ) } + this.workerChoiceStrategyContext.update(workerNodeKey) } } } @@ -800,11 +829,17 @@ export abstract class AbstractPool< * @returns The worker nodes length. */ private pushWorkerNode (worker: Worker): number { - return this.workerNodes.push({ + this.workerNodes.push({ worker, - workerUsage: this.getWorkerUsage(worker), + workerUsage: this.getWorkerUsage(), tasksQueue: new Queue>() }) + const workerNodeKey = this.getWorkerNodeKey(worker) + this.setWorkerNodeTasksUsage( + this.workerNodes[workerNodeKey], + this.getWorkerUsage(workerNodeKey) + ) + return this.workerNodes.length } // /** @@ -858,6 +893,10 @@ export abstract class AbstractPool< return this.workerNodes[workerNodeKey].tasksQueue.size } + private tasksMaxQueueSize (workerNodeKey: number): number { + return this.workerNodes[workerNodeKey].tasksQueue.maxSize + } + private flushTasksQueue (workerNodeKey: number): void { if (this.tasksQueueSize(workerNodeKey) > 0) { for (let i = 0; i < this.tasksQueueSize(workerNodeKey); i++) { @@ -867,6 +906,7 @@ export abstract class AbstractPool< ) } } + this.workerNodes[workerNodeKey].tasksQueue.clear() } private flushTasksQueues (): void { @@ -887,9 +927,25 @@ export abstract class AbstractPool< }) } - private getWorkerUsage (worker: Worker): WorkerUsage { + private getWorkerUsage (workerNodeKey?: number): WorkerUsage { + const getTasksQueueSize = (workerNodeKey?: number): number => { + return workerNodeKey != null ? this.tasksQueueSize(workerNodeKey) : 0 + } + const getTasksMaxQueueSize = (workerNodeKey?: number): number => { + return workerNodeKey != null ? this.tasksMaxQueueSize(workerNodeKey) : 0 + } return { - tasks: this.getTaskStatistics(worker), + tasks: { + executed: 0, + executing: 0, + get queued (): number { + return getTasksQueueSize(workerNodeKey) + }, + get maxQueued (): number { + return getTasksMaxQueueSize(workerNodeKey) + }, + failed: 0 + }, runTime: { aggregate: 0, average: 0, @@ -919,17 +975,4 @@ export abstract class AbstractPool< } } } - - private getTaskStatistics (worker: Worker): TaskStatistics { - const queueSize = - this.workerNodes[this.getWorkerNodeKey(worker)]?.tasksQueue?.size - return { - executed: 0, - executing: 0, - get queued (): number { - return queueSize ?? 0 - }, - failed: 0 - } - } }