- private updateWorkerTasksRunTime (
- worker: Worker,
- taskRunTime: number | undefined
- ): void {
- if (
- this.workerChoiceStrategyContext.getWorkerChoiceStrategy()
- .requiredStatistics.runTime === true
- ) {
- const tasksUsage = this.workersTasksUsage.get(worker)
- if (tasksUsage !== undefined) {
- tasksUsage.runTime += taskRunTime ?? 0
- if (tasksUsage.run !== 0) {
- tasksUsage.avgRunTime = tasksUsage.runTime / tasksUsage.run
- }
- this.workersTasksUsage.set(worker, tasksUsage)
- } else {
- throw new Error(WORKER_NOT_FOUND_TASKS_USAGE_MAP)
+ private removeWorkerNode (worker: Worker): void {
+ const workerNodeKey = this.getWorkerNodeKey(worker)
+ this.workerNodes.splice(workerNodeKey, 1)
+ this.workerChoiceStrategyContext.remove(workerNodeKey)
+ }
+
+ private executeTask (workerNodeKey: number, task: Task<Data>): void {
+ this.beforeTaskExecutionHook(workerNodeKey)
+ this.sendToWorker(this.workerNodes[workerNodeKey].worker, task)
+ }
+
+ private enqueueTask (workerNodeKey: number, task: Task<Data>): void {
+ this.workerNodes[workerNodeKey].tasksQueue.push(task)
+ }
+
+ private dequeueTask (workerNodeKey: number): Task<Data> | undefined {
+ return this.workerNodes[workerNodeKey].tasksQueue.shift()
+ }
+
+ private tasksQueueSize (workerNodeKey: number): number {
+ return this.workerNodes[workerNodeKey].tasksQueue.length
+ }
+
+ private flushTasksQueue (workerNodeKey: number): void {
+ if (this.tasksQueueSize(workerNodeKey) > 0) {
+ for (const task of this.workerNodes[workerNodeKey].tasksQueue) {
+ this.executeTask(workerNodeKey, task)