Commit 9f5db87e authored by Dominik Charousset's avatar Dominik Charousset

Consistently use policy factories in actor pool

parent d8b30f1c
...@@ -20,8 +20,6 @@ ...@@ -20,8 +20,6 @@
#ifndef CAF_ACTOR_POOL_HPP #ifndef CAF_ACTOR_POOL_HPP
#define CAF_ACTOR_POOL_HPP #define CAF_ACTOR_POOL_HPP
#include <atomic>
#include <random>
#include <vector> #include <vector>
#include <functional> #include <functional>
...@@ -63,44 +61,27 @@ public: ...@@ -63,44 +61,27 @@ public:
using policy = std::function<void (uplock&, const actor_vec&, using policy = std::function<void (uplock&, const actor_vec&,
mailbox_element_ptr&, execution_unit*)>; mailbox_element_ptr&, execution_unit*)>;
/// Default policy class implementing simple round robin dispatching. /// Returns a simple round robin dispatching policy.
class round_robin { static policy round_robin();
public:
round_robin(); /// Returns a broadcast dispatching policy.
round_robin(const round_robin&); static policy broadcast();
void operator()(uplock&, const actor_vec&,
mailbox_element_ptr&, execution_unit*); /// Returns a random dispatching policy.
static policy random();
private:
std::atomic<size_t> pos_; /// Returns a split/join dispatching policy. The function object `sf`
}; /// distributes a work item to all workers (split step) and the function
/// object `jf` joins individual results into a single one with `init`
/// Default policy class implementing broadcast dispatching. /// as initial value of the operation.
class broadcast { /// @tparam T Result type of the join step.
public: /// @tparam Join Function object with signature `void (T&, message&)`.
void operator()(uplock&, const actor_vec&, /// @tparam Split Function object with signature
mailbox_element_ptr&, execution_unit*); /// `void (vector<pair<actor, message>>&, message&)`. The first
}; /// argument is a mapping from actors (workers) to tasks
/// (messages). The second argument is the input message.
/// Default policy class implementing random dispatching. /// The default split policy broadcasts the work item to all
class random { /// workers.
public:
random();
random(const random&);
void operator()(uplock&, const actor_vec&,
mailbox_element_ptr&, execution_unit*);
private:
std::random_device rd_;
};
/// Default policy class implementing broadcast dispatching (split step)
/// followed by a join operation `F` combining all individual results to
/// a single result of type `T`.
/// @tparam T Result type received by the original sender.
/// @tparam Join Functor with signature `void (T&, message&)`.
/// @tparam Split Functor with signature
/// `void (vector<pair<actor, message>>&, message&)`.
template <class T, class Join, class Split = detail::nop_split> template <class T, class Join, class Split = detail::nop_split>
static policy split_join(Join jf, Split sf = Split(), T init = T()) { static policy split_join(Join jf, Split sf = Split(), T init = T()) {
using impl = detail::split_join<T, Split, Join>; using impl = detail::split_join<T, Split, Join>;
......
...@@ -19,6 +19,9 @@ ...@@ -19,6 +19,9 @@
#include "caf/actor_pool.hpp" #include "caf/actor_pool.hpp"
#include <atomic>
#include <random>
#include "caf/send.hpp" #include "caf/send.hpp"
#include "caf/default_attachable.hpp" #include "caf/default_attachable.hpp"
...@@ -26,49 +29,63 @@ ...@@ -26,49 +29,63 @@
namespace caf { namespace caf {
actor_pool::round_robin::round_robin() : pos_(0) { actor_pool::policy actor_pool::round_robin() {
// nop struct impl {
} impl() : pos_(0) {
// nop
actor_pool::round_robin::round_robin(const round_robin&) : pos_(0) { }
// nop impl(const impl&) : pos_(0) {
// nop
}
void operator()(uplock& guard, const actor_vec& vec,
mailbox_element_ptr& ptr, execution_unit* host) {
CAF_ASSERT(!vec.empty());
actor selected = vec[pos_++ % vec.size()];
guard.unlock();
selected->enqueue(std::move(ptr), host);
}
std::atomic<size_t> pos_;
};
return impl{};
} }
void actor_pool::round_robin::operator()(uplock& guard, const actor_vec& vec, namespace {
mailbox_element_ptr& ptr,
execution_unit* host) {
CAF_ASSERT(! vec.empty());
actor selected = vec[pos_++ % vec.size()];
guard.unlock();
selected->enqueue(std::move(ptr), host);
}
void actor_pool::broadcast::operator()(uplock&, const actor_vec& vec, void broadcast_dispatch(actor_pool::uplock&, const actor_pool::actor_vec& vec,
mailbox_element_ptr& ptr, mailbox_element_ptr& ptr, execution_unit* host) {
execution_unit* host) { CAF_ASSERT(!vec.empty());
CAF_ASSERT(! vec.empty());
for (size_t i = 1; i < vec.size(); ++i) { for (size_t i = 1; i < vec.size(); ++i) {
vec[i]->enqueue(ptr->sender, ptr->mid, ptr->msg, host); vec[i]->enqueue(ptr->sender, ptr->mid, ptr->msg, host);
} }
vec.front()->enqueue(std::move(ptr), host); vec.front()->enqueue(std::move(ptr), host);
} }
actor_pool::random::random() { } // namespace <anonymous>
// nop
}
actor_pool::random::random(const random&) { actor_pool::policy actor_pool::broadcast() {
// nop return broadcast_dispatch;
} }
void actor_pool::random::operator()(uplock& guard, const actor_vec& vec,
mailbox_element_ptr& ptr, actor_pool::policy actor_pool::random() {
execution_unit* host) { struct impl {
std::uniform_int_distribution<size_t> dis(0, vec.size() - 1); impl() {
upgrade_to_unique_lock<detail::shared_spinlock> unique_guard{guard}; // nop
actor selected = vec[dis(rd_)]; }
unique_guard.unlock(); impl(const impl&) {
selected->enqueue(std::move(ptr), host); // nop
}
void operator()(uplock& guard, const actor_vec& vec,
mailbox_element_ptr& ptr, execution_unit* host) {
std::uniform_int_distribution<size_t> dis(0, vec.size() - 1);
upgrade_to_unique_lock<detail::shared_spinlock> unique_guard{guard};
actor selected = vec[dis(rd_)];
unique_guard.unlock();
selected->enqueue(std::move(ptr), host);
}
std::random_device rd_;
};
return impl{};
} }
actor_pool::~actor_pool() { actor_pool::~actor_pool() {
......
...@@ -73,7 +73,7 @@ CAF_TEST_FIXTURE_SCOPE(actor_pool_tests, fixture) ...@@ -73,7 +73,7 @@ CAF_TEST_FIXTURE_SCOPE(actor_pool_tests, fixture)
CAF_TEST(round_robin_actor_pool) { CAF_TEST(round_robin_actor_pool) {
scoped_actor self; scoped_actor self;
auto w = actor_pool::make(5, spawn_worker, actor_pool::round_robin{}); auto w = actor_pool::make(5, spawn_worker, actor_pool::round_robin());
self->monitor(w); self->monitor(w);
self->send(w, sys_atom::value, put_atom::value, spawn_worker()); self->send(w, sys_atom::value, put_atom::value, spawn_worker());
std::vector<actor_addr> workers; std::vector<actor_addr> workers;
...@@ -148,9 +148,9 @@ CAF_TEST(round_robin_actor_pool) { ...@@ -148,9 +148,9 @@ CAF_TEST(round_robin_actor_pool) {
CAF_TEST(broadcast_actor_pool) { CAF_TEST(broadcast_actor_pool) {
scoped_actor self; scoped_actor self;
auto spawn5 = []() { auto spawn5 = []() {
return actor_pool::make(5, spawn_worker, actor_pool::broadcast{}); return actor_pool::make(5, spawn_worker, actor_pool::broadcast());
}; };
auto w = actor_pool::make(5, spawn5, actor_pool::broadcast{}); auto w = actor_pool::make(5, spawn5, actor_pool::broadcast());
self->send(w, 1, 2); self->send(w, 1, 2);
std::vector<int> results; std::vector<int> results;
int i = 0; int i = 0;
...@@ -170,7 +170,7 @@ CAF_TEST(broadcast_actor_pool) { ...@@ -170,7 +170,7 @@ CAF_TEST(broadcast_actor_pool) {
CAF_TEST(random_actor_pool) { CAF_TEST(random_actor_pool) {
scoped_actor self; scoped_actor self;
auto w = actor_pool::make(5, spawn_worker, actor_pool::random{}); auto w = actor_pool::make(5, spawn_worker, actor_pool::random());
for (int i = 0; i < 5; ++i) { for (int i = 0; i < 5; ++i) {
self->sync_send(w, 1, 2).await( self->sync_send(w, 1, 2).await(
[&](int res) { [&](int res) {
......
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