Merge branch 'master' into combined-prs-branch

[poolifier.git] / src / pools / utils.ts

diff --git a/src/pools/utils.ts b/src/pools/utils.ts
index 3fa17312d1fe24f54fb121eeda4d9dee6124e0c9..d4649d70a56cc31dbb51ddf62aa3b79eca3ef4ef 100644 (file)
--- a/src/pools/utils.ts
+++ b/src/pools/utils.ts
@@ -43,7 +43,7 @@ export const getDefaultTasksQueueOptions = (
     size: Math.pow(poolMaxSize, 2),
     concurrency: 1,
     taskStealing: true,
-    tasksStealingOnBackPressure: true,
+    tasksStealingOnBackPressure: false,
     tasksFinishedTimeout: 2000
   }
 }
@@ -169,7 +169,7 @@ export const checkWorkerNodeArguments = (
   }
   if (!isPlainObject(opts)) {
     throw new TypeError(
-      'Cannot construct a worker node with invalid options: must be a plain object'
+      'Cannot construct a worker node with invalid worker node options: must be a plain object'
     )
   }
   if (opts.tasksQueueBackPressureSize == null) {
@@ -187,6 +187,31 @@ export const checkWorkerNodeArguments = (
       'Cannot construct a worker node with a tasks queue back pressure size option that is not a positive integer'
     )
   }
+  if (opts.tasksQueueBucketSize == null) {
+    throw new TypeError(
+      'Cannot construct a worker node without a tasks queue bucket size option'
+    )
+  }
+  if (!Number.isSafeInteger(opts.tasksQueueBucketSize)) {
+    throw new TypeError(
+      'Cannot construct a worker node with a tasks queue bucket size option that is not an integer'
+    )
+  }
+  if (opts.tasksQueueBucketSize <= 0) {
+    throw new RangeError(
+      'Cannot construct a worker node with a tasks queue bucket size option that is not a positive integer'
+    )
+  }
+  if (opts.tasksQueuePriority == null) {
+    throw new TypeError(
+      'Cannot construct a worker node without a tasks queue priority option'
+    )
+  }
+  if (typeof opts.tasksQueuePriority !== 'boolean') {
+    throw new TypeError(
+      'Cannot construct a worker node with a tasks queue priority option that is not a boolean'
+    )
+  }
 }
 
 /**
@@ -211,21 +236,25 @@ const updateMeasurementStatistics = (
       (measurementStatistics.aggregate ?? 0) + measurementValue
     measurementStatistics.minimum = min(
       measurementValue,
-      measurementStatistics.minimum ?? Infinity
+      measurementStatistics.minimum ?? Number.POSITIVE_INFINITY
     )
     measurementStatistics.maximum = max(
       measurementValue,
-      measurementStatistics.maximum ?? -Infinity
+      measurementStatistics.maximum ?? Number.NEGATIVE_INFINITY
     )
     if (measurementRequirements.average || measurementRequirements.median) {
-      measurementStatistics.history.push(measurementValue)
+      measurementStatistics.history.put(measurementValue)
       if (measurementRequirements.average) {
-        measurementStatistics.average = average(measurementStatistics.history)
+        measurementStatistics.average = average(
+          measurementStatistics.history.toArray()
+        )
       } else if (measurementStatistics.average != null) {
         delete measurementStatistics.average
       }
       if (measurementRequirements.median) {
-        measurementStatistics.median = median(measurementStatistics.history)
+        measurementStatistics.median = median(
+          measurementStatistics.history.toArray()
+        )
       } else if (measurementStatistics.median != null) {
         delete measurementStatistics.median
       }
@@ -242,7 +271,7 @@ export const updateWaitTimeWorkerUsage = <
   Data = unknown,
   Response = unknown
 >(
-    workerChoiceStrategyContext:
+    workerChoiceStrategiesContext:
     | WorkerChoiceStrategiesContext<Worker, Data, Response>
     | undefined,
     workerUsage: WorkerUsage,
@@ -252,7 +281,7 @@ export const updateWaitTimeWorkerUsage = <
   const taskWaitTime = timestamp - (task.timestamp ?? timestamp)
   updateMeasurementStatistics(
     workerUsage.waitTime,
-    workerChoiceStrategyContext?.getTaskStatisticsRequirements().waitTime,
+    workerChoiceStrategiesContext?.getTaskStatisticsRequirements().waitTime,
     taskWaitTime
   )
 }
@@ -281,7 +310,7 @@ export const updateRunTimeWorkerUsage = <
   Data = unknown,
   Response = unknown
 >(
-    workerChoiceStrategyContext:
+    workerChoiceStrategiesContext:
     | WorkerChoiceStrategiesContext<Worker, Data, Response>
     | undefined,
     workerUsage: WorkerUsage,
@@ -292,7 +321,7 @@ export const updateRunTimeWorkerUsage = <
   }
   updateMeasurementStatistics(
     workerUsage.runTime,
-    workerChoiceStrategyContext?.getTaskStatisticsRequirements().runTime,
+    workerChoiceStrategiesContext?.getTaskStatisticsRequirements().runTime,
     message.taskPerformance?.runTime ?? 0
   )
 }
@@ -302,7 +331,7 @@ export const updateEluWorkerUsage = <
   Data = unknown,
   Response = unknown
 >(
-    workerChoiceStrategyContext:
+    workerChoiceStrategiesContext:
     | WorkerChoiceStrategiesContext<Worker, Data, Response>
     | undefined,
     workerUsage: WorkerUsage,
@@ -312,7 +341,7 @@ export const updateEluWorkerUsage = <
     return
   }
   const eluTaskStatisticsRequirements =
-    workerChoiceStrategyContext?.getTaskStatisticsRequirements().elu
+    workerChoiceStrategiesContext?.getTaskStatisticsRequirements().elu
   updateMeasurementStatistics(
     workerUsage.elu.active,
     eluTaskStatisticsRequirements,
@@ -401,12 +430,20 @@ export const waitWorkerNodeEvents = async <
       resolve(events)
       return
     }
-    workerNode.on(workerNodeEvent, () => {
-      ++events
-      if (events === numberOfEventsToWait) {
-        resolve(events)
-      }
-    })
+    switch (workerNodeEvent) {
+      case 'idle':
+      case 'backPressure':
+      case 'taskFinished':
+        workerNode.on(workerNodeEvent, () => {
+          ++events
+          if (events === numberOfEventsToWait) {
+            resolve(events)
+          }
+        })
+        break
+      default:
+        throw new Error('Invalid worker node event')
+    }
     if (timeout >= 0) {
       setTimeout(() => {
         resolve(events)