-
Notifications
You must be signed in to change notification settings - Fork 1.2k
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
[New Scheduler] Add memory queue for the new scheduler #5110
Conversation
According to the order defined in the wiki, this component requires other components to be merged. But there is no big difference in the memory queue logic, so I opened it for reviews. Since this logic is the core logic and changes on this component are highly likely to introduce huge impacts on the system, I will write down a comprehensive design document. Once dependent modules are merged into the master branch, I would rebase this PR. |
* | ||
* @param maxSize the maximum size of the buffer | ||
*/ | ||
class AverageRingBuffer(private val maxSize: Int) { |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
This circular buffer is used to calculate the average execution time of recent N activations for a given action.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
What was the reasoning for picking average as the heuristic. Would median be a better heuristic here? i.e. if all activations take 100 milliseconds and then one activation has a slow call to a db that takes 10 seconds it will heavily skew the average.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Yes, that makes sense.
The ring size in our case was relatively small such as 10, so we calculated the average of recent 10 activations. Even if there was a skew at some point, it quickly gets back to the "normal" average if the slow activation is a transient issue.
Also, it only affects the timing to add more containers, and activations are still being processed by existing containers. So there was no critical impact.
Basically, the median would be better, but the average is a much simpler and cheaper solution so I chose it.
import scala.concurrent.{ExecutionContext, Future} | ||
import scala.util.{Failure, Success} | ||
|
||
class SchedulingDecisionMaker( |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
This component decides whether to add more containers or not for the given action.
It uses the average execution time of an action.
There are three cases to calculate the average execution.
-
Initial execution.
In this case, we just assume the execution time as tick time(100ms by default). -
Queue is newly created.
In this case, activations exist in the DB, but no data in the AverageRingBuffer.
So we fetch the average execution time from DB. -
An action is continuously executed.
Each container sends the last execution time along withFetchRequest
when it pulls a new activation from a queue.
And our scheduler keeps N execution times in the AverageRingBuffer.
So we can easily calculate the average execution time of recent N executions.
Codecov Report
@@ Coverage Diff @@
## master #5110 +/- ##
===========================================
+ Coverage 43.88% 74.60% +30.72%
===========================================
Files 231 234 +3
Lines 12807 13381 +574
Branches 528 513 -15
===========================================
+ Hits 5620 9983 +4363
+ Misses 7187 3398 -3789
Continue to review full report at Codecov.
|
common/scala/src/main/scala/org/apache/openwhisk/common/AverageRingBuffer.scala
Show resolved
Hide resolved
It is better to add below configuration to scheduler's application.conf, e.g.
Or when open another new pr which include ansible/roles/schedulers/tasks/deploy.yml, add there. |
core/scheduler/src/main/scala/org/apache/openwhisk/core/scheduler/queue/MemoryQueue.scala
Show resolved
Hide resolved
.travis.yml
Outdated
@@ -27,6 +27,7 @@ env: | |||
global: | |||
- ANSIBLE_CMD="ansible-playbook -i environments/local -e docker_image_prefix=testing" | |||
- GRADLE_PROJS_SKIP="" | |||
- OW_SCALA_VERSION=2.13 |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Great!
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
The scala compilation version was not the culprit.
Need rebase. |
case Event(msg: DecisionResults, _) => | ||
val DecisionResults(result, num) = msg | ||
result match { | ||
case AddInitialContainer if num > 0 => |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Should this be if num = 0
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
AFAIK, there is no case where SchedulingDecisionMaker sends AddInitialContainer
with num == 0
?
takeUncompletedRequest() | ||
.map { res => | ||
res.trySuccess(Right(msg)) | ||
in -= 1 |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Pretty sure this isn't thread safe if takeUncompletedRequest
is a promise
} | ||
} | ||
|
||
private def addServersIfPossible(existing: Int, |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Should this be called addContainersIfPossible
This is probably the most complicated PR, but did the best I could to understand at a high level I think LGTM apart from my comments. One comment on the persistence nature of the queue. I think queueing with kafka doesn't add anything to openwhisk if openwhisk does not provide at least once guarantee of activation execution. It's unnecessary overhead to queue up through kafka if the activation can get lost elsewhere anyways. |
364cc8e
to
65cd5dc
Compare
Agree. I think the immediate next step could be removing Kafka from the critical path completely as currently, container creation messages and activation messages from controller to scheduler are delivered through Kafka.
I mentioned above that if we really need to support the at-least-once semantic, we can consider this approach. |
After rebasing this PR, it started failing to run compile |
It’s ready to merge. |
I will merge this in 48 hours. |
Description
This PR adds one of the core implementations of the new scheduler.
This queue is dynamically created for each action and buffer activations in an internal queue.
It provisions more containers according to the number of activations in the queue.
This process is called "scheduling" and it is handled by another component
SchedulingDecisionMaker
.The queue communicates with
ContainerManager
to add more containers.ContainerManager
selects proper invokers and sends container creation messages to them.This queue is designed in a way that activations are not persistent.
But it could be improved in the future by taking a similar approach with what Kafka does.
My changes affect the following components
Types of changes
Checklist: