fix: worker index identification at tasks stealing under back pressure

Signed-off-by: Jérôme Benoit <jerome.benoit@piment-noir.org>

Author: jerome-benoit

deno.json

@@ -3,9 +3,7 @@
   "version": "0.5.8",
   "exports": "./src/mod.ts",
   "compilerOptions": {
-    "lib": [
-      "deno.worker"
-    ],
+    "lib": ["deno.worker"],
     "strict": true
   },
   "tasks": {
@@ -27,9 +25,7 @@
     "documentation": "deno doc ./src/mod.ts"
   },
   "test": {
-    "include": [
-      "./tests/**/*.test.mjs"
-    ]
+    "include": ["./tests/**/*.test.mjs"]
   },
   "fmt": {
     "semiColons": false,
@@ -42,18 +38,8 @@
     "@std/testing": "jsr:@std/testing@^1.0.15"
   },
   "publish": {
-    "include": [
-      "LICENSE",
-      "README.md",
-      "deno.json",
-      "src/**/*.ts"
-    ]
+    "include": ["LICENSE", "README.md", "deno.json", "src/**/*.ts"]
   },
   "lock": false,
-  "exclude": [
-    "./coverage",
-    "./dist/browser",
-    "./dist/esm",
-    "./npm"
-  ]
+  "exclude": ["./coverage", "./dist/browser", "./dist/esm", "./npm"]
 }

src/pools/abstract-pool.ts

@@ -300,6 +300,8 @@ export abstract class AbstractPool<
 
   /** @inheritDoc */
   public get info(): PoolInfo {
+    const taskStatisticsRequirements = this.workerChoiceStrategiesContext
+      ?.getTaskStatisticsRequirements()
     return {
       version,
       type: this.type,
@@ -310,13 +312,10 @@ export abstract class AbstractPool<
       strategyRetries: this.workerChoiceStrategiesContext?.retriesCount ?? 0,
       minSize: this.minimumNumberOfWorkers,
       maxSize: this.maximumNumberOfWorkers ?? this.minimumNumberOfWorkers,
-      ...(this.workerChoiceStrategiesContext?.getTaskStatisticsRequirements()
-            .runTime.aggregate === true &&
-        this.workerChoiceStrategiesContext.getTaskStatisticsRequirements()
-          .waitTime.aggregate &&
-        {
-          utilization: round(this.utilization),
-        }),
+      ...(taskStatisticsRequirements?.runTime.aggregate === true &&
+        taskStatisticsRequirements.waitTime.aggregate && {
+        utilization: round(this.utilization),
+      }),
       workerNodes: this.workerNodes.length,
       ...(this.type === PoolTypes.dynamic && {
         dynamicWorkerNodes: this.workerNodes.reduce(
@@ -382,8 +381,7 @@ export abstract class AbstractPool<
           0,
         ),
       }),
-      ...(this.workerChoiceStrategiesContext?.getTaskStatisticsRequirements()
-            .runTime.aggregate === true && {
+      ...(taskStatisticsRequirements?.runTime.aggregate === true && {
         runTime: {
           minimum: round(
             min(
@@ -401,8 +399,7 @@ export abstract class AbstractPool<
               ),
             ),
           ),
-          ...(this.workerChoiceStrategiesContext.getTaskStatisticsRequirements()
-            .runTime.average && {
+          ...(taskStatisticsRequirements?.runTime.average && {
             average: round(
               average(
                 this.workerNodes.reduce<number[]>(
@@ -415,8 +412,7 @@ export abstract class AbstractPool<
               ),
             ),
           }),
-          ...(this.workerChoiceStrategiesContext.getTaskStatisticsRequirements()
-            .runTime.median && {
+          ...(taskStatisticsRequirements?.runTime.median && {
             median: round(
               median(
                 this.workerNodes.reduce<number[]>(
@@ -431,8 +427,7 @@ export abstract class AbstractPool<
           }),
         },
       }),
-      ...(this.workerChoiceStrategiesContext?.getTaskStatisticsRequirements()
-            .waitTime.aggregate === true && {
+      ...(taskStatisticsRequirements?.waitTime.aggregate === true && {
         waitTime: {
           minimum: round(
             min(
@@ -450,8 +445,7 @@ export abstract class AbstractPool<
               ),
             ),
           ),
-          ...(this.workerChoiceStrategiesContext.getTaskStatisticsRequirements()
-            .waitTime.average && {
+          ...(taskStatisticsRequirements?.waitTime.average && {
             average: round(
               average(
                 this.workerNodes.reduce<number[]>(
@@ -464,8 +458,7 @@ export abstract class AbstractPool<
               ),
             ),
           }),
-          ...(this.workerChoiceStrategiesContext.getTaskStatisticsRequirements()
-            .waitTime.median && {
+          ...(taskStatisticsRequirements?.waitTime.median && {
             median: round(
               median(
                 this.workerNodes.reduce<number[]>(
@@ -514,6 +507,9 @@ export abstract class AbstractPool<
     }
     const poolTimeCapacity = (performance.now() - this.startTimestamp) *
       (this.maximumNumberOfWorkers ?? this.minimumNumberOfWorkers)
+    if (!Number.isFinite(poolTimeCapacity) || poolTimeCapacity <= 0) {
+      return 0
+    }
     const totalTasksRunTime = this.workerNodes.reduce(
       (accumulator, workerNode) =>
         accumulator + (workerNode.usage.runTime.aggregate ?? 0),
@@ -747,6 +743,7 @@ export abstract class AbstractPool<
    * @returns Worker nodes busyness boolean status.
    */
   protected internalBusy(): boolean {
+    if (this.workerNodes.length === 0) return false
     return (
       this.workerNodes.reduce(
         (accumulator, _, workerNodeKey) =>
@@ -855,12 +852,12 @@ export abstract class AbstractPool<
     message: MessageValue<Data>,
   ): Promise<boolean> {
     const targetWorkerNodeKeys = [...this.workerNodes.keys()]
+    const responsesReceived: MessageValue<Response>[] = []
     const taskFunctionOperationsListener = (
       message: MessageValue<Response>,
       resolve: (value: boolean | PromiseLike<boolean>) => void,
       reject: (reason?: unknown) => void,
     ): void => {
-      const responsesReceived: MessageValue<Response>[] = []
       this.checkMessageWorkerId(message)
       if (
         message.taskFunctionOperationStatus != null &&
@@ -1119,7 +1116,7 @@ export abstract class AbstractPool<
     data?: Data,
     name?: string,
     abortSignal?: AbortSignal,
-    transferList?: Transferable[],
+    transferList?: readonly Transferable[],
   ): Promise<Response> {
     return await new Promise<Response>((resolve, reject) => {
       const timestamp = performance.now()
@@ -1572,32 +1569,25 @@ export abstract class AbstractPool<
    * @param workerNode - The worker node.
    */
   private initWorkerNodeUsage(workerNode: IWorkerNode<Worker, Data>): void {
-    if (
-      this.workerChoiceStrategiesContext?.getTaskStatisticsRequirements()
-        .runTime.aggregate === true
-    ) {
+    const taskStatisticsRequirements = this.workerChoiceStrategiesContext
+      ?.getTaskStatisticsRequirements()
+    if (taskStatisticsRequirements?.runTime.aggregate === true) {
       workerNode.usage.runTime.aggregate = min(
         ...this.workerNodes.map(
           (workerNode) =>
             workerNode.usage.runTime.aggregate ?? Number.POSITIVE_INFINITY,
         ),
       )
     }
-    if (
-      this.workerChoiceStrategiesContext?.getTaskStatisticsRequirements()
-        .waitTime.aggregate === true
-    ) {
+    if (taskStatisticsRequirements?.waitTime.aggregate === true) {
       workerNode.usage.waitTime.aggregate = min(
         ...this.workerNodes.map(
           (workerNode) =>
             workerNode.usage.waitTime.aggregate ?? Number.POSITIVE_INFINITY,
         ),
       )
     }
-    if (
-      this.workerChoiceStrategiesContext?.getTaskStatisticsRequirements().elu
-        .aggregate === true
-    ) {
+    if (taskStatisticsRequirements?.elu.aggregate === true) {
       workerNode.usage.elu.active.aggregate = min(
         ...this.workerNodes.map(
           (workerNode) =>
@@ -1822,13 +1812,12 @@ export abstract class AbstractPool<
    * @param workerNodeKey - The worker node key.
    */
   private sendStatisticsMessageToWorker(workerNodeKey: number): void {
+    const taskStatisticsRequirements = this.workerChoiceStrategiesContext
+      ?.getTaskStatisticsRequirements()
     this.sendToWorker(workerNodeKey, {
       statistics: {
-        runTime:
-          this.workerChoiceStrategiesContext?.getTaskStatisticsRequirements()
-            .runTime.aggregate ?? false,
-        // elu: this.workerChoiceStrategiesContext?.getTaskStatisticsRequirements()
-        //   .elu.aggregate ?? false,
+        runTime: taskStatisticsRequirements?.runTime.aggregate ?? false,
+        // elu: taskStatisticsRequirements?.elu.aggregate ?? false,
       },
     })
   }
@@ -1857,15 +1846,22 @@ export abstract class AbstractPool<
     while (this.tasksQueueSize(sourceWorkerNodeKey) > 0) {
       const destinationWorkerNodeKey = this.workerNodes.reduce(
         (minWorkerNodeKey, workerNode, workerNodeKey, workerNodes) => {
-          return workerNodeKey !== sourceWorkerNodeKey &&
-              workerNode.info.ready &&
-              workerNode.usage.tasks.queued <
-                workerNodes[minWorkerNodeKey].usage.tasks.queued
+          if (workerNodeKey === sourceWorkerNodeKey || !workerNode.info.ready) {
+            return minWorkerNodeKey
+          }
+          if (minWorkerNodeKey === -1) {
+            return workerNodeKey
+          }
+          return workerNode.usage.tasks.queued <
+              workerNodes[minWorkerNodeKey].usage.tasks.queued
             ? workerNodeKey
             : minWorkerNodeKey
         },
-        0,
+        -1,
       )
+      if (destinationWorkerNodeKey === -1) {
+        break
+      }
       this.handleTask(
         destinationWorkerNodeKey,
         this.dequeueTask(sourceWorkerNodeKey) as Task<Data>,
@@ -1961,7 +1957,12 @@ export abstract class AbstractPool<
     }
     destinationWorkerNode.info.stealing = true
     sourceWorkerNode.info.stolen = true
-    const stolenTask = sourceWorkerNode.dequeueLastPrioritizedTask()!
+    const stolenTask = sourceWorkerNode.dequeueLastPrioritizedTask()
+    if (stolenTask == null) {
+      sourceWorkerNode.info.stolen = false
+      destinationWorkerNode.info.stealing = false
+      return
+    }
     sourceWorkerNode.info.stolen = false
     destinationWorkerNode.info.stealing = false
     this.handleTask(destinationWorkerNodeKey, stolenTask)
@@ -2038,15 +2039,17 @@ export abstract class AbstractPool<
   private readonly workerNodeStealTask = (
     workerNodeKey: number,
   ): Task<Data> | undefined => {
+    const workerNode = this.workerNodes[workerNodeKey]
+    if (workerNode == null) return
     const workerNodes = this.workerNodes
       .slice()
       .sort(
         (workerNodeA, workerNodeB) =>
           workerNodeB.usage.tasks.queued - workerNodeA.usage.tasks.queued,
       )
     const sourceWorkerNode = workerNodes.find(
-      (sourceWorkerNode, sourceWorkerNodeKey) =>
-        sourceWorkerNodeKey !== workerNodeKey &&
+      (sourceWorkerNode) =>
+        sourceWorkerNode !== workerNode &&
         sourceWorkerNode.usage.tasks.queued > 0,
     )
     if (sourceWorkerNode != null) {
@@ -2080,7 +2083,7 @@ export abstract class AbstractPool<
         (workerNodeA, workerNodeB) =>
           workerNodeA.usage.tasks.queued - workerNodeB.usage.tasks.queued,
       )
-    for (const [workerNodeKey, workerNode] of workerNodes.entries()) {
+    for (const workerNode of workerNodes) {
       if (sourceWorkerNode.usage.tasks.queued === 0) {
         break
       }
@@ -2090,6 +2093,7 @@ export abstract class AbstractPool<
         workerNode.usage.tasks.queued <
           this.opts.tasksQueueOptions!.size! - sizeOffset
       ) {
+        const workerNodeKey = this.workerNodes.indexOf(workerNode)
         workerNode.info.backPressureStealing = true
         this.stealTask(sourceWorkerNode, workerNodeKey)
         workerNode.info.backPressureStealing = false
@@ -2392,6 +2396,7 @@ export abstract class AbstractPool<
    * @returns Worker nodes back pressure boolean status.
    */
   protected internalBackPressure(): boolean {
+    if (this.workerNodes.length === 0) return false
     return (
       this.workerNodes.reduce(
         (accumulator, _, workerNodeKey) =>

src/utility-types.ts

@@ -124,7 +124,7 @@ export interface Task<Data = unknown> {
   /**
    * Array of transferable objects.
    */
-  readonly transferList?: Transferable[]
+  readonly transferList?: readonly Transferable[]
   /**
    * Timestamp.
    */

src/utils.ts

@@ -62,10 +62,10 @@ export const exponentialDelay = (
  * @internal
  */
 export const average = (dataSet: number[]): number => {
-  if (Array.isArray(dataSet) && dataSet.length === 0) {
+  if (!Array.isArray(dataSet) || dataSet.length === 0) {
     return 0
   }
-  if (Array.isArray(dataSet) && dataSet.length === 1) {
+  if (dataSet.length === 1) {
     return dataSet[0]
   }
   return (
@@ -82,10 +82,10 @@ export const average = (dataSet: number[]): number => {
  * @internal
  */
 export const median = (dataSet: number[]): number => {
-  if (Array.isArray(dataSet) && dataSet.length === 0) {
+  if (!Array.isArray(dataSet) || dataSet.length === 0) {
     return 0
   }
-  if (Array.isArray(dataSet) && dataSet.length === 1) {
+  if (dataSet.length === 1) {
     return dataSet[0]
   }
   const sortedDataSet = dataSet.slice().sort((a, b) => a - b)
@@ -98,7 +98,6 @@ export const median = (dataSet: number[]): number => {
 
 /**
  * Rounds the given number to the given scale.
- * The rounding is done using the "round half away from zero" method.
  *
  * @param num - The number to round.
  * @param scale - The scale to round to.
@@ -107,7 +106,7 @@ export const median = (dataSet: number[]): number => {
  */
 export const round = (num: number, scale = 2): number => {
   const rounder = 10 ** scale
-  return Math.round(num * rounder * (1 + Number.EPSILON)) / rounder
+  return Math.round((num + Math.sign(num) * Number.EPSILON) * rounder) / rounder
 }
 
 /**