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:02 UTC
[incubator-openwhisk] branch master updated (e9d5c50 -> c9768ea)
This is an automated email from the ASF dual-hosted git repository.
cbickel pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-openwhisk.git.
from e9d5c50 Fixed typo (#2407)
new f79ae51 Keep inactive containers around even in a fully loaded system.
new 85e8c8f Retype scheduling code to get rid of unsafeness.
new c9768ea Adjust pause grace to a less dangerous value
The 3 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../whisk/core/containerpool/ContainerPool.scala | 146 +++++++++++----------
.../whisk/core/containerpool/ContainerProxy.scala | 17 ++-
.../scala/whisk/core/invoker/InvokerReactive.scala | 1 +
.../containerpool/test/ContainerPoolTests.scala | 123 +++++++++--------
.../containerpool/test/ContainerProxyTests.scala | 24 ++--
5 files changed, 157 insertions(+), 154 deletions(-)
--
To stop receiving notification emails like this one, please contact
['"commits@openwhisk.apache.org" <co...@openwhisk.apache.org>'].
[incubator-openwhisk] 02/03: Retype scheduling code to get rid of
unsafeness.
Posted by cb...@apache.org.
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 85e8c8f1d49dac3c04fe0f58480d652f52729c54
Author: Markus Thoemmes <ma...@de.ibm.com>
AuthorDate: Tue Jun 6 14:38:18 2017 +0200
Retype scheduling code to get rid of unsafeness.
The pool used to look up the data of a container it has just chosen/created which is unnecessary. Removed also a non-reachable error-condition.
Also: Take out unnecessary prewarmConfig checking.
---
.../whisk/core/containerpool/ContainerPool.scala | 64 +++++++++++-----------
.../containerpool/test/ContainerPoolTests.scala | 6 +-
2 files changed, 34 insertions(+), 36 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 012373f..9a796ee 100644
--- a/core/invoker/src/main/scala/whisk/core/containerpool/ContainerPool.scala
+++ b/core/invoker/src/main/scala/whisk/core/containerpool/ContainerPool.scala
@@ -100,17 +100,12 @@ class ContainerPool(
} else None
container match {
- case Some(actor) =>
- 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 => self ! r
+ case Some((actor, data)) =>
+ busyPool.update(actor, data)
+ freePool.remove(actor)
+ actor ! r // forwards the run request to the container
+ case None =>
+ self ! r
}
// Container is free to take more work
@@ -133,15 +128,17 @@ class ContainerPool(
}
/** Creates a new container and updates state accordingly. */
- def createContainer() = {
+ def createContainer(): (ActorRef, ContainerData) = {
val ref = childFactory(context)
- freePool.update(ref, NoData())
- ref
+ val data = NoData()
+ freePool.update(ref, data)
+
+ (ref, data)
}
/** Creates a new prewarmed container */
def prewarmContainer(exec: CodeExec[_], memoryLimit: ByteSize) =
- prewarmConfig.foreach(config => childFactory(context) ! Start(exec, memoryLimit))
+ childFactory(context) ! Start(exec, memoryLimit)
/**
* Takes a prewarm container out of the prewarmed pool
@@ -150,23 +147,24 @@ class ContainerPool(
* @param kind the kind you want to invoke
* @return the container iff found
*/
- def takePrewarmContainer(action: ExecutableWhiskAction) = prewarmConfig.flatMap { config =>
- val kind = action.exec.kind
- val memory = action.limits.memory.megabytes.MB
- prewarmedPool.find {
- case (_, PreWarmedData(_, `kind`, `memory`)) => true
- case _ => false
- }.map {
- case (ref, data) =>
- // Move the container to the usual pool
- freePool.update(ref, data)
- prewarmedPool.remove(ref)
- // Create a new prewarm container
- prewarmContainer(config.exec, config.memoryLimit)
-
- ref
+ def takePrewarmContainer(action: ExecutableWhiskAction): Option[(ActorRef, ContainerData)] =
+ prewarmConfig.flatMap { config =>
+ val kind = action.exec.kind
+ val memory = action.limits.memory.megabytes.MB
+ prewarmedPool.find {
+ case (_, PreWarmedData(_, `kind`, `memory`)) => true
+ case _ => false
+ }.map {
+ case (ref, data) =>
+ // Move the container to the usual pool
+ freePool.update(ref, data)
+ prewarmedPool.remove(ref)
+ // Create a new prewarm container
+ prewarmContainer(config.exec, config.memoryLimit)
+
+ (ref, data)
+ }
}
- }
/** Removes a container and updates state accordingly. */
def removeContainer(toDelete: ActorRef) = {
@@ -192,11 +190,11 @@ 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, ContainerData]): Option[A] = {
+ def schedule[A](action: ExecutableWhiskAction, invocationNamespace: EntityName, idles: Map[A, ContainerData]): Option[(A, ContainerData)] = {
idles.find {
case (_, WarmedData(_, `invocationNamespace`, `action`, _)) => true
case _ => false
- }.map(_._1)
+ }
}
/**
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 7a3454b..e262525 100644
--- a/tests/src/test/scala/whisk/core/containerpool/test/ContainerPoolTests.scala
+++ b/tests/src/test/scala/whisk/core/containerpool/test/ContainerPoolTests.scala
@@ -340,7 +340,7 @@ class ContainerPoolObjectTests extends FlatSpec with Matchers with MockFactory {
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)
+ ContainerPool.schedule(data.action.copy(), data.invocationNamespace, pool) shouldBe Some('name, data)
}
it should "reuse an applicable warm container from idle pool with several applicable containers" in {
@@ -349,7 +349,7 @@ class ContainerPoolObjectTests extends FlatSpec with Matchers with MockFactory {
'first -> data,
'second -> data)
- ContainerPool.schedule(data.action.copy(), data.invocationNamespace, pool) should contain oneOf ('first, 'second)
+ ContainerPool.schedule(data.action.copy(), data.invocationNamespace, pool) should (be(Some('first, data)) or be(Some('second, data)))
}
it should "reuse an applicable warm container from idle pool with several different containers" in {
@@ -359,7 +359,7 @@ class ContainerPoolObjectTests extends FlatSpec with Matchers with MockFactory {
'pre -> preWarmedData(),
'warm -> matchingData)
- ContainerPool.schedule(matchingData.action.copy(), matchingData.invocationNamespace, pool) shouldBe Some('warm)
+ ContainerPool.schedule(matchingData.action.copy(), matchingData.invocationNamespace, pool) shouldBe Some('warm, matchingData)
}
it should "not reuse a container from idle pool with non-warm containers" in {
--
To stop receiving notification emails like this one, please contact
"commits@openwhisk.apache.org" <co...@openwhisk.apache.org>.
[incubator-openwhisk] 03/03: Adjust pause grace to a less dangerous
value
Posted by cb...@apache.org.
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 c9768ea96e9561ee42f4a2b7795afa46cc6b6e0a
Author: Markus Thoemmes <ma...@de.ibm.com>
AuthorDate: Thu Jun 22 08:32:10 2017 +0200
Adjust pause grace to a less dangerous value
---
.../whisk/core/containerpool/ContainerProxy.scala | 17 ++++++++-------
.../containerpool/test/ContainerProxyTests.scala | 24 +++++++++++-----------
2 files changed, 20 insertions(+), 21 deletions(-)
diff --git a/core/invoker/src/main/scala/whisk/core/containerpool/ContainerProxy.scala b/core/invoker/src/main/scala/whisk/core/containerpool/ContainerProxy.scala
index 796624b..55c7999 100644
--- a/core/invoker/src/main/scala/whisk/core/containerpool/ContainerProxy.scala
+++ b/core/invoker/src/main/scala/whisk/core/containerpool/ContainerProxy.scala
@@ -91,21 +91,18 @@ case object ActivationCompleted
* @param factory a function generating a Container
* @param sendActiveAck a function sending the activation via active ack
* @param storeActivation a function storing the activation in a persistent store
+ * @param unusedTimeout time after which the container is automatically thrown away
+ * @param pauseGrace time to wait for new work before pausing the container
*/
class ContainerProxy(
factory: (TransactionId, String, ImageName, Boolean, ByteSize) => Future[Container],
sendActiveAck: (TransactionId, WhiskActivation) => Future[Any],
- storeActivation: (TransactionId, WhiskActivation) => Future[Any]) extends FSM[ContainerState, ContainerData] with Stash {
+ storeActivation: (TransactionId, WhiskActivation) => Future[Any],
+ unusedTimeout: FiniteDuration,
+ pauseGrace: FiniteDuration) extends FSM[ContainerState, ContainerData] with Stash {
implicit val ec = context.system.dispatcher
val logging = new AkkaLogging(context.system.log)
- // The container is destroyed after this period of time
- val unusedTimeout = 10.minutes
-
- // The container is not paused for this period of time
- // after an activation has finished successfully
- val pauseGrace = 1.second
-
startWith(Uninitialized, NoData())
when(Uninitialized) {
@@ -384,7 +381,9 @@ class ContainerProxy(
object ContainerProxy {
def props(factory: (TransactionId, String, ImageName, Boolean, ByteSize) => Future[Container],
ack: (TransactionId, WhiskActivation) => Future[Any],
- store: (TransactionId, WhiskActivation) => Future[Any]) = Props(new ContainerProxy(factory, ack, store))
+ store: (TransactionId, WhiskActivation) => Future[Any],
+ unusedTimeout: FiniteDuration = 10.minutes,
+ pauseGrace: FiniteDuration = 50.milliseconds) = Props(new ContainerProxy(factory, ack, store, unusedTimeout, pauseGrace))
// Needs to be thread-safe as it's used by multiple proxies concurrently.
private val containerCount = new Counter
diff --git a/tests/src/test/scala/whisk/core/containerpool/test/ContainerProxyTests.scala b/tests/src/test/scala/whisk/core/containerpool/test/ContainerProxyTests.scala
index a49c766..f7a7e12 100644
--- a/tests/src/test/scala/whisk/core/containerpool/test/ContainerProxyTests.scala
+++ b/tests/src/test/scala/whisk/core/containerpool/test/ContainerProxyTests.scala
@@ -148,7 +148,7 @@ class ContainerProxyTests extends TestKit(ActorSystem("ContainerProxys"))
val container = new TestContainer
val factory = createFactory(Future.successful(container))
- val machine = childActorOf(ContainerProxy.props(factory, createAcker, store))
+ val machine = childActorOf(ContainerProxy.props(factory, createAcker, store, pauseGrace = timeout))
registerCallback(machine)
preWarm(machine)
@@ -164,7 +164,7 @@ class ContainerProxyTests extends TestKit(ActorSystem("ContainerProxys"))
val factory = createFactory(Future.successful(container))
val acker = createAcker
- val machine = childActorOf(ContainerProxy.props(factory, acker, store))
+ val machine = childActorOf(ContainerProxy.props(factory, acker, store, pauseGrace = timeout))
registerCallback(machine)
preWarm(machine)
@@ -196,7 +196,7 @@ class ContainerProxyTests extends TestKit(ActorSystem("ContainerProxys"))
val factory = createFactory(Future.successful(container))
val acker = createAcker
- val machine = childActorOf(ContainerProxy.props(factory, acker, store))
+ val machine = childActorOf(ContainerProxy.props(factory, acker, store, pauseGrace = timeout))
registerCallback(machine)
preWarm(machine)
@@ -220,7 +220,7 @@ class ContainerProxyTests extends TestKit(ActorSystem("ContainerProxys"))
val factory = createFactory(Future.successful(container))
val acker = createAcker
- val machine = childActorOf(ContainerProxy.props(factory, acker, store))
+ val machine = childActorOf(ContainerProxy.props(factory, acker, store, pauseGrace = timeout))
registerCallback(machine)
preWarm(machine)
@@ -246,7 +246,7 @@ class ContainerProxyTests extends TestKit(ActorSystem("ContainerProxys"))
val factory = createFactory(Future.successful(container))
val acker = createAcker
- val machine = childActorOf(ContainerProxy.props(factory, acker, store))
+ val machine = childActorOf(ContainerProxy.props(factory, acker, store, pauseGrace = timeout))
registerCallback(machine)
run(machine, Uninitialized)
@@ -268,7 +268,7 @@ class ContainerProxyTests extends TestKit(ActorSystem("ContainerProxys"))
val factory = createFactory(Future.failed(new Exception()))
val acker = createAcker
- val machine = childActorOf(ContainerProxy.props(factory, acker, store))
+ val machine = childActorOf(ContainerProxy.props(factory, acker, store, pauseGrace = timeout))
registerCallback(machine)
machine ! Run(action, message)
expectMsg(Transition(machine, Uninitialized, Running))
@@ -297,7 +297,7 @@ class ContainerProxyTests extends TestKit(ActorSystem("ContainerProxys"))
val factory = createFactory(Future.successful(container))
val acker = createAcker
- val machine = childActorOf(ContainerProxy.props(factory, acker, store))
+ val machine = childActorOf(ContainerProxy.props(factory, acker, store, pauseGrace = timeout))
registerCallback(machine)
machine ! Run(action, message)
expectMsg(Transition(machine, Uninitialized, Running))
@@ -326,7 +326,7 @@ class ContainerProxyTests extends TestKit(ActorSystem("ContainerProxys"))
val factory = createFactory(Future.successful(container))
val acker = createAcker
- val machine = childActorOf(ContainerProxy.props(factory, acker, store))
+ val machine = childActorOf(ContainerProxy.props(factory, acker, store, pauseGrace = timeout))
registerCallback(machine)
machine ! Run(action, message)
expectMsg(Transition(machine, Uninitialized, Running))
@@ -355,7 +355,7 @@ class ContainerProxyTests extends TestKit(ActorSystem("ContainerProxys"))
val factory = createFactory(Future.successful(container))
val acker = createAcker
- val machine = childActorOf(ContainerProxy.props(factory, acker, store))
+ val machine = childActorOf(ContainerProxy.props(factory, acker, store, pauseGrace = timeout))
registerCallback(machine)
run(machine, Uninitialized) // first run an activation
timeout(machine) // times out Ready state so container suspends
@@ -387,7 +387,7 @@ class ContainerProxyTests extends TestKit(ActorSystem("ContainerProxys"))
val factory = createFactory(Future.successful(container))
val acker = createAcker
- val machine = childActorOf(ContainerProxy.props(factory, acker, store))
+ val machine = childActorOf(ContainerProxy.props(factory, acker, store, pauseGrace = timeout))
registerCallback(machine)
run(machine, Uninitialized)
timeout(machine) // times out Ready state so container suspends
@@ -418,7 +418,7 @@ class ContainerProxyTests extends TestKit(ActorSystem("ContainerProxys"))
val factory = createFactory(Future.successful(container))
val acker = createAcker
- val machine = childActorOf(ContainerProxy.props(factory, acker, store))
+ val machine = childActorOf(ContainerProxy.props(factory, acker, store, pauseGrace = timeout))
registerCallback(machine)
// Start running the action
@@ -466,7 +466,7 @@ class ContainerProxyTests extends TestKit(ActorSystem("ContainerProxys"))
val factory = createFactory(Future.successful(container))
val acker = createAcker
- val machine = childActorOf(ContainerProxy.props(factory, acker, store))
+ val machine = childActorOf(ContainerProxy.props(factory, acker, store, pauseGrace = timeout))
registerCallback(machine)
run(machine, Uninitialized)
timeout(machine)
--
To stop receiving notification emails like this one, please contact
"commits@openwhisk.apache.org" <co...@openwhisk.apache.org>.
[incubator-openwhisk] 01/03: Keep inactive containers around even
in a fully loaded system.
Posted by cb...@apache.org.
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>.