You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@openwhisk.apache.org by cb...@apache.org on 2017/06/22 13:58:03 UTC
[incubator-openwhisk] 01/03: Keep inactive containers around even
in a fully loaded system.
This is an automated email from the ASF dual-hosted git repository.
cbickel pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-openwhisk.git
commit f79ae51ae46ce2cc19d164684c620081c7d3cc1c
Author: Markus Thoemmes <ma...@de.ibm.com>
AuthorDate: Mon Jun 5 11:08:05 2017 +0200
Keep inactive containers around even in a fully loaded system.
The ContainerPool will trash it's whole pool of containers if running containers are exhausting the maximum size of the pool.
This implements a scheme where the ContainerPool differs between "maximum containers that are allowed do work" and "maximum containers in the pool at all", where the latter can be greater than the first. This will keep containers longer around allowing for a higher warm-container hit rate with heterogeneous load.
---
.../whisk/core/containerpool/ContainerPool.scala | 106 ++++++++++---------
.../scala/whisk/core/invoker/InvokerReactive.scala | 1 +
.../containerpool/test/ContainerPoolTests.scala | 117 ++++++++++-----------
3 files changed, 115 insertions(+), 109 deletions(-)
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 fca7a99..012373f 100644
--- a/core/invoker/src/main/scala/whisk/core/containerpool/ContainerPool.scala
+++ b/core/invoker/src/main/scala/whisk/core/containerpool/ContainerPool.scala
@@ -52,20 +52,23 @@ case class WorkerData(data: ContainerData, state: WorkerState)
* Prewarm containers are only used, if they have matching arguments
* (kind, memory) and there is space in the pool.
*
- * @param childFactory method to create new container proxy actors
+ * @param childFactory method to create new container proxy actor
+ * @param maxActiveContainers maximum amount of containers doing work
* @param maxPoolSize maximum size of containers allowed in the pool
* @param feed actor to request more work from
* @param prewarmConfig optional settings for container prewarming
*/
class ContainerPool(
childFactory: ActorRefFactory => ActorRef,
+ maxActiveContainers: Int,
maxPoolSize: Int,
feed: ActorRef,
prewarmConfig: Option[PrewarmingConfig] = None) extends Actor {
- val logging = new AkkaLogging(context.system.log)
+ implicit val logging = new AkkaLogging(context.system.log)
- val pool = new mutable.HashMap[ActorRef, WorkerData]
- val prewarmedPool = new mutable.HashMap[ActorRef, WorkerData]
+ val freePool = mutable.Map[ActorRef, ContainerData]()
+ val busyPool = mutable.Map[ActorRef, ContainerData]()
+ val prewarmedPool = mutable.Map[ActorRef, ContainerData]()
prewarmConfig.foreach { config =>
logging.info(this, s"pre-warming ${config.count} ${config.exec.kind} containers")
@@ -77,52 +80,62 @@ class ContainerPool(
def receive: Receive = {
// A job to run on a container
case r: Run =>
- // Schedule a job to a warm container
- ContainerPool.schedule(r.action, r.msg.user.namespace, pool.toMap).orElse {
- // Create a cold container iff there's space in the pool
- if (pool.size < maxPoolSize) {
- takePrewarmContainer(r.action).orElse {
- Some(createContainer())
+ val container = if (busyPool.size < maxActiveContainers) {
+ // Schedule a job to a warm container
+ ContainerPool.schedule(r.action, r.msg.user.namespace, freePool.toMap).orElse {
+ if (busyPool.size + freePool.size < maxPoolSize) {
+ takePrewarmContainer(r.action).orElse {
+ Some(createContainer())
+ }
+ } else None
+ }.orElse {
+ // Remove a container and create a new one for the given job
+ ContainerPool.remove(r.action, r.msg.user.namespace, freePool.toMap).map { toDelete =>
+ removeContainer(toDelete)
+ takePrewarmContainer(r.action).getOrElse {
+ createContainer()
+ }
}
- } else None
- }.orElse {
- // Remove a container and create a new one for the given job
- ContainerPool.remove(r.action, r.msg.user.namespace, pool.toMap).map { toDelete =>
- removeContainer(toDelete)
- createContainer()
}
- } match {
+ } else None
+
+ container match {
case Some(actor) =>
- pool.get(actor) match {
- case Some(w) =>
- pool.update(actor, WorkerData(w.data, Busy))
+ freePool.get(actor) match {
+ case Some(data) =>
+ busyPool.update(actor, data)
+ freePool.remove(actor)
actor ! r // forwards the run request to the container
case None =>
logging.error(this, "actor data not found")
self ! r
}
- case None =>
- // "reenqueue" the request to find a container at a later point in time
- self ! r
+ case None => self ! r
}
// Container is free to take more work
- case NeedWork(data: WarmedData) => pool.update(sender(), WorkerData(data, Free))
+ case NeedWork(data: WarmedData) =>
+ freePool.update(sender(), data)
+ busyPool.remove(sender())
// Container is prewarmed and ready to take work
- case NeedWork(data: PreWarmedData) => prewarmedPool.update(sender(), WorkerData(data, Free))
+ case NeedWork(data: PreWarmedData) =>
+ prewarmedPool.update(sender(), data)
// Container got removed
- case ContainerRemoved => pool.remove(sender())
+ case ContainerRemoved =>
+ freePool.remove(sender())
+ busyPool.remove(sender())
// Activation completed
- case ActivationCompleted => feed ! ContainerReleased
+ case ActivationCompleted =>
+ feed ! ContainerReleased
}
/** Creates a new container and updates state accordingly. */
def createContainer() = {
val ref = childFactory(context)
- pool.update(ref, WorkerData(NoData(), Free))
+ freePool.update(ref, NoData())
ref
}
@@ -141,12 +154,12 @@ class ContainerPool(
val kind = action.exec.kind
val memory = action.limits.memory.megabytes.MB
prewarmedPool.find {
- case (_, WorkerData(PreWarmedData(_, `kind`, `memory`), _)) => true
- case _ => false
+ case (_, PreWarmedData(_, `kind`, `memory`)) => true
+ case _ => false
}.map {
case (ref, data) =>
// Move the container to the usual pool
- pool.update(ref, data)
+ freePool.update(ref, data)
prewarmedPool.remove(ref)
// Create a new prewarm container
prewarmContainer(config.exec, config.memoryLimit)
@@ -158,7 +171,8 @@ class ContainerPool(
/** Removes a container and updates state accordingly. */
def removeContainer(toDelete: ActorRef) = {
toDelete ! Remove
- pool.remove(toDelete)
+ freePool.remove(toDelete)
+ busyPool.remove(toDelete)
}
}
@@ -178,9 +192,9 @@ object ContainerPool {
* @param idles a map of idle containers, awaiting work
* @return a container if one found
*/
- def schedule[A](action: ExecutableWhiskAction, invocationNamespace: EntityName, idles: Map[A, WorkerData]): Option[A] = {
+ def schedule[A](action: ExecutableWhiskAction, invocationNamespace: EntityName, idles: Map[A, ContainerData]): Option[A] = {
idles.find {
- case (_, WorkerData(WarmedData(_, `invocationNamespace`, `action`, _), Free)) => true
+ case (_, WarmedData(_, `invocationNamespace`, `action`, _)) => true
case _ => false
}.map(_._1)
}
@@ -188,34 +202,30 @@ object ContainerPool {
/**
* Finds the best container to remove to make space for the job passed to run.
*
- * Determines which namespace consumes most resources in the current pool and
- * takes away one of their containers iff the namespace placing the new job is
- * not already the most consuming one.
+ * Determines the least recently used Free container in the pool.
*
* @param action the action that wants to get a container
* @param invocationNamespace the namespace, that wants to run the action
- * @param pool a map of all containers in the pool
+ * @param pool a map of all free containers in the pool
* @return a container to be removed iff found
*/
- def remove[A](action: ExecutableWhiskAction, invocationNamespace: EntityName, pool: Map[A, WorkerData]): Option[A] = {
- //try to find a Free container that is initialized with any OTHER action
- val grouped = pool.collect {
- case (ref, WorkerData(w: WarmedData, _)) if (w.action != action || w.invocationNamespace != invocationNamespace) => ref -> w
- }.groupBy {
- case (ref, data) => data.invocationNamespace
+ def remove[A](action: ExecutableWhiskAction, invocationNamespace: EntityName, pool: Map[A, ContainerData]): Option[A] = {
+ // Try to find a Free container that is initialized with any OTHER action
+ val freeContainers = pool.collect {
+ case (ref, w: WarmedData) if (w.action != action || w.invocationNamespace != invocationNamespace) => ref -> w
}
- if (!grouped.isEmpty) {
- val (maxConsumer, containersToDelete) = grouped.maxBy(_._2.size)
- val (ref, _) = containersToDelete.minBy(_._2.lastUsed)
+ if (freeContainers.nonEmpty) {
+ val (ref, _) = freeContainers.minBy(_._2.lastUsed)
Some(ref)
} else None
}
def props(factory: ActorRefFactory => ActorRef,
+ maxActive: Int,
size: Int,
feed: ActorRef,
- prewarmConfig: Option[PrewarmingConfig] = None) = Props(new ContainerPool(factory, size, feed, prewarmConfig))
+ prewarmConfig: Option[PrewarmingConfig] = None) = Props(new ContainerPool(factory, maxActive, size, feed, prewarmConfig))
}
/** Contains settings needed to perform container prewarming */
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 5f25521..cf7e3ae 100644
--- a/core/invoker/src/main/scala/whisk/core/invoker/InvokerReactive.scala
+++ b/core/invoker/src/main/scala/whisk/core/invoker/InvokerReactive.scala
@@ -135,6 +135,7 @@ class InvokerReactive(
val pool = actorSystem.actorOf(ContainerPool.props(
childFactory,
OldContainerPool.getDefaultMaxActive(config),
+ OldContainerPool.getDefaultMaxActive(config),
activationFeed,
Some(PrewarmingConfig(2, prewarmExec, 256.MB))))
diff --git a/tests/src/test/scala/whisk/core/containerpool/test/ContainerPoolTests.scala b/tests/src/test/scala/whisk/core/containerpool/test/ContainerPoolTests.scala
index e7fac0f..7a3454b 100644
--- a/tests/src/test/scala/whisk/core/containerpool/test/ContainerPoolTests.scala
+++ b/tests/src/test/scala/whisk/core/containerpool/test/ContainerPoolTests.scala
@@ -111,7 +111,7 @@ class ContainerPoolTests extends TestKit(ActorSystem("ContainerPool"))
val (containers, factory) = testContainers(1)
val feed = TestProbe()
- val pool = system.actorOf(ContainerPool.props(factory, 0, feed.ref))
+ val pool = system.actorOf(ContainerPool.props(factory, 0, 0, feed.ref))
containers(0).send(pool, ActivationCompleted)
feed.expectMsg(ContainerReleased)
}
@@ -125,7 +125,7 @@ class ContainerPoolTests extends TestKit(ActorSystem("ContainerPool"))
it should "reuse a warm container" in within(timeout) {
val (containers, factory) = testContainers(2)
val feed = TestProbe()
- val pool = system.actorOf(ContainerPool.props(factory, 2, feed.ref))
+ val pool = system.actorOf(ContainerPool.props(factory, 2, 2, feed.ref))
pool ! runMessage
containers(0).expectMsg(runMessage)
@@ -140,7 +140,7 @@ class ContainerPoolTests extends TestKit(ActorSystem("ContainerPool"))
val (containers, factory) = testContainers(2)
val feed = TestProbe()
- val pool = system.actorOf(ContainerPool.props(factory, 2, feed.ref))
+ val pool = system.actorOf(ContainerPool.props(factory, 2, 2, feed.ref))
pool ! runMessage
containers(0).expectMsg(runMessage)
// Note that the container doesn't respond, thus it's not free to take work
@@ -153,7 +153,7 @@ class ContainerPoolTests extends TestKit(ActorSystem("ContainerPool"))
val feed = TestProbe()
// a pool with only 1 slot
- val pool = system.actorOf(ContainerPool.props(factory, 1, feed.ref))
+ val pool = system.actorOf(ContainerPool.props(factory, 1, 1, feed.ref))
pool ! runMessage
containers(0).expectMsg(runMessage)
containers(0).send(pool, NeedWork(warmedData()))
@@ -164,12 +164,40 @@ class ContainerPoolTests extends TestKit(ActorSystem("ContainerPool"))
containers(1).expectMsg(runMessageDifferentEverything)
}
+ it should "cache a container if there is still space in the pool" in within(timeout) {
+ val (containers, factory) = testContainers(2)
+ val feed = TestProbe()
+
+ // a pool with only 1 active slot but 2 slots in total
+ val pool = system.actorOf(ContainerPool.props(factory, 1, 2, feed.ref))
+
+ // Run the first container
+ pool ! runMessage
+ containers(0).expectMsg(runMessage)
+ containers(0).send(pool, NeedWork(warmedData()))
+ containers(0).send(pool, ActivationCompleted)
+ feed.expectMsg(ContainerReleased)
+
+ // Run the second container, don't remove the first one
+ pool ! runMessageDifferentEverything
+ containers(1).expectMsg(runMessageDifferentEverything)
+ containers(1).send(pool, NeedWork(warmedData()))
+ containers(1).send(pool, ActivationCompleted)
+ feed.expectMsg(ContainerReleased)
+
+ pool ! runMessageDifferentNamespace
+ containers(2).expectMsg(runMessageDifferentNamespace)
+
+ // 2 Slots exhausted, remove the first container to make space
+ containers(0).expectMsg(Remove)
+ }
+
it should "remove a container to make space in the pool if it is already full and another action with different invocation namespace arrives" in within(timeout) {
val (containers, factory) = testContainers(2)
val feed = TestProbe()
// a pool with only 1 slot
- val pool = system.actorOf(ContainerPool.props(factory, 1, feed.ref))
+ val pool = system.actorOf(ContainerPool.props(factory, 1, 1, feed.ref))
pool ! runMessage
containers(0).expectMsg(runMessage)
containers(0).send(pool, NeedWork(warmedData()))
@@ -185,7 +213,7 @@ class ContainerPoolTests extends TestKit(ActorSystem("ContainerPool"))
val feed = TestProbe()
// a pool with only 1 slot
- val pool = system.actorOf(ContainerPool.props(factory, 1, feed.ref))
+ val pool = system.actorOf(ContainerPool.props(factory, 1, 1, feed.ref))
pool ! runMessage
containers(0).expectMsg(runMessage)
containers(0).send(pool, NeedWork(warmedData()))
@@ -205,7 +233,7 @@ class ContainerPoolTests extends TestKit(ActorSystem("ContainerPool"))
val (containers, factory) = testContainers(1)
val feed = TestProbe()
- val pool = system.actorOf(ContainerPool.props(factory, 0, feed.ref, Some(PrewarmingConfig(1, exec, memoryLimit))))
+ val pool = system.actorOf(ContainerPool.props(factory, 0, 0, feed.ref, Some(PrewarmingConfig(1, exec, memoryLimit))))
containers(0).expectMsg(Start(exec, memoryLimit))
}
@@ -213,7 +241,7 @@ class ContainerPoolTests extends TestKit(ActorSystem("ContainerPool"))
val (containers, factory) = testContainers(2)
val feed = TestProbe()
- val pool = system.actorOf(ContainerPool.props(factory, 1, feed.ref, Some(PrewarmingConfig(1, exec, memoryLimit))))
+ val pool = system.actorOf(ContainerPool.props(factory, 1, 1, feed.ref, Some(PrewarmingConfig(1, exec, memoryLimit))))
containers(0).expectMsg(Start(exec, memoryLimit))
containers(0).send(pool, NeedWork(preWarmedData(exec.kind)))
pool ! runMessage
@@ -226,7 +254,7 @@ class ContainerPoolTests extends TestKit(ActorSystem("ContainerPool"))
val alternativeExec = CodeExecAsString(RuntimeManifest("anotherKind", ImageName("testImage")), "testCode", None)
- val pool = system.actorOf(ContainerPool.props(factory, 1, feed.ref, Some(PrewarmingConfig(1, alternativeExec, memoryLimit))))
+ val pool = system.actorOf(ContainerPool.props(factory, 1, 1, feed.ref, Some(PrewarmingConfig(1, alternativeExec, memoryLimit))))
containers(0).expectMsg(Start(alternativeExec, memoryLimit)) // container0 was prewarmed
containers(0).send(pool, NeedWork(preWarmedData(alternativeExec.kind)))
pool ! runMessage
@@ -239,7 +267,7 @@ class ContainerPoolTests extends TestKit(ActorSystem("ContainerPool"))
val alternativeLimit = 128.MB
- val pool = system.actorOf(ContainerPool.props(factory, 1, feed.ref, Some(PrewarmingConfig(1, exec, alternativeLimit))))
+ val pool = system.actorOf(ContainerPool.props(factory, 1, 1, feed.ref, Some(PrewarmingConfig(1, exec, alternativeLimit))))
containers(0).expectMsg(Start(exec, alternativeLimit)) // container0 was prewarmed
containers(0).send(pool, NeedWork(preWarmedData(exec.kind, alternativeLimit)))
pool ! runMessage
@@ -253,7 +281,7 @@ class ContainerPoolTests extends TestKit(ActorSystem("ContainerPool"))
val (containers, factory) = testContainers(2)
val feed = TestProbe()
- val pool = system.actorOf(ContainerPool.props(factory, 2, feed.ref))
+ val pool = system.actorOf(ContainerPool.props(factory, 2, 2, feed.ref))
// container0 is created and used
pool ! runMessage
@@ -272,7 +300,6 @@ class ContainerPoolTests extends TestKit(ActorSystem("ContainerPool"))
pool ! runMessage
containers(1).expectMsg(runMessage)
}
-
}
/**
@@ -302,12 +329,6 @@ class ContainerPoolObjectTests extends FlatSpec with Matchers with MockFactory {
/** Helper to create NoData */
def noData() = NoData()
- /** Helper to create a free Worker, for shorter notation */
- def freeWorker(data: ContainerData) = WorkerData(data, Free)
-
- /** Helper to create a busy Worker, for shorter notation */
- def busyWorker(data: ContainerData) = WorkerData(data, Busy)
-
behavior of "ContainerPool schedule()"
it should "not provide a container if idle pool is empty" in {
@@ -316,7 +337,7 @@ class ContainerPoolObjectTests extends FlatSpec with Matchers with MockFactory {
it should "reuse an applicable warm container from idle pool with one container" in {
val data = warmedData()
- val pool = Map('name -> freeWorker(data))
+ val pool = Map('name -> data)
// copy to make sure, referencial equality doesn't suffice
ContainerPool.schedule(data.action.copy(), data.invocationNamespace, pool) shouldBe Some('name)
@@ -325,8 +346,8 @@ class ContainerPoolObjectTests extends FlatSpec with Matchers with MockFactory {
it should "reuse an applicable warm container from idle pool with several applicable containers" in {
val data = warmedData()
val pool = Map(
- 'first -> freeWorker(data),
- 'second -> freeWorker(data))
+ 'first -> data,
+ 'second -> data)
ContainerPool.schedule(data.action.copy(), data.invocationNamespace, pool) should contain oneOf ('first, 'second)
}
@@ -334,9 +355,9 @@ class ContainerPoolObjectTests extends FlatSpec with Matchers with MockFactory {
it should "reuse an applicable warm container from idle pool with several different containers" in {
val matchingData = warmedData()
val pool = Map(
- 'none -> freeWorker(noData()),
- 'pre -> freeWorker(preWarmedData()),
- 'warm -> freeWorker(matchingData))
+ 'none -> noData(),
+ 'pre -> preWarmedData(),
+ 'warm -> matchingData)
ContainerPool.schedule(matchingData.action.copy(), matchingData.invocationNamespace, pool) shouldBe Some('warm)
}
@@ -345,15 +366,15 @@ class ContainerPoolObjectTests extends FlatSpec with Matchers with MockFactory {
val data = warmedData()
// data is **not** in the pool!
val pool = Map(
- 'none -> freeWorker(noData()),
- 'pre -> freeWorker(preWarmedData()))
+ 'none -> noData(),
+ 'pre -> preWarmedData())
ContainerPool.schedule(data.action.copy(), data.invocationNamespace, pool) shouldBe None
}
it should "not reuse a warm container with different invocation namespace" in {
val data = warmedData()
- val pool = Map('warm -> freeWorker(data))
+ val pool = Map('warm -> data)
val differentNamespace = EntityName(data.invocationNamespace.asString + "butDifferent")
data.invocationNamespace should not be differentNamespace
@@ -363,8 +384,7 @@ class ContainerPoolObjectTests extends FlatSpec with Matchers with MockFactory {
it should "not reuse a warm container with different action name" in {
val data = warmedData()
val differentAction = data.action.copy(name = EntityName(data.action.name.asString + "butDifferent"))
- val pool = Map(
- 'warm -> freeWorker(data))
+ val pool = Map('warm -> data)
data.action.name should not be differentAction.name
ContainerPool.schedule(differentAction, data.invocationNamespace, pool) shouldBe None
@@ -373,8 +393,7 @@ class ContainerPoolObjectTests extends FlatSpec with Matchers with MockFactory {
it should "not reuse a warm container with different action version" in {
val data = warmedData()
val differentAction = data.action.copy(version = data.action.version.upMajor)
- val pool = Map(
- 'warm -> freeWorker(data))
+ val pool = Map('warm -> data)
data.action.version should not be differentAction.version
ContainerPool.schedule(differentAction, data.invocationNamespace, pool) shouldBe None
@@ -388,22 +407,14 @@ class ContainerPoolObjectTests extends FlatSpec with Matchers with MockFactory {
it should "not provide a container from busy pool with non-warm containers" in {
val pool = Map(
- 'none -> freeWorker(noData()),
- 'pre -> freeWorker(preWarmedData()))
- ContainerPool.remove(createAction(), standardNamespace, pool) shouldBe None
- }
-
- it should "not provide a container from busy pool with warm Busy containers" in {
- val pool = Map(
- 'none -> freeWorker(noData()),
- 'pre -> freeWorker(preWarmedData()),
- 'busy -> busyWorker(warmedData()))
+ 'none -> noData(),
+ 'pre -> preWarmedData())
ContainerPool.remove(createAction(), standardNamespace, pool) shouldBe None
}
it should "not provide a container from pool with one single free container with the same action and namespace" in {
val data = warmedData()
- val pool = Map('warm -> freeWorker(data))
+ val pool = Map('warm -> data)
// same data --> no removal
ContainerPool.remove(data.action, data.invocationNamespace, pool) shouldBe None
@@ -427,26 +438,10 @@ class ContainerPoolObjectTests extends FlatSpec with Matchers with MockFactory {
val oldest = warmedData(namespace = commonNamespace, lastUsed = Instant.ofEpochMilli(0))
val pool = Map(
- 'first -> freeWorker(first),
- 'second -> freeWorker(second),
- 'oldest -> freeWorker(oldest))
+ 'first -> first,
+ 'second -> second,
+ 'oldest -> oldest)
ContainerPool.remove(createAction(), standardNamespace, pool) shouldBe Some('oldest)
}
-
- it should "provide oldest container of largest namespace group from busy pool with multiple containers" in {
- val smallNamespace = "smallNamespace"
- val mediumNamespace = "mediumNamespace"
- val largeNamespace = "largeNamespace"
-
- // Note: We choose the oldest from the **largest** pool, although all other containers are even older.
- val myData = warmedData(namespace = smallNamespace, lastUsed = Instant.ofEpochMilli(0))
- val pool = Map(
- 'my -> freeWorker(myData),
- 'other -> freeWorker(warmedData(namespace = mediumNamespace, lastUsed = Instant.ofEpochMilli(1))),
- 'largeYoung -> freeWorker(warmedData(namespace = largeNamespace, lastUsed = Instant.ofEpochMilli(3))),
- 'largeOld -> freeWorker(warmedData(namespace = largeNamespace, lastUsed = Instant.ofEpochMilli(2))))
-
- ContainerPool.remove(createAction(), standardNamespace, pool) shouldBe Some('largeOld)
- }
}
--
To stop receiving notification emails like this one, please contact
"commits@openwhisk.apache.org" <co...@openwhisk.apache.org>.