fix bug in average ring buffer and add negative protection to scheduling decision maker

Author: Brendan Doyle

common/scala/src/main/scala/org/apache/openwhisk/common/AverageRingBuffer.scala

@@ -32,18 +32,12 @@ object AverageRingBuffer {
 class AverageRingBuffer(private val maxSize: Int) {
   private var elements = Vector.empty[Double]
   private var sum = 0.0
-  private var max = 0.0
-  private var min = 0.0
 
   def nonEmpty: Boolean = elements.nonEmpty
 
   def average: Double = {
     val size = elements.size
-    if (size > 2) {
-      (sum - max - min) / (size - 2)
-    } else {
-      sum / size
-    }
+    sum / size
   }
 
   def add(el: Double): Unit = synchronized {
@@ -54,12 +48,6 @@ class AverageRingBuffer(private val maxSize: Int) {
       sum += el
       elements = elements :+ el
     }
-    if (el > max) {
-      max = el
-    }
-    if (el < min) {
-      min = el
-    }
   }
 
   def size(): Int = elements.size

core/scheduler/src/main/scala/org/apache/openwhisk/core/scheduler/queue/SchedulingDecisionMaker.scala

@@ -153,19 +153,24 @@ class SchedulingDecisionMaker(
               Running)
           // need more containers and a message is already processed
           case (Running, Some(duration)) =>
-            // we can safely get the value as we already checked the existence
-            val containerThroughput = (staleThreshold / duration) * maxActionConcurrency
+            // we can safely get the value as we already checked the existence, have extra protection in case duration is somehow negative
+            val containerThroughput = if (duration <= 0) {
+              maxActionConcurrency
+            } else {
+              (staleThreshold / duration) * maxActionConcurrency
+            }
             val expectedTps = containerThroughput * (existing + inProgress)
             val availableNonStaleActivations = availableMsg - staleActivationNum
 
             var staleContainerProvision = 0
             if (staleActivationNum > 0) {
+
               val num = ceiling(staleActivationNum.toDouble / containerThroughput)
               // if it tries to create more containers than existing messages, we just create shortage
               staleContainerProvision = (if (num > staleActivationNum) staleActivationNum else num) - inProgress
             }
 
-            if (availableNonStaleActivations >= expectedTps && existing + inProgress < availableNonStaleActivations) {
+            if (availableNonStaleActivations >= expectedTps && existing + inProgress < availableNonStaleActivations && duration > 0) {
               val num = ceiling((availableNonStaleActivations / containerThroughput) - existing - inProgress)
               // if it tries to create more containers than existing messages, we just create shortage
               val actualNum =
@@ -202,7 +207,11 @@ class SchedulingDecisionMaker(
           // this case is for that as a last resort.
           case (Removing, Some(duration)) if staleActivationNum > 0 =>
             // we can safely get the value as we already checked the existence
-            val containerThroughput = (staleThreshold / duration) * maxActionConcurrency
+            val containerThroughput = if (duration <= 0) {
+              maxActionConcurrency
+            } else {
+              (staleThreshold / duration) * maxActionConcurrency
+            }
             val num = ceiling(staleActivationNum.toDouble / containerThroughput)
             // if it tries to create more containers than existing messages, we just create shortage
             val actualNum = (if (num > staleActivationNum) staleActivationNum else num) - inProgress