Review

ansible/group_vars/all

@@ -173,7 +173,7 @@ invoker:
  port: 12001
  heap: "{{ invoker_heap | default('2g') }}"
  arguments: "{{ invoker_arguments | default('') }}"
- userMemory: "{{ invoker_user_memory | default('4096 m') }}"
+ userMemory: "{{ invoker_user_memory | default('1024 m') }}"
  instances: "{{ groups['invokers'] | length }}"
  # Specify if it is allowed to deploy more than 1 invoker on a single machine.
  allowMultipleInstances: "{{ invoker_allow_multiple_instances | default(false) }}"

core/controller/src/main/resources/reference.conf

@@ -6,6 +6,7 @@ whisk {
  use-cluster-bootstrap: false
  }
  loadbalancer {
+ user-memory: 1024 m
  blackbox-fraction: 10%
  # factor to increase the timeout for forced active acks
  # timeout = time-limit.std * timeoutfactor + 1m

core/controller/src/main/scala/whisk/core/loadBalancer/ShardingContainerPoolBalancer.scala

@@ -594,6 +594,12 @@ case class ShardingContainerPoolBalancerState(
  val actualSize = newSize max 1 // if a cluster size < 1 is reported, falls back to a size of 1 (alone)
  if (_clusterSize != actualSize) {
  _clusterSize = actualSize
+ if (totalInvokerThreshold / actualSize < MemoryLimit.minMemory) {
+ logging.warn(
+ this,
+ s"registered controllers: ${_clusterSize}: the slots per invoker fall below the min memory of one action.")(
+ TransactionId.loadbalancer)
+ }
  val newTreshold = (totalInvokerThreshold / actualSize) max MemoryLimit.minMemory // letting this fall below minMemory doesn't make sense
  currentInvokerThreshold = newTreshold
  _invokerSlots = _invokerSlots.map(_ => new ForcibleSemaphore(currentInvokerThreshold.toMB.toInt))

core/invoker/src/main/scala/whisk/core/containerpool/ContainerPool.scala

@@ -95,58 +95,54 @@ class ContainerPool(childFactory: ActorRefFactory => ActorRef,
  // their requests and send them back to the pool for rescheduling (this may happen if "docker" operations
  // fail for example, or a container has aged and was destroying itself when a new request was assigned)
  case r: Run =>
+ val isFirstMessageOnBuffer = runBuffer.dequeueOption.exists(_._1.msg == r.msg)
+ val isResentFromBuffer = runBuffer.nonEmpty && isFirstMessageOnBuffer
+
  // Only process request, if there are no other requests waiting for free slots, or if the current request is the next request to process
- if (runBuffer.isEmpty || runBuffer.dequeueOption.exists(_._1.msg == r.msg)) {
+ if (runBuffer.isEmpty || isFirstMessageOnBuffer) {
  val createdContainer =
- if (busyPool
- .map(_._2.memoryLimit.toMB)
- .sum + r.action.limits.memory.megabytes <= poolConfig.userMemory.toMB) {
+ // Is there enough space on the invoker for this action to be executed.
+ if (hasPoolSpaceFor(busyPool, r.action.limits.memory.megabytes.MB)) {
  // Schedule a job to a warm container
  ContainerPool
  .schedule(r.action, r.msg.user.namespace.name, freePool)
- .map(container => {
- (container, "warm")
- })
- .orElse {
- if (busyPool
- .map(_._2.memoryLimit.toMB)
- .sum + freePool.map(_._2.memoryLimit.toMB).sum < poolConfig.userMemory.toMB) {
+ .map(container => (container, "warm"))
+ .orElse(
+ // There was no warm container. Try to take a prewarm container or a cold container.
+
+ // Is there enough space to create a new container or do other containers have to be removed?
+ if (hasPoolSpaceFor(busyPool ++ freePool, r.action.limits.memory.megabytes.MB)) {
  takePrewarmContainer(r.action)
- .map(container => {
- (container, "prewarmed")
- })
- .orElse {
- Some(createContainer(r.action.limits.memory.megabytes.MB), "cold")
- }
- } else None
- }
- .orElse {
+ .map(container => (container, "prewarmed"))
+ .orElse(Some(createContainer(r.action.limits.memory.megabytes.MB), "cold"))
+ } else None)
+ .orElse(
  // Remove a container and create a new one for the given job
  ContainerPool
- .remove(freePool, r.action.limits.memory.megabytes.MB)
+ // Only free up the amount, that is really needed to free up
+ .remove(
+ freePool,
+ Math.min(r.action.limits.memory.megabytes, freePool.map(_._2.memoryLimit.toMB).sum).MB)
  .map(removeContainer)
  // If the list had at least one entry, enough containers were removed to start the new container. After
  // removing the containers, we are not interested anymore in the containers that have been removed.
  .headOption
- .map { _ =>
+ .map(_ =>
  takePrewarmContainer(r.action)
- .map(container => {
- (container, "recreated")
- })
- .getOrElse {
- (createContainer(r.action.limits.memory.megabytes.MB), "recreated")
- }
- }
- }
+ .map(container => (container, "recreated prewarm"))
+ .getOrElse(createContainer(r.action.limits.memory.megabytes.MB), "recreated")))
  } else None
 
  createdContainer match {
  case Some(((actor, data), containerState)) =>
  busyPool = busyPool + (actor -> data)
  freePool = freePool - actor
  // Remove the action that get's executed now from the buffer and execute the next one afterwards.
- runBuffer = runBuffer.dequeueOption.map(_._2).getOrElse(runBuffer)
- runBuffer.dequeueOption.foreach { case (run, _) => self ! run }
+ if (isResentFromBuffer) {
+ // It is guaranteed that the currently executed messages is the head of the queue, if the message comes from the buffer
+ runBuffer = runBuffer.dequeue._2
+ runBuffer.dequeueOption.foreach { case (run, _) => self ! run }
+ }
  actor ! r // forwards the run request to the container
  logContainerStart(r, containerState)
  case None =>
@@ -167,7 +163,7 @@ class ContainerPool(childFactory: ActorRefFactory => ActorRef,
  } else {
  r.retryLogDeadline
  }
- if (!runBuffer.exists(_.msg == r.msg)) {
+ if (!isResentFromBuffer) {
  // Add this request to the buffer, as it is not there yet.
  runBuffer = runBuffer.enqueue(r)
  }
@@ -256,6 +252,10 @@ class ContainerPool(childFactory: ActorRefFactory => ActorRef,
  freePool = freePool - toDelete
  busyPool = busyPool - toDelete
  }
+
+ def hasPoolSpaceFor[A](pool: Map[A, ContainerData], memory: ByteSize): Boolean = {
+ pool.map(_._2.memoryLimit.toMB).sum + memory.toMB <= poolConfig.userMemory.toMB
+ }
 }
 
 object ContainerPool {
@@ -299,18 +299,20 @@ object ContainerPool {
  case (ref, w: WarmedData) => ref -> w
  }
 
- if (freeContainers.nonEmpty && freeContainers.map(_._2.memoryLimit.toMB).sum >= memory.toMB) {
- if (memory > 0.B) {
- val (ref, data) = freeContainers.minBy(_._2.lastUsed)
- // Catch exception if remaining memory will be negative
- val remainingMemory = Try(memory - data.memoryLimit).getOrElse(0.B)
- List(ref) ++ remove(freeContainers.filterKeys(_ != ref), remainingMemory)
- } else {
- // Enough containers are found to get the memory, that is necessary. -> Abort recursion
- List.empty
- }
- // else case: All containers are in use currently, or there is more memory needed than containers can be removed.
- } else List.empty
+ if (memory > 0.B && freeContainers.nonEmpty && freeContainers.map(_._2.memoryLimit.toMB).sum >= memory.toMB) {
+ // Remove the oldest container if:
+ // - there is more memory required
+ // - there are still containers that can be removed
+ // - there are enough free containers that can be removed
+ val (ref, data) = freeContainers.minBy(_._2.lastUsed)
+ // Catch exception if remaining memory will be negative
+ val remainingMemory = Try(memory - data.memoryLimit).getOrElse(0.B)
+ List(ref) ++ remove(freeContainers.filterKeys(_ != ref), remainingMemory)
+ } else {
+ // If this is the first call: All containers are in use currently, or there is more memory needed than containers can be removed.
+ // Or, if this is one of the recurstions: Enough containers are found to get the memory, that is necessary. -> Abort recursion
+ List.empty
+ }
  }
 
  def props(factory: ActorRefFactory => ActorRef,

tests/src/test/scala/whisk/core/containerpool/test/ContainerPoolTests.scala

@@ -131,6 +131,7 @@ class ContainerPoolTests
  it should "reuse a warm container" in within(timeout) {
  val (containers, factory) = testContainers(2)
  val feed = TestProbe()
+ // Actions are created with default memory limit (MemoryLimit.stdMemory). This means 4 actions can be scheduled.
  val pool = system.actorOf(ContainerPool.props(factory, poolConfig(MemoryLimit.stdMemory * 4), feed.ref))
 
  pool ! runMessage
@@ -145,6 +146,7 @@ class ContainerPoolTests
  it should "reuse a warm container when action is the same even if revision changes" in within(timeout) {
  val (containers, factory) = testContainers(2)
  val feed = TestProbe()
+ // Actions are created with default memory limit (MemoryLimit.stdMemory). This means 4 actions can be scheduled.
  val pool = system.actorOf(ContainerPool.props(factory, poolConfig(MemoryLimit.stdMemory * 4), feed.ref))
 
  pool ! runMessage
@@ -160,6 +162,7 @@ class ContainerPoolTests
  val (containers, factory) = testContainers(2)
  val feed = TestProbe()
 
+ // Actions are created with default memory limit (MemoryLimit.stdMemory). This means 4 actions can be scheduled.
  val pool = system.actorOf(ContainerPool.props(factory, poolConfig(MemoryLimit.stdMemory * 4), feed.ref))
  pool ! runMessage
  containers(0).expectMsg(runMessage)
@@ -189,19 +192,19 @@ class ContainerPoolTests
  val (containers, factory) = testContainers(3)
  val feed = TestProbe()
 
- // a pool with slots for 512MB
+ // a pool with slots for 2 actions with default memory limit.
  val pool = system.actorOf(ContainerPool.props(factory, poolConfig(512.MB), feed.ref))
  pool ! runMessage
  containers(0).expectMsg(runMessage)
- pool ! runMessageDifferentAction
+ pool ! runMessageDifferentAction // 2 * stdMemory taken -> full
  containers(1).expectMsg(runMessageDifferentAction)
 
- containers(0).send(pool, NeedWork(warmedData()))
+ containers(0).send(pool, NeedWork(warmedData())) // first action finished -> 1 * stdMemory taken
  feed.expectMsg(MessageFeed.Processed)
- containers(1).send(pool, NeedWork(warmedData()))
+ containers(1).send(pool, NeedWork(warmedData())) // second action finished -> 1 * stdMemory taken
  feed.expectMsg(MessageFeed.Processed)
 
- pool ! runMessageLarge
+ pool ! runMessageLarge // need to remove both action to make space for the large action (needs 2 * stdMemory)
  containers(0).expectMsg(Remove)
  containers(1).expectMsg(Remove)
  containers(2).expectMsg(runMessageLarge)
@@ -269,15 +272,15 @@ class ContainerPoolTests
  val pool = system.actorOf(ContainerPool.props(factory, poolConfig(MemoryLimit.stdMemory * 2), feed.ref))
 
  // Start first action
- pool ! runMessage
+ pool ! runMessage // 1 * stdMemory taken
  containers(0).expectMsg(runMessage)
 
  // Send second action to the pool
- pool ! runMessageLarge
+ pool ! runMessageLarge // message is too large to be processed immediately.
  containers(1).expectNoMessage(100.milliseconds)
 
  // First action is finished
- containers(0).send(pool, NeedWork(warmedData()))
+ containers(0).send(pool, NeedWork(warmedData())) // pool is empty again.
  feed.expectMsg(MessageFeed.Processed)
 
  // Second action should run now
@@ -388,7 +391,8 @@ class ContainerPoolTests
  val feed = TestProbe()
 
  // Pool with 512 MB usermemory
- val pool = system.actorOf(ContainerPool.props(factory, ContainerPoolConfig(MemoryLimit.stdMemory * 2), feed.ref))
+ val pool =
+ system.actorOf(ContainerPool.props(factory, poolConfig(MemoryLimit.stdMemory * 2), feed.ref))
 
  // Send action that blocks the pool
  pool ! runMessageLarge
@@ -420,7 +424,7 @@ class ContainerPoolTests
  val feed = TestProbe()
 
  // Pool with 512 MB usermemory
- val pool = system.actorOf(ContainerPool.props(factory, ContainerPoolConfig(MemoryLimit.stdMemory * 2), feed.ref))
+ val pool = system.actorOf(ContainerPool.props(factory, poolConfig(MemoryLimit.stdMemory * 2), feed.ref))
 
  // Send 4 actions to the ContainerPool (Action 0, Action 2 and Action 3 with each 265 MB and Action 1 with 512 MB)
  pool ! runMessage
@@ -576,6 +580,12 @@ class ContainerPoolObjectTests extends FlatSpec with Matchers with MockFactory {
  ContainerPool.remove(pool, MemoryLimit.stdMemory) shouldBe List.empty
  }
 
+ it should "not provide a container from pool if there is not enough capacity" in {
+ val pool = Map('first -> warmedData())
+
+ ContainerPool.remove(pool, MemoryLimit.stdMemory * 2) shouldBe List.empty
+ }
+
  it should "provide a container from pool with one single free container" in {
  val data = warmedData()
  val pool = Map('warm -> data)