[GitHub] [openwhisk] ningyougang commented on a change in pull request #5102: [New Scheduler] Implement FunctionPullingContainerPool
ningyougang commented on a change in pull request #5102:
URL: https://github.com/apache/openwhisk/pull/5102#discussion_r622878434
##
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)
Review comment:
This is for retry logic.
Let's assume the max retry limit is `5`,
* If 1th, 2th, 3th creation prewarm failed, but the 4th creation prewarm
success, the prewarmCreateFailedCount would be reset to 0. (the count would be
reinitialized when creation is succeeded at any time.)
* If 1th, 2th, 3th, 4th, 5th creation
[GitHub] [openwhisk] ningyougang commented on a change in pull request #5102: [New Scheduler] Implement FunctionPullingContainerPool
ningyougang commented on a change in pull request #5102:
URL: https://github.com/apache/openwhisk/pull/5102#discussion_r627863269
##
File path: core/invoker/src/main/resources/application.conf
##
@@ -60,9 +60,13 @@ whisk {
user-memory: 1024 m
concurrent-peek-factor: 0.5 #factor used to limit message peeking: 0 <
factor <= 1.0 - larger number improves concurrent processing, but increases
risk of message loss during invoker crash
akka-client: false # if true, use PoolingContainerClient for HTTP from
invoker to action container (otherwise use ApacheBlockingContainerClient)
-prewarm-expiration-check-interval: 1 minute # period to check for prewarm
expiration
+prewarm-expiration-check-init-delay: 10 minute # the init delay time for
the first check
+prewarm-expiration-check-interval: 10 minute # period to check for prewarm
expiration
prewarm-expiration-check-interval-variance: 10 seconds # varies expiration
across invokers to avoid many concurrent expirations
prewarm-expiration-limit: 100 # number of prewarms to expire in one
expiration cycle (remaining expired will be considered for expiration in next
cycle)
+prewarm-max-retry-limit: 5 # max retry limit for create prewarm
+prewarm-promotion: false # if true, action can take prewarm container
which has bigger memory
Review comment:
Yes, but for `prewarm-promotion`, is used in
FunctionPullingContainerPool only, and the configuration is `false` which mean,
didn't take prewarm container which has bigger memory
--
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
[GitHub] [openwhisk] ningyougang commented on a change in pull request #5102: [New Scheduler] Implement FunctionPullingContainerPool
ningyougang commented on a change in pull request #5102:
URL: https://github.com/apache/openwhisk/pull/5102#discussion_r627862698
##
File path: core/invoker/src/main/resources/application.conf
##
@@ -60,9 +60,13 @@ whisk {
user-memory: 1024 m
concurrent-peek-factor: 0.5 #factor used to limit message peeking: 0 <
factor <= 1.0 - larger number improves concurrent processing, but increases
risk of message loss during invoker crash
akka-client: false # if true, use PoolingContainerClient for HTTP from
invoker to action container (otherwise use ApacheBlockingContainerClient)
-prewarm-expiration-check-interval: 1 minute # period to check for prewarm
expiration
+prewarm-expiration-check-init-delay: 10 minute # the init delay time for
the first check
+prewarm-expiration-check-interval: 10 minute # period to check for prewarm
expiration
prewarm-expiration-check-interval-variance: 10 seconds # varies expiration
across invokers to avoid many concurrent expirations
prewarm-expiration-limit: 100 # number of prewarms to expire in one
expiration cycle (remaining expired will be considered for expiration in next
cycle)
+prewarm-max-retry-limit: 5 # max retry limit for create prewarm
Review comment:
Already changed the comment to `max subsequent retry limit to create
prewarm containers`
--
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
[GitHub] [openwhisk] ningyougang commented on a change in pull request #5102: [New Scheduler] Implement FunctionPullingContainerPool
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"),
+
[GitHub] [openwhisk] ningyougang commented on a change in pull request #5102: [New Scheduler] Implement FunctionPullingContainerPool
ningyougang commented on a change in pull request #5102:
URL: https://github.com/apache/openwhisk/pull/5102#discussion_r622880300
##
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"),
+
[GitHub] [openwhisk] ningyougang commented on a change in pull request #5102: [New Scheduler] Implement FunctionPullingContainerPool
ningyougang commented on a change in pull request #5102:
URL: https://github.com/apache/openwhisk/pull/5102#discussion_r622879797
##
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"),
+
[GitHub] [openwhisk] ningyougang commented on a change in pull request #5102: [New Scheduler] Implement FunctionPullingContainerPool
ningyougang commented on a change in pull request #5102:
URL: https://github.com/apache/openwhisk/pull/5102#discussion_r622878434
##
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)
Review comment:
This is for retry logic.
Let's assume the max retry limit is `5`,
* If 1th, 2th, 3th creation prewarm failed, but the 4th creation prewarm
success, the prewarmCreateFailedCount would be reset to 0.
* If 1th, 2th, 3th, 4th, 5th creation prewarm failed, it would stop all
creation. wait next round.
--
[GitHub] [openwhisk] ningyougang commented on a change in pull request #5102: [New Scheduler] Implement FunctionPullingContainerPool
ningyougang commented on a change in pull request #5102:
URL: https://github.com/apache/openwhisk/pull/5102#discussion_r622878434
##
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)
Review comment:
This is for retry.
Let's assume the max retry limit is `5`,
* If 1th, 2th, 3th creation prewarm failed, but the 4th creation prewarm
success, the prewarmCreateFailedCount would be reset to 0.
* If 1th, 2th, 3th, 4th, 5th creation prewarm failed, it would stop all
creation. wait next round.
--
This is
[GitHub] [openwhisk] ningyougang commented on a change in pull request #5102: [New Scheduler] Implement FunctionPullingContainerPool
ningyougang commented on a change in pull request #5102:
URL: https://github.com/apache/openwhisk/pull/5102#discussion_r622877409
##
File path:
core/invoker/src/main/scala/org/apache/openwhisk/core/containerpool/ContainerPool.scala
##
@@ -590,9 +596,9 @@ object ContainerPool {
}
.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) {
+ // emit expired container counter metric with memory + kind
+
MetricEmitter.emitCounterMetric(LoggingMarkers.CONTAINER_POOL_PREWARM_EXPIRED(memory.toString,
kind))
Review comment:
If expiredPrewarmedContainer is empty, have no need to emit this counter
metric.
--
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
[GitHub] [openwhisk] ningyougang commented on a change in pull request #5102: [New Scheduler] Implement FunctionPullingContainerPool
ningyougang commented on a change in pull request #5102:
URL: https://github.com/apache/openwhisk/pull/5102#discussion_r622876853
##
File path:
core/invoker/src/main/scala/org/apache/openwhisk/core/containerpool/ContainerPool.scala
##
@@ -91,7 +91,13 @@ class ContainerPool(childFactory: ActorRefFactory =>
ActorRef,
.nextInt(v.toSeconds.toInt))
.getOrElse(0)
.seconds
- context.system.scheduler.schedule(2.seconds, interval, self,
AdjustPrewarmedContainer)
+ if (prewarmConfig.exists(!_.reactive.isEmpty)) {
Review comment:
Have no need to backfill the prewarm periodically if `reactive`
configuration is not included in `runtimes.json`, because if doesn't exist
`reactive` in runtimes.json, the prewarm container will not be expired forever.
fyi, the reactive configuration in runtimes.json like below
```
...
"stemCells": [
{
"initialCount": 2,
"memory": "256 MB",
"reactive": {
"minCount": 1,
"maxCount": 4,
"ttl": "2 minutes",
"threshold": 1,
"increment": 1
}
}
]
...
```
--
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
[GitHub] [openwhisk] ningyougang commented on a change in pull request #5102: [New Scheduler] Implement FunctionPullingContainerPool
ningyougang commented on a change in pull request #5102:
URL: https://github.com/apache/openwhisk/pull/5102#discussion_r621782870
##
File path:
core/invoker/src/main/scala/org/apache/openwhisk/core/containerpool/v2/FunctionPullingContainerPool.scala
##
@@ -0,0 +1,866 @@
+/*
+ * 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 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 Creation(creationMessage: ContainerCreationMessage, action:
WhiskAction)
+case class Deletion(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 var prewarmCreateFailedCount = immutable.Map.empty[(String,
ByteSize), Int]
+
+ 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))
+
[GitHub] [openwhisk] ningyougang commented on a change in pull request #5102: [New Scheduler] Implement FunctionPullingContainerPool
ningyougang commented on a change in pull request #5102:
URL: https://github.com/apache/openwhisk/pull/5102#discussion_r621817799
##
File path:
core/invoker/src/main/scala/org/apache/openwhisk/core/containerpool/v2/FunctionPullingContainerPool.scala
##
@@ -0,0 +1,866 @@
+/*
+ * 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 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 Creation(creationMessage: ContainerCreationMessage, action:
WhiskAction)
+case class Deletion(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 var prewarmCreateFailedCount = immutable.Map.empty[(String,
ByteSize), Int]
+
+ 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))
+
