X-Git-Url: https://git.piment-noir.org/?a=blobdiff_plain;f=src%2Fpools%2Fabstract-pool.ts;h=0069fb7fadc61c95196982365adfaf60561069fa;hb=c05f0d5094264db0ff50636e794201e60ba7f448;hp=24e8a5b0c670f275a91a93dec22227fa12659d25;hpb=5931725336d0de8a06146641ae6feef22bf60e6e;p=poolifier.git diff --git a/src/pools/abstract-pool.ts b/src/pools/abstract-pool.ts index 24e8a5b0..0069fb7f 100644 --- a/src/pools/abstract-pool.ts +++ b/src/pools/abstract-pool.ts @@ -253,7 +253,10 @@ export abstract class AbstractPool< worker: this.worker, minSize: this.minSize, maxSize: this.maxSize, - utilization: round(this.utilization), + ...(this.workerChoiceStrategyContext.getTaskStatisticsRequirements() + .runTime.aggregate && + this.workerChoiceStrategyContext.getTaskStatisticsRequirements().runTime + .aggregate && { utilization: round(this.utilization) }), workerNodes: this.workerNodes.length, idleWorkerNodes: this.workerNodes.reduce( (accumulator, workerNode) => @@ -816,42 +819,53 @@ export abstract class AbstractPool< return message => { if (message.workerId != null && message.started != null) { // Worker started message received - const worker = this.getWorkerById(message.workerId) - if (worker != null) { - this.workerNodes[this.getWorkerNodeKey(worker)].info.started = - message.started - } else { - throw new Error( - `Worker started message received from unknown worker '${message.workerId}'` - ) - } + this.handleWorkerStartedMessage(message) } else if (message.id != null) { // Task execution response received - const promiseResponse = this.promiseResponseMap.get(message.id) - if (promiseResponse != null) { - if (message.taskError != null) { - if (this.emitter != null) { - this.emitter.emit(PoolEvents.taskError, message.taskError) - } - promiseResponse.reject(message.taskError.message) - } else { - promiseResponse.resolve(message.data as Response) - } - this.afterTaskExecutionHook(promiseResponse.worker, message) - this.promiseResponseMap.delete(message.id) - const workerNodeKey = this.getWorkerNodeKey(promiseResponse.worker) - if ( - this.opts.enableTasksQueue === true && - this.tasksQueueSize(workerNodeKey) > 0 - ) { - this.executeTask( - workerNodeKey, - this.dequeueTask(workerNodeKey) as Task - ) - } - this.workerChoiceStrategyContext.update(workerNodeKey) + this.handleTaskExecutionResponse(message) + } + } + } + + private handleWorkerStartedMessage (message: MessageValue): void { + // Worker started message received + const worker = this.getWorkerById(message.workerId as number) + if (worker != null) { + this.workerNodes[this.getWorkerNodeKey(worker)].info.started = + message.started as boolean + } else { + throw new Error( + `Worker started message received from unknown worker '${ + message.workerId as number + }'` + ) + } + } + + private handleTaskExecutionResponse (message: MessageValue): void { + const promiseResponse = this.promiseResponseMap.get(message.id as string) + if (promiseResponse != null) { + if (message.taskError != null) { + if (this.emitter != null) { + this.emitter.emit(PoolEvents.taskError, message.taskError) } + promiseResponse.reject(message.taskError.message) + } else { + promiseResponse.resolve(message.data as Response) + } + this.afterTaskExecutionHook(promiseResponse.worker, message) + this.promiseResponseMap.delete(message.id as string) + const workerNodeKey = this.getWorkerNodeKey(promiseResponse.worker) + if ( + this.opts.enableTasksQueue === true && + this.tasksQueueSize(workerNodeKey) > 0 + ) { + this.executeTask( + workerNodeKey, + this.dequeueTask(workerNodeKey) as Task + ) } + this.workerChoiceStrategyContext.update(workerNodeKey) } }