+ private getWorkerId (worker: Worker): number | undefined {
+ if (this.worker === WorkerTypes.thread) {
+ return worker.threadId
+ } else if (this.worker === WorkerTypes.cluster) {
+ return worker.id
+ }
+ }
+
+ // /**
+ // * Sets the given worker in the pool worker nodes.
+ // *
+ // * @param workerNodeKey - The worker node key.
+ // * @param worker - The worker.
+ // * @param workerInfo - The worker info.
+ // * @param workerUsage - The worker usage.
+ // * @param tasksQueue - The worker task queue.
+ // */
+ // private setWorkerNode (
+ // workerNodeKey: number,
+ // worker: Worker,
+ // workerInfo: WorkerInfo,
+ // workerUsage: WorkerUsage,
+ // tasksQueue: Queue<Task<Data>>
+ // ): void {
+ // this.workerNodes[workerNodeKey] = {
+ // worker,
+ // info: workerInfo,
+ // usage: workerUsage,
+ // tasksQueue
+ // }
+ // }
+
+ /**
+ * Removes the given worker from the pool worker nodes.
+ *
+ * @param worker - The worker.
+ */
+ private removeWorkerNode (worker: Worker): void {
+ const workerNodeKey = this.getWorkerNodeKey(worker)
+ if (workerNodeKey !== -1) {
+ this.workerNodes.splice(workerNodeKey, 1)
+ this.workerChoiceStrategyContext.remove(workerNodeKey)
+ }
+ }
+
+ 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 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++) {
+ this.executeTask(
+ workerNodeKey,
+ this.dequeueTask(workerNodeKey) as Task<Data>
+ )
+ }
+ }
+ this.workerNodes[workerNodeKey].tasksQueue.clear()
+ }
+
+ 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 (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: {
+ executed: 0,
+ executing: 0,
+ get queued (): number {
+ return getTasksQueueSize(workerNodeKey)
+ },
+ get maxQueued (): number {
+ return getTasksMaxQueueSize(workerNodeKey)
+ },
+ failed: 0
+ },
+ 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
+ }