Commit 4e7c04bb authored by Lingxi-Li's avatar Lingxi-Li

Optimize work stealing performance

parent dc7e3f4f
......@@ -21,6 +21,7 @@
#define CAF_POLICY_SCHEDULER_POLICY_HPP
#include "caf/fwd.hpp"
#include "caf/scheduler/abstract_coordinator.hpp"
namespace caf {
namespace policy {
......@@ -30,10 +31,14 @@ namespace policy {
class scheduler_policy {
public:
/// Policy-specific data fields for the coordinator.
struct coordinator_data { };
struct coordinator_data {
explicit coordinator_data(scheduler::abstract_coordinator*);
};
/// Policy-specific data fields for the worker.
struct worker_data { };
struct worker_data {
explicit worker_data(scheduler::abstract_coordinator*);
};
/// Enqueues a new job to coordinator.
template <class Coordinator>
......
......@@ -20,6 +20,8 @@
#ifndef CAF_POLICY_UNPROFILED_HPP
#define CAF_POLICY_UNPROFILED_HPP
#include "caf/scheduler/abstract_coordinator.hpp"
namespace caf {
namespace policy {
......@@ -33,10 +35,18 @@ public:
virtual ~unprofiled() = default;
/// Policy-specific data fields for the coordinator.
struct coordinator_data { };
struct coordinator_data {
inline explicit coordinator_data(scheduler::abstract_coordinator*) {
// nop
}
};
/// Policy-specific data fields for the worker.
struct worker_data { };
struct worker_data {
inline explicit worker_data(scheduler::abstract_coordinator*) {
// nop
}
};
/// Performs cleanup action before a shutdown takes place.
template <class Worker>
......
......@@ -38,6 +38,10 @@ public:
using queue_type = std::list<resumable*>;
struct coordinator_data {
inline explicit coordinator_data(scheduler::abstract_coordinator*) {
// nop
}
queue_type queue;
std::mutex lock;
std::condition_variable cv;
......
......@@ -57,23 +57,30 @@ public:
// The coordinator has only a counter for round-robin enqueue to its workers.
struct coordinator_data {
std::atomic<size_t> next_worker;
inline coordinator_data() : next_worker(0) {
inline explicit coordinator_data(scheduler::abstract_coordinator*)
: next_worker(0) {
// nop
}
std::atomic<size_t> next_worker;
};
// Holds job job queue of a worker and a random number generator.
struct worker_data {
inline explicit worker_data(scheduler::abstract_coordinator* p)
: rengine(std::random_device{}()),
// no need to worry about wrap-around; if `p->num_workers() < 2`,
// `uniform` will not be used anyway
uniform(0, p->num_workers() - 2) {
// nop
}
// This queue is exposed to other workers that may attempt to steal jobs
// from it and the central scheduling unit can push new jobs to the queue.
queue_type queue;
// needed to generate pseudo random numbers
std::default_random_engine rengine;
// initialize random engine
inline worker_data() : rengine(std::random_device{}()) {
// nop
}
std::uniform_int_distribution<size_t> uniform;
};
// Goes on a raid in quest for a shiny new job.
......@@ -85,7 +92,7 @@ public:
return nullptr;
}
// roll the dice to pick a victim other than ourselves
size_t victim = d(self).rengine() % (p->num_workers() - 1);
auto victim = d(self).uniform(d(self).rengine);
if (victim == self->id())
victim = p->num_workers() - 1;
// steal oldest element from the victim's queue
......
......@@ -41,7 +41,7 @@ public:
using policy_data = typename Policy::coordinator_data;
coordinator(actor_system& sys) : super(sys) {
coordinator(actor_system& sys) : super(sys), data_(this) {
// nop
}
......
......@@ -46,7 +46,8 @@ public:
: execution_unit(&worker_parent->system()),
max_throughput_(throughput),
id_(worker_id),
parent_(worker_parent) {
parent_(worker_parent),
data_(worker_parent) {
// nop
}
......
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