Commit fe717fe8 authored by massimotorquati's avatar massimotorquati

Modify relaxed work-stealing to reduce latency

This patch modifies the work-stealing scheduler to let the runtime
worker threads sleep on a wait_for call instead of using a sleep_for
call, only involving the relaxed strategy.
The relaxed-sleep-duration value is used as the timeout value for
the wait_for call.
This patch improves message latency at low message rates without
introducing significant extra overheads at high message rates.
parent 24368f18
...@@ -25,6 +25,8 @@ ...@@ -25,6 +25,8 @@
#include <thread> #include <thread>
#include <random> #include <random>
#include <cstddef> #include <cstddef>
#include <mutex>
#include <condition_variable>
#include "caf/resumable.hpp" #include "caf/resumable.hpp"
#include "caf/actor_system_config.hpp" #include "caf/actor_system_config.hpp"
...@@ -55,6 +57,13 @@ public: ...@@ -55,6 +57,13 @@ public:
usec sleep_duration; usec sleep_duration;
}; };
// what is needed to implement the waiting strategy.
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. // The coordinator has only a counter for round-robin enqueue to its workers.
struct coordinator_data { struct coordinator_data {
inline explicit coordinator_data(scheduler::abstract_coordinator*) inline explicit coordinator_data(scheduler::abstract_coordinator*)
...@@ -92,6 +101,7 @@ public: ...@@ -92,6 +101,7 @@ public:
std::default_random_engine rengine; std::default_random_engine rengine;
std::uniform_int_distribution<size_t> uniform; std::uniform_int_distribution<size_t> uniform;
poll_strategy strategies[3]; poll_strategy strategies[3];
wait_strategy waitdata;
}; };
// Goes on a raid in quest for a shiny new job. // Goes on a raid in quest for a shiny new job.
...@@ -119,6 +129,14 @@ public: ...@@ -119,6 +129,14 @@ public:
template <class Worker> template <class Worker>
void external_enqueue(Worker* self, resumable* job) { void external_enqueue(Worker* self, resumable* job) {
d(self).queue.append(job); d(self).queue.append(job);
auto& lock = d(self).waitdata.lock;
auto& cv = d(self).waitdata.cv;
{ // guard scope
std::unique_lock<std::mutex> guard(lock);
// check if the worker is sleeping
if (d(self).waitdata.sleeping && !d(self).queue.empty() )
cv.notify_one();
}
} }
template <class Worker> template <class Worker>
...@@ -138,30 +156,52 @@ public: ...@@ -138,30 +156,52 @@ public:
// we wait for new jobs by polling our external queue: first, we // we wait for new jobs by polling our external queue: first, we
// assume an active work load on the machine and perform aggresive // assume an active work load on the machine and perform aggresive
// polling, then we relax our polling a bit and wait 50 us between // polling, then we relax our polling a bit and wait 50 us between
// dequeue attempts, finally we assume pretty much nothing is going // dequeue attempts
// 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
auto& strategies = d(self).strategies; auto& strategies = d(self).strategies;
resumable* job = nullptr; resumable* job = nullptr;
for (auto& strat : strategies) { for (int k = 0; k < 2; ++k) { // iterate over the first two strategies
for (size_t i = 0; i < strat.attempts; i += strat.step_size) { for (size_t i = 0; i < strategies[k].attempts; i += strategies[k].step_size) {
job = d(self).queue.take_head(); job = d(self).queue.take_head();
if (job) if (job)
return job; return job;
// try to steal every X poll attempts // try to steal every X poll attempts
if ((i % strat.steal_interval) == 0) { if ((i % strategies[k].steal_interval) == 0) {
job = try_steal(self); job = try_steal(self);
if (job) if (job)
return job; return job;
} }
if (strat.sleep_duration.count() > 0) if (strategies[k].sleep_duration.count() > 0)
std::this_thread::sleep_for(strat.sleep_duration); std::this_thread::sleep_for(strategies[k].sleep_duration);
} }
} }
// unreachable, because the last strategy loops // we assume pretty much nothing is going on so we can relax polling
// until a job has been dequeued // and falling to sleep on a condition variable whose timeout is the one
return nullptr; // 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> 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