ningyougang commented on a change in pull request #5102: URL: https://github.com/apache/openwhisk/pull/5102#discussion_r627083252
########## File path: core/invoker/src/main/scala/org/apache/openwhisk/core/containerpool/v2/FunctionPullingContainerPool.scala ########## @@ -0,0 +1,857 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.openwhisk.core.containerpool.v2 + +import java.util.concurrent.atomic.AtomicInteger + +import akka.actor.{Actor, ActorRef, ActorRefFactory, Cancellable, Props} +import org.apache.kafka.clients.producer.RecordMetadata +import org.apache.openwhisk.common._ +import org.apache.openwhisk.core.connector.ContainerCreationError._ +import org.apache.openwhisk.core.connector.{ + ContainerCreationAckMessage, + ContainerCreationMessage, + ContainerDeletionMessage +} +import org.apache.openwhisk.core.containerpool.{ + AdjustPrewarmedContainer, + BlackboxStartupError, + ColdStartKey, + ContainerPool, + ContainerPoolConfig, + ContainerRemoved, + PrewarmingConfig, + WhiskContainerStartupError +} +import org.apache.openwhisk.core.entity._ +import org.apache.openwhisk.core.entity.size._ +import org.apache.openwhisk.http.Messages + +import scala.annotation.tailrec +import scala.collection.concurrent.TrieMap +import scala.collection.immutable +import scala.concurrent.Future +import scala.concurrent.duration._ +import scala.util.{Random, Try} +import scala.collection.immutable.Queue + +case class CreationContainer(creationMessage: ContainerCreationMessage, action: WhiskAction) +case class DeletionContainer(deletionMessage: ContainerDeletionMessage) +case object Remove +case class Keep(timeout: FiniteDuration) +case class PrewarmContainer(maxConcurrent: Int) + +/** + * A pool managing containers to run actions on. + * + * This pool fulfills the other half of the ContainerProxy contract. Only + * one job (either Start or Run) is sent to a child-actor at any given + * time. The pool then waits for a response of that container, indicating + * the container is done with the job. Only then will the pool send another + * request to that container. + * + * Upon actor creation, the pool will start to prewarm containers according + * to the provided prewarmConfig, iff set. Those containers will **not** be + * part of the poolsize calculation, which is capped by the poolSize parameter. + * 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 actor + * @param prewarmConfig optional settings for container prewarming + * @param poolConfig config for the ContainerPool + */ +class FunctionPullingContainerPool( + childFactory: ActorRefFactory => ActorRef, + invokerHealthService: ActorRef, + poolConfig: ContainerPoolConfig, + instance: InvokerInstanceId, + prewarmConfig: List[PrewarmingConfig] = List.empty, + sendAckToScheduler: (SchedulerInstanceId, ContainerCreationAckMessage) => Future[RecordMetadata])( + implicit val logging: Logging) + extends Actor { + import ContainerPoolV2.memoryConsumptionOf + + implicit val ec = context.system.dispatcher + + private var busyPool = immutable.Map.empty[ActorRef, Data] + private var inProgressPool = immutable.Map.empty[ActorRef, Data] + private var warmedPool = immutable.Map.empty[ActorRef, WarmData] + private var prewarmedPool = immutable.Map.empty[ActorRef, PreWarmData] + private var prewarmStartingPool = immutable.Map.empty[ActorRef, (String, ByteSize)] + + private var shuttingDown = false + + private val creationMessages = TrieMap[ActorRef, ContainerCreationMessage]() + + private var preWarmScheduler: Option[Cancellable] = None + private var prewarmConfigQueue = Queue.empty[(CodeExec[_], ByteSize, Option[FiniteDuration])] + private val prewarmCreateFailedCount = new AtomicInteger(0) + + val logScheduler = context.system.scheduler.schedule(0.seconds, 1.seconds) { + MetricEmitter.emitHistogramMetric( + LoggingMarkers.INVOKER_CONTAINERPOOL_MEMORY("inprogress"), + memoryConsumptionOf(inProgressPool)) + MetricEmitter.emitHistogramMetric( + LoggingMarkers.INVOKER_CONTAINERPOOL_MEMORY("busy"), + memoryConsumptionOf(busyPool)) + MetricEmitter.emitHistogramMetric( + LoggingMarkers.INVOKER_CONTAINERPOOL_MEMORY("prewarmed"), + memoryConsumptionOf(prewarmedPool)) + MetricEmitter.emitHistogramMetric(LoggingMarkers.INVOKER_CONTAINERPOOL_MEMORY("max"), poolConfig.userMemory.toMB) + } + + // Key is ColdStartKey, value is the number of cold Start in minute + var coldStartCount = immutable.Map.empty[ColdStartKey, Int] + + adjustPrewarmedContainer(true, false) + + // check periodically, adjust prewarmed container(delete if unused for some time and create some increment containers) + // add some random amount to this schedule to avoid a herd of container removal + creation + val interval = poolConfig.prewarmExpirationCheckInterval + poolConfig.prewarmExpirationCheckIntervalVariance + .map(v => + Random + .nextInt(v.toSeconds.toInt)) + .getOrElse(0) + .seconds + + if (prewarmConfig.exists(!_.reactive.isEmpty)) { + context.system.scheduler.schedule( + poolConfig.prewarmExpirationCheckInitDelay, + interval, + self, + AdjustPrewarmedContainer) + } + + val resourceSubmitter = context.system.scheduler.schedule(0.seconds, poolConfig.memorySyncInterval) { + syncMemoryInfo + } + + private def logContainerStart(c: ContainerCreationMessage, action: WhiskAction, containerState: String): Unit = { + val FQN = c.action + if (FQN.namespace.name == "whisk.system" && FQN.fullPath.segments > 2) { + MetricEmitter.emitCounterMetric(LoggingMarkers.INVOKER_SHAREDPACKAGE(FQN.fullPath.asString)) + } + + MetricEmitter.emitCounterMetric( + LoggingMarkers.INVOKER_CONTAINER_START( + containerState, + c.invocationNamespace, + c.action.namespace.toString, + c.action.name.toString)) + } + + def receive: Receive = { + case PrewarmContainer(maxConcurrent) => + if (prewarmConfigQueue.isEmpty) { + preWarmScheduler.map(_.cancel()) + preWarmScheduler = None + } else { + for (_ <- 1 to maxConcurrent if !prewarmConfigQueue.isEmpty) { + val ((codeExec, byteSize, ttl), newQueue) = prewarmConfigQueue.dequeue + prewarmConfigQueue = newQueue + prewarmContainer(codeExec, byteSize, ttl) + } + } + + case CreationContainer(create: ContainerCreationMessage, action: WhiskAction) => + if (shuttingDown) { + val message = + s"creationId: ${create.creationId}, invoker is shutting down, reschedule ${action.fullyQualifiedName(false)}" + val ack = ContainerCreationAckMessage( + create.transid, + create.creationId, + create.invocationNamespace, + create.action, + create.revision, + create.whiskActionMetaData, + instance, + create.schedulerHost, + create.rpcPort, + create.retryCount, + Some(ShuttingDownError), + Some(message)) + logging.warn(this, message) + sendAckToScheduler(create.rootSchedulerIndex, ack) + } else { + logging.info(this, s"received a container creation message: ${create.creationId}") + action.toExecutableWhiskAction match { + case Some(executable) => + val createdContainer = + takeWarmedContainer(executable, create.invocationNamespace, create.revision) + .map(container => (container, "warmed")) + .orElse { + takeContainer(executable) + } + handleChosenContainer(create, executable, createdContainer) + case None => + val message = + s"creationId: ${create.creationId}, non-executable action reached the container pool ${action.fullyQualifiedName(false)}" + logging.error(this, message) + val ack = ContainerCreationAckMessage( + create.transid, + create.creationId, + create.invocationNamespace, + create.action, + create.revision, + create.whiskActionMetaData, + instance, + create.schedulerHost, + create.rpcPort, + create.retryCount, + Some(NonExecutableActionError), + Some(message)) + sendAckToScheduler(create.rootSchedulerIndex, ack) + } + } + + case DeletionContainer(deletionMessage: ContainerDeletionMessage) => + val oldRevision = deletionMessage.revision + val invocationNamespace = deletionMessage.invocationNamespace + val fqn = deletionMessage.action.copy(version = None) + + warmedPool.foreach(warmed => { + val proxy = warmed._1 + val data = warmed._2 + + if (data.invocationNamespace == invocationNamespace + && data.action.fullyQualifiedName(withVersion = false) == fqn.copy(version = None) + && data.revision <= oldRevision) { + proxy ! GracefulShutdown + } + }) + + busyPool.foreach(f = busy => { + val proxy = busy._1 + busy._2 match { + case warmData: WarmData + if warmData.invocationNamespace == invocationNamespace + && warmData.action.fullyQualifiedName(withVersion = false) == fqn.copy(version = None) + && warmData.revision <= oldRevision => + proxy ! GracefulShutdown + case initializedData: InitializedData + if initializedData.invocationNamespace == invocationNamespace + && initializedData.action.fullyQualifiedName(withVersion = false) == fqn.copy(version = None) => + proxy ! GracefulShutdown + case _ => // Other actions are ignored. + } + }) + + case ReadyToWork(data) => + prewarmStartingPool = prewarmStartingPool - sender() + prewarmedPool = prewarmedPool + (sender() -> data) + // after create prewarm successfully, reset the value to 0 + if (prewarmCreateFailedCount.get() > 0) { + prewarmCreateFailedCount.set(0) + } + + // Container is initialized + case Initialized(data) => + busyPool = busyPool + (sender() -> data) + inProgressPool = inProgressPool - sender() + // container init completed, send creationAck(success) to scheduler + creationMessages.remove(sender()).foreach { msg => + val ack = ContainerCreationAckMessage( + msg.transid, + msg.creationId, + msg.invocationNamespace, + msg.action, + msg.revision, + msg.whiskActionMetaData, + instance, + msg.schedulerHost, + msg.rpcPort, + msg.retryCount) + sendAckToScheduler(msg.rootSchedulerIndex, ack) + } + + case Resumed(data) => + busyPool = busyPool + (sender() -> data) + inProgressPool = inProgressPool - sender() + // container init completed, send creationAck(success) to scheduler + creationMessages.remove(sender()).foreach { msg => + val ack = ContainerCreationAckMessage( + msg.transid, + msg.creationId, + msg.invocationNamespace, + msg.action, + msg.revision, + msg.whiskActionMetaData, + instance, + msg.schedulerHost, + msg.rpcPort, + msg.retryCount) + sendAckToScheduler(msg.rootSchedulerIndex, ack) + } + + // if warmed containers is failed to resume, we should try to use other container or create a new one + case ResumeFailed(data) => + inProgressPool = inProgressPool - sender() + creationMessages.remove(sender()).foreach { msg => + val container = takeWarmedContainer(data.action, data.invocationNamespace, data.revision) + .map(container => (container, "warmed")) + .orElse { + takeContainer(data.action) + } + handleChosenContainer(msg, data.action, container) + } + + case ContainerCreationFailed(t) => + val (error, message) = t match { + case WhiskContainerStartupError(msg) => (WhiskError, msg) + case BlackboxStartupError(msg) => (BlackBoxError, msg) + case _ => (WhiskError, Messages.resourceProvisionError) + } + creationMessages.remove(sender()).foreach { msg => + val ack = ContainerCreationAckMessage( + msg.transid, + msg.creationId, + msg.invocationNamespace, + msg.action, + msg.revision, + msg.whiskActionMetaData, + instance, + msg.schedulerHost, + msg.rpcPort, + msg.retryCount, + Some(error), + Some(message)) + sendAckToScheduler(msg.rootSchedulerIndex, ack) + } + + case ContainerIsPaused(data) => + warmedPool = warmedPool + (sender() -> data) + busyPool = busyPool - sender() // remove container from busy pool + + // Container got removed + case ContainerRemoved(replacePrewarm) => + inProgressPool.get(sender()).foreach { _ => + inProgressPool = inProgressPool - sender() + } + + warmedPool.get(sender()).foreach { _ => + warmedPool = warmedPool - sender() + } + + // container was busy (busy indicates at full capacity), so there is capacity to accept another job request + busyPool.get(sender()).foreach { _ => + busyPool = busyPool - sender() + } + + //in case this was a prewarm + prewarmedPool.get(sender()).foreach { data => + prewarmedPool = prewarmedPool - sender() + logging.info( + this, + s"${if (replacePrewarm) "failed" else "expired"} prewarm [kind: ${data.kind}, memory: ${data.memoryLimit.toString}] removed") + } + + //in case this was a starting prewarm + prewarmStartingPool.get(sender()).foreach { data => + logging.info(this, s"failed starting prewarm [kind: ${data._1}, memory: ${data._2.toString}] removed") + prewarmStartingPool = prewarmStartingPool - sender() + prewarmCreateFailedCount.incrementAndGet() + } + + //backfill prewarms on every ContainerRemoved, just in case + if (replacePrewarm) { + adjustPrewarmedContainer(false, false) //in case a prewarm is removed due to health failure or crash + } + + // there maybe a chance that container create failed or init grpc client failed, + // send creationAck(reschedule) to scheduler + creationMessages.remove(sender()).foreach { msg => + val ack = ContainerCreationAckMessage( + msg.transid, + msg.creationId, + msg.invocationNamespace, + msg.action, + msg.revision, + msg.whiskActionMetaData, + instance, + msg.schedulerHost, + msg.rpcPort, + msg.retryCount, + Some(UnknownError), + Some("ContainerProxy init failed.")) + sendAckToScheduler(msg.rootSchedulerIndex, ack) + } + + case GracefulShutdown => + shuttingDown = true + waitForPoolToClear() + + case Enable => + shuttingDown = false + + case AdjustPrewarmedContainer => + // Reset the prewarmCreateCount value when do expiration check and backfill prewarm if possible + prewarmCreateFailedCount.set(0) + adjustPrewarmedContainer(false, true) + } + + /** Install prewarm containers up to the configured requirements for each kind/memory combination or specified kind/memory */ + private def adjustPrewarmedContainer(init: Boolean, scheduled: Boolean): Unit = { + if (!shuttingDown) { + if (scheduled) { + //on scheduled time, remove expired prewarms + ContainerPoolV2.removeExpired(poolConfig, prewarmConfig, prewarmedPool).foreach { p => + prewarmedPool = prewarmedPool - p + p ! Remove + } + //on scheduled time, emit cold start counter metric with memory + kind + coldStartCount foreach { coldStart => + val coldStartKey = coldStart._1 + MetricEmitter.emitCounterMetric( + LoggingMarkers.CONTAINER_POOL_PREWARM_COLDSTART(coldStartKey.memory.toString, coldStartKey.kind)) + } + } + + ContainerPoolV2 + .increasePrewarms( + init, + scheduled, + coldStartCount, + prewarmConfig, + prewarmedPool, + prewarmStartingPool, + prewarmConfigQueue) + .foreach { c => + val config = c._1 + val currentCount = c._2._1 + val desiredCount = c._2._2 + if (prewarmCreateFailedCount.get() > poolConfig.prewarmMaxRetryLimit) { + logging.warn( + this, + s"[kind: ${config.exec.kind}, memory: ${config.memoryLimit.toString}] prewarm create failed count exceeds max retry limit: ${poolConfig.prewarmMaxRetryLimit}, currentCount: ${currentCount}, desiredCount: ${desiredCount}") + } else { + if (currentCount < desiredCount) { + (currentCount until desiredCount).foreach { _ => + poolConfig.prewarmContainerCreationConfig match { + case Some(_) => + prewarmConfigQueue = + prewarmConfigQueue.enqueue((config.exec, config.memoryLimit, config.reactive.map(_.ttl))) + case None => + prewarmContainer(config.exec, config.memoryLimit, config.reactive.map(_.ttl)) + } + } + } + } + } + + // run queue consumer + poolConfig.prewarmContainerCreationConfig.foreach(config => { + logging.info( + this, + s"prewarm container creation is starting with creation delay configuration [maxConcurrent: ${config.maxConcurrent}, creationDelay: ${config.creationDelay.toMillis} millisecond]") + if (preWarmScheduler.isEmpty) { + preWarmScheduler = Some( + context.system.scheduler + .schedule(0.seconds, config.creationDelay, self, PrewarmContainer(config.maxConcurrent))) + } + }) + + if (scheduled) { + // lastly, clear coldStartCounts each time scheduled event is processed to reset counts + coldStartCount = immutable.Map.empty[ColdStartKey, Int] + } + } + } + + private def syncMemoryInfo: Unit = { + val busyMemory = memoryConsumptionOf(busyPool) + val inProgressMemory = memoryConsumptionOf(inProgressPool) + invokerHealthService ! MemoryInfo( + poolConfig.userMemory.toMB - busyMemory - inProgressMemory, + busyMemory, + inProgressMemory) + } + + /** Creates a new container and updates state accordingly. */ + private def createContainer(memoryLimit: ByteSize): (ActorRef, Data) = { + val ref = childFactory(context) + val data = MemoryData(memoryLimit) + ref -> data + } + + /** Creates a new prewarmed container */ + private def prewarmContainer(exec: CodeExec[_], memoryLimit: ByteSize, ttl: Option[FiniteDuration]): Unit = { + val newContainer = childFactory(context) + prewarmStartingPool = prewarmStartingPool + (newContainer -> (exec.kind, memoryLimit)) + newContainer ! Start(exec, memoryLimit, ttl) + } + + /** this is only for cold start statistics of prewarm configs, e.g. not blackbox or other configs. */ + def incrementColdStartCount(kind: String, memoryLimit: ByteSize): Unit = { + prewarmConfig + .filter { config => + kind == config.exec.kind && memoryLimit == config.memoryLimit + } + .foreach { _ => + val coldStartKey = ColdStartKey(kind, memoryLimit) + coldStartCount.get(coldStartKey) match { + case Some(value) => coldStartCount = coldStartCount + (coldStartKey -> (value + 1)) + case None => coldStartCount = coldStartCount + (coldStartKey -> 1) + } + } + } + + /** + * Takes a warmed container out of the warmed pool + * iff a container with a matching revision is found + * + * @param action the action. + * @param invocationNamespace the invocation namespace for shared package. + * @param revision the DocRevision. + * @return the container iff found + */ + private def takeWarmedContainer(action: ExecutableWhiskAction, + invocationNamespace: String, + revision: DocRevision): Option[(ActorRef, Data)] = { + warmedPool + .find { + case (_, WarmData(_, `invocationNamespace`, `action`, `revision`, _, _)) => true + case _ => false + } + .map { + case (ref, data) => + warmedPool = warmedPool - ref + logging.info(this, s"Choose warmed container ${data.container.containerId}") + (ref, data) + } + } + + /** + * Takes a prewarm container out of the prewarmed pool + * iff a container with a matching kind and suitable memory is found. + * + * @param action the action that holds the kind and the required memory. + * @param maximumMemory the maximum memory container can have + * @return the container iff found + */ + private def takePrewarmContainer(action: ExecutableWhiskAction, maximumMemory: ByteSize): Option[(ActorRef, Data)] = { + val kind = action.exec.kind + val memory = action.limits.memory.megabytes.MB + + // find a container with same kind and smallest memory + prewarmedPool.filter { + case (_, PreWarmData(_, `kind`, preMemory, _)) if preMemory >= memory && preMemory <= maximumMemory => true + case _ => false + }.toList match { + case Nil => + None + case res => + val (ref, data) = res.minBy(_._2.memoryLimit) + prewarmedPool = prewarmedPool - ref + + //get the appropriate ttl from prewarm configs + val ttl = + prewarmConfig.find(pc => pc.memoryLimit == memory && pc.exec.kind == kind).flatMap(_.reactive.map(_.ttl)) + prewarmContainer(action.exec, data.memoryLimit, ttl) + Some(ref, data) + } + } + + /** Removes a container and updates state accordingly. */ + def removeContainer(toDelete: ActorRef) = { + toDelete ! Remove + warmedPool = warmedPool - toDelete + } + + /** + * Calculate if there is enough free memory within a given pool. + * + * @param pool The pool, that has to be checked, if there is enough free memory. + * @param memory The amount of memory to check. + * @return true, if there is enough space for the given amount of memory. + */ + private def hasPoolSpaceFor[A](pool: Map[A, Data], memory: ByteSize): Boolean = { + memoryConsumptionOf(pool) + memory.toMB <= poolConfig.userMemory.toMB + } + + /** + * Make all busyPool's memoryQueue actor shutdown gracefully + */ + private def waitForPoolToClear(): Unit = { + busyPool.keys.foreach(_ ! GracefulShutdown) + warmedPool.keys.foreach(_ ! GracefulShutdown) + if (inProgressPool.nonEmpty) { + context.system.scheduler.scheduleOnce(5.seconds) { + waitForPoolToClear() + } + } + } + + /** + * take a prewarmed container or create a new one + * + * @param executable The executable whisk action + * @return + */ + private def takeContainer(executable: ExecutableWhiskAction) = { + val freeMemory = poolConfig.userMemory.toMB - memoryConsumptionOf(busyPool ++ warmedPool ++ inProgressPool) + val deletableMemory = memoryConsumptionOf(warmedPool) + val requiredMemory = executable.limits.memory.megabytes + if (requiredMemory > freeMemory + deletableMemory) { + None + } else { + // try to take a preWarmed container whose memory doesn't exceed the max `usable` memory + takePrewarmContainer( + executable, + if (poolConfig.prewarmPromotion) (freeMemory + deletableMemory).MB else requiredMemory.MB) match { + // there is a suitable preWarmed container but not enough free memory for it, delete some warmed container first + case Some(container) if container._2.memoryLimit > freeMemory.MB => + ContainerPoolV2 + .remove(warmedPool, container._2.memoryLimit - freeMemory.MB) + .map(removeContainer) + .headOption + .map { _ => + (container, "prewarmed") + } + // there is a suitable preWarmed container and enough free memory for it + case Some(container) => + Some((container, "prewarmed")) + // there is no suitable preWarmed container and not enough free memory for the action + case None if executable.limits.memory.megabytes > freeMemory => + ContainerPoolV2 + .remove(warmedPool, executable.limits.memory.megabytes.MB - freeMemory.MB) + .map(removeContainer) + .headOption + .map { _ => + incrementColdStartCount(executable.exec.kind, executable.limits.memory.megabytes.MB) + (createContainer(executable.limits.memory.megabytes.MB), "cold") + } + // there is no suitable preWarmed container and enough free memory + case None => + incrementColdStartCount(executable.exec.kind, executable.limits.memory.megabytes.MB) + Some(createContainer(executable.limits.memory.megabytes.MB), "cold") + + // this should not happen, but just for safety + case _ => + None + } + } + } + + private def handleChosenContainer(create: ContainerCreationMessage, + executable: ExecutableWhiskAction, + container: Option[((ActorRef, Data), String)]) = { + container match { + case Some(((proxy, data), containerState)) => + // record creationMessage so when container created failed, we can send failed message to scheduler + creationMessages.getOrElseUpdate(proxy, create) + proxy ! Initialize(create.invocationNamespace, executable, create.schedulerHost, create.rpcPort, create.transid) + inProgressPool = inProgressPool + (proxy -> data) + logContainerStart(create, executable.toWhiskAction, containerState) + + case None => + val message = + s"creationId: ${create.creationId}, invoker[$instance] doesn't have enough resource for container: ${create.action}" + logging.info(this, message) + syncMemoryInfo + val ack = ContainerCreationAckMessage( + create.transid, + create.creationId, + create.invocationNamespace, + create.action, + create.revision, + create.whiskActionMetaData, + instance, + create.schedulerHost, + create.rpcPort, + create.retryCount, + Some(ResourceNotEnoughError), + Some(message)) + sendAckToScheduler(create.rootSchedulerIndex, ack) + } + } +} + +object ContainerPoolV2 { + + /** + * Calculate the memory of a given pool. + * + * @param pool The pool with the containers. + * @return The memory consumption of all containers in the pool in Megabytes. + */ + protected[containerpool] def memoryConsumptionOf[A](pool: Map[A, Data]): Long = { + pool.map(_._2.memoryLimit.toMB).sum + } + + /** + * Finds the oldest previously used container to remove to make space for the job passed to run. + * Depending on the space that has to be allocated, several containers might be removed. + * + * NOTE: This method is never called to remove an action that is in the pool already, + * since this would be picked up earlier in the scheduler and the container reused. + * + * @param pool a map of all free containers in the pool + * @param memory the amount of memory that has to be freed up + * @return a list of containers to be removed iff found + */ + @tailrec + protected[containerpool] def remove[A](pool: Map[A, WarmData], + memory: ByteSize, + toRemove: List[A] = List.empty): List[A] = { + if (memory > 0.B && pool.nonEmpty && memoryConsumptionOf(pool) >= 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) = pool.minBy(_._2.lastUsed) + // Catch exception if remaining memory will be negative + val remainingMemory = Try(memory - data.memoryLimit).getOrElse(0.B) + remove(pool - ref, remainingMemory, toRemove ++ List(ref)) + } 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 recursions: Enough containers are found to get the memory, that is + // necessary. -> Abort recursion + toRemove + } + } + + /** + * Find the expired actor in prewarmedPool + * + * @param poolConfig + * @param prewarmConfig + * @param prewarmedPool + * @param logging + * @return a list of expired actor + */ + def removeExpired[A](poolConfig: ContainerPoolConfig, + prewarmConfig: List[PrewarmingConfig], + prewarmedPool: Map[A, PreWarmData])(implicit logging: Logging): List[A] = { + val now = Deadline.now + val expireds = prewarmConfig + .flatMap { config => + val kind = config.exec.kind + val memory = config.memoryLimit + config.reactive + .map { c => + val expiredPrewarmedContainer = prewarmedPool.toSeq + .filter { warmInfo => + warmInfo match { + case (_, p @ PreWarmData(_, `kind`, `memory`, _)) if p.isExpired() => true + case _ => false + } + } + .sortBy(_._2.expires.getOrElse(now)) + + // emit expired container counter metric with memory + kind + MetricEmitter.emitCounterMetric(LoggingMarkers.CONTAINER_POOL_PREWARM_EXPIRED(memory.toString, kind)) + if (expiredPrewarmedContainer.nonEmpty) { + logging.info( + this, + s"[kind: ${kind} memory: ${memory.toString}] ${expiredPrewarmedContainer.size} expired prewarmed containers") + } + expiredPrewarmedContainer.map(e => (e._1, e._2.expires.getOrElse(now))) + } + .getOrElse(List.empty) + } + .sortBy(_._2) //need to sort these so that if the results are limited, we take the oldest + .map(_._1) + if (expireds.nonEmpty) { + logging.info(this, s"removing up to ${poolConfig.prewarmExpirationLimit} of ${expireds.size} expired containers") + expireds.take(poolConfig.prewarmExpirationLimit).foreach { e => + prewarmedPool.get(e).map { d => + logging.info(this, s"removing expired prewarm of kind ${d.kind} with container ${d.container} ") + } + } + } + expireds.take(poolConfig.prewarmExpirationLimit) + } + + /** + * Find the increased number for the prewarmed kind + * + * @param init + * @param scheduled + * @param coldStartCount + * @param prewarmConfig + * @param prewarmedPool + * @param prewarmStartingPool + * @param logging + * @return the current number and increased number for the kind in the Map + */ + def increasePrewarms(init: Boolean, + scheduled: Boolean, + coldStartCount: Map[ColdStartKey, Int], + prewarmConfig: List[PrewarmingConfig], + prewarmedPool: Map[ActorRef, PreWarmData], + prewarmStartingPool: Map[ActorRef, (String, ByteSize)], + prewarmQueue: Queue[(CodeExec[_], ByteSize, Option[FiniteDuration])])( + implicit logging: Logging): Map[PrewarmingConfig, (Int, Int)] = { + prewarmConfig.map { config => + val kind = config.exec.kind + val memory = config.memoryLimit + + val runningCount = prewarmedPool.count { + // done starting (include expired, since they may not have been removed yet) + case (_, p @ PreWarmData(_, `kind`, `memory`, _)) => true + // started but not finished starting (or expired) + case _ => false + } + val startingCount = prewarmStartingPool.count(p => p._2._1 == kind && p._2._2 == memory) + val queuingCount = prewarmQueue.count(p => p._1.kind == kind && p._2 == memory) Review comment: This is to `avoid create a lot of prewarm in a very short time` -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org
