Commit b9881e7c authored by Dominik Charousset's avatar Dominik Charousset

Merge branch 'Lingxi-Li-topic/work_stealing' into topic/actor-system

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