diff --git a/common/scala/src/main/scala/whisk/common/Scheduler.scala
b/common/scala/src/main/scala/whisk/common/Scheduler.scala
index 28d150260e..bf73ad7d66 100644
--- a/common/scala/src/main/scala/whisk/common/Scheduler.scala
+++ b/common/scala/src/main/scala/whisk/common/Scheduler.scala
@@ -22,11 +22,7 @@ import scala.concurrent.duration._
import scala.util.Failure
import scala.util.Success
import scala.util.Try
-
-import akka.actor.Actor
-import akka.actor.ActorSystem
-import akka.actor.Cancellable
-import akka.actor.Props
+import akka.actor.{Actor, ActorRef, ActorSystem, Cancellable, Props}
/**
* Scheduler utility functions to execute tasks in a repetitive way with
controllable behavior
@@ -99,7 +95,7 @@ object Scheduler {
name: String = "Scheduler")(f: () =>
Future[Any])(implicit system: ActorSystem,
logging: Logging,
transid: TransactionId =
-
TransactionId.unknown) = {
+
TransactionId.unknown): ActorRef = {
require(interval > Duration.Zero)
system.actorOf(Props(new Worker(initialDelay, interval, false, name, f)))
}
diff --git
a/common/scala/src/main/scala/whisk/core/connector/MessageConsumer.scala
b/common/scala/src/main/scala/whisk/core/connector/MessageConsumer.scala
index 14af69e4a9..753c19a36d 100644
--- a/common/scala/src/main/scala/whisk/core/connector/MessageConsumer.scala
+++ b/common/scala/src/main/scala/whisk/core/connector/MessageConsumer.scala
@@ -60,6 +60,7 @@ object MessageFeed {
protected[connector] case object Idle extends FeedState
protected[connector] case object FillingPipeline extends FeedState
protected[connector] case object DrainingPipeline extends FeedState
+ protected[connector] case object GracefulShutdownDrain extends FeedState
protected sealed trait FeedData
private case object NoData extends FeedData
@@ -70,6 +71,10 @@ object MessageFeed {
/** Steady state message, indicates capacity in downstream process to
receive more messages. */
object Processed
+ object GracefulShutdown
+
+ object Busy
+
/** Indicates the fill operation has completed. */
private case class FillCompleted(messages: Seq[(String, Int, Long,
Array[Byte])])
}
@@ -117,6 +122,7 @@ class MessageFeed(description: String,
// Best practice dictates a mutable variable pointing at an immutable
collection for this reason
private var outstandingMessages = immutable.Queue.empty[(String, Int, Long,
Array[Byte])]
private var handlerCapacity = maximumHandlerCapacity
+ private var fillingQueue = false
private implicit val tid = TransactionId.dispatcher
@@ -126,6 +132,7 @@ class MessageFeed(description: String,
when(Idle) {
case Event(Ready, _) =>
+ fillingQueue = true
fillPipeline()
goto(FillingPipeline)
@@ -143,14 +150,19 @@ class MessageFeed(description: String,
case Event(FillCompleted(messages), _) =>
outstandingMessages = outstandingMessages ++ messages
sendOutstandingMessages()
+ fillingQueue = false
if (shouldFillQueue()) {
+ fillingQueue = true
fillPipeline()
stay
} else {
goto(DrainingPipeline)
}
+ case Event(GracefulShutdown, _) =>
+ goto(GracefulShutdownDrain)
+
case _ => stay
}
@@ -158,11 +170,33 @@ class MessageFeed(description: String,
case Event(Processed, _) =>
updateHandlerCapacity()
sendOutstandingMessages()
+
if (shouldFillQueue()) {
+ fillingQueue = true
fillPipeline()
goto(FillingPipeline)
} else stay
+ case Event(GracefulShutdown, _) =>
+ goto(GracefulShutdownDrain)
+
+ case _ => stay
+ }
+
+ when(GracefulShutdownDrain) {
+ case Event(Processed, _) =>
+ updateHandlerCapacity()
+ sendOutstandingMessages()
+ stay
+
+ case Event(FillCompleted(messages), _) =>
+ outstandingMessages = outstandingMessages ++ messages
+ sendOutstandingMessages()
+ stay
+
+ case Event(Busy, _) =>
+ stay() replying (handlerCapacity != maximumHandlerCapacity &&
outstandingMessages.nonEmpty && fillingQueue)
+
case _ => stay
}
@@ -207,7 +241,7 @@ class MessageFeed(description: String,
val occupancy = outstandingMessages.size
if (occupancy > 0 && handlerCapacity > 0) {
// Easiest way with an immutable queue to cleanly dequeue
- // Head is the first elemeent of the queue, desugared w/ an assignment
pattern
+ // Head is the first element of the queue, desugared w/ an assignment
pattern
// Tail is everything but the first element, thus mutating the
collection variable
val (topic, partition, offset, bytes) = outstandingMessages.head
outstandingMessages = outstandingMessages.tail
diff --git
a/core/invoker/src/main/scala/whisk/core/containerpool/ContainerPool.scala
b/core/invoker/src/main/scala/whisk/core/containerpool/ContainerPool.scala
index d26ebdc2f9..70a3b98855 100644
--- a/core/invoker/src/main/scala/whisk/core/containerpool/ContainerPool.scala
+++ b/core/invoker/src/main/scala/whisk/core/containerpool/ContainerPool.scala
@@ -139,7 +139,7 @@ class ContainerPool(childFactory: ActorRefFactory =>
ActorRef,
val retryLogDeadline = if (isErrorLogged) {
logging.error(
this,
- s"Rescheduling Run message, too many message in the pool,
freePoolSize: ${freePool.size}, " +
+ s"Rescheduling Run message, too many messages in the pool,
freePoolSize: ${freePool.size}, " +
s"busyPoolSize: ${busyPool.size}, maxActiveContainers
${poolConfig.maxActiveContainers}, " +
s"userNamespace: ${r.msg.user.namespace}, action:
${r.action}")(r.msg.transid)
Some(logMessageInterval.fromNow)
@@ -178,6 +178,9 @@ class ContainerPool(childFactory: ActorRefFactory =>
ActorRef,
case RescheduleJob =>
freePool = freePool - sender()
busyPool = busyPool - sender()
+
+ case Busy =>
+ sender ! (busyPool.nonEmpty)
}
/** Creates a new container and updates state accordingly. */
diff --git a/core/invoker/src/main/scala/whisk/core/invoker/Invoker.scala
b/core/invoker/src/main/scala/whisk/core/invoker/Invoker.scala
index 4a69d3cf99..32cdc4a06e 100644
--- a/core/invoker/src/main/scala/whisk/core/invoker/Invoker.scala
+++ b/core/invoker/src/main/scala/whisk/core/invoker/Invoker.scala
@@ -20,22 +20,18 @@ package whisk.core.invoker
import scala.concurrent.Await
import scala.concurrent.duration._
import scala.concurrent.Future
-import scala.util.Failure
import scala.util.Try
import kamon.Kamon
import org.apache.curator.retry.RetryUntilElapsed
import org.apache.curator.framework.CuratorFrameworkFactory
import org.apache.curator.framework.recipes.shared.SharedCount
import akka.Done
-import akka.actor.ActorSystem
-import akka.actor.CoordinatedShutdown
+import akka.actor.{ActorSystem, CoordinatedShutdown}
import akka.stream.ActorMaterializer
import whisk.common.AkkaLogging
-import whisk.common.Scheduler
import whisk.core.WhiskConfig
import whisk.core.WhiskConfig._
import whisk.core.connector.MessagingProvider
-import whisk.core.connector.PingMessage
import whisk.core.entity._
import whisk.core.entity.ExecManifest
import whisk.core.entity.InstanceId
@@ -178,15 +174,10 @@ object Invoker {
case e: Exception => abort(s"Failed to initialize reactive invoker:
${e.getMessage}")
}
- Scheduler.scheduleWaitAtMost(1.seconds)(() => {
- producer.send("health", PingMessage(invokerInstance)).andThen {
- case Failure(t) => logger.error(this, s"failed to ping the controller:
$t")
- }
- })
-
val port = config.servicePort.toInt
BasicHttpService.startHttpService(new BasicRasService {}.route, port)(
actorSystem,
ActorMaterializer.create(actorSystem))
+
}
}
diff --git
a/core/invoker/src/main/scala/whisk/core/invoker/InvokerReactive.scala
b/core/invoker/src/main/scala/whisk/core/invoker/InvokerReactive.scala
index 20cbbd46cd..5b0335817f 100644
--- a/core/invoker/src/main/scala/whisk/core/invoker/InvokerReactive.scala
+++ b/core/invoker/src/main/scala/whisk/core/invoker/InvokerReactive.scala
@@ -20,12 +20,16 @@ package whisk.core.invoker
import java.nio.charset.StandardCharsets
import java.time.Instant
-import akka.actor.{ActorRefFactory, ActorSystem, Props}
+import akka.actor.{ActorRef, ActorRefFactory, ActorSystem, Props}
import akka.event.Logging.InfoLevel
import akka.stream.ActorMaterializer
+import akka.pattern.{after, ask, gracefulStop}
+import akka.util.Timeout
+
import org.apache.kafka.common.errors.RecordTooLargeException
import pureconfig._
import spray.json._
+
import whisk.common._
import whisk.core.{ConfigKeys, WhiskConfig}
import whisk.core.connector._
@@ -36,10 +40,13 @@ import whisk.core.entity._
import whisk.http.Messages
import whisk.spi.SpiLoader
-import scala.concurrent.{ExecutionContext, Future}
+import scala.concurrent.{Await, ExecutionContext, Future}
import scala.concurrent.duration._
import scala.util.{Failure, Success}
+import scala.language.postfixOps
+
import DefaultJsonProtocol._
+import sun.misc.{Signal, SignalHandler}
class InvokerReactive(
config: WhiskConfig,
@@ -76,7 +83,6 @@ class InvokerReactive(
"--ulimit" -> Set("nofile=1024:1024"),
"--pids-limit" -> Set("1024")) ++ logsProvider.containerParameters)
containerFactory.init()
- sys.addShutdownHook(containerFactory.cleanup())
/** Initialize needed databases */
private val entityStore = WhiskEntityStore.datastore()
@@ -284,4 +290,84 @@ class InvokerReactive(
})
}
+ val healthScheduler = Scheduler.scheduleWaitAtMost(1.seconds)(() => {
+ producer.send("health", PingMessage(instance)).andThen {
+ case Failure(t) => logging.error(this, s"failed to ping the
controller(s): $t")
+ }
+ })
+
+ /** Polls the pool's status and returns a future which completes once the
pool is idle. */
+ def waitForContainerPoolIdle(pool: ActorRef): Future[Unit] = {
+ implicit val timeout = Timeout(5 seconds)
+ val delay = 1.second
+
+ (pool ? Busy)
+ .mapTo[Boolean]
+ .flatMap {
+ case true =>
+ logging.info(this, "Container pool is not idle.")
+ after(delay, actorSystem.scheduler)(waitForContainerPoolIdle(pool))
+ case false =>
+ Future.successful(())
+ }
+ .recoverWith { case _ => after(delay,
actorSystem.scheduler)(waitForContainerPoolIdle(pool)) }
+ }
+
+ /** Polls the feed's status and returns a future which completes once the
feed is idle. */
+ def waitForActivationFeedIdle(feed: ActorRef): Future[Unit] = {
+ implicit val timeout = Timeout(5 seconds)
+ val delay = 1.second
+
+ activationFeed ! MessageFeed.GracefulShutdown
+ (feed ? MessageFeed.Busy)
+ .mapTo[Boolean]
+ .flatMap {
+ case true =>
+ logging.info(this, "Activation feed is not idle.")
+ after(delay, actorSystem.scheduler)(waitForActivationFeedIdle(feed))
+ case false =>
+ Future.successful(())
+ }
+ .recoverWith { case _ => after(delay,
actorSystem.scheduler)(waitForActivationFeedIdle(feed)) }
+ }
+
+ // Capture SIGTERM signals to gracefully shutdown the invoker. When
gracefully shutting down, the health scheduler is
+ // shutdown preventing additional actions from being scheduler to the
invoker, then the invoker processes its buffered
+ // messages from the activation feed, and waits for its user containers to
finish running before the process exits.
+ Signal.handle(new Signal("TERM"), new SignalHandler() {
+ override def handle(signal: Signal) = {
+ logging.info(this, s"Starting graceful shutdown")
+
+ // Order is important here so futures are ran sequentially
+ val shutdowns = for {
+ _ <- gracefulStop(healthScheduler, 5.seconds)
+ _ <- waitForActivationFeedIdle(activationFeed)
+ _ <- waitForContainerPoolIdle(pool)
+ } yield {
+ logging.info(this, "Successfully shutdown health scheduler, activation
feed, and container pool")
+ }
+
+ // Allow the shutdown to take a maximum of 3 times the maximum action
runtime since the feed can be
+ // buffered and we want to allow for some grace period. If a graceful
shutdown is not successful, the
+ // the invoker will continue running and a graceful shutdown can be
attempted again.
+ Await.result(shutdowns, TimeLimit.MAX_DURATION * 3)
+ containerFactory.cleanup()
+ logging.info(this, "Shutting down invoker")
+ System.exit(0)
+ }
+ })
+
+ // Capture SIGUSR2 signals to put the invoker into drain mode. When
draining, the health scheduler is shutdown
+ // preventing additional actions from being scheduled to the invoker
allowing the invoker to process its current
+ // queue.
+ Signal.handle(new Signal("USR2"), new SignalHandler() {
+ override def handle(signal: Signal) = {
+ logging.info(this, "Draining invoker")
+
+ gracefulStop(healthScheduler, 5.seconds).recover {
+ case _ => logging.info(this, "Health communication failed to shutdown
gracefully")
+ }
+ }
+ })
+
}
With regards,
Apache Git Services
