Commit c417ddd7 authored by Sebastian Woelke's avatar Sebastian Woelke

Add low latency scheduler

parent ec6430b8
......@@ -25,6 +25,8 @@
#include <thread>
#include <random>
#include <cstddef>
#include <mutex>
#include <condition_variable>
#include "caf/resumable.hpp"
#include "caf/actor_system_config.hpp"
......@@ -55,13 +57,18 @@ public:
usec sleep_duration;
};
struct wait_strategy {
std::mutex lock;
std::condition_variable cv;
bool sleeping{false};
};
// The coordinator has only a counter for round-robin enqueue to its workers.
struct coordinator_data {
inline explicit coordinator_data(scheduler::abstract_coordinator*)
: next_worker(0) {
// nop
}
std::atomic<size_t> next_worker;
};
......@@ -92,6 +99,7 @@ public:
std::default_random_engine rengine;
std::uniform_int_distribution<size_t> uniform;
poll_strategy strategies[3];
wait_strategy waitdata;
};
// Goes on a raid in quest for a shiny new job.
......@@ -119,6 +127,13 @@ public:
template <class Worker>
void external_enqueue(Worker* self, resumable* job) {
d(self).queue.append(job);
auto& lock = d(self).waitdata.lock;
auto& cv = d(self).waitdata.cv;
{
std::unique_lock<std::mutex> guard(lock);
if (d(self).waitdata.sleeping && !d(self).queue.empty() )
cv.notify_one();
}
}
template <class Worker>
......@@ -138,30 +153,53 @@ public:
// we wait for new jobs by polling our external queue: first, we
// assume an active work load on the machine and perform aggresive
// polling, then we relax our polling a bit and wait 50 us between
// dequeue attempts, finally we assume pretty much nothing is going
// on and poll every 10 ms; this strategy strives to minimize the
// downside of "busy waiting", which still performs much better than a
// "signalizing" implementation based on mutexes and conition variables
// dequeue attempts
auto& strategies = d(self).strategies;
resumable* job = nullptr;
for (auto& strat : strategies) {
for (size_t i = 0; i < strat.attempts; i += strat.step_size) {
for (int k = 0; k < 2; ++k) { // iterate over the first two strategies
for (size_t i = 0; i < strategies[k].attempts;
i += strategies[k].step_size) {
job = d(self).queue.take_head();
if (job)
return job;
// try to steal every X poll attempts
if ((i % strat.steal_interval) == 0) {
if ((i % strategies[k].steal_interval) == 0) {
job = try_steal(self);
if (job)
return job;
}
if (strat.sleep_duration.count() > 0)
std::this_thread::sleep_for(strat.sleep_duration);
if (strategies[k].sleep_duration.count() > 0)
std::this_thread::sleep_for(strategies[k].sleep_duration);
}
}
// unreachable, because the last strategy loops
// until a job has been dequeued
return nullptr;
// we assume pretty much nothing is going on so we can relax polling
// and falling to sleep on a condition variable whose timeout is the one
// of the relaxed polling strategy
auto& relaxed = strategies[2];
auto& sleeping = d(self).waitdata.sleeping;
auto& lock = d(self).waitdata.lock;
auto& cv = d(self).waitdata.cv;
bool notimeout = true;
size_t i=1;
do {
{ // guard scope
std::unique_lock<std::mutex> guard(lock);
sleeping=true;
if (!cv.wait_for(guard, relaxed.sleep_duration,
[&] { return !d(self).queue.empty(); }))
notimeout = false;
sleeping=false;
}
if (notimeout) {
job = d(self).queue.take_head();
} else {
notimeout=true;
if ((i % relaxed.steal_interval) == 0)
job = try_steal(self);
}
++i;
} while(job == nullptr);
return job;
}
template <class Worker, class UnaryFunction>
......
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment