You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@openwhisk.apache.org by GitBox <gi...@apache.org> on 2021/07/12 06:34:12 UTC
[GitHub] [openwhisk] style95 commented on a change in pull request #5110: [New Scheduler] Add memory queue for the new scheduler
style95 commented on a change in pull request #5110:
URL: https://github.com/apache/openwhisk/pull/5110#discussion_r667633847
##########
File path: core/scheduler/src/main/scala/org/apache/openwhisk/core/scheduler/queue/MemoryQueue.scala
##########
@@ -0,0 +1,1097 @@
+package org.apache.openwhisk.core.scheduler.queue
+
+import java.time.{Duration, Instant}
+
+import akka.actor.Status.{Failure => FailureMessage}
+import akka.actor.{ActorRef, ActorSystem, Cancellable, FSM, Props, Stash}
+import akka.util.Timeout
+import org.apache.openwhisk.common._
+import org.apache.openwhisk.core.ack.ActiveAck
+import org.apache.openwhisk.core.connector.ContainerCreationError.{TooManyConcurrentRequests, ZeroNamespaceLimit}
+import org.apache.openwhisk.core.connector._
+import org.apache.openwhisk.core.database.{NoDocumentException, UserContext}
+import org.apache.openwhisk.core.entity._
+import org.apache.openwhisk.core.entity.size._
+import org.apache.openwhisk.core.etcd.EtcdClient
+import org.apache.openwhisk.core.etcd.EtcdKV.ContainerKeys.containerPrefix
+import org.apache.openwhisk.core.etcd.EtcdKV.{ContainerKeys, QueueKeys, ThrottlingKeys}
+import org.apache.openwhisk.core.scheduler.SchedulerEndpoints
+import org.apache.openwhisk.core.scheduler.grpc.{GetActivation, ActivationResponse => GetActivationResponse}
+import org.apache.openwhisk.core.scheduler.message.{
+ ContainerCreation,
+ ContainerDeletion,
+ FailedCreationJob,
+ SuccessfulCreationJob
+}
+import org.apache.openwhisk.core.service._
+import org.apache.openwhisk.core.{ConfigKeys, WhiskConfig}
+import org.apache.openwhisk.http.Messages.{namespaceLimitUnderZero, tooManyConcurrentRequests}
+import pureconfig.generic.auto._
+import pureconfig.loadConfigOrThrow
+import spray.json._
+
+import scala.annotation.tailrec
+import scala.collection.immutable.Queue
+import scala.collection.mutable
+import scala.concurrent.duration._
+import scala.concurrent.{duration, ExecutionContextExecutor, Future, Promise}
+import scala.language.postfixOps
+import scala.util.{Failure, Success}
+
+// States
+sealed trait MemoryQueueState
+case object Uninitialized extends MemoryQueueState
+case object Running extends MemoryQueueState
+case object Idle extends MemoryQueueState
+case object Flushing extends MemoryQueueState
+case object Removing extends MemoryQueueState
+case object Removed extends MemoryQueueState
+case object ActionThrottled extends MemoryQueueState
+case object NamespaceThrottled extends MemoryQueueState
+
+// Data
+sealed abstract class MemoryQueueData()
+case class NoData() extends MemoryQueueData()
+case class NoActors() extends MemoryQueueData()
+case class RunningData(schedulerActor: ActorRef, droppingActor: ActorRef) extends MemoryQueueData()
+case class ThrottledData(schedulerActor: ActorRef, droppingActor: ActorRef) extends MemoryQueueData()
+case class FlushingData(schedulerActor: ActorRef,
+ droppingActor: ActorRef,
+ error: ContainerCreationError,
+ reason: String,
+ activeDuringFlush: Boolean = false)
+ extends MemoryQueueData()
+case class RemovingData(schedulerActor: ActorRef, droppingActor: ActorRef, outdated: Boolean) extends MemoryQueueData()
+
+// Events sent by the actor
+case class QueueRemoved(invocationNamespace: String, action: DocInfo, leaderKey: Option[String])
+case class QueueReactivated(invocationNamespace: String, action: FullyQualifiedEntityName, docInfo: DocInfo)
+case class CancelPoll(promise: Promise[Either[MemoryQueueError, ActivationMessage]])
+case object QueueRemovedCompleted
+case object FlushPulse
+
+// Events received by the actor
+case object Start
+case object VersionUpdated
+case object StopSchedulingAsOutdated
+
+sealed trait RequiredAction
+case object Skip extends RequiredAction
+case object AddInitialContainer extends RequiredAction
+case object AddContainer extends RequiredAction
+case class EnableNamespaceThrottling(dropMsg: Boolean) extends RequiredAction
+case object DisableNamespaceThrottling extends RequiredAction
+case object EnableActionThrottling extends RequiredAction
+case object DisableActionThrottling extends RequiredAction
+case object Pausing extends RequiredAction
+case class DecisionResults(required: RequiredAction, num: Int)
+
+case class TimeSeriesActivationEntry(timestamp: Instant, msg: ActivationMessage)
+
+class MemoryQueue(private val etcdClient: EtcdClient,
+ private val durationChecker: DurationChecker,
+ private val action: FullyQualifiedEntityName,
+ messagingProducer: MessageProducer,
+ config: WhiskConfig,
+ invocationNamespace: String,
+ revision: DocRevision,
+ endpoints: SchedulerEndpoints,
+ actionMetaData: WhiskActionMetaData,
+ dataManagementService: ActorRef,
+ watcherService: ActorRef,
+ containerManager: ActorRef,
+ decisionMaker: ActorRef,
+ schedulerId: SchedulerInstanceId,
+ ack: ActiveAck,
+ store: (TransactionId, WhiskActivation, UserContext) => Future[Any],
+ getUserLimit: String => Future[Int],
+ checkToDropStaleActivation: (Queue[TimeSeriesActivationEntry],
+ Long,
+ String,
+ FullyQualifiedEntityName,
+ MemoryQueueState,
+ ActorRef) => Unit,
+ queueConfig: QueueConfig)(implicit logging: Logging)
+ extends FSM[MemoryQueueState, MemoryQueueData]
+ with Stash {
+
+ private implicit val ec: ExecutionContextExecutor = context.dispatcher
+ private implicit val actorSystem: ActorSystem = context.system
+ private implicit val timeout = Timeout(5.seconds)
+ private implicit val order: Ordering[BufferedRequest] = Ordering.by(_.containerId)
+
+ private val unversionedAction = action.copy(version = None)
+ private val checkInterval: FiniteDuration = 100 milliseconds
+ private val StaleThreshold: Double = 100.0
+ private val StaleDuration = Duration.ofMillis(StaleThreshold.toLong)
+ private val dropInterval: FiniteDuration = 10 seconds
+ private val leaderKey = QueueKeys.queue(invocationNamespace, unversionedAction, leader = true)
+ private val inProgressContainerPrefixKey =
+ containerPrefix(ContainerKeys.inProgressPrefix, invocationNamespace, action, Some(revision))
+ private val existingContainerPrefixKey =
+ containerPrefix(ContainerKeys.namespacePrefix, invocationNamespace, action, Some(revision))
+ private val namespaceThrottlingKey = ThrottlingKeys.namespace(EntityName(invocationNamespace))
+ private val actionThrottlingKey = ThrottlingKeys.action(invocationNamespace, unversionedAction)
+ private val pollTimeOut = 1.seconds
+ private var requestBuffer = mutable.PriorityQueue.empty[BufferedRequest]
+ private val memory = actionMetaData.limits.memory.megabytes.MB
+ private val queueRemovedMsg = QueueRemoved(invocationNamespace, action.toDocId.asDocInfo(revision), Some(leaderKey))
+ private val staleQueueRemovedMsg = QueueRemoved(invocationNamespace, action.toDocId.asDocInfo(revision), None)
+
+ private[queue] var containers = Set.empty[String]
+ private[queue] var creationIds = Set.empty[String]
+
+ private[queue] var queue = Queue.empty[TimeSeriesActivationEntry]
+ private[queue] var in: Int = 0
+ private[queue] val namespaceContainerCount = NamespaceContainerCount(invocationNamespace, etcdClient, watcherService)
+ private[queue] var averageDuration: Option[Double] = None
+ private[queue] var averageDurationBuffer = AverageRingBuffer(queueConfig.durationBufferSize)
+ private[queue] var limit: Option[Int] = None
+ private[queue] var initialized = false
+
+ private val logScheduler: Cancellable = context.system.scheduler.scheduleWithFixedDelay(0.seconds, 1.seconds) { () =>
+ MetricEmitter.emitGaugeMetric(
+ LoggingMarkers.SCHEDULER_QUEUE_WAITING_ACTIVATION(s"$invocationNamespace/$action"),
+ queue.size)
+
+ MetricEmitter.emitGaugeMetric(
+ LoggingMarkers.SCHEDULER_NAMESPACE_CONTAINER(invocationNamespace),
+ namespaceContainerCount.existingContainerNumByNamespace)
+ MetricEmitter.emitGaugeMetric(
+ LoggingMarkers.SCHEDULER_NAMESPACE_INPROGRESS_CONTAINER(invocationNamespace),
+ namespaceContainerCount.inProgressContainerNumByNamespace)
+
+ MetricEmitter.emitGaugeMetric(
+ LoggingMarkers.SCHEDULER_ACTION_CONTAINER(invocationNamespace, action.asString),
+ containers.size)
+ MetricEmitter.emitGaugeMetric(
+ LoggingMarkers.SCHEDULER_ACTION_INPROGRESS_CONTAINER(invocationNamespace, action.asString),
+ creationIds.size)
+ }
+
+ getAverageDuration()
+
+ private val watcherName = s"memory-queue-$action-$revision"
+ // watch existing containers for action and namespace
+ private val watchedKeys = Seq(inProgressContainerPrefixKey, existingContainerPrefixKey)
+
+ watchedKeys.foreach { key =>
+ watcherService ! WatchEndpoint(key, "", isPrefix = true, watcherName, Set(PutEvent, DeleteEvent))
+ }
+
+ startWith(Uninitialized, NoData())
+
+ when(Uninitialized) {
+ case Event(Start, _) =>
+ logging.info(this, s"[$invocationNamespace:$action:$stateName] a new queue is created.")
+ val (schedulerActor, droppingActor) = startMonitoring()
+ initializeThrottling()
+
+ watcherService ! WatchEndpoint(leaderKey, endpoints.serialize, isPrefix = false, watcherName, Set(DeleteEvent))
+
+ goto(Running) using RunningData(schedulerActor, droppingActor)
+
+ // this is the case that the action version is updated, so no data needs to be stored
+ case Event(VersionUpdated, _) =>
+ val (schedulerActor, droppingActor) = startMonitoring()
+
+ goto(Running) using RunningData(schedulerActor, droppingActor)
+
+ // other messages should not be handled in this state.
+ case _ =>
+ stash()
+ stay
+ }
+
+ when(Running, stateTimeout = queueConfig.idleGrace) {
+ case Event(EnableNamespaceThrottling(dropMsg), data: RunningData) =>
+ logging.info(this, s"[$invocationNamespace:$action:$stateName] Enable namespace throttling.")
+ enableNamespaceThrottling()
+
+ // if no container could be created, it is same with Flushing state.
+ if (dropMsg) {
+ completeAllActivations(tooManyConcurrentRequests, isWhiskError = false)
+ goto(Flushing) using FlushingData(
+ data.schedulerActor,
+ data.droppingActor,
+ TooManyConcurrentRequests,
+ tooManyConcurrentRequests)
+ } else {
+ // if there are already some containers running, activations can still be processed so goto the NamespaceThrottled state.
+ goto(NamespaceThrottled) using ThrottledData(data.schedulerActor, data.droppingActor)
+ }
+
+ case Event(StateTimeout, data: RunningData) =>
+ if (queue.isEmpty && (containers.size + creationIds.size) <= 0) {
+ logging.info(
+ this,
+ s"[$invocationNamespace:$action:$stateName] No activations coming in ${queueConfig.idleGrace}")
+ actorSystem.stop(data.schedulerActor)
+ actorSystem.stop(data.droppingActor)
+
+ goto(Idle) using NoActors()
+ } else {
+ logging.info(
+ this,
+ s"[$invocationNamespace:$action:$stateName] The queue is timed out but there are still ${queue.size} activation messages or (running: ${containers.size}, in-progress: ${creationIds.size}) containers")
+ stay
+ }
+
+ case Event(FailedCreationJob(creationId, _, _, _, error, message), RunningData(schedulerActor, droppingActor)) =>
+ creationIds -= creationId.asString
+ // when there is no container, it moves to the Flushing state as no activations can be invoked
+ if (containers.size <= 0) {
+ val isWhiskError = ContainerCreationError.whiskErrors.contains(error)
+ completeAllActivations(message, isWhiskError)
+ logging.error(
+ this,
+ s"[$invocationNamespace:$action:$stateName] Failed to create an initial container due to ${if (isWhiskError) "whiskError"
+ else "developerError"}, reason: $message.")
+
+ goto(Flushing) using FlushingData(schedulerActor, droppingActor, error, message)
+ } else
+ // if there are already some containers running, activations can be handled anyway.
+ stay
+ }
+
+ // there is no timeout for this state as when there is no further message, it would move to the Running state again.
+ when(NamespaceThrottled) {
+ case Event(msg: ActivationMessage, _: ThrottledData) =>
+ handleActivationMessage(msg)
+ stay
+
+ case Event(DisableNamespaceThrottling, data: ThrottledData) =>
+ logging.info(this, s"[$invocationNamespace:$action:$stateName] Disable namespace throttling.")
+ disableNamespaceThrottling()
+ goto(Running) using RunningData(data.schedulerActor, data.schedulerActor)
+ }
+
+ // there is no timeout for this state as when there is no further message, it would move to the Running state again.
+ when(ActionThrottled) {
+ // since there are already too many activation messages, it drops the new messages
+ case Event(msg: ActivationMessage, ThrottledData(_, _)) =>
+ completeErrorActivation(msg, tooManyConcurrentRequests, isWhiskError = false)
+ stay
+ }
+
+ when(Idle, stateTimeout = queueConfig.stopGrace) {
+ case Event(msg: ActivationMessage, _: NoActors) =>
+ val (schedulerActor, droppingActor) = startMonitoring()
+ handleActivationMessage(msg)
+ goto(Running) using RunningData(schedulerActor, droppingActor)
+
+ case Event(request: GetActivation, _) if request.action == action =>
+ sender ! GetActivationResponse(Left(NoActivationMessage()))
+ stay
+
+ case Event(StateTimeout, _: NoActors) =>
+ logging.info(this, s"[$invocationNamespace:$action:$stateName] The queue is timed out, stop the queue.")
+ cleanUpDataAndGotoRemoved()
+
+ case Event(GracefulShutdown, _: NoActors) =>
+ logging.info(this, s"[$invocationNamespace:$action:$stateName] Received GracefulShutdown, stop the queue.")
+ cleanUpDataAndGotoRemoved()
+
+ case Event(StopSchedulingAsOutdated, _: NoActors) =>
+ logging.info(this, s"[$invocationNamespace:$action:$stateName] stop further scheduling.")
+
+ cleanUpWatcher()
+
+ // let QueueManager know this queue is no longer in charge.
+ context.parent ! staleQueueRemovedMsg
+
+ // since the queue is outdated and there is no activation, delete all old containers.
+ containerManager ! ContainerDeletion(invocationNamespace, action, revision, actionMetaData)
+
+ goto(Removed) using NoData()
+ }
+
+ when(Flushing) {
+ // an initial container is successfully created.
+ case Event(SuccessfulCreationJob(creationId, _, _, _), FlushingData(schedulerActor, droppingActor, _, _, _)) =>
+ creationIds -= creationId.asString
+
+ goto(Running) using RunningData(schedulerActor, droppingActor)
+
+ // log the failed information
+ case Event(FailedCreationJob(creationId, _, _, _, _, message), data: FlushingData) =>
+ creationIds -= creationId.asString
+ logging.info(
+ this,
+ s"[$invocationNamespace:$action:$stateName][$creationId] Failed to create a container due to $message")
+
+ // keep updating the reason
+ stay using data.copy(reason = message)
+
+ // since there is no container, activations cannot be handled.
+ case Event(msg: ActivationMessage, data: FlushingData) =>
+ completeErrorActivation(msg, data.reason, ContainerCreationError.whiskErrors.contains(data.error))
+ stay() using data.copy(activeDuringFlush = true)
+
+ // Since SchedulingDecisionMaker keep sending a message to create a container, this state is not automatically timed out.
+ // Instead, StateTimeout message will be sent by a timer.
+ case Event(StateTimeout, data: FlushingData) =>
+ completeAllActivations(data.reason, ContainerCreationError.whiskErrors.contains(data.error))
+ if (data.activeDuringFlush)
+ stay using data.copy(activeDuringFlush = false)
+ else
+ cleanUpActorsAndGotoRemoved(data)
+
+ case Event(GracefulShutdown, data: FlushingData) =>
+ completeAllActivations(data.reason, ContainerCreationError.whiskErrors.contains(data.error))
+ logging.info(this, s"[$invocationNamespace:$action:$stateName] Received GracefulShutdown, stop the queue.")
+ cleanUpActorsAndGotoRemoved(data)
+ }
+
+ // in case there is any activation in the queue, it waits until all of them are handled.
+ when(Removing, stateTimeout = queueConfig.gracefulShutdownTimeout) {
+ // When there is no message in the queue, SchedulingDecisionMaker would stop sending any message
+ // So the queue can be timed out on every gracefulShutdownTimeout
+ case Event(QueueRemovedCompleted | StateTimeout, data: RemovingData) =>
+ cleanUpActorsAndGotoRemovedIfPossible(data)
+
+ case Event(GracefulShutdown, data: RemovingData) =>
+ logging.info(
+ this,
+ s"[$invocationNamespace:$action:$stateName] The queue received GracefulShutdown trying to stop the queue.")
+ cleanUpActorsAndGotoRemovedIfPossible(data)
+
+ case Event(StopSchedulingAsOutdated, data: RemovingData) =>
+ logging.info(
+ this,
+ s"[$invocationNamespace:$action:$stateName] The queue received StopSchedulingAsOutdated trying to stop the queue.")
+ cleanUpActorsAndGotoRemovedIfPossible(data.copy(outdated = true))
+ }
+
+ when(Removed, stateTimeout = queueConfig.gracefulShutdownTimeout) {
+ // since this Queue will be terminated, rescheduling the msg
+ case Event(msg: ActivationMessage, _: NoData) =>
+ context.parent ! msg
+ stay()
+
+ // this queue is going to stop so let client connect to a new queue
+ case Event(request: GetActivation, _: NoData) if request.action == action =>
+ implicit val tid = request.transactionId
+ logging.info(
+ this,
+ s"[$invocationNamespace:$action:$stateName] Get activation request ${request.containerId}, let client connect to a new queue.")
+ forwardAllActivations(context.parent)
+ sender ! GetActivationResponse(Left(NoMemoryQueue()))
+
+ stay
+
+ // actors and data are already wiped
+ case Event(QueueRemovedCompleted, _: NoData) =>
+ stop()
+
+ // This is not supposed to happen. This will ensure the queue does not run forever.
+ // This can happen when QueueManager could not respond with QueueRemovedCompleted for some reason.
+ case Event(StateTimeout, _: NoData) =>
+ context.parent ! queueRemovedMsg
+
+ stop()
+
+ // This queue is going to stop, do nothing
+ case Event(msg @ (StopSchedulingAsOutdated | GracefulShutdown), _: NoData) =>
+ logging.info(
+ this,
+ s"[$invocationNamespace:$action:$stateName] The queue received $msg but do nothing as it is going to stop.")
+ stay
+ }
+
+ whenUnhandled {
+ // The queue endpoint is removed, trying to restore it.
+ case Event(WatchEndpointRemoved(_, `leaderKey`, value, false), data) =>
+ data match {
+ case RemovingData(_, _, _) =>
+ logging.info(
+ this,
+ s"[$invocationNamespace:$action:$stateName] This queue is shutdown by `/disable` api, do nothing here.")
+ case _ =>
+ dataManagementService ! RegisterInitialData(leaderKey, value, failoverEnabled = false, Some(self)) // the watcher is already setup
+ }
+ stay
+
+ // we don't care the storage results for namespaceThrottlingKey
+ case Event(InitialDataStorageResults(`namespaceThrottlingKey`, _), _) =>
+ stay
+
+ // The queue endpoint is restored
+ case Event(InitialDataStorageResults(`leaderKey`, Right(_)), _) =>
+ stay
+
+ // this can be a case that there is another queue already running.
+ // it can happen if a node is segregated by the temporal network rupture and the queue endpoint is removed.
+ case Event(InitialDataStorageResults(`leaderKey`, Left(_)), data) =>
+ logging.warn(this, s"[$invocationNamespace:$action:$stateName] the queue is superseded by a new queue.")
+ // let QueueManager know this queue is no longer in charge.
+ context.parent ! queueRemovedMsg
+
+ // forward all activations to the parent queue manager.
+ // parent queue manager is supposed to removed the reference of this queue and forward messages to a new queue
+ forwardAllActivations(context.parent)
+
+ // only clean up actors because etcd data is already being used by another queue
+ cleanUpActors(data)
+
+ goto(Removed) using NoData()
+
+ case Event(WatchEndpointRemoved(watchKey, key, _, true), _) =>
+ watchKey match {
+ case `inProgressContainerPrefixKey` =>
+ creationIds -= key.split("/").last
+ case `existingContainerPrefixKey` =>
+ containers -= key.split("/").last
+ case _ =>
+ }
+ stay
+
+ case Event(WatchEndpointInserted(watchKey, key, _, true), _) =>
+ watchKey match {
+ case `inProgressContainerPrefixKey` =>
+ creationIds += key.split("/").last
+ case `existingContainerPrefixKey` =>
+ containers += key.split("/").last
+ case _ =>
+ }
+ stay
+
+ // common case for Running, NamespaceThrottled, ActionThrottled
+ case Event(SuccessfulCreationJob(creationId, _, _, _), _) =>
+ creationIds -= creationId.asString
+ stay()
+
+ // for other cases
+ case Event(FailedCreationJob(creationId, invocationNamespace, action, revision, _, message), _) =>
+ creationIds -= creationId.asString
+ logging.info(
+ this,
+ s"[$invocationNamespace:$action:$stateName][$creationId] Got failed creation job with revision $revision and error $message.")
+ stay()
+
+ // common case for Running, NamespaceThrottled, ActionThrottled, Removing
+ case Event(cancel: CancelPoll, _) =>
+ cancel.promise.trySuccess(Left(NoActivationMessage()))
+
+ stay
+
+ // common case for Running, NamespaceThrottled, ActionThrottled, Removing
+ case Event(msg: ActivationMessage, _) =>
+ handleActivationMessage(msg)
+
+ // common case for Running, NamespaceThrottled, ActionThrottled, Removing
+ case Event(request: GetActivation, _) if request.action == action =>
+ implicit val tid = request.transactionId
+ if (request.alive) {
+ containers += request.containerId
+ handleActivationRequest(request)
+ } else {
+ logging.info(this, s"Remove containerId because ${request.containerId} is not alive")
+ sender ! GetActivationResponse(Left(NoActivationMessage()))
+ containers -= request.containerId
+ stay
+ }
+
+ // common case for Running, NamespaceThrottled, ActionThrottled, Removing
+ case Event(request: GetActivation, _) if request.action != action =>
+ implicit val tid = request.transactionId
+ logging.warn(this, s"[$invocationNamespace:$action:$stateName] version mismatch ${request.action}")
+ sender ! GetActivationResponse(Left(ActionMismatch()))
+
+ stay
+
+ case Event(DropOld, _) =>
+ if (queue.nonEmpty && Duration
+ .between(queue.head.timestamp, Instant.now)
+ .compareTo(Duration.ofMillis(queueConfig.maxRetentionMs)) < 0) {
+ logging.error(
+ this,
+ s"[$invocationNamespace:$action:$stateName] Drop some stale activations for $revision, existing container is ${containers.size}, inProgress container is ${creationIds.size}, state data: $stateData, in is $in, current: ${queue.size}.")
+ logging.error(
+ this,
+ s"[$invocationNamespace:$action:$stateName] the head stale message: ${queue.head.msg.activationId}")
+ }
+ queue = MemoryQueue.dropOld(queue, Duration.ofMillis(queueConfig.maxRetentionMs), completeErrorActivation)
+
+ stay
+
+ // common case for all statuses
+ case Event(StatusQuery, _) =>
+ sender ! StatusData(invocationNamespace, action.asString, queue.size, stateName.toString, stateData.toString)
+ stay
+
+ // Common case for all cases
+ case Event(GracefulShutdown, data) =>
+ logging.info(this, s"[$invocationNamespace:$action:$stateName] Gracefully shutdown the memory queue.")
+ // delete relative data, e.g leaderKey, namespaceThrottlingKey, actionThrottlingKey
+ cleanUpData()
+
+ // let queue manager knows this queue is going to stop and let it forward incoming activations to a new queue
+ context.parent ! queueRemovedMsg
+
+ goto(Removing) using getRemovingData(data, outdated = false)
+
+ // the version is updated. it's a shared case for all states
+ case Event(StopSchedulingAsOutdated, data) =>
+ logging.info(this, s"[$invocationNamespace:$action:$stateName] stop further scheduling.")
+ // let QueueManager know this queue is no longer in charge.
+ context.parent ! staleQueueRemovedMsg
+
+ goto(Removing) using getRemovingData(data, outdated = true)
+
+ case Event(t: FailureMessage, _) =>
+ logging.error(this, s"[$invocationNamespace:$action:$stateName] got an unexpected failure message: $t")
+
+ stay
+
+ case Event(msg: DecisionResults, _) =>
+ val DecisionResults(result, num) = msg
+ result match {
+ case AddInitialContainer if num > 0 =>
Review comment:
AFAIK, there is no case where SchedulingDecisionMaker sends `AddInitialContainer` with `num == 0`?
##########
File path: common/scala/src/main/scala/org/apache/openwhisk/common/AverageRingBuffer.scala
##########
@@ -0,0 +1,34 @@
+package org.apache.openwhisk.common
+
+object AverageRingBuffer {
+ def apply(maxSize: Int) = new AverageRingBuffer(maxSize)
+}
+
+/**
+ * This buffer provides the average of the given elements.
+ * The number of elements are limited and the first element is removed if the maximum size is reached.
+ * Since it is based on the Vector, its operation takes effectively constant time.
+ * For more details, please visit https://docs.scala-lang.org/overviews/collections/performance-characteristics.html
+ *
+ * @param maxSize the maximum size of the buffer
+ */
+class AverageRingBuffer(private val maxSize: Int) {
Review comment:
Yes, that makes sense.
The ring size in our case was relatively small such as 10, so we calculated the average of recent 10 activations. Even if there was a skew at some point, it quickly gets back to the "normal" average if the slow activation is a transient issue.
Also, it only affects the timing to add more containers, and activations are still being processed by existing containers. So there was no critical impact.
Basically, the median would be better, but the average is a much simpler and cheaper solution so I chose it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@openwhisk.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org