Commit 3c3f268a authored by Dominik Charousset's avatar Dominik Charousset

Add new split/join actor pool policy

parent de6e9a2d
...@@ -30,6 +30,7 @@ ...@@ -30,6 +30,7 @@
#include "caf/abstract_actor.hpp" #include "caf/abstract_actor.hpp"
#include "caf/mailbox_element.hpp" #include "caf/mailbox_element.hpp"
#include "caf/detail/split_join.hpp"
#include "caf/detail/shared_spinlock.hpp" #include "caf/detail/shared_spinlock.hpp"
namespace caf { namespace caf {
...@@ -101,6 +102,21 @@ class actor_pool : public abstract_actor { ...@@ -101,6 +102,21 @@ class actor_pool : public abstract_actor {
std::random_device m_rd; std::random_device m_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>
static policy split_join(Join jf, Split sf = Split(), T init = T()) {
using impl = detail::split_join<T, Split, Join>;
return impl{std::move(init), std::move(sf), std::move(jf)};
}
~actor_pool(); ~actor_pool();
/** /**
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2015 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#ifndef CAF_DETAIL_SPLIT_JOIN_HPP
#define CAF_DETAIL_SPLIT_JOIN_HPP
#include <vector>
#include "caf/actor.hpp"
#include "caf/locks.hpp"
#include "caf/spawn.hpp"
#include "caf/event_based_actor.hpp"
#include "caf/detail/shared_spinlock.hpp"
namespace caf {
namespace detail {
using actor_msg_vec = std::vector<std::pair<actor, message>>;
template <class T, class Split, class Join>
class split_join_collector : public event_based_actor {
public:
split_join_collector(T init_value, Split s, Join j, actor_msg_vec xs)
: m_workset(std::move(xs)),
m_awaited_results(m_workset.size()),
m_join(std::move(j)),
m_split(std::move(s)),
m_value(std::move(init_value)) {
// nop
}
behavior make_behavior() override {
return {
others >> [=] {
m_rp = this->make_response_promise();
m_split(m_workset, this->current_message());
// first message is the forwarded request
for (auto& x : m_workset) {
this->sync_send(x.first, std::move(x.second)).then(
others >> [=] {
m_join(m_value, this->current_message());
if (--m_awaited_results == 0) {
m_rp.deliver(make_message(m_value));
quit();
}
}
);
}
// no longer needed
m_workset.clear();
}
};
}
private:
actor_msg_vec m_workset;
size_t m_awaited_results;
Join m_join;
Split m_split;
T m_value;
response_promise m_rp;
};
struct nop_split {
inline void operator()(actor_msg_vec& xs, message& y) const {
for (auto& x : xs) {
x.second = y;
}
}
};
template <class T, class Split, class Join>
class split_join {
public:
split_join(T init_value, Split s, Join j)
: m_init(std::move(init_value)),
m_sf(std::move(s)),
m_jf(std::move(j)) {
// nop
}
void operator()(upgrade_lock<detail::shared_spinlock>& ulock,
const std::vector<actor>& workers,
mailbox_element_ptr& ptr,
execution_unit* host) {
if (ptr->sender == invalid_actor_addr) {
return;
}
actor_msg_vec xs(workers.size());
for (size_t i = 0; i < workers.size(); ++i) {
xs[i].first = workers[i];
}
ulock.unlock();
using collector_t = split_join_collector<T, Split, Join>;
auto hdl = spawn<collector_t, lazy_init>(m_init, m_sf, m_jf, std::move(xs));
hdl->enqueue(std::move(ptr), host);
}
private:
T m_init;
Split m_sf; // split function
Join m_jf; // join function
};
} // namespace detail
} // namespace caf
#endif // CAF_DETAIL_SPLIT_JOIN_HPP
...@@ -172,11 +172,50 @@ void test_random_actor_pool() { ...@@ -172,11 +172,50 @@ void test_random_actor_pool() {
self->await_all_other_actors_done(); self->await_all_other_actors_done();
} }
void test_split_join_actor_pool() {
CAF_CHECKPOINT();
auto spawn_split_worker = [] {
return spawn<lazy_init>([]() -> behavior {
return {
[](size_t pos, std::vector<int> xs) {
return xs[pos];
}
};
});
};
auto split_fun = [](std::vector<std::pair<actor, message>>& xs, message& y) {
for (size_t i = 0; i < xs.size(); ++i) {
xs[i].second = make_message(i) + y;
}
};
auto join_fun = [](int& res, message& msg) {
msg.apply([&](int x) {
res += x;
});
};
scoped_actor self;
auto w = actor_pool::make(5, spawn_split_worker,
actor_pool::split_join<int>(join_fun, split_fun));
self->sync_send(w, std::vector<int>{1, 2, 3, 4, 5}).await(
[&](int res) {
CAF_CHECK_EQUAL(res, 15);
}
);
self->sync_send(w, std::vector<int>{6, 7, 8, 9, 10}).await(
[&](int res) {
CAF_CHECK_EQUAL(res, 40);
}
);
self->send_exit(w, exit_reason::user_shutdown);
self->await_all_other_actors_done();
}
int main() { int main() {
CAF_TEST(test_actor_pool); CAF_TEST(test_actor_pool);
test_actor_pool(); test_actor_pool();
test_broadcast_actor_pool(); test_broadcast_actor_pool();
test_random_actor_pool(); test_random_actor_pool();
test_split_join_actor_pool();
await_all_actors_done(); await_all_actors_done();
shutdown(); shutdown();
CAF_CHECK_EQUAL(s_dtors.load(), s_ctors.load()); CAF_CHECK_EQUAL(s_dtors.load(), s_ctors.load());
......
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