You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@openwhisk.apache.org by GitBox <gi...@apache.org> on 2022/11/01 01:27:06 UTC
[GitHub] [openwhisk] style95 commented on a diff in pull request #5284: Add scheduler overprovision for new actions before namespace throttling
style95 commented on code in PR #5284:
URL: https://github.com/apache/openwhisk/pull/5284#discussion_r1009995134
##########
tests/src/test/scala/org/apache/openwhisk/core/scheduler/queue/test/SchedulingDecisionMakerTests.scala:
##########
@@ -197,7 +197,142 @@ class SchedulingDecisionMakerTests
testProbe.expectMsg(DecisionResults(EnableNamespaceThrottling(dropMsg = false), 0))
}
- it should "add an initial container if there is no any" in {
+ it should "add one container when there is no container, and namespace over-provision has capacity" in {
+ val schedulingConfigNamespaceOverProvisioning =
+ SchedulingConfig(100.milliseconds, 100.milliseconds, 10.seconds, true, 1.5)
+ val decisionMaker =
+ system.actorOf(SchedulingDecisionMaker.props(testNamespace, action, schedulingConfigNamespaceOverProvisioning))
+ val testProbe = TestProbe()
+
+ val msg = QueueSnapshot(
+ initialized = false,
+ incomingMsgCount = new AtomicInteger(0),
+ currentMsgCount = 0,
+ existingContainerCount = 0, // there is no container for this action
+ inProgressContainerCount = 0,
+ staleActivationNum = 0,
+ existingContainerCountInNamespace = 1, // but there are already 2 containers in this namespace
+ inProgressContainerCountInNamespace = 1,
+ averageDuration = None,
+ limit = 2,
+ stateName = Running,
+ recipient = testProbe.ref)
+
+ decisionMaker ! msg
+
+ // this queue cannot create an initial container so enable throttling and drop messages.
+ testProbe.expectMsg(DecisionResults(AddInitialContainer, 1))
+ }
+
+ it should "enable namespace throttling with dropping msg when there is no container, and namespace over-provision has no capacity" in {
+ val schedulingConfigNamespaceOverProvisioning =
+ SchedulingConfig(100.milliseconds, 100.milliseconds, 10.seconds, true, 1.0)
+ val decisionMaker =
+ system.actorOf(SchedulingDecisionMaker.props(testNamespace, action, schedulingConfigNamespaceOverProvisioning))
+ val testProbe = TestProbe()
+
+ val msg = QueueSnapshot(
+ initialized = true,
+ incomingMsgCount = new AtomicInteger(0),
+ currentMsgCount = 0,
+ existingContainerCount = 0, // there is no container for this action
+ inProgressContainerCount = 0,
+ staleActivationNum = 0,
+ existingContainerCountInNamespace = 1, // but there are already 2 containers in this namespace
+ inProgressContainerCountInNamespace = 1,
+ averageDuration = None,
+ limit = 2,
+ stateName = Running,
+ recipient = testProbe.ref)
+
+ decisionMaker ! msg
+
+ // this queue cannot create an initial container so enable throttling and drop messages.
+ testProbe.expectMsg(DecisionResults(EnableNamespaceThrottling(dropMsg = true), 0))
+ }
+
+ it should "disable namespace throttling when namespace over-provision has capacity again" in {
+ val schedulingConfigNamespaceOverProvisioning =
+ SchedulingConfig(100.milliseconds, 100.milliseconds, 10.seconds, true, 1.1)
+ val decisionMaker =
+ system.actorOf(SchedulingDecisionMaker.props(testNamespace, action, schedulingConfigNamespaceOverProvisioning))
+ val testProbe = TestProbe()
+
+ val msg = QueueSnapshot(
+ initialized = true,
+ incomingMsgCount = new AtomicInteger(0),
+ currentMsgCount = 0,
+ existingContainerCount = 1, // there is no container for this action
Review Comment:
```suggestion
existingContainerCount = 1, // there is one container for this action
```
##########
tests/src/test/scala/org/apache/openwhisk/core/scheduler/queue/test/SchedulingDecisionMakerTests.scala:
##########
@@ -197,7 +197,142 @@ class SchedulingDecisionMakerTests
testProbe.expectMsg(DecisionResults(EnableNamespaceThrottling(dropMsg = false), 0))
}
- it should "add an initial container if there is no any" in {
+ it should "add one container when there is no container, and namespace over-provision has capacity" in {
+ val schedulingConfigNamespaceOverProvisioning =
+ SchedulingConfig(100.milliseconds, 100.milliseconds, 10.seconds, true, 1.5)
+ val decisionMaker =
+ system.actorOf(SchedulingDecisionMaker.props(testNamespace, action, schedulingConfigNamespaceOverProvisioning))
+ val testProbe = TestProbe()
+
+ val msg = QueueSnapshot(
+ initialized = false,
+ incomingMsgCount = new AtomicInteger(0),
+ currentMsgCount = 0,
+ existingContainerCount = 0, // there is no container for this action
+ inProgressContainerCount = 0,
+ staleActivationNum = 0,
+ existingContainerCountInNamespace = 1, // but there are already 2 containers in this namespace
+ inProgressContainerCountInNamespace = 1,
+ averageDuration = None,
+ limit = 2,
+ stateName = Running,
+ recipient = testProbe.ref)
+
+ decisionMaker ! msg
+
+ // this queue cannot create an initial container so enable throttling and drop messages.
+ testProbe.expectMsg(DecisionResults(AddInitialContainer, 1))
+ }
+
+ it should "enable namespace throttling with dropping msg when there is no container, and namespace over-provision has no capacity" in {
+ val schedulingConfigNamespaceOverProvisioning =
+ SchedulingConfig(100.milliseconds, 100.milliseconds, 10.seconds, true, 1.0)
+ val decisionMaker =
+ system.actorOf(SchedulingDecisionMaker.props(testNamespace, action, schedulingConfigNamespaceOverProvisioning))
+ val testProbe = TestProbe()
+
+ val msg = QueueSnapshot(
+ initialized = true,
+ incomingMsgCount = new AtomicInteger(0),
+ currentMsgCount = 0,
+ existingContainerCount = 0, // there is no container for this action
+ inProgressContainerCount = 0,
+ staleActivationNum = 0,
+ existingContainerCountInNamespace = 1, // but there are already 2 containers in this namespace
+ inProgressContainerCountInNamespace = 1,
+ averageDuration = None,
+ limit = 2,
+ stateName = Running,
+ recipient = testProbe.ref)
+
+ decisionMaker ! msg
+
+ // this queue cannot create an initial container so enable throttling and drop messages.
+ testProbe.expectMsg(DecisionResults(EnableNamespaceThrottling(dropMsg = true), 0))
+ }
+
+ it should "disable namespace throttling when namespace over-provision has capacity again" in {
+ val schedulingConfigNamespaceOverProvisioning =
+ SchedulingConfig(100.milliseconds, 100.milliseconds, 10.seconds, true, 1.1)
+ val decisionMaker =
+ system.actorOf(SchedulingDecisionMaker.props(testNamespace, action, schedulingConfigNamespaceOverProvisioning))
+ val testProbe = TestProbe()
+
+ val msg = QueueSnapshot(
+ initialized = true,
+ incomingMsgCount = new AtomicInteger(0),
+ currentMsgCount = 0,
+ existingContainerCount = 1, // there is no container for this action
+ inProgressContainerCount = 0,
+ staleActivationNum = 0,
+ existingContainerCountInNamespace = 1, // but there are already 2 containers in this namespace
+ inProgressContainerCountInNamespace = 1,
+ averageDuration = None,
+ limit = 2,
+ stateName = NamespaceThrottled,
+ recipient = testProbe.ref)
+
+ decisionMaker ! msg
+
+ // this queue cannot create an initial container so enable throttling and drop messages.
+ testProbe.expectMsg(DecisionResults(DisableNamespaceThrottling, 0))
+ }
+
+ it should "enable namespace throttling with dropping msg when there is a container, and namespace over-provision has no additional capacity" in {
Review Comment:
```suggestion
it should "enable namespace throttling without dropping msg when there is a container, and namespace over-provision has no additional capacity" in {
```
##########
core/scheduler/src/main/scala/org/apache/openwhisk/core/scheduler/queue/SchedulingDecisionMaker.scala:
##########
@@ -79,12 +89,15 @@ class SchedulingDecisionMaker(
*
* However, if the container exists(totalContainers != 0), the activation is not treated as a failure and the activation is delivered to the container.
*/
- case Running =>
+ case Running
+ if !schedulingConfig.allowOverProvisionBeforeThrottle || (schedulingConfig.allowOverProvisionBeforeThrottle && ceiling(
+ limit * schedulingConfig.namespaceOverProvisionBeforeThrottleRatio) - existingContainerCountInNs - inProgressContainerCountInNs <= 0) =>
logging.info(
this,
s"there is no capacity activations will be dropped or throttled, (availableMsg: $availableMsg totalContainers: $totalContainers, limit: $limit, namespaceContainers: ${existingContainerCountInNs}, namespaceInProgressContainer: ${inProgressContainerCountInNs}) [$invocationNamespace:$action]")
Future.successful(DecisionResults(EnableNamespaceThrottling(dropMsg = totalContainers == 0), 0))
-
+ case NamespaceThrottled if schedulingConfig.allowOverProvisionBeforeThrottle && ceiling(limit * schedulingConfig.namespaceOverProvisionBeforeThrottleRatio) - existingContainerCountInNs - inProgressContainerCountInNs > 0 =>
Review Comment:
👍
##########
tests/src/test/scala/org/apache/openwhisk/core/scheduler/queue/test/SchedulingDecisionMakerTests.scala:
##########
@@ -197,7 +197,142 @@ class SchedulingDecisionMakerTests
testProbe.expectMsg(DecisionResults(EnableNamespaceThrottling(dropMsg = false), 0))
}
- it should "add an initial container if there is no any" in {
+ it should "add one container when there is no container, and namespace over-provision has capacity" in {
+ val schedulingConfigNamespaceOverProvisioning =
+ SchedulingConfig(100.milliseconds, 100.milliseconds, 10.seconds, true, 1.5)
+ val decisionMaker =
+ system.actorOf(SchedulingDecisionMaker.props(testNamespace, action, schedulingConfigNamespaceOverProvisioning))
+ val testProbe = TestProbe()
+
+ val msg = QueueSnapshot(
+ initialized = false,
+ incomingMsgCount = new AtomicInteger(0),
+ currentMsgCount = 0,
+ existingContainerCount = 0, // there is no container for this action
+ inProgressContainerCount = 0,
+ staleActivationNum = 0,
+ existingContainerCountInNamespace = 1, // but there are already 2 containers in this namespace
+ inProgressContainerCountInNamespace = 1,
+ averageDuration = None,
+ limit = 2,
+ stateName = Running,
+ recipient = testProbe.ref)
+
+ decisionMaker ! msg
+
+ // this queue cannot create an initial container so enable throttling and drop messages.
+ testProbe.expectMsg(DecisionResults(AddInitialContainer, 1))
+ }
+
+ it should "enable namespace throttling with dropping msg when there is no container, and namespace over-provision has no capacity" in {
+ val schedulingConfigNamespaceOverProvisioning =
+ SchedulingConfig(100.milliseconds, 100.milliseconds, 10.seconds, true, 1.0)
+ val decisionMaker =
+ system.actorOf(SchedulingDecisionMaker.props(testNamespace, action, schedulingConfigNamespaceOverProvisioning))
+ val testProbe = TestProbe()
+
+ val msg = QueueSnapshot(
+ initialized = true,
+ incomingMsgCount = new AtomicInteger(0),
+ currentMsgCount = 0,
+ existingContainerCount = 0, // there is no container for this action
+ inProgressContainerCount = 0,
+ staleActivationNum = 0,
+ existingContainerCountInNamespace = 1, // but there are already 2 containers in this namespace
+ inProgressContainerCountInNamespace = 1,
+ averageDuration = None,
+ limit = 2,
+ stateName = Running,
+ recipient = testProbe.ref)
+
+ decisionMaker ! msg
+
+ // this queue cannot create an initial container so enable throttling and drop messages.
+ testProbe.expectMsg(DecisionResults(EnableNamespaceThrottling(dropMsg = true), 0))
+ }
+
+ it should "disable namespace throttling when namespace over-provision has capacity again" in {
+ val schedulingConfigNamespaceOverProvisioning =
+ SchedulingConfig(100.milliseconds, 100.milliseconds, 10.seconds, true, 1.1)
+ val decisionMaker =
+ system.actorOf(SchedulingDecisionMaker.props(testNamespace, action, schedulingConfigNamespaceOverProvisioning))
+ val testProbe = TestProbe()
+
+ val msg = QueueSnapshot(
+ initialized = true,
+ incomingMsgCount = new AtomicInteger(0),
+ currentMsgCount = 0,
+ existingContainerCount = 1, // there is no container for this action
+ inProgressContainerCount = 0,
+ staleActivationNum = 0,
+ existingContainerCountInNamespace = 1, // but there are already 2 containers in this namespace
+ inProgressContainerCountInNamespace = 1,
+ averageDuration = None,
+ limit = 2,
+ stateName = NamespaceThrottled,
+ recipient = testProbe.ref)
+
+ decisionMaker ! msg
+
+ // this queue cannot create an initial container so enable throttling and drop messages.
+ testProbe.expectMsg(DecisionResults(DisableNamespaceThrottling, 0))
+ }
+
+ it should "enable namespace throttling with dropping msg when there is a container, and namespace over-provision has no additional capacity" in {
+ val schedulingConfigNamespaceOverProvisioning =
+ SchedulingConfig(100.milliseconds, 100.milliseconds, 10.seconds, true, 1.0)
+ val decisionMaker =
+ system.actorOf(SchedulingDecisionMaker.props(testNamespace, action, schedulingConfigNamespaceOverProvisioning))
+ val testProbe = TestProbe()
+
+ val msg = QueueSnapshot(
+ initialized = true,
+ incomingMsgCount = new AtomicInteger(0),
+ currentMsgCount = 0,
+ existingContainerCount = 1,
+ inProgressContainerCount = 0,
+ staleActivationNum = 0,
+ existingContainerCountInNamespace = 1, // but there are already 2 containers in this namespace
+ inProgressContainerCountInNamespace = 1,
+ averageDuration = None,
+ limit = 2,
+ stateName = Running,
+ recipient = testProbe.ref)
+
+ decisionMaker ! msg
+
+ // this queue cannot create an initial container so enable throttling and drop messages.
Review Comment:
```suggestion
```
--
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.
To unsubscribe, e-mail: issues-unsubscribe@openwhisk.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org