- /**
- * Removes the given worker from the pool.
- *
- * @param worker - The worker that will be removed.
- */
- protected removeWorker (worker: Worker): void {
- const workerKey = this.getWorkerKey(worker)
- this.workers.splice(workerKey, 1)
- this.workerChoiceStrategyContext.remove(workerKey)
+ private executeTask (workerNodeKey: number, task: Task<Data>): void {
+ this.beforeTaskExecutionHook(workerNodeKey, task)
+ this.sendToWorker(this.workerNodes[workerNodeKey].worker, task)
+ }
+
+ private enqueueTask (workerNodeKey: number, task: Task<Data>): number {
+ return this.workerNodes[workerNodeKey].tasksQueue.enqueue(task)
+ }
+
+ private dequeueTask (workerNodeKey: number): Task<Data> | undefined {
+ return this.workerNodes[workerNodeKey].tasksQueue.dequeue()
+ }
+
+ private tasksQueueSize (workerNodeKey: number): number {
+ return this.workerNodes[workerNodeKey].tasksQueue.size
+ }
+
+ private flushTasksQueue (workerNodeKey: number): void {
+ if (this.tasksQueueSize(workerNodeKey) > 0) {
+ for (let i = 0; i < this.tasksQueueSize(workerNodeKey); i++) {
+ this.executeTask(
+ workerNodeKey,
+ this.dequeueTask(workerNodeKey) as Task<Data>
+ )
+ }
+ }
+ }
+
+ private flushTasksQueues (): void {
+ for (const [workerNodeKey] of this.workerNodes.entries()) {
+ this.flushTasksQueue(workerNodeKey)
+ }
+ }
+
+ private setWorkerStatistics (worker: Worker): void {
+ this.sendToWorker(worker, {
+ statistics: {
+ runTime:
+ this.workerChoiceStrategyContext.getTaskStatisticsRequirements()
+ .runTime.aggregate,
+ elu: this.workerChoiceStrategyContext.getTaskStatisticsRequirements()
+ .elu.aggregate
+ }
+ })
+ }
+
+ private getWorkerUsage (worker: Worker): WorkerUsage {
+ return {
+ tasks: this.getTaskStatistics(worker),
+ runTime: {
+ aggregate: 0,
+ average: 0,
+ median: 0,
+ history: new CircularArray()
+ },
+ waitTime: {
+ aggregate: 0,
+ average: 0,
+ median: 0,
+ history: new CircularArray()
+ },
+ elu: {
+ idle: {
+ aggregate: 0,
+ average: 0,
+ median: 0,
+ history: new CircularArray()
+ },
+ active: {
+ aggregate: 0,
+ average: 0,
+ median: 0,
+ history: new CircularArray()
+ },
+ utilization: 0
+ }
+ }
+ }
+
+ 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
+ }