X-Git-Url: https://git.piment-noir.org/?a=blobdiff_plain;f=src%2Fpools%2Fabstract-pool.ts;h=e4dac676ab95acae7ee60a82d332084766735e46;hb=ec8ed549f5422b9c23ff5b446ec885f39656373d;hp=2f91df9d88ebcb1661ac661726d5050eb2fda4ab;hpb=7aa2f4763d2ca23123ffadaf7c83f624f8374d84;p=poolifier.git

diff --git a/src/pools/abstract-pool.ts b/src/pools/abstract-pool.ts
index 2f91df9d..e4dac676 100644
--- a/src/pools/abstract-pool.ts
+++ b/src/pools/abstract-pool.ts
@@ -4,11 +4,12 @@ import { EventEmitterAsyncResource } from 'node:events'
 import { performance } from 'node:perf_hooks'
 import type { TransferListItem } from 'node:worker_threads'
 
+import { defaultBucketSize } from '../priority-queue.js'
 import type {
   MessageValue,
   PromiseResponseWrapper,
   Task,
-  TaskFunctionProperties
+  TaskFunctionProperties,
 } from '../utility-types.js'
 import {
   average,
@@ -22,11 +23,11 @@ import {
   median,
   min,
   round,
-  sleep
+  sleep,
 } from '../utils.js'
 import type {
   TaskFunction,
-  TaskFunctionObject
+  TaskFunctionObject,
 } from '../worker/task-functions.js'
 import { KillBehaviors } from '../worker/worker-options.js'
 import {
@@ -36,13 +37,13 @@ import {
   type PoolOptions,
   type PoolType,
   PoolTypes,
-  type TasksQueueOptions
+  type TasksQueueOptions,
 } from './pool.js'
 import {
   Measurements,
   WorkerChoiceStrategies,
   type WorkerChoiceStrategy,
-  type WorkerChoiceStrategyOptions
+  type WorkerChoiceStrategyOptions,
 } from './selection-strategies/selection-strategies-types.js'
 import { WorkerChoiceStrategiesContext } from './selection-strategies/worker-choice-strategies-context.js'
 import {
@@ -55,7 +56,7 @@ import {
   updateRunTimeWorkerUsage,
   updateTaskStatisticsWorkerUsage,
   updateWaitTimeWorkerUsage,
-  waitWorkerNodeEvents
+  waitWorkerNodeEvents,
 } from './utils.js'
 import { version } from './version.js'
 import type {
@@ -63,13 +64,12 @@ import type {
   IWorkerNode,
   WorkerInfo,
   WorkerNodeEventDetail,
-  WorkerType
+  WorkerType,
 } from './worker.js'
 import { WorkerNode } from './worker-node.js'
 
 /**
  * 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 structured-cloneable data.
  * @typeParam Response - Type of execution response. This can only be structured-cloneable data.
@@ -80,7 +80,7 @@ export abstract class AbstractPool<
   Response = unknown
 > implements IPool<Worker, Data, Response> {
   /** @inheritDoc */
-  public readonly workerNodes: Array<IWorkerNode<Worker, Data>> = []
+  public readonly workerNodes: IWorkerNode<Worker, Data>[] = []
 
   /** @inheritDoc */
   public emitter?: EventEmitterAsyncResource
@@ -94,19 +94,19 @@ export abstract class AbstractPool<
    */
   protected promiseResponseMap: Map<
     `${string}-${string}-${string}-${string}-${string}`,
-  PromiseResponseWrapper<Response>
+    PromiseResponseWrapper<Response>
   > = new Map<
     `${string}-${string}-${string}-${string}-${string}`,
     PromiseResponseWrapper<Response>
-    >()
+  >()
 
   /**
    * Worker choice strategies context referencing worker choice algorithms implementation.
    */
   protected workerChoiceStrategiesContext?: WorkerChoiceStrategiesContext<
-  Worker,
-  Data,
-  Response
+    Worker,
+    Data,
+    Response
   >
 
   /**
@@ -115,8 +115,8 @@ export abstract class AbstractPool<
    * - `value`: The task function object.
    */
   private readonly taskFunctions: Map<
-  string,
-  TaskFunctionObject<Data, Response>
+    string,
+    TaskFunctionObject<Data, Response>
   >
 
   /**
@@ -146,7 +146,6 @@ export abstract class AbstractPool<
 
   /**
    * Constructs a new poolifier pool.
-   *
    * @param minimumNumberOfWorkers - Minimum number of workers that this pool manages.
    * @param filePath - Path to the worker file.
    * @param opts - Options for the pool.
@@ -176,9 +175,9 @@ export abstract class AbstractPool<
       this.initEventEmitter()
     }
     this.workerChoiceStrategiesContext = new WorkerChoiceStrategiesContext<
-    Worker,
-    Data,
-    Response
+      Worker,
+      Data,
+      Response
     >(
       this,
       // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
@@ -288,7 +287,7 @@ export abstract class AbstractPool<
 
   private initEventEmitter (): void {
     this.emitter = new EventEmitterAsyncResource({
-      name: `poolifier:${this.type}-${this.worker}-pool`
+      name: `poolifier:${this.type}-${this.worker}-pool`,
     })
   }
 
@@ -309,7 +308,7 @@ export abstract class AbstractPool<
         .runTime.aggregate === true &&
         this.workerChoiceStrategiesContext.getTaskStatisticsRequirements()
           .waitTime.aggregate && {
-        utilization: round(this.utilization)
+        utilization: round(this.utilization),
       }),
       workerNodes: this.workerNodes.length,
       idleWorkerNodes: this.workerNodes.reduce(
@@ -324,10 +323,10 @@ export abstract class AbstractPool<
           (accumulator, workerNode) =>
             workerNode.info.stealing ? accumulator + 1 : accumulator,
           0
-        )
+        ),
       }),
       busyWorkerNodes: this.workerNodes.reduce(
-        (accumulator, _workerNode, workerNodeKey) =>
+        (accumulator, _, workerNodeKey) =>
           this.isWorkerNodeBusy(workerNodeKey) ? accumulator + 1 : accumulator,
         0
       ),
@@ -346,24 +345,24 @@ export abstract class AbstractPool<
           (accumulator, workerNode) =>
             accumulator + workerNode.usage.tasks.queued,
           0
-        )
+        ),
       }),
       ...(this.opts.enableTasksQueue === true && {
         maxQueuedTasks: this.workerNodes.reduce(
           (accumulator, workerNode) =>
             accumulator + (workerNode.usage.tasks.maxQueued ?? 0),
           0
-        )
+        ),
       }),
       ...(this.opts.enableTasksQueue === true && {
-        backPressure: this.hasBackPressure()
+        backPressure: this.hasBackPressure(),
       }),
       ...(this.opts.enableTasksQueue === true && {
         stolenTasks: this.workerNodes.reduce(
           (accumulator, workerNode) =>
             accumulator + workerNode.usage.tasks.stolen,
           0
-        )
+        ),
       }),
       failedTasks: this.workerNodes.reduce(
         (accumulator, workerNode) =>
@@ -376,14 +375,16 @@ export abstract class AbstractPool<
           minimum: round(
             min(
               ...this.workerNodes.map(
-                workerNode => workerNode.usage.runTime.minimum ?? Infinity
+                workerNode =>
+                  workerNode.usage.runTime.minimum ?? Number.POSITIVE_INFINITY
               )
             )
           ),
           maximum: round(
             max(
               ...this.workerNodes.map(
-                workerNode => workerNode.usage.runTime.maximum ?? -Infinity
+                workerNode =>
+                  workerNode.usage.runTime.maximum ?? Number.NEGATIVE_INFINITY
               )
             )
           ),
@@ -393,11 +394,13 @@ export abstract class AbstractPool<
               average(
                 this.workerNodes.reduce<number[]>(
                   (accumulator, workerNode) =>
-                    accumulator.concat(workerNode.usage.runTime.history),
+                    accumulator.concat(
+                      workerNode.usage.runTime.history.toArray()
+                    ),
                   []
                 )
               )
-            )
+            ),
           }),
           ...(this.workerChoiceStrategiesContext.getTaskStatisticsRequirements()
             .runTime.median && {
@@ -405,13 +408,15 @@ export abstract class AbstractPool<
               median(
                 this.workerNodes.reduce<number[]>(
                   (accumulator, workerNode) =>
-                    accumulator.concat(workerNode.usage.runTime.history),
+                    accumulator.concat(
+                      workerNode.usage.runTime.history.toArray()
+                    ),
                   []
                 )
               )
-            )
-          })
-        }
+            ),
+          }),
+        },
       }),
       ...(this.workerChoiceStrategiesContext?.getTaskStatisticsRequirements()
         .waitTime.aggregate === true && {
@@ -419,14 +424,16 @@ export abstract class AbstractPool<
           minimum: round(
             min(
               ...this.workerNodes.map(
-                workerNode => workerNode.usage.waitTime.minimum ?? Infinity
+                workerNode =>
+                  workerNode.usage.waitTime.minimum ?? Number.POSITIVE_INFINITY
               )
             )
           ),
           maximum: round(
             max(
               ...this.workerNodes.map(
-                workerNode => workerNode.usage.waitTime.maximum ?? -Infinity
+                workerNode =>
+                  workerNode.usage.waitTime.maximum ?? Number.NEGATIVE_INFINITY
               )
             )
           ),
@@ -436,11 +443,13 @@ export abstract class AbstractPool<
               average(
                 this.workerNodes.reduce<number[]>(
                   (accumulator, workerNode) =>
-                    accumulator.concat(workerNode.usage.waitTime.history),
+                    accumulator.concat(
+                      workerNode.usage.waitTime.history.toArray()
+                    ),
                   []
                 )
               )
-            )
+            ),
           }),
           ...(this.workerChoiceStrategiesContext.getTaskStatisticsRequirements()
             .waitTime.median && {
@@ -448,14 +457,133 @@ export abstract class AbstractPool<
               median(
                 this.workerNodes.reduce<number[]>(
                   (accumulator, workerNode) =>
-                    accumulator.concat(workerNode.usage.waitTime.history),
+                    accumulator.concat(
+                      workerNode.usage.waitTime.history.toArray()
+                    ),
                   []
                 )
               )
-            )
-          })
-        }
-      })
+            ),
+          }),
+        },
+      }),
+      ...(this.workerChoiceStrategiesContext?.getTaskStatisticsRequirements()
+        .elu.aggregate === true && {
+        elu: {
+          idle: {
+            minimum: round(
+              min(
+                ...this.workerNodes.map(
+                  workerNode =>
+                    workerNode.usage.elu.idle.minimum ??
+                    Number.POSITIVE_INFINITY
+                )
+              )
+            ),
+            maximum: round(
+              max(
+                ...this.workerNodes.map(
+                  workerNode =>
+                    workerNode.usage.elu.idle.maximum ??
+                    Number.NEGATIVE_INFINITY
+                )
+              )
+            ),
+            ...(this.workerChoiceStrategiesContext.getTaskStatisticsRequirements()
+              .elu.average && {
+              average: round(
+                average(
+                  this.workerNodes.reduce<number[]>(
+                    (accumulator, workerNode) =>
+                      accumulator.concat(
+                        workerNode.usage.elu.idle.history.toArray()
+                      ),
+                    []
+                  )
+                )
+              ),
+            }),
+            ...(this.workerChoiceStrategiesContext.getTaskStatisticsRequirements()
+              .elu.median && {
+              median: round(
+                median(
+                  this.workerNodes.reduce<number[]>(
+                    (accumulator, workerNode) =>
+                      accumulator.concat(
+                        workerNode.usage.elu.idle.history.toArray()
+                      ),
+                    []
+                  )
+                )
+              ),
+            }),
+          },
+          active: {
+            minimum: round(
+              min(
+                ...this.workerNodes.map(
+                  workerNode =>
+                    workerNode.usage.elu.active.minimum ??
+                    Number.POSITIVE_INFINITY
+                )
+              )
+            ),
+            maximum: round(
+              max(
+                ...this.workerNodes.map(
+                  workerNode =>
+                    workerNode.usage.elu.active.maximum ??
+                    Number.NEGATIVE_INFINITY
+                )
+              )
+            ),
+            ...(this.workerChoiceStrategiesContext.getTaskStatisticsRequirements()
+              .elu.average && {
+              average: round(
+                average(
+                  this.workerNodes.reduce<number[]>(
+                    (accumulator, workerNode) =>
+                      accumulator.concat(
+                        workerNode.usage.elu.active.history.toArray()
+                      ),
+                    []
+                  )
+                )
+              ),
+            }),
+            ...(this.workerChoiceStrategiesContext.getTaskStatisticsRequirements()
+              .elu.median && {
+              median: round(
+                median(
+                  this.workerNodes.reduce<number[]>(
+                    (accumulator, workerNode) =>
+                      accumulator.concat(
+                        workerNode.usage.elu.active.history.toArray()
+                      ),
+                    []
+                  )
+                )
+              ),
+            }),
+          },
+          utilization: {
+            average: round(
+              average(
+                this.workerNodes.map(
+                  workerNode => workerNode.usage.elu.utilization ?? 0
+                )
+              )
+            ),
+            median: round(
+              median(
+                this.workerNodes.map(
+                  workerNode => workerNode.usage.elu.utilization ?? 0
+                )
+              )
+            ),
+          },
+        },
+      }),
     }
   }
 
@@ -486,7 +614,6 @@ export abstract class AbstractPool<
 
   /**
    * The approximate pool utilization.
-   *
    * @returns The pool utilization.
    */
   private get utilization (): number {
@@ -523,7 +650,6 @@ export abstract class AbstractPool<
 
   /**
    * Checks if the worker id sent in the received message from a worker is valid.
-   *
    * @param message - The received message.
    * @throws {@link https://nodejs.org/api/errors.html#class-error} If the worker id is invalid.
    */
@@ -539,7 +665,6 @@ export abstract class AbstractPool<
 
   /**
    * Gets the worker node key given its worker id.
-   *
    * @param workerId - The worker id.
    * @returns The worker node key if the worker id is found in the pool worker nodes, `-1` otherwise.
    */
@@ -571,7 +696,7 @@ export abstract class AbstractPool<
     }
     if (requireSync) {
       this.workerChoiceStrategiesContext?.syncWorkerChoiceStrategies(
-        this.getWorkerWorkerChoiceStrategies(),
+        this.getWorkerChoiceStrategies(),
         this.opts.workerChoiceStrategyOptions
       )
       for (const workerNodeKey of this.workerNodes.keys()) {
@@ -591,7 +716,7 @@ export abstract class AbstractPool<
         this.opts.workerChoiceStrategyOptions
       )
       this.workerChoiceStrategiesContext?.syncWorkerChoiceStrategies(
-        this.getWorkerWorkerChoiceStrategies(),
+        this.getWorkerChoiceStrategies(),
         this.opts.workerChoiceStrategyOptions
       )
       for (const workerNodeKey of this.workerNodes.keys()) {
@@ -650,7 +775,7 @@ export abstract class AbstractPool<
       ...getDefaultTasksQueueOptions(
         this.maximumNumberOfWorkers ?? this.minimumNumberOfWorkers
       ),
-      ...tasksQueueOptions
+      ...tasksQueueOptions,
     }
   }
 
@@ -714,7 +839,6 @@ export abstract class AbstractPool<
 
   /**
    * Whether worker nodes are executing concurrently their tasks quota or not.
-   *
    * @returns Worker nodes busyness boolean status.
    */
   protected internalBusy (): boolean {
@@ -867,11 +991,11 @@ export abstract class AbstractPool<
     const opResult = await this.sendTaskFunctionOperationToWorkers({
       taskFunctionOperation: 'add',
       taskFunctionProperties: buildTaskFunctionProperties(name, fn),
-      taskFunction: fn.taskFunction.toString()
+      taskFunction: fn.taskFunction.toString(),
     })
     this.taskFunctions.set(name, fn)
     this.workerChoiceStrategiesContext?.syncWorkerChoiceStrategies(
-      this.getWorkerWorkerChoiceStrategies()
+      this.getWorkerChoiceStrategies()
     )
     for (const workerNodeKey of this.workerNodes.keys()) {
       this.sendStatisticsMessageToWorker(workerNodeKey)
@@ -891,14 +1015,14 @@ export abstract class AbstractPool<
       taskFunctionProperties: buildTaskFunctionProperties(
         name,
         this.taskFunctions.get(name)
-      )
+      ),
     })
     for (const workerNode of this.workerNodes) {
       workerNode.deleteTaskFunctionWorkerUsage(name)
     }
     this.taskFunctions.delete(name)
     this.workerChoiceStrategiesContext?.syncWorkerChoiceStrategies(
-      this.getWorkerWorkerChoiceStrategies()
+      this.getWorkerChoiceStrategies()
     )
     for (const workerNodeKey of this.workerNodes.keys()) {
       this.sendStatisticsMessageToWorker(workerNodeKey)
@@ -920,47 +1044,77 @@ export abstract class AbstractPool<
   }
 
   /**
-   * Gets task function strategy, if any.
-   *
+   * Gets task function worker choice strategy, if any.
    * @param name - The task function name.
    * @returns The task function worker choice strategy if the task function worker choice strategy is defined, `undefined` otherwise.
    */
-  private readonly getTaskFunctionWorkerWorkerChoiceStrategy = (
+  private readonly getTaskFunctionWorkerChoiceStrategy = (
     name?: string
   ): WorkerChoiceStrategy | undefined => {
-    if (name != null) {
-      return this.listTaskFunctionsProperties().find(
-        (taskFunctionProperties: TaskFunctionProperties) =>
-          taskFunctionProperties.name === name
-      )?.strategy
+    name = name ?? DEFAULT_TASK_NAME
+    const taskFunctionsProperties = this.listTaskFunctionsProperties()
+    if (name === DEFAULT_TASK_NAME) {
+      name = taskFunctionsProperties[1]?.name
     }
+    return taskFunctionsProperties.find(
+      (taskFunctionProperties: TaskFunctionProperties) =>
+        taskFunctionProperties.name === name
+    )?.strategy
+  }
+
+  /**
+   * Gets worker node task function worker choice strategy, if any.
+   * @param workerNodeKey - The worker node key.
+   * @param name - The task function name.
+   * @returns The worker node task function worker choice strategy if the worker node task function worker choice strategy is defined, `undefined` otherwise.
+   */
+  private readonly getWorkerNodeTaskFunctionWorkerChoiceStrategy = (
+    workerNodeKey: number,
+    name?: string
+  ): WorkerChoiceStrategy | undefined => {
+    const workerInfo = this.getWorkerInfo(workerNodeKey)
+    if (workerInfo == null) {
+      return
+    }
+    name = name ?? DEFAULT_TASK_NAME
+    if (name === DEFAULT_TASK_NAME) {
+      name = workerInfo.taskFunctionsProperties?.[1]?.name
+    }
+    return workerInfo.taskFunctionsProperties?.find(
+      (taskFunctionProperties: TaskFunctionProperties) =>
+        taskFunctionProperties.name === name
+    )?.strategy
   }
 
   /**
    * Gets worker node task function priority, if any.
-   *
    * @param workerNodeKey - The worker node key.
    * @param name - The task function name.
-   * @returns The task function worker choice priority if the task function worker choice priority is defined, `undefined` otherwise.
+   * @returns The worker node task function priority if the worker node task function priority is defined, `undefined` otherwise.
    */
   private readonly getWorkerNodeTaskFunctionPriority = (
     workerNodeKey: number,
     name?: string
   ): number | undefined => {
-    if (name != null) {
-      return this.getWorkerInfo(workerNodeKey)?.taskFunctionsProperties?.find(
-        (taskFunctionProperties: TaskFunctionProperties) =>
-          taskFunctionProperties.name === name
-      )?.priority
+    const workerInfo = this.getWorkerInfo(workerNodeKey)
+    if (workerInfo == null) {
+      return
+    }
+    name = name ?? DEFAULT_TASK_NAME
+    if (name === DEFAULT_TASK_NAME) {
+      name = workerInfo.taskFunctionsProperties?.[1]?.name
     }
+    return workerInfo.taskFunctionsProperties?.find(
+      (taskFunctionProperties: TaskFunctionProperties) =>
+        taskFunctionProperties.name === name
+    )?.priority
   }
 
   /**
    * Gets the worker choice strategies registered in this pool.
-   *
    * @returns The worker choice strategies.
    */
-  private readonly getWorkerWorkerChoiceStrategies =
+  private readonly getWorkerChoiceStrategies =
     (): Set<WorkerChoiceStrategy> => {
       return new Set([
         // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
@@ -972,7 +1126,7 @@ export abstract class AbstractPool<
           )
           .filter(
             (strategy: WorkerChoiceStrategy | undefined) => strategy != null
-          ) as WorkerChoiceStrategy[])
+          ) as WorkerChoiceStrategy[]),
       ])
     }
 
@@ -983,7 +1137,7 @@ export abstract class AbstractPool<
       taskFunctionProperties: buildTaskFunctionProperties(
         name,
         this.taskFunctions.get(name)
-      )
+      ),
     })
   }
 
@@ -1028,18 +1182,18 @@ export abstract class AbstractPool<
         return
       }
       const timestamp = performance.now()
-      const taskFunctionStrategy =
-        this.getTaskFunctionWorkerWorkerChoiceStrategy(name)
-      const workerNodeKey = this.chooseWorkerNode(taskFunctionStrategy)
+      const workerNodeKey = this.chooseWorkerNode(name)
       const task: Task<Data> = {
         name: name ?? DEFAULT_TASK_NAME,
-        // eslint-disable-next-line @typescript-eslint/consistent-type-assertions
         data: data ?? ({} as Data),
         priority: this.getWorkerNodeTaskFunctionPriority(workerNodeKey, name),
-        strategy: taskFunctionStrategy,
+        strategy: this.getWorkerNodeTaskFunctionWorkerChoiceStrategy(
+          workerNodeKey,
+          name
+        ),
         transferList,
         timestamp,
-        taskId: randomUUID()
+        taskId: randomUUID(),
       }
       // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
       this.promiseResponseMap.set(task.taskId!, {
@@ -1049,9 +1203,9 @@ export abstract class AbstractPool<
         ...(this.emitter != null && {
           asyncResource: new AsyncResource('poolifier:task', {
             triggerAsyncId: this.emitter.asyncId,
-            requireManualDestroy: true
-          })
-        })
+            requireManualDestroy: true,
+          }),
+        }),
       })
       if (
         this.opts.enableTasksQueue === false ||
@@ -1067,6 +1221,7 @@ export abstract class AbstractPool<
 
   /**
    * Starts the minimum number of workers.
+   * @param initWorkerNodeUsage - Whether to initialize the worker node usage or not. @defaultValue false
    */
   private startMinimumNumberOfWorkers (initWorkerNodeUsage = false): void {
     this.startingMinimumNumberOfWorkers = true
@@ -1115,7 +1270,7 @@ export abstract class AbstractPool<
     }
     this.destroying = true
     await Promise.all(
-      this.workerNodes.map(async (_workerNode, workerNodeKey) => {
+      this.workerNodes.map(async (_, workerNodeKey) => {
         await this.destroyWorkerNode(workerNodeKey)
       })
     )
@@ -1129,7 +1284,6 @@ export abstract class AbstractPool<
 
   private async sendKillMessageToWorker (workerNodeKey: number): Promise<void> {
     await new Promise<void>((resolve, reject) => {
-      // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
       if (this.workerNodes[workerNodeKey] == null) {
         resolve()
         return
@@ -1154,7 +1308,6 @@ export abstract class AbstractPool<
 
   /**
    * Terminates the worker node given its worker node key.
-   *
    * @param workerNodeKey - The worker node key.
    */
   protected async destroyWorkerNode (workerNodeKey: number): Promise<void> {
@@ -1177,8 +1330,6 @@ export abstract class AbstractPool<
   /**
    * Setup hook to execute code before worker nodes are created in the abstract constructor.
    * Can be overridden.
-   *
-   * @virtual
    */
   protected setupHook (): void {
     /* Intentionally empty */
@@ -1186,7 +1337,6 @@ export abstract class AbstractPool<
 
   /**
    * Returns whether the worker is the main worker or not.
-   *
    * @returns `true` if the worker is the main worker, `false` otherwise.
    */
   protected abstract isMain (): boolean
@@ -1194,7 +1344,6 @@ export abstract class AbstractPool<
   /**
    * Hook executed before the worker task execution.
    * Can be overridden.
-   *
    * @param workerNodeKey - The worker node key.
    * @param task - The task to execute.
    */
@@ -1202,7 +1351,6 @@ export abstract class AbstractPool<
     workerNodeKey: number,
     task: Task<Data>
   ): void {
-    // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
     if (this.workerNodes[workerNodeKey]?.usage != null) {
       const workerUsage = this.workerNodes[workerNodeKey].usage
       ++workerUsage.tasks.executing
@@ -1235,7 +1383,6 @@ export abstract class AbstractPool<
   /**
    * Hook executed after the worker task execution.
    * Can be overridden.
-   *
    * @param workerNodeKey - The worker node key.
    * @param message - The received message.
    */
@@ -1244,7 +1391,6 @@ export abstract class AbstractPool<
     message: MessageValue<Response>
   ): void {
     let needWorkerChoiceStrategiesUpdate = false
-    // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
     if (this.workerNodes[workerNodeKey]?.usage != null) {
       const workerUsage = this.workerNodes[workerNodeKey].usage
       updateTaskStatisticsWorkerUsage(workerUsage, message)
@@ -1292,7 +1438,6 @@ export abstract class AbstractPool<
 
   /**
    * Whether the worker node shall update its task function worker usage or not.
-   *
    * @param workerNodeKey - The worker node key.
    * @returns `true` if the worker node shall update its task function worker usage, `false` otherwise.
    */
@@ -1306,14 +1451,11 @@ export abstract class AbstractPool<
   }
 
   /**
-   * Chooses a worker node for the next task given the worker choice strategy.
-   *
-   * @param workerChoiceStrategy - The worker choice strategy.
-   * @returns The chosen worker node key
+   * Chooses a worker node for the next task.
+   * @param name - The task function name.
+   * @returns The chosen worker node key.
    */
-  private chooseWorkerNode (
-    workerChoiceStrategy?: WorkerChoiceStrategy
-  ): number {
+  private chooseWorkerNode (name?: string): number {
     if (this.shallCreateDynamicWorker()) {
       const workerNodeKey = this.createAndSetupDynamicWorkerNode()
       if (
@@ -1324,19 +1466,19 @@ export abstract class AbstractPool<
       }
     }
     // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
-    return this.workerChoiceStrategiesContext!.execute(workerChoiceStrategy)
+    return this.workerChoiceStrategiesContext!.execute(
+      this.getTaskFunctionWorkerChoiceStrategy(name)
+    )
   }
 
   /**
    * Conditions for dynamic worker creation.
-   *
    * @returns Whether to create a dynamic worker or not.
    */
   protected abstract shallCreateDynamicWorker (): boolean
 
   /**
    * Sends a message to worker given its worker node key.
-   *
    * @param workerNodeKey - The worker node key.
    * @param message - The message.
    * @param transferList - The optional array of transferable objects.
@@ -1349,7 +1491,6 @@ export abstract class AbstractPool<
 
   /**
    * Initializes the worker node usage with sensible default values gathered during runtime.
-   *
    * @param workerNode - The worker node.
    */
   private initWorkerNodeUsage (workerNode: IWorkerNode<Worker, Data>): void {
@@ -1359,7 +1500,8 @@ export abstract class AbstractPool<
     ) {
       workerNode.usage.runTime.aggregate = min(
         ...this.workerNodes.map(
-          workerNode => workerNode.usage.runTime.aggregate ?? Infinity
+          workerNode =>
+            workerNode.usage.runTime.aggregate ?? Number.POSITIVE_INFINITY
         )
       )
     }
@@ -1369,7 +1511,8 @@ export abstract class AbstractPool<
     ) {
       workerNode.usage.waitTime.aggregate = min(
         ...this.workerNodes.map(
-          workerNode => workerNode.usage.waitTime.aggregate ?? Infinity
+          workerNode =>
+            workerNode.usage.waitTime.aggregate ?? Number.POSITIVE_INFINITY
         )
       )
     }
@@ -1379,7 +1522,8 @@ export abstract class AbstractPool<
     ) {
       workerNode.usage.elu.active.aggregate = min(
         ...this.workerNodes.map(
-          workerNode => workerNode.usage.elu.active.aggregate ?? Infinity
+          workerNode =>
+            workerNode.usage.elu.active.aggregate ?? Number.POSITIVE_INFINITY
         )
       )
     }
@@ -1387,7 +1531,6 @@ export abstract class AbstractPool<
 
   /**
    * Creates a new, completely set up worker node.
-   *
    * @returns New, completely set up worker node key.
    */
   protected createAndSetupWorkerNode (): number {
@@ -1425,7 +1568,6 @@ export abstract class AbstractPool<
       ) {
         this.redistributeQueuedTasks(this.workerNodes.indexOf(workerNode))
       }
-      // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
       workerNode?.terminate().catch((error: unknown) => {
         this.emitter?.emit(PoolEvents.error, error)
       })
@@ -1451,7 +1593,6 @@ export abstract class AbstractPool<
 
   /**
    * Creates a new, completely set up dynamic worker node.
-   *
    * @returns New, completely set up dynamic worker node key.
    */
   protected createAndSetupDynamicWorkerNode (): number {
@@ -1461,6 +1602,7 @@ export abstract class AbstractPool<
       const localWorkerNodeKey = this.getWorkerNodeKeyByWorkerId(
         message.workerId
       )
+      const workerInfo = this.getWorkerInfo(localWorkerNodeKey)
       const workerUsage = this.workerNodes[localWorkerNodeKey]?.usage
       // Kill message received from worker
       if (
@@ -1469,6 +1611,8 @@ export abstract class AbstractPool<
           ((this.opts.enableTasksQueue === false &&
             workerUsage.tasks.executing === 0) ||
             (this.opts.enableTasksQueue === true &&
+              workerInfo != null &&
+              !workerInfo.stealing &&
               workerUsage.tasks.executing === 0 &&
               this.tasksQueueSize(localWorkerNodeKey) === 0)))
       ) {
@@ -1480,7 +1624,7 @@ export abstract class AbstractPool<
       }
     })
     this.sendToWorker(workerNodeKey, {
-      checkActive: true
+      checkActive: true,
     })
     if (this.taskFunctions.size > 0) {
       for (const [taskFunctionName, taskFunctionObject] of this.taskFunctions) {
@@ -1490,7 +1634,7 @@ export abstract class AbstractPool<
             taskFunctionName,
             taskFunctionObject
           ),
-          taskFunction: taskFunctionObject.taskFunction.toString()
+          taskFunction: taskFunctionObject.taskFunction.toString(),
         }).catch((error: unknown) => {
           this.emitter?.emit(PoolEvents.error, error)
         })
@@ -1513,7 +1657,6 @@ export abstract class AbstractPool<
 
   /**
    * Registers a listener callback on the worker given its worker node key.
-   *
    * @param workerNodeKey - The worker node key.
    * @param listener - The message listener callback.
    */
@@ -1526,7 +1669,6 @@ export abstract class AbstractPool<
 
   /**
    * Registers once a listener callback on the worker given its worker node key.
-   *
    * @param workerNodeKey - The worker node key.
    * @param listener - The message listener callback.
    */
@@ -1539,7 +1681,6 @@ export abstract class AbstractPool<
 
   /**
    * Deregisters a listener callback on the worker given its worker node key.
-   *
    * @param workerNodeKey - The worker node key.
    * @param listener - The message listener callback.
    */
@@ -1553,7 +1694,6 @@ export abstract class AbstractPool<
   /**
    * Method hooked up after a worker node has been newly created.
    * Can be overridden.
-   *
    * @param workerNodeKey - The newly created worker node key.
    */
   protected afterWorkerNodeSetup (workerNodeKey: number): void {
@@ -1584,14 +1724,12 @@ export abstract class AbstractPool<
 
   /**
    * Sends the startup message to worker given its worker node key.
-   *
    * @param workerNodeKey - The worker node key.
    */
   protected abstract sendStartupMessageToWorker (workerNodeKey: number): void
 
   /**
    * Sends the statistics message to worker given its worker node key.
-   *
    * @param workerNodeKey - The worker node key.
    */
   private sendStatisticsMessageToWorker (workerNodeKey: number): void {
@@ -1602,8 +1740,8 @@ export abstract class AbstractPool<
             .runTime.aggregate ?? false,
         elu:
           this.workerChoiceStrategiesContext?.getTaskStatisticsRequirements()
-            .elu.aggregate ?? false
-      }
+            .elu.aggregate ?? false,
+      },
     })
   }
 
@@ -1648,7 +1786,6 @@ export abstract class AbstractPool<
     taskName: string
   ): void {
     const workerNode = this.workerNodes[workerNodeKey]
-    // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
     if (workerNode?.usage != null) {
       ++workerNode.usage.tasks.stolen
     }
@@ -1656,38 +1793,37 @@ export abstract class AbstractPool<
       this.shallUpdateTaskFunctionWorkerUsage(workerNodeKey) &&
       workerNode.getTaskFunctionWorkerUsage(taskName) != null
     ) {
-      const taskFunctionWorkerUsage =
-        // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
-        workerNode.getTaskFunctionWorkerUsage(taskName)!
-      ++taskFunctionWorkerUsage.tasks.stolen
+      // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
+      ++workerNode.getTaskFunctionWorkerUsage(taskName)!.tasks.stolen
     }
   }
 
   private updateTaskSequentiallyStolenStatisticsWorkerUsage (
     workerNodeKey: number,
     taskName: string,
-    previousStolenTaskName: string
+    previousTaskName?: string
   ): void {
     const workerNode = this.workerNodes[workerNodeKey]
-    // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
     if (workerNode?.usage != null) {
       ++workerNode.usage.tasks.sequentiallyStolen
     }
-    const taskFunctionWorkerUsage =
-      workerNode.getTaskFunctionWorkerUsage(taskName)
     if (
       this.shallUpdateTaskFunctionWorkerUsage(workerNodeKey) &&
-      taskFunctionWorkerUsage != null &&
-      (taskFunctionWorkerUsage.tasks.sequentiallyStolen === 0 ||
-        (previousStolenTaskName === taskName &&
-          taskFunctionWorkerUsage.tasks.sequentiallyStolen > 0))
-    ) {
-      ++taskFunctionWorkerUsage.tasks.sequentiallyStolen
-    } else if (
-      this.shallUpdateTaskFunctionWorkerUsage(workerNodeKey) &&
-      taskFunctionWorkerUsage != null
+      workerNode.getTaskFunctionWorkerUsage(taskName) != null
     ) {
-      taskFunctionWorkerUsage.tasks.sequentiallyStolen = 0
+      const taskFunctionWorkerUsage =
+        // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
+        workerNode.getTaskFunctionWorkerUsage(taskName)!
+      if (
+        taskFunctionWorkerUsage.tasks.sequentiallyStolen === 0 ||
+        (previousTaskName != null &&
+          previousTaskName === taskName &&
+          taskFunctionWorkerUsage.tasks.sequentiallyStolen > 0)
+      ) {
+        ++taskFunctionWorkerUsage.tasks.sequentiallyStolen
+      } else if (taskFunctionWorkerUsage.tasks.sequentiallyStolen > 0) {
+        taskFunctionWorkerUsage.tasks.sequentiallyStolen = 0
+      }
     }
   }
 
@@ -1696,7 +1832,6 @@ export abstract class AbstractPool<
     taskName: string
   ): void {
     const workerNode = this.workerNodes[workerNodeKey]
-    // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
     if (workerNode?.usage != null) {
       workerNode.usage.tasks.sequentiallyStolen = 0
     }
@@ -1704,10 +1839,10 @@ export abstract class AbstractPool<
       this.shallUpdateTaskFunctionWorkerUsage(workerNodeKey) &&
       workerNode.getTaskFunctionWorkerUsage(taskName) != null
     ) {
-      const taskFunctionWorkerUsage =
-        // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
-        workerNode.getTaskFunctionWorkerUsage(taskName)!
-      taskFunctionWorkerUsage.tasks.sequentiallyStolen = 0
+      // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
+      workerNode.getTaskFunctionWorkerUsage(
+        taskName
+      )!.tasks.sequentiallyStolen = 0
     }
   }
 
@@ -1722,12 +1857,17 @@ export abstract class AbstractPool<
       )
     }
     const workerInfo = this.getWorkerInfo(workerNodeKey)
+    if (workerInfo == null) {
+      throw new Error(
+        `Worker node with key '${workerNodeKey}' not found in pool`
+      )
+    }
     if (
       this.cannotStealTask() ||
       (this.info.stealingWorkerNodes ?? 0) >
         Math.floor(this.workerNodes.length / 2)
     ) {
-      if (workerInfo != null && previousStolenTask != null) {
+      if (previousStolenTask != null) {
         workerInfo.stealing = false
         this.resetTaskSequentiallyStolenStatisticsWorkerUsage(
           workerNodeKey,
@@ -1739,7 +1879,6 @@ export abstract class AbstractPool<
     }
     const workerNodeTasksUsage = this.workerNodes[workerNodeKey].usage.tasks
     if (
-      workerInfo != null &&
       previousStolenTask != null &&
       (workerNodeTasksUsage.executing > 0 ||
         this.tasksQueueSize(workerNodeKey) > 0)
@@ -1752,20 +1891,14 @@ export abstract class AbstractPool<
       )
       return
     }
-    if (workerInfo == null) {
-      throw new Error(
-        `Worker node with key '${workerNodeKey}' not found in pool`
-      )
-    }
     workerInfo.stealing = true
     const stolenTask = this.workerNodeStealTask(workerNodeKey)
-    if (stolenTask != null && previousStolenTask != null) {
+    if (stolenTask != null) {
       this.updateTaskSequentiallyStolenStatisticsWorkerUsage(
         workerNodeKey,
         // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
         stolenTask.name!,
-        // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
-        previousStolenTask.name!
+        previousStolenTask?.name
       )
     }
     sleep(exponentialDelay(workerNodeTasksUsage.sequentiallyStolen))
@@ -1815,12 +1948,12 @@ export abstract class AbstractPool<
     ) {
       return
     }
-    const { workerId } = eventDetail
     const sizeOffset = 1
     // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
     if (this.opts.tasksQueueOptions!.size! <= sizeOffset) {
       return
     }
+    const { workerId } = eventDetail
     const sourceWorkerNode =
       this.workerNodes[this.getWorkerNodeKeyByWorkerId(workerId)]
     const workerNodes = this.workerNodes
@@ -1856,8 +1989,15 @@ export abstract class AbstractPool<
     }
   }
 
+  private setTasksQueuePriority (workerNodeKey: number): void {
+    this.workerNodes[workerNodeKey].setTasksQueuePriority(
+      this.getTasksQueuePriority()
+    )
+  }
+
   /**
    * This method is the message listener registered on each worker.
+   * @param message - The message received from the worker.
    */
   protected readonly workerMessageListener = (
     message: MessageValue<Response>
@@ -1874,6 +2014,7 @@ export abstract class AbstractPool<
       if (workerInfo != null) {
         workerInfo.taskFunctionsProperties = taskFunctionsProperties
         this.sendStatisticsMessageToWorker(workerNodeKey)
+        this.setTasksQueuePriority(workerNodeKey)
       }
     } else if (taskId != null) {
       // Task execution response received from worker
@@ -1898,6 +2039,7 @@ export abstract class AbstractPool<
     workerNode.info.ready = ready
     workerNode.info.taskFunctionsProperties = taskFunctionsProperties
     this.sendStatisticsMessageToWorker(workerNodeKey)
+    this.setTasksQueuePriority(workerNodeKey)
     this.checkAndEmitReadyEvent()
   }
 
@@ -1926,12 +2068,10 @@ export abstract class AbstractPool<
       this.afterTaskExecutionHook(workerNodeKey, message)
       // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
       this.promiseResponseMap.delete(taskId!)
-      // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
       workerNode?.emit('taskFinished', taskId)
       if (
         this.opts.enableTasksQueue === true &&
         !this.destroying &&
-        // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
         workerNode != null
       ) {
         const workerNodeTasksUsage = workerNode.usage.tasks
@@ -1951,7 +2091,7 @@ export abstract class AbstractPool<
         ) {
           workerNode.emit('idle', {
             workerId,
-            workerNodeKey
+            workerNodeKey,
           })
         }
       }
@@ -1977,7 +2117,6 @@ export abstract class AbstractPool<
 
   /**
    * Gets the worker information given its worker node key.
-   *
    * @param workerNodeKey - The worker node key.
    * @returns The worker information.
    */
@@ -1985,9 +2124,14 @@ export abstract class AbstractPool<
     return this.workerNodes[workerNodeKey]?.info
   }
 
+  private getTasksQueuePriority (): boolean {
+    return this.listTaskFunctionsProperties().some(
+      taskFunctionProperties => taskFunctionProperties.priority != null
+    )
+  }
+
   /**
    * Creates a worker node.
-   *
    * @returns The created worker node.
    */
   private createWorkerNode (): IWorkerNode<Worker, Data> {
@@ -2002,8 +2146,8 @@ export abstract class AbstractPool<
           getDefaultTasksQueueOptions(
             this.maximumNumberOfWorkers ?? this.minimumNumberOfWorkers
           ).size,
-        tasksQueueBucketSize:
-          (this.maximumNumberOfWorkers ?? this.minimumNumberOfWorkers) * 2
+        tasksQueueBucketSize: defaultBucketSize,
+        tasksQueuePriority: this.getTasksQueuePriority(),
       }
     )
     // Flag the worker node as ready at pool startup.
@@ -2015,7 +2159,6 @@ export abstract class AbstractPool<
 
   /**
    * Adds the given worker node in the pool worker nodes.
-   *
    * @param workerNode - The worker node.
    * @returns The added worker node key.
    * @throws {@link https://nodejs.org/api/errors.html#class-error} If the added worker node is not found.
@@ -2038,7 +2181,6 @@ export abstract class AbstractPool<
 
   /**
    * Removes the worker node from the pool worker nodes.
-   *
    * @param workerNode - The worker node.
    */
   private removeWorkerNode (workerNode: IWorkerNode<Worker, Data>): void {
@@ -2068,7 +2210,6 @@ export abstract class AbstractPool<
 
   /**
    * Executes the given task on the worker given its worker node key.
-   *
    * @param workerNodeKey - The worker node key.
    * @param task - The task to execute.
    */