@@ -85,7 +85,6 @@ case class QueueRemoved(invocationNamespace: String, action: DocInfo, leaderKey:
8585case class QueueReactivated (invocationNamespace : String , action : FullyQualifiedEntityName , docInfo : DocInfo )
8686case class CancelPoll (promise : Promise [Either [MemoryQueueError , ActivationMessage ]])
8787case object QueueRemovedCompleted
88- case object FlushPulse
8988
9089// Events received by the actor
9190case object Start
@@ -125,7 +124,7 @@ class MemoryQueue(private val etcdClient: EtcdClient,
125124 checkToDropStaleActivation : (Queue [TimeSeriesActivationEntry ],
126125 Long ,
127126 String ,
128- FullyQualifiedEntityName ,
127+ WhiskActionMetaData ,
129128 MemoryQueueState ,
130129 ActorRef ) => Unit ,
131130 queueConfig : QueueConfig )(implicit logging : Logging )
@@ -151,6 +150,7 @@ class MemoryQueue(private val etcdClient: EtcdClient,
151150 private val memory = actionMetaData.limits.memory.megabytes.MB
152151 private val queueRemovedMsg = QueueRemoved (invocationNamespace, action.toDocId.asDocInfo(revision), Some (leaderKey))
153152 private val staleQueueRemovedMsg = QueueRemoved (invocationNamespace, action.toDocId.asDocInfo(revision), None )
153+ private val actionRetentionTimeout = MemoryQueue .getRetentionTimeout(actionMetaData, queueConfig)
154154
155155 private [queue] var containers = Set .empty[String ]
156156 private [queue] var creationIds = Set .empty[String ]
@@ -197,7 +197,9 @@ class MemoryQueue(private val etcdClient: EtcdClient,
197197
198198 when(Uninitialized ) {
199199 case Event (Start , _) =>
200- logging.info(this , s " [ $invocationNamespace: $action: $stateName] a new queue is created. " )
200+ logging.info(
201+ this ,
202+ s " [ $invocationNamespace: $action: $stateName] a new queue is created, retentionTimeout: $actionRetentionTimeout, kind: ${actionMetaData.exec.kind}. " )
201203 val (schedulerActor, droppingActor) = startMonitoring()
202204 initializeThrottling()
203205
@@ -256,7 +258,9 @@ class MemoryQueue(private val etcdClient: EtcdClient,
256258 // when there is no container, it moves to the Flushing state as no activations can be invoked
257259 if (containers.size <= 0 ) {
258260 val isWhiskError = ContainerCreationError .whiskErrors.contains(error)
259- completeAllActivations(message, isWhiskError)
261+ if (! isWhiskError) {
262+ completeAllActivations(message, isWhiskError)
263+ }
260264 logging.error(
261265 this ,
262266 s " [ $invocationNamespace: $action: $stateName] Failed to create an initial container due to ${if (isWhiskError) " whiskError"
@@ -271,7 +275,11 @@ class MemoryQueue(private val etcdClient: EtcdClient,
271275 // there is no timeout for this state as when there is no further message, it would move to the Running state again.
272276 when(NamespaceThrottled ) {
273277 case Event (msg : ActivationMessage , _ : ThrottledData ) =>
274- handleActivationMessage(msg)
278+ if (containers.size + creationIds.size == 0 ) {
279+ completeErrorActivation(msg, tooManyConcurrentRequests, isWhiskError = false )
280+ } else {
281+ handleActivationMessage(msg)
282+ }
275283 stay
276284
277285 case Event (DisableNamespaceThrottling , data : ThrottledData ) =>
@@ -328,33 +336,51 @@ class MemoryQueue(private val etcdClient: EtcdClient,
328336 goto(Running ) using RunningData (schedulerActor, droppingActor)
329337
330338 // log the failed information
331- case Event (FailedCreationJob (creationId, _, _, _, _ , message), data : FlushingData ) =>
339+ case Event (FailedCreationJob (creationId, _, _, _, error , message), data : FlushingData ) =>
332340 creationIds -= creationId.asString
333341 logging.info(
334342 this ,
335343 s " [ $invocationNamespace: $action: $stateName][ $creationId] Failed to create a container due to $message" )
336344
337345 // keep updating the reason
338- stay using data.copy(reason = message)
346+ stay using data.copy(error = error, reason = message)
339347
340348 // since there is no container, activations cannot be handled.
341349 case Event (msg : ActivationMessage , data : FlushingData ) =>
342- completeErrorActivation(msg, data.reason, ContainerCreationError .whiskErrors.contains(data.error))
350+ logging.info(this , s " [ $invocationNamespace: $action: $stateName] got a new activation message ${msg.activationId}" )(
351+ msg.transid)
352+ val whiskError = isWhiskError(data.error)
353+ if (whiskError)
354+ queue = queue.enqueue(TimeSeriesActivationEntry (Instant .now, msg))
355+ else
356+ completeErrorActivation(msg, data.reason, whiskError)
343357 stay() using data.copy(activeDuringFlush = true )
344358
345359 // Since SchedulingDecisionMaker keep sending a message to create a container, this state is not automatically timed out.
346360 // Instead, StateTimeout message will be sent by a timer.
347- case Event (StateTimeout , data : FlushingData ) =>
348- completeAllActivations(data.reason, ContainerCreationError .whiskErrors.contains(data.error))
349- if (data.activeDuringFlush)
361+ case Event (StateTimeout | DropOld , data : FlushingData ) =>
362+ logging.info(this , s " [ $invocationNamespace: $action: $stateName] Received StateTimeout, drop stale messages. " )
363+ queue =
364+ MemoryQueue .dropOld(queue, Duration .ofMillis(actionRetentionTimeout), data.reason, completeErrorActivation)
365+ if (data.activeDuringFlush || queue.nonEmpty)
350366 stay using data.copy(activeDuringFlush = false )
351367 else
352368 cleanUpActorsAndGotoRemoved(data)
353369
354370 case Event (GracefulShutdown , data : FlushingData ) =>
355- completeAllActivations(data.reason, ContainerCreationError .whiskErrors.contains (data.error))
371+ completeAllActivations(data.reason, isWhiskError (data.error))
356372 logging.info(this , s " [ $invocationNamespace: $action: $stateName] Received GracefulShutdown, stop the queue. " )
357373 cleanUpActorsAndGotoRemoved(data)
374+
375+ case Event (StopSchedulingAsOutdated , data : FlushingData ) =>
376+ logging.info(this , s " [ $invocationNamespace: $action: $stateName] stop further scheduling. " )
377+ completeAllActivations(data.reason, isWhiskError(data.error))
378+ // let QueueManager know this queue is no longer in charge.
379+ context.parent ! staleQueueRemovedMsg
380+ cleanUpActors(data)
381+ cleanUpData()
382+
383+ goto(Removed ) using NoData ()
358384 }
359385
360386 // in case there is any activation in the queue, it waits until all of them are handled.
@@ -399,6 +425,7 @@ class MemoryQueue(private val etcdClient: EtcdClient,
399425
400426 // actors and data are already wiped
401427 case Event (QueueRemovedCompleted , _ : NoData ) =>
428+ logging.info(this , " stop fsm"
402429 stop()
403430
404431 // This is not supposed to happen. This will ensure the queue does not run forever.
@@ -523,15 +550,19 @@ class MemoryQueue(private val etcdClient: EtcdClient,
523550 case Event (DropOld , _) =>
524551 if (queue.nonEmpty && Duration
525552 .between(queue.head.timestamp, Instant .now)
526- .compareTo(Duration .ofMillis(queueConfig.maxRetentionMs )) < 0 ) {
553+ .compareTo(Duration .ofMillis(actionRetentionTimeout )) < 0 ) {
527554 logging.error(
528555 this ,
529556 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}. " )
530557 logging.error(
531558 this ,
532559 s " [ $invocationNamespace: $action: $stateName] the head stale message: ${queue.head.msg.activationId}" )
533560 }
534- queue = MemoryQueue .dropOld(queue, Duration .ofMillis(queueConfig.maxRetentionMs), completeErrorActivation)
561+ queue = MemoryQueue .dropOld(
562+ queue,
563+ Duration .ofMillis(actionRetentionTimeout),
564+ s " Activation processing is not initiated for $actionRetentionTimeout ms " ,
565+ completeErrorActivation)
535566
536567 stay
537568
@@ -861,7 +892,7 @@ class MemoryQueue(private val etcdClient: EtcdClient,
861892 // these schedulers will run forever and stop when the memory queue stops
862893 private def startMonitoring (): (ActorRef , ActorRef ) = {
863894 val droppingScheduler = Scheduler .scheduleWaitAtLeast(schedulingConfig.dropInterval) { () =>
864- checkToDropStaleActivation(queue, queueConfig.maxRetentionMs , invocationNamespace, action , stateName, self)
895+ checkToDropStaleActivation(queue, actionRetentionTimeout , invocationNamespace, actionMetaData , stateName, self)
865896 Future .successful(())
866897 }
867898
@@ -1055,11 +1086,12 @@ class MemoryQueue(private val etcdClient: EtcdClient,
10551086 causedBy ++ limits ++ binding
10561087 })
10571088 }
1089+
1090+ private def isWhiskError (error : ContainerCreationError ): Boolean = ContainerCreationError .whiskErrors.contains(error)
10581091}
10591092
10601093object MemoryQueue {
10611094 private [queue] val queueConfig = loadConfigOrThrow[QueueConfig ](ConfigKeys .schedulerQueue)
1062- private [queue] val MaxRetentionTime = queueConfig.maxRetentionMs
10631095
10641096 def props (etcdClient : EtcdClient ,
10651097 durationChecker : DurationChecker ,
@@ -1105,21 +1137,27 @@ object MemoryQueue {
11051137 def dropOld (
11061138 queue : Queue [TimeSeriesActivationEntry ],
11071139 retention : Duration ,
1140+ reason : String ,
11081141 completeErrorActivation : (ActivationMessage , String , Boolean ) => Future [Any ]): Queue [TimeSeriesActivationEntry ] = {
11091142 if (queue.isEmpty || Duration .between(queue.head.timestamp, Instant .now).compareTo(retention) < 0 )
11101143 queue
11111144 else {
1112- completeErrorActivation(queue.head.msg, s " activation processing is not initiated for $MaxRetentionTime ms " , true )
1113- dropOld(queue.tail, retention, completeErrorActivation)
1145+ completeErrorActivation(queue.head.msg, reason , true )
1146+ dropOld(queue.tail, retention, reason, completeErrorActivation)
11141147 }
11151148 }
11161149
11171150 def checkToDropStaleActivation (queue : Queue [TimeSeriesActivationEntry ],
11181151 maxRetentionMs : Long ,
11191152 invocationNamespace : String ,
1120- action : FullyQualifiedEntityName ,
1153+ actionMetaData : WhiskActionMetaData ,
11211154 stateName : MemoryQueueState ,
11221155 queueRef : ActorRef )(implicit logging : Logging ) = {
1156+ val action = actionMetaData.fullyQualifiedName(true )
1157+ logging.debug(
1158+ this ,
1159+ s " [ $invocationNamespace: $action: $stateName] use the given retention timeout: $maxRetentionMs for this action kind: ${actionMetaData.exec.kind}. " )
1160+
11231161 if (queue.nonEmpty && Duration
11241162 .between(queue.head.timestamp, Instant .now)
11251163 .compareTo(Duration .ofMillis(maxRetentionMs)) >= 0 ) {
@@ -1130,6 +1168,14 @@ object MemoryQueue {
11301168 queueRef ! DropOld
11311169 }
11321170 }
1171+
1172+ private def getRetentionTimeout (actionMetaData : WhiskActionMetaData , queueConfig : QueueConfig ): Long = {
1173+ if (actionMetaData.exec.kind == ExecMetaDataBase .BLACKBOX ) {
1174+ queueConfig.maxBlackboxRetentionMs
1175+ } else {
1176+ queueConfig.maxRetentionMs
1177+ }
1178+ }
11331179}
11341180
11351181case class QueueSnapshot (initialized : Boolean ,
@@ -1151,6 +1197,7 @@ case class QueueConfig(idleGrace: FiniteDuration,
11511197 gracefulShutdownTimeout : FiniteDuration ,
11521198 maxRetentionSize : Int ,
11531199 maxRetentionMs : Long ,
1200+ maxBlackboxRetentionMs : Long ,
11541201 throttlingFraction : Double ,
11551202 durationBufferSize : Int )
11561203
0 commit comments