Commit d2909f6a authored by Dominik Charousset's avatar Dominik Charousset

Prototype new fan_in policy for collecting results

parent 043e740d
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2019 Dominik Charousset *
* *
* 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. *
******************************************************************************/
#pragma once
#include <cstddef>
#include <functional>
#include <limits>
#include <memory>
#include <vector>
#include "caf/behavior.hpp"
#include "caf/config.hpp"
#include "caf/detail/type_list.hpp"
#include "caf/detail/type_traits.hpp"
#include "caf/detail/typed_actor_util.hpp"
#include "caf/error.hpp"
#include "caf/message_id.hpp"
namespace caf {
namespace detail {
template <class F, class T>
struct fan_in_helper {
std::vector<T> results;
std::shared_ptr<size_t> pending;
F f;
void operator()(T& x) {
if (*pending > 0) {
results.emplace_back(std::move(x));
if (--*pending == 0)
f(std::move(results));
}
}
template <class Fun>
fan_in_helper(size_t pending, Fun&& f)
: pending(std::make_shared<size_t>(pending)), f(std::forward<Fun>(f)) {
results.reserve(pending);
}
// TODO: return 'auto' from this function when switching to C++17
std::function<void(T&)> wrap() {
return [this](T& x) { (*this)(x); };
}
};
template <class F, class... Ts>
struct fan_in_tuple_helper {
using value_type = std::tuple<Ts...>;
std::vector<value_type> results;
std::shared_ptr<size_t> pending;
F f;
void operator()(Ts&... xs) {
if (*pending > 0) {
results.emplace_back(std::move(xs)...);
if (--*pending == 0)
f(std::move(results));
}
}
template <class Fun>
fan_in_tuple_helper(size_t pending, Fun&& f)
: pending(std::make_shared<size_t>(pending)), f(std::forward<Fun>(f)) {
results.reserve(pending);
}
// TODO: return 'auto' from this function when switching to C++17
std::function<void(Ts&...)> wrap() {
return [this](Ts&... xs) { (*this)(xs...); };
}
};
template <class F, class = typename detail::get_callable_trait<F>::arg_types>
struct select_fan_in_helper;
template <class F, class... Ts>
struct select_fan_in_helper<F,
detail::type_list<std::vector<std::tuple<Ts...>>>> {
using type = fan_in_tuple_helper<F, Ts...>;
};
template <class F, class T>
struct select_fan_in_helper<F, detail::type_list<std::vector<T>>> {
using type = fan_in_helper<F, T>;
};
template <class F>
using fan_in_helper_t = typename select_fan_in_helper<F>::type;
// TODO: Replace with a lambda when switching to C++17 (move g into lambda).
template <class G>
class fan_in_error_handler {
public:
template <class Fun>
fan_in_error_handler(Fun&& fun, std::shared_ptr<size_t> pending)
: handler(std::forward<Fun>(fun)), pending(std::move(pending)) {
// nop
}
void operator()(error& err) {
if (*pending > 0) {
*pending = 0;
handler(err);
}
}
private:
G handler;
std::shared_ptr<size_t> pending;
};
} // namespace detail
} // namespace caf
namespace caf {
namespace policy {
/// Enables a `response_handle` to fan-in multiple responses into a single
/// result (a `vector` of Individual values) for the client.
/// @relates mixin::requester
/// @relates response_handle
template <class ResponseType>
class fan_in {
public:
fan_in(std::vector<message_id> ids) : ids_(std::move(ids)) {
CAF_ASSERT(ids_.size()
<= static_cast<size_t>(std::numeric_limits<int>::max()));
}
template <class Self, class F, class OnError>
void await(Self* self, F&& f, OnError&& g) {
auto bhvr = make_behavior(std::forward<F>(f), std::forward<OnError>(g));
for (auto id : ids_)
self->add_awaited_response_handler(id, bhvr);
}
template <class Self, class F, class OnError>
void then(Self* self, F&& f, OnError&& g) {
auto bhvr = make_behavior(std::forward<F>(f), std::forward<OnError>(g));
for (auto id : ids_)
self->add_multiplexed_response_handler(id, bhvr);
}
template <class Self, class F, class OnError>
void receive(Self* self, F&& f, OnError&& g) {
using helper_type = detail::fan_in_helper_t<detail::decay_t<F>>;
helper_type helper{ids_.size(), std::forward<F>(f)};
detail::type_checker<ResponseType, helper_type>::check();
auto error_handler = [&](error& err) {
if (*helper.pending > 0) {
*helper.pending = 0;
helper.results.clear();
g(err);
}
};
auto wrapped_helper = helper.wrap();
for (auto id : ids_) {
typename Self::accept_one_cond rc;
self->varargs_receive(rc, id, wrapped_helper, error_handler);
}
}
private:
template <class F, class OnError>
behavior make_behavior(F&& f, OnError&& g) {
using namespace detail;
using helper_type = fan_in_helper_t<decay_t<F>>;
using error_handler_type = fan_in_error_handler<decay_t<OnError>>;
helper_type helper{ids_.size(), std::move(f)};
type_checker<ResponseType, helper_type>::check();
error_handler_type err_helper{std::forward<OnError>(g), helper.pending};
return {
std::move(helper),
std::move(err_helper),
};
}
std::vector<message_id> ids_;
};
} // namespace policy
} // namespace caf
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2019 Dominik Charousset *
* *
* 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. *
******************************************************************************/
#define CAF_SUITE policy.fan_in
#include "caf/policy/fan_in.hpp"
#include "caf/test/dsl.hpp"
#include <tuple>
#include "caf/actor_system.hpp"
#include "caf/event_based_actor.hpp"
#include "caf/sec.hpp"
using caf::policy::fan_in;
using namespace caf;
namespace {
struct fixture : test_coordinator_fixture<> {
template <class F>
actor make_server(F f) {
auto init = [f]() -> behavior {
return {
[f](int x, int y) { return f(x, y); },
};
};
return sys.spawn(init);
}
std::function<void(const error&)> make_error_handler() {
return [this](const error& err) {
CAF_FAIL("unexpected error: " << sys.render(err));
};
}
std::function<void(const error&)> make_counting_error_handler(size_t* count) {
return [count](const error&) { *count += 1; };
}
};
} // namespace
#define SUBTEST(message) \
run(); \
CAF_MESSAGE("subtest: " message); \
for (int subtest_dummy = 0; subtest_dummy < 1; ++subtest_dummy)
CAF_TEST_FIXTURE_SCOPE(fan_in_tests, fixture)
CAF_TEST(fan_in combines two integer results into one vector) {
using int_list = std::vector<int>;
auto f = [](int x, int y) { return x + y; };
auto server1 = make_server(f);
auto server2 = make_server(f);
SUBTEST("request.receive") {
SUBTEST("vector of int") {
auto r1 = self->request(server1, infinite, 1, 2);
auto r2 = self->request(server2, infinite, 2, 3);
fan_in<detail::type_list<int>> merge{{r1.id(), r2.id()}};
run();
merge.receive(
self.ptr(),
[](int_list results) {
std::sort(results.begin(), results.end());
CAF_CHECK_EQUAL(results, int_list({3, 5}));
},
make_error_handler());
}
SUBTEST("vector of tuples") {
using std::make_tuple;
auto r1 = self->request(server1, infinite, 1, 2);
auto r2 = self->request(server2, infinite, 2, 3);
fan_in<detail::type_list<int>> merge{{r1.id(), r2.id()}};
run();
using results_vector = std::vector<std::tuple<int>>;
merge.receive(
self.ptr(),
[](results_vector results) {
std::sort(results.begin(), results.end());
CAF_CHECK_EQUAL(results,
results_vector({make_tuple(3), make_tuple(5)}));
},
make_error_handler());
}
}
SUBTEST("request.then") {
int_list results;
auto client = sys.spawn([=, &results](event_based_actor* client_ptr) {
auto r1 = client_ptr->request(server1, infinite, 1, 2);
auto r2 = client_ptr->request(server2, infinite, 2, 3);
fan_in<detail::type_list<int>> merge{{r1.id(), r2.id()}};
merge.then(
client_ptr, [&results](int_list xs) { results = std::move(xs); },
make_error_handler());
});
run_once();
expect((int, int), from(client).to(server1).with(1, 2));
expect((int, int), from(client).to(server2).with(2, 3));
expect((int), from(server1).to(client).with(3));
expect((int), from(server2).to(client).with(5));
CAF_MESSAGE("request.then stores results in arrival order");
CAF_CHECK_EQUAL(results, int_list({3, 5}));
}
SUBTEST("request.await") {
int_list results;
auto client = sys.spawn([=, &results](event_based_actor* client_ptr) {
auto r1 = client_ptr->request(server1, infinite, 1, 2);
auto r2 = client_ptr->request(server2, infinite, 2, 3);
fan_in<detail::type_list<int>> merge{{r1.id(), r2.id()}};
merge.await(
client_ptr, [&results](int_list xs) { results = std::move(xs); },
make_error_handler());
});
run_once();
expect((int, int), from(client).to(server1).with(1, 2));
expect((int, int), from(client).to(server2).with(2, 3));
// TODO: DSL (mailbox.peek) cannot deal with receivers that skip messages.
// expect((int), from(server1).to(client).with(3));
// expect((int), from(server2).to(client).with(5));
run();
CAF_MESSAGE("request.await froces responses into reverse request order");
CAF_CHECK_EQUAL(results, int_list({5, 3}));
}
}
CAF_TEST(fan_in calls the error handler at most once) {
using int_list = std::vector<int>;
auto f = [](int, int) -> result<int> { return sec::invalid_argument; };
auto server1 = make_server(f);
auto server2 = make_server(f);
SUBTEST("request.receive") {
auto r1 = self->request(server1, infinite, 1, 2);
auto r2 = self->request(server2, infinite, 2, 3);
fan_in<detail::type_list<int>> merge{{r1.id(), r2.id()}};
run();
size_t errors = 0;
merge.receive(
self.ptr(),
[](int_list) { CAF_FAIL("fan-in policy called the result handler"); },
make_counting_error_handler(&errors));
CAF_CHECK_EQUAL(errors, 1u);
}
SUBTEST("request.then") {
size_t errors = 0;
auto client = sys.spawn([=, &errors](event_based_actor* client_ptr) {
auto r1 = client_ptr->request(server1, infinite, 1, 2);
auto r2 = client_ptr->request(server2, infinite, 2, 3);
fan_in<detail::type_list<int>> merge{{r1.id(), r2.id()}};
merge.then(
client_ptr,
[](int_list) { CAF_FAIL("fan-in policy called the result handler"); },
make_counting_error_handler(&errors));
});
run_once();
expect((int, int), from(client).to(server1).with(1, 2));
expect((int, int), from(client).to(server2).with(2, 3));
expect((error), from(server1).to(client).with(sec::invalid_argument));
expect((error), from(server2).to(client).with(sec::invalid_argument));
CAF_CHECK_EQUAL(errors, 1u);
}
SUBTEST("request.await") {
size_t errors = 0;
auto client = sys.spawn([=, &errors](event_based_actor* client_ptr) {
auto r1 = client_ptr->request(server1, infinite, 1, 2);
auto r2 = client_ptr->request(server2, infinite, 2, 3);
fan_in<detail::type_list<int>> merge{{r1.id(), r2.id()}};
merge.await(
client_ptr,
[](int_list) { CAF_FAIL("fan-in policy called the result handler"); },
make_counting_error_handler(&errors));
});
run_once();
expect((int, int), from(client).to(server1).with(1, 2));
expect((int, int), from(client).to(server2).with(2, 3));
// TODO: DSL (mailbox.peek) cannot deal with receivers that skip messages.
// expect((int), from(server1).to(client).with(3));
// expect((int), from(server2).to(client).with(5));
run();
CAF_CHECK_EQUAL(errors, 1u);
}
}
CAF_TEST_FIXTURE_SCOPE_END()
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