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/10/26 22:25:25 UTC
[GitHub] [openwhisk] bdoyle0182 opened a new pull request, #5344: optimize scheduling decision when there are stale activations
bdoyle0182 opened a new pull request, #5344:
URL: https://github.com/apache/openwhisk/pull/5344
## Description
I believe this optimizes scheduling decisions without sacrificing any performance. The reasoning here is if a queue has stale activations, you risk over-provisioning if you make the calculation based on all available messages, not just stale ones. Here is an example:
t0ms: 5 activation arrives and for whatever reason it takes 100ms before the activation is processed even if there are containers existing (random network latency or whatever the brief blip may be). Let's say container throughput is 10 (10ms duration) and there's already 10 containers.
t50ms: 100 activations arrive and are available
t100ms: 5 activation is stale and enters the decision case when there is a stale activation. However it uses all 105 activations available to make a scheduling decision. 105 / 10 will create 11 containers. But it's not yet certain that the available messages can't be processed before going stale. You've therefore doubled containers for the action to 21 unnecessarily when 11 would be satisfactory. What this pr does is use the 5 stale activations at decision time to determine how much additional capacity is needed, in this case it would be ceil(5 / 10) = 1.
## Related issue and scope
<!--- Please include a link to a related issue if there is one. -->
- [ ] I opened an issue to propose and discuss this change (#????)
## My changes affect the following components
<!--- Select below all system components are affected by your change. -->
<!--- Enter an `x` in all applicable boxes. -->
- [ ] API
- [ ] Controller
- [ ] Message Bus (e.g., Kafka)
- [ ] Loadbalancer
- [X] Scheduler
- [ ] Invoker
- [ ] Intrinsic actions (e.g., sequences, conductors)
- [ ] Data stores (e.g., CouchDB)
- [ ] Tests
- [ ] Deployment
- [ ] CLI
- [ ] General tooling
- [ ] Documentation
## Types of changes
- [X] Bug fix (generally a non-breaking change which closes an issue).
- [ ] Enhancement or new feature (adds new functionality).
- [ ] Breaking change (a bug fix or enhancement which changes existing behavior).
## Checklist:
- [X] I signed an [Apache CLA](https://github.com/apache/openwhisk/blob/master/CONTRIBUTING.md).
- [X] I reviewed the [style guides](https://github.com/apache/openwhisk/blob/master/CONTRIBUTING.md#coding-standards) and followed the recommendations (Travis CI will check :).
- [ ] I added tests to cover my changes.
- [ ] My changes require further changes to the documentation.
- [ ] I updated the documentation where necessary.
--
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
[GitHub] [openwhisk] style95 commented on a diff in pull request #5344: optimize scheduling decision when there are stale activations
Posted by GitBox <gi...@apache.org>.
style95 commented on code in PR #5344:
URL: https://github.com/apache/openwhisk/pull/5344#discussion_r1009997561
##########
tests/src/test/scala/org/apache/openwhisk/core/scheduler/queue/test/SchedulingDecisionMakerTests.scala:
##########
@@ -683,6 +683,54 @@ class SchedulingDecisionMakerTests
testProbe.expectMsg(DecisionResults(AddContainer, 2))
}
+ it should "add more containers when there are stale messages and non-stale messages and both message classes need more containers" in {
+ val decisionMaker = system.actorOf(SchedulingDecisionMaker.props(testNamespace, action, schedulingConfig))
+ val testProbe = TestProbe()
+
+ val msg = QueueSnapshot(
+ initialized = true,
+ incomingMsgCount = new AtomicInteger(0),
+ currentMsgCount = 5,
+ existingContainerCount = 2,
+ inProgressContainerCount = 0,
+ staleActivationNum = 2,
+ existingContainerCountInNamespace = 2,
+ inProgressContainerCountInNamespace = 0,
+ averageDuration = Some(1000), // the average duration exists
+ limit = 10,
+ stateName = Running,
+ recipient = testProbe.ref)
+
+ decisionMaker ! msg
+
+ //should add two for the stale messages and one to increase tps of non-stale available messages
+ testProbe.expectMsg(DecisionResults(AddContainer, 3))
+ }
+
+ it should "add more containers when there are stale messages and non-stale messages have needed tps" in {
+ val decisionMaker = system.actorOf(SchedulingDecisionMaker.props(testNamespace, action, schedulingConfig))
+ val testProbe = TestProbe()
+
+ val msg = QueueSnapshot(
+ initialized = true,
+ incomingMsgCount = new AtomicInteger(0),
+ currentMsgCount = 5,
+ existingContainerCount = 2,
+ inProgressContainerCount = 0,
+ staleActivationNum = 2,
+ existingContainerCountInNamespace = 2,
+ inProgressContainerCountInNamespace = 0,
+ averageDuration = Some(50), // the average duration gives container throughput of 2
Review Comment:
```suggestion
averageDuration = Some(50), // the average duration gives container throughput of 20
```
--
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
[GitHub] [openwhisk] style95 commented on a diff in pull request #5344: optimize scheduling decision when there are stale activations
Posted by GitBox <gi...@apache.org>.
style95 commented on code in PR #5344:
URL: https://github.com/apache/openwhisk/pull/5344#discussion_r1006487212
##########
core/scheduler/src/main/scala/org/apache/openwhisk/core/scheduler/queue/SchedulingDecisionMaker.scala:
##########
@@ -139,9 +139,9 @@ class SchedulingDecisionMaker(
case (Running, Some(duration)) if staleActivationNum > 0 =>
// we can safely get the value as we already checked the existence
val containerThroughput = staleThreshold / duration
- val num = ceiling(availableMsg.toDouble / containerThroughput)
+ val num = ceiling(staleActivationNum.toDouble / containerThroughput)
// if it tries to create more containers than existing messages, we just create shortage
- val actualNum = (if (num > availableMsg) availableMsg else num) - inProgress
+ val actualNum = (if (num > staleActivationNum) staleActivationNum else num) - inProgress
Review Comment:
I still believe we need to consider all available messages.
This is because there can be both stale activations and non-stale(but about to be staled) activations.
It takes some time to provision more containers. Even if we can add more containers based on stale activations but we would easily need to add more containers if the container throughput itself is not enough.
To avoid too much over-provision, I think we can just substitute `existing` as well.
```
val actualNum = (if (num > availableMsg) availableMsg else num) - inProgress - existing
```
--
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
[GitHub] [openwhisk] bdoyle0182 commented on pull request #5344: optimize scheduling decision when there are stale activations
Posted by GitBox <gi...@apache.org>.
bdoyle0182 commented on PR #5344:
URL: https://github.com/apache/openwhisk/pull/5344#issuecomment-1292752659
I think an additional optimization would be to do the stale calculation and normal available message case in the same decision such that we take this code:
```
case (Running, Some(duration)) if staleActivationNum > 0 =>
// we can safely get the value as we already checked the existence
val containerThroughput = staleThreshold / duration
val num = ceiling(staleActivationNum.toDouble / containerThroughput)
// if it tries to create more containers than existing messages, we just create shortage
val actualNum = (if (num > staleActivationNum) staleActivationNum else num) - inProgress
addServersIfPossible(
existing,
inProgress,
containerThroughput,
availableMsg,
capacity,
actualNum,
staleActivationNum,
duration,
Running)
// need more containers and a message is already processed
case (Running, Some(duration)) =>
// we can safely get the value as we already checked the existence
val containerThroughput = staleThreshold / duration
val expectedTps = containerThroughput * (existing + inProgress)
if (availableMsg >= expectedTps && existing + inProgress < availableMsg) {
val num = ceiling((availableMsg / containerThroughput) - existing - inProgress)
// if it tries to create more containers than existing messages, we just create shortage
val actualNum = if (num + totalContainers > availableMsg) availableMsg - totalContainers else num
addServersIfPossible(
existing,
inProgress,
containerThroughput,
availableMsg,
capacity,
actualNum,
staleActivationNum,
duration,
Running)
} else {
Future.successful(DecisionResults(Skip, 0))
}
```
and consolidate it to this:
```
// need more containers and a message is already processed
case (Running, Some(duration)) =>
// we can safely get the value as we already checked the existence
val containerThroughput = staleThreshold / duration
val expectedTps = containerThroughput * (existing + inProgress)
val availableNonStaleActivations = availableMsg - staleActivationNum
var staleContainerProvision = 0
if (staleActivationNum > 0) {
val num = ceiling(staleActivationNum.toDouble / containerThroughput)
// if it tries to create more containers than existing messages, we just create shortage
staleContainerProvision = (if (num > staleActivationNum) staleActivationNum else num) - inProgress
}
if (availableNonStaleActivations >= expectedTps && existing + inProgress < availableMsg) {
val num = ceiling((availableNonStaleActivations / containerThroughput) - existing - inProgress)
// if it tries to create more containers than existing messages, we just create shortage
val actualNum = if (num + totalContainers > availableNonStaleActivations) availableNonStaleActivations - totalContainers else num
addServersIfPossible(
existing,
inProgress,
containerThroughput,
availableMsg,
capacity,
actualNum + staleContainerProvision,
staleActivationNum,
duration,
Running)
} else if (staleContainerProvision > 0) {
addServersIfPossible(
existing,
inProgress,
containerThroughput,
availableMsg,
capacity,
staleContainerProvision,
staleActivationNum,
duration,
Running)
} else {
Future.successful(DecisionResults(Skip, 0))
}
```
such that if you do need more containers for available msg, you don't need to wait an additional scheduling interval to create them. So take the same example from the description and say that there are 6 existing containers. The second code block will create 1 container for the stale activations and 5 containers for the 100 available messages that are not stale bringing us to the correct 11 capacity in one scheduling interval. If there were already 10 containers available like in the original example and we just need 1 additional container for the stale activations it would enter the `else if` case and just add the 1 container to account for the stale activations to bring it to 11.
--
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
[GitHub] [openwhisk] bdoyle0182 commented on a diff in pull request #5344: optimize scheduling decision when there are stale activations
Posted by GitBox <gi...@apache.org>.
bdoyle0182 commented on code in PR #5344:
URL: https://github.com/apache/openwhisk/pull/5344#discussion_r1010009171
##########
tests/src/test/scala/org/apache/openwhisk/core/scheduler/queue/test/SchedulingDecisionMakerTests.scala:
##########
@@ -683,6 +683,54 @@ class SchedulingDecisionMakerTests
testProbe.expectMsg(DecisionResults(AddContainer, 2))
}
+ it should "add more containers when there are stale messages and non-stale messages and both message classes need more containers" in {
+ val decisionMaker = system.actorOf(SchedulingDecisionMaker.props(testNamespace, action, schedulingConfig))
+ val testProbe = TestProbe()
+
+ val msg = QueueSnapshot(
+ initialized = true,
+ incomingMsgCount = new AtomicInteger(0),
+ currentMsgCount = 5,
+ existingContainerCount = 2,
+ inProgressContainerCount = 0,
+ staleActivationNum = 2,
+ existingContainerCountInNamespace = 2,
+ inProgressContainerCountInNamespace = 0,
+ averageDuration = Some(1000), // the average duration exists
+ limit = 10,
+ stateName = Running,
+ recipient = testProbe.ref)
+
+ decisionMaker ! msg
+
+ //should add two for the stale messages and one to increase tps of non-stale available messages
+ testProbe.expectMsg(DecisionResults(AddContainer, 3))
+ }
+
+ it should "add more containers when there are stale messages and non-stale messages have needed tps" in {
+ val decisionMaker = system.actorOf(SchedulingDecisionMaker.props(testNamespace, action, schedulingConfig))
+ val testProbe = TestProbe()
+
+ val msg = QueueSnapshot(
+ initialized = true,
+ incomingMsgCount = new AtomicInteger(0),
+ currentMsgCount = 5,
+ existingContainerCount = 2,
+ inProgressContainerCount = 0,
+ staleActivationNum = 2,
+ existingContainerCountInNamespace = 2,
+ inProgressContainerCountInNamespace = 0,
+ averageDuration = Some(50), // the average duration gives container throughput of 2
Review Comment:
So should 1s be the proper default `staleThreshold` or remain 100ms?
--
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
[GitHub] [openwhisk] codecov-commenter commented on pull request #5344: optimize scheduling decision when there are stale activations
Posted by GitBox <gi...@apache.org>.
codecov-commenter commented on PR #5344:
URL: https://github.com/apache/openwhisk/pull/5344#issuecomment-1292901981
# [Codecov](https://codecov.io/gh/apache/openwhisk/pull/5344?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) Report
> Merging [#5344](https://codecov.io/gh/apache/openwhisk/pull/5344?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (2ac471b) into [master](https://codecov.io/gh/apache/openwhisk/commit/ef725a653ab112391f79c274d8e3dcfb915d59a3?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (ef725a6) will **decrease** coverage by `76.95%`.
> The diff coverage is `0.00%`.
```diff
@@ Coverage Diff @@
## master #5344 +/- ##
==========================================
- Coverage 81.49% 4.54% -76.96%
==========================================
Files 240 240
Lines 14313 14313
Branches 610 610
==========================================
- Hits 11664 650 -11014
- Misses 2649 13663 +11014
```
| [Impacted Files](https://codecov.io/gh/apache/openwhisk/pull/5344?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) | Coverage Δ | |
|---|---|---|
| [...core/scheduler/queue/SchedulingDecisionMaker.scala](https://codecov.io/gh/apache/openwhisk/pull/5344/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-Y29yZS9zY2hlZHVsZXIvc3JjL21haW4vc2NhbGEvb3JnL2FwYWNoZS9vcGVud2hpc2svY29yZS9zY2hlZHVsZXIvcXVldWUvU2NoZWR1bGluZ0RlY2lzaW9uTWFrZXIuc2NhbGE=) | `0.00% <0.00%> (-97.60%)` | :arrow_down: |
| [.../main/scala/org/apache/openwhisk/core/WarmUp.scala](https://codecov.io/gh/apache/openwhisk/pull/5344/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-Y29tbW9uL3NjYWxhL3NyYy9tYWluL3NjYWxhL29yZy9hcGFjaGUvb3BlbndoaXNrL2NvcmUvV2FybVVwLnNjYWxh) | `0.00% <0.00%> (-100.00%)` | :arrow_down: |
| [...ain/scala/org/apache/openwhisk/spi/SpiLoader.scala](https://codecov.io/gh/apache/openwhisk/pull/5344/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-Y29tbW9uL3NjYWxhL3NyYy9tYWluL3NjYWxhL29yZy9hcGFjaGUvb3BlbndoaXNrL3NwaS9TcGlMb2FkZXIuc2NhbGE=) | `0.00% <0.00%> (-100.00%)` | :arrow_down: |
| [...n/scala/org/apache/openwhisk/utils/JsHelpers.scala](https://codecov.io/gh/apache/openwhisk/pull/5344/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-Y29tbW9uL3NjYWxhL3NyYy9tYWluL3NjYWxhL29yZy9hcGFjaGUvb3BlbndoaXNrL3V0aWxzL0pzSGVscGVycy5zY2FsYQ==) | `0.00% <0.00%> (-100.00%)` | :arrow_down: |
| [...scala/org/apache/openwhisk/common/Prometheus.scala](https://codecov.io/gh/apache/openwhisk/pull/5344/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-Y29tbW9uL3NjYWxhL3NyYy9tYWluL3NjYWxhL29yZy9hcGFjaGUvb3BlbndoaXNrL2NvbW1vbi9Qcm9tZXRoZXVzLnNjYWxh) | `0.00% <0.00%> (-100.00%)` | :arrow_down: |
| [...scala/org/apache/openwhisk/common/time/Clock.scala](https://codecov.io/gh/apache/openwhisk/pull/5344/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-Y29tbW9uL3NjYWxhL3NyYy9tYWluL3NjYWxhL29yZy9hcGFjaGUvb3BlbndoaXNrL2NvbW1vbi90aW1lL0Nsb2NrLnNjYWxh) | `0.00% <0.00%> (-100.00%)` | :arrow_down: |
| [...scala/org/apache/openwhisk/core/FeatureFlags.scala](https://codecov.io/gh/apache/openwhisk/pull/5344/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-Y29tbW9uL3NjYWxhL3NyYy9tYWluL3NjYWxhL29yZy9hcGFjaGUvb3BlbndoaXNrL2NvcmUvRmVhdHVyZUZsYWdzLnNjYWxh) | `0.00% <0.00%> (-100.00%)` | :arrow_down: |
| [...ala/org/apache/openwhisk/common/ConfigMXBean.scala](https://codecov.io/gh/apache/openwhisk/pull/5344/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-Y29tbW9uL3NjYWxhL3NyYy9tYWluL3NjYWxhL29yZy9hcGFjaGUvb3BlbndoaXNrL2NvbW1vbi9Db25maWdNWEJlYW4uc2NhbGE=) | `0.00% <0.00%> (-100.00%)` | :arrow_down: |
| [...la/org/apache/openwhisk/http/BasicRasService.scala](https://codecov.io/gh/apache/openwhisk/pull/5344/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-Y29tbW9uL3NjYWxhL3NyYy9tYWluL3NjYWxhL29yZy9hcGFjaGUvb3BlbndoaXNrL2h0dHAvQmFzaWNSYXNTZXJ2aWNlLnNjYWxh) | `0.00% <0.00%> (-100.00%)` | :arrow_down: |
| [...a/org/apache/openwhisk/common/ConfigMapValue.scala](https://codecov.io/gh/apache/openwhisk/pull/5344/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-Y29tbW9uL3NjYWxhL3NyYy9tYWluL3NjYWxhL29yZy9hcGFjaGUvb3BlbndoaXNrL2NvbW1vbi9Db25maWdNYXBWYWx1ZS5zY2FsYQ==) | `0.00% <0.00%> (-100.00%)` | :arrow_down: |
| ... and [205 more](https://codecov.io/gh/apache/openwhisk/pull/5344/diff?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) | |
:mega: We’re building smart automated test selection to slash your CI/CD build times. [Learn more](https://about.codecov.io/iterative-testing/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation)
--
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
[GitHub] [openwhisk] bdoyle0182 commented on a diff in pull request #5344: optimize scheduling decision when there are stale activations
Posted by GitBox <gi...@apache.org>.
bdoyle0182 commented on code in PR #5344:
URL: https://github.com/apache/openwhisk/pull/5344#discussion_r1010006968
##########
tests/src/test/scala/org/apache/openwhisk/core/scheduler/queue/test/SchedulingDecisionMakerTests.scala:
##########
@@ -683,6 +683,54 @@ class SchedulingDecisionMakerTests
testProbe.expectMsg(DecisionResults(AddContainer, 2))
}
+ it should "add more containers when there are stale messages and non-stale messages and both message classes need more containers" in {
+ val decisionMaker = system.actorOf(SchedulingDecisionMaker.props(testNamespace, action, schedulingConfig))
+ val testProbe = TestProbe()
+
+ val msg = QueueSnapshot(
+ initialized = true,
+ incomingMsgCount = new AtomicInteger(0),
+ currentMsgCount = 5,
+ existingContainerCount = 2,
+ inProgressContainerCount = 0,
+ staleActivationNum = 2,
+ existingContainerCountInNamespace = 2,
+ inProgressContainerCountInNamespace = 0,
+ averageDuration = Some(1000), // the average duration exists
+ limit = 10,
+ stateName = Running,
+ recipient = testProbe.ref)
+
+ decisionMaker ! msg
+
+ //should add two for the stale messages and one to increase tps of non-stale available messages
+ testProbe.expectMsg(DecisionResults(AddContainer, 3))
+ }
+
+ it should "add more containers when there are stale messages and non-stale messages have needed tps" in {
+ val decisionMaker = system.actorOf(SchedulingDecisionMaker.props(testNamespace, action, schedulingConfig))
+ val testProbe = TestProbe()
+
+ val msg = QueueSnapshot(
+ initialized = true,
+ incomingMsgCount = new AtomicInteger(0),
+ currentMsgCount = 5,
+ existingContainerCount = 2,
+ inProgressContainerCount = 0,
+ staleActivationNum = 2,
+ existingContainerCountInNamespace = 2,
+ inProgressContainerCountInNamespace = 0,
+ averageDuration = Some(50), // the average duration gives container throughput of 2
Review Comment:
Isn't the default stale threshold 100ms so 100 / 50 = 2. Is that a mistake should the `staleThreshold` configuration really be 1000ms? It always seemed odd to me that the `staleThreshold` and `checkInterval` were the same value
--
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
[GitHub] [openwhisk] bdoyle0182 commented on a diff in pull request #5344: optimize scheduling decision when there are stale activations
Posted by GitBox <gi...@apache.org>.
bdoyle0182 commented on code in PR #5344:
URL: https://github.com/apache/openwhisk/pull/5344#discussion_r1009833309
##########
core/scheduler/src/main/scala/org/apache/openwhisk/core/scheduler/queue/SchedulingDecisionMaker.scala:
##########
@@ -135,41 +135,44 @@ class SchedulingDecisionMaker(
staleActivationNum,
0.0,
Running)
-
- case (Running, Some(duration)) if staleActivationNum > 0 =>
- // we can safely get the value as we already checked the existence
- val containerThroughput = staleThreshold / duration
- val num = ceiling(availableMsg.toDouble / containerThroughput)
- // if it tries to create more containers than existing messages, we just create shortage
- val actualNum = (if (num > availableMsg) availableMsg else num) - inProgress
- addServersIfPossible(
- existing,
- inProgress,
- containerThroughput,
- availableMsg,
- capacity,
- actualNum,
- staleActivationNum,
- duration,
- Running)
-
// need more containers and a message is already processed
case (Running, Some(duration)) =>
// we can safely get the value as we already checked the existence
val containerThroughput = staleThreshold / duration
val expectedTps = containerThroughput * (existing + inProgress)
+ val availableNonStaleActivations = availableMsg - staleActivationNum
+
+ var staleContainerProvision = 0
+ if (staleActivationNum > 0) {
+ val num = ceiling(staleActivationNum.toDouble / containerThroughput)
+ // if it tries to create more containers than existing messages, we just create shortage
+ staleContainerProvision = (if (num > staleActivationNum) staleActivationNum else num) - inProgress
+ }
- if (availableMsg >= expectedTps && existing + inProgress < availableMsg) {
- val num = ceiling((availableMsg / containerThroughput) - existing - inProgress)
+ if (availableNonStaleActivations >= expectedTps && existing + inProgress < availableMsg) {
+ val num = ceiling((availableNonStaleActivations / containerThroughput) - existing - inProgress)
// if it tries to create more containers than existing messages, we just create shortage
- val actualNum = if (num + totalContainers > availableMsg) availableMsg - totalContainers else num
+ val actualNum =
+ if (num + totalContainers > availableNonStaleActivations) availableNonStaleActivations - totalContainers
+ else num
+ addServersIfPossible(
Review Comment:
Added two test cases to cover 1. when need to add containers both for stale messages and increase tps for non-stale available messages 2. when only need to add container to cover stale messages and tps is still met for non-stale messages.
Would appreciate approval on this one too when you have some time @style95
--
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
[GitHub] [openwhisk] bdoyle0182 merged pull request #5344: optimize scheduling decision when there are stale activations
Posted by GitBox <gi...@apache.org>.
bdoyle0182 merged PR #5344:
URL: https://github.com/apache/openwhisk/pull/5344
--
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
[GitHub] [openwhisk] bdoyle0182 commented on a diff in pull request #5344: optimize scheduling decision when there are stale activations
Posted by GitBox <gi...@apache.org>.
bdoyle0182 commented on code in PR #5344:
URL: https://github.com/apache/openwhisk/pull/5344#discussion_r1010006968
##########
tests/src/test/scala/org/apache/openwhisk/core/scheduler/queue/test/SchedulingDecisionMakerTests.scala:
##########
@@ -683,6 +683,54 @@ class SchedulingDecisionMakerTests
testProbe.expectMsg(DecisionResults(AddContainer, 2))
}
+ it should "add more containers when there are stale messages and non-stale messages and both message classes need more containers" in {
+ val decisionMaker = system.actorOf(SchedulingDecisionMaker.props(testNamespace, action, schedulingConfig))
+ val testProbe = TestProbe()
+
+ val msg = QueueSnapshot(
+ initialized = true,
+ incomingMsgCount = new AtomicInteger(0),
+ currentMsgCount = 5,
+ existingContainerCount = 2,
+ inProgressContainerCount = 0,
+ staleActivationNum = 2,
+ existingContainerCountInNamespace = 2,
+ inProgressContainerCountInNamespace = 0,
+ averageDuration = Some(1000), // the average duration exists
+ limit = 10,
+ stateName = Running,
+ recipient = testProbe.ref)
+
+ decisionMaker ! msg
+
+ //should add two for the stale messages and one to increase tps of non-stale available messages
+ testProbe.expectMsg(DecisionResults(AddContainer, 3))
+ }
+
+ it should "add more containers when there are stale messages and non-stale messages have needed tps" in {
+ val decisionMaker = system.actorOf(SchedulingDecisionMaker.props(testNamespace, action, schedulingConfig))
+ val testProbe = TestProbe()
+
+ val msg = QueueSnapshot(
+ initialized = true,
+ incomingMsgCount = new AtomicInteger(0),
+ currentMsgCount = 5,
+ existingContainerCount = 2,
+ inProgressContainerCount = 0,
+ staleActivationNum = 2,
+ existingContainerCountInNamespace = 2,
+ inProgressContainerCountInNamespace = 0,
+ averageDuration = Some(50), // the average duration gives container throughput of 2
Review Comment:
Isn't the default stale threshold 100ms so 100 / 50 = 2. Is that a mistake should the `staleThreshold` really be 1000ms?
##########
tests/src/test/scala/org/apache/openwhisk/core/scheduler/queue/test/SchedulingDecisionMakerTests.scala:
##########
@@ -683,6 +683,54 @@ class SchedulingDecisionMakerTests
testProbe.expectMsg(DecisionResults(AddContainer, 2))
}
+ it should "add more containers when there are stale messages and non-stale messages and both message classes need more containers" in {
+ val decisionMaker = system.actorOf(SchedulingDecisionMaker.props(testNamespace, action, schedulingConfig))
+ val testProbe = TestProbe()
+
+ val msg = QueueSnapshot(
+ initialized = true,
+ incomingMsgCount = new AtomicInteger(0),
+ currentMsgCount = 5,
+ existingContainerCount = 2,
+ inProgressContainerCount = 0,
+ staleActivationNum = 2,
+ existingContainerCountInNamespace = 2,
+ inProgressContainerCountInNamespace = 0,
+ averageDuration = Some(1000), // the average duration exists
+ limit = 10,
+ stateName = Running,
+ recipient = testProbe.ref)
+
+ decisionMaker ! msg
+
+ //should add two for the stale messages and one to increase tps of non-stale available messages
+ testProbe.expectMsg(DecisionResults(AddContainer, 3))
+ }
+
+ it should "add more containers when there are stale messages and non-stale messages have needed tps" in {
+ val decisionMaker = system.actorOf(SchedulingDecisionMaker.props(testNamespace, action, schedulingConfig))
+ val testProbe = TestProbe()
+
+ val msg = QueueSnapshot(
+ initialized = true,
+ incomingMsgCount = new AtomicInteger(0),
+ currentMsgCount = 5,
+ existingContainerCount = 2,
+ inProgressContainerCount = 0,
+ staleActivationNum = 2,
+ existingContainerCountInNamespace = 2,
+ inProgressContainerCountInNamespace = 0,
+ averageDuration = Some(50), // the average duration gives container throughput of 2
Review Comment:
Isn't the default stale threshold 100ms so 100 / 50 = 2. Is that a mistake should the `staleThreshold` configuration really be 1000ms?
--
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
[GitHub] [openwhisk] style95 commented on a diff in pull request #5344: optimize scheduling decision when there are stale activations
Posted by GitBox <gi...@apache.org>.
style95 commented on code in PR #5344:
URL: https://github.com/apache/openwhisk/pull/5344#discussion_r1007731211
##########
core/scheduler/src/main/scala/org/apache/openwhisk/core/scheduler/queue/SchedulingDecisionMaker.scala:
##########
@@ -135,41 +135,44 @@ class SchedulingDecisionMaker(
staleActivationNum,
0.0,
Running)
-
- case (Running, Some(duration)) if staleActivationNum > 0 =>
- // we can safely get the value as we already checked the existence
- val containerThroughput = staleThreshold / duration
- val num = ceiling(availableMsg.toDouble / containerThroughput)
- // if it tries to create more containers than existing messages, we just create shortage
- val actualNum = (if (num > availableMsg) availableMsg else num) - inProgress
- addServersIfPossible(
- existing,
- inProgress,
- containerThroughput,
- availableMsg,
- capacity,
- actualNum,
- staleActivationNum,
- duration,
- Running)
-
// need more containers and a message is already processed
case (Running, Some(duration)) =>
// we can safely get the value as we already checked the existence
val containerThroughput = staleThreshold / duration
val expectedTps = containerThroughput * (existing + inProgress)
+ val availableNonStaleActivations = availableMsg - staleActivationNum
+
+ var staleContainerProvision = 0
+ if (staleActivationNum > 0) {
+ val num = ceiling(staleActivationNum.toDouble / containerThroughput)
+ // if it tries to create more containers than existing messages, we just create shortage
+ staleContainerProvision = (if (num > staleActivationNum) staleActivationNum else num) - inProgress
+ }
- if (availableMsg >= expectedTps && existing + inProgress < availableMsg) {
- val num = ceiling((availableMsg / containerThroughput) - existing - inProgress)
+ if (availableNonStaleActivations >= expectedTps && existing + inProgress < availableMsg) {
+ val num = ceiling((availableNonStaleActivations / containerThroughput) - existing - inProgress)
// if it tries to create more containers than existing messages, we just create shortage
- val actualNum = if (num + totalContainers > availableMsg) availableMsg - totalContainers else num
+ val actualNum =
+ if (num + totalContainers > availableNonStaleActivations) availableNonStaleActivations - totalContainers
+ else num
+ addServersIfPossible(
Review Comment:
Can we add a unit test to cover this case?
--
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
[GitHub] [openwhisk] bdoyle0182 commented on a diff in pull request #5344: optimize scheduling decision when there are stale activations
Posted by GitBox <gi...@apache.org>.
bdoyle0182 commented on code in PR #5344:
URL: https://github.com/apache/openwhisk/pull/5344#discussion_r1007084325
##########
core/scheduler/src/main/scala/org/apache/openwhisk/core/scheduler/queue/SchedulingDecisionMaker.scala:
##########
@@ -139,9 +139,9 @@ class SchedulingDecisionMaker(
case (Running, Some(duration)) if staleActivationNum > 0 =>
// we can safely get the value as we already checked the existence
val containerThroughput = staleThreshold / duration
- val num = ceiling(availableMsg.toDouble / containerThroughput)
+ val num = ceiling(staleActivationNum.toDouble / containerThroughput)
// if it tries to create more containers than existing messages, we just create shortage
- val actualNum = (if (num > availableMsg) availableMsg else num) - inProgress
+ val actualNum = (if (num > staleActivationNum) staleActivationNum else num) - inProgress
Review Comment:
I think the code in my comment above covers this. I just pushed it so it's the code in the pr now. In the same decision interval, it will both make the calculation for additional containers needed for activations that are stale and increase containers based on # of available messages to meet the needed transactions per second. I think this is the most optimal case.
--
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
[GitHub] [openwhisk] style95 commented on a diff in pull request #5344: optimize scheduling decision when there are stale activations
Posted by GitBox <gi...@apache.org>.
style95 commented on code in PR #5344:
URL: https://github.com/apache/openwhisk/pull/5344#discussion_r1010007263
##########
tests/src/test/scala/org/apache/openwhisk/core/scheduler/queue/test/SchedulingDecisionMakerTests.scala:
##########
@@ -683,6 +683,54 @@ class SchedulingDecisionMakerTests
testProbe.expectMsg(DecisionResults(AddContainer, 2))
}
+ it should "add more containers when there are stale messages and non-stale messages and both message classes need more containers" in {
+ val decisionMaker = system.actorOf(SchedulingDecisionMaker.props(testNamespace, action, schedulingConfig))
+ val testProbe = TestProbe()
+
+ val msg = QueueSnapshot(
+ initialized = true,
+ incomingMsgCount = new AtomicInteger(0),
+ currentMsgCount = 5,
+ existingContainerCount = 2,
+ inProgressContainerCount = 0,
+ staleActivationNum = 2,
+ existingContainerCountInNamespace = 2,
+ inProgressContainerCountInNamespace = 0,
+ averageDuration = Some(1000), // the average duration exists
+ limit = 10,
+ stateName = Running,
+ recipient = testProbe.ref)
+
+ decisionMaker ! msg
+
+ //should add two for the stale messages and one to increase tps of non-stale available messages
+ testProbe.expectMsg(DecisionResults(AddContainer, 3))
+ }
+
+ it should "add more containers when there are stale messages and non-stale messages have needed tps" in {
+ val decisionMaker = system.actorOf(SchedulingDecisionMaker.props(testNamespace, action, schedulingConfig))
+ val testProbe = TestProbe()
+
+ val msg = QueueSnapshot(
+ initialized = true,
+ incomingMsgCount = new AtomicInteger(0),
+ currentMsgCount = 5,
+ existingContainerCount = 2,
+ inProgressContainerCount = 0,
+ staleActivationNum = 2,
+ existingContainerCountInNamespace = 2,
+ inProgressContainerCountInNamespace = 0,
+ averageDuration = Some(50), // the average duration gives container throughput of 2
Review Comment:
oh yes, I got confused.
I used 1s as a threshold.
--
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
[GitHub] [openwhisk] style95 commented on a diff in pull request #5344: optimize scheduling decision when there are stale activations
Posted by GitBox <gi...@apache.org>.
style95 commented on code in PR #5344:
URL: https://github.com/apache/openwhisk/pull/5344#discussion_r1010011304
##########
tests/src/test/scala/org/apache/openwhisk/core/scheduler/queue/test/SchedulingDecisionMakerTests.scala:
##########
@@ -683,6 +683,54 @@ class SchedulingDecisionMakerTests
testProbe.expectMsg(DecisionResults(AddContainer, 2))
}
+ it should "add more containers when there are stale messages and non-stale messages and both message classes need more containers" in {
+ val decisionMaker = system.actorOf(SchedulingDecisionMaker.props(testNamespace, action, schedulingConfig))
+ val testProbe = TestProbe()
+
+ val msg = QueueSnapshot(
+ initialized = true,
+ incomingMsgCount = new AtomicInteger(0),
+ currentMsgCount = 5,
+ existingContainerCount = 2,
+ inProgressContainerCount = 0,
+ staleActivationNum = 2,
+ existingContainerCountInNamespace = 2,
+ inProgressContainerCountInNamespace = 0,
+ averageDuration = Some(1000), // the average duration exists
+ limit = 10,
+ stateName = Running,
+ recipient = testProbe.ref)
+
+ decisionMaker ! msg
+
+ //should add two for the stale messages and one to increase tps of non-stale available messages
+ testProbe.expectMsg(DecisionResults(AddContainer, 3))
+ }
+
+ it should "add more containers when there are stale messages and non-stale messages have needed tps" in {
+ val decisionMaker = system.actorOf(SchedulingDecisionMaker.props(testNamespace, action, schedulingConfig))
+ val testProbe = TestProbe()
+
+ val msg = QueueSnapshot(
+ initialized = true,
+ incomingMsgCount = new AtomicInteger(0),
+ currentMsgCount = 5,
+ existingContainerCount = 2,
+ inProgressContainerCount = 0,
+ staleActivationNum = 2,
+ existingContainerCountInNamespace = 2,
+ inProgressContainerCountInNamespace = 0,
+ averageDuration = Some(50), // the average duration gives container throughput of 2
Review Comment:
No, I used to calculate the throughput based on 1s. It's not related to the systems default.
##########
tests/src/test/scala/org/apache/openwhisk/core/scheduler/queue/test/SchedulingDecisionMakerTests.scala:
##########
@@ -683,6 +683,54 @@ class SchedulingDecisionMakerTests
testProbe.expectMsg(DecisionResults(AddContainer, 2))
}
+ it should "add more containers when there are stale messages and non-stale messages and both message classes need more containers" in {
+ val decisionMaker = system.actorOf(SchedulingDecisionMaker.props(testNamespace, action, schedulingConfig))
+ val testProbe = TestProbe()
+
+ val msg = QueueSnapshot(
+ initialized = true,
+ incomingMsgCount = new AtomicInteger(0),
+ currentMsgCount = 5,
+ existingContainerCount = 2,
+ inProgressContainerCount = 0,
+ staleActivationNum = 2,
+ existingContainerCountInNamespace = 2,
+ inProgressContainerCountInNamespace = 0,
+ averageDuration = Some(1000), // the average duration exists
+ limit = 10,
+ stateName = Running,
+ recipient = testProbe.ref)
+
+ decisionMaker ! msg
+
+ //should add two for the stale messages and one to increase tps of non-stale available messages
+ testProbe.expectMsg(DecisionResults(AddContainer, 3))
+ }
+
+ it should "add more containers when there are stale messages and non-stale messages have needed tps" in {
+ val decisionMaker = system.actorOf(SchedulingDecisionMaker.props(testNamespace, action, schedulingConfig))
+ val testProbe = TestProbe()
+
+ val msg = QueueSnapshot(
+ initialized = true,
+ incomingMsgCount = new AtomicInteger(0),
+ currentMsgCount = 5,
+ existingContainerCount = 2,
+ inProgressContainerCount = 0,
+ staleActivationNum = 2,
+ existingContainerCountInNamespace = 2,
+ inProgressContainerCountInNamespace = 0,
+ averageDuration = Some(50), // the average duration gives container throughput of 2
Review Comment:
No, I used to calculate the throughput based on 1s. It's not related to the system default.
--
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