Commit 24a13d0e authored by Dominik Charousset's avatar Dominik Charousset

Merge branch issue/932 into topic/novaquark

parents 307ca984 09da4942
...@@ -36,8 +36,9 @@ MacroBlockBegin: "^BEGIN_STATE$" ...@@ -36,8 +36,9 @@ MacroBlockBegin: "^BEGIN_STATE$"
MacroBlockEnd: "^END_STATE$" MacroBlockEnd: "^END_STATE$"
MaxEmptyLinesToKeep: 1 MaxEmptyLinesToKeep: 1
NamespaceIndentation: None NamespaceIndentation: None
PenaltyBreakAssignment: 1000 PenaltyBreakAssignment: 10
PenaltyBreakBeforeFirstCallParameter: 1000 PenaltyBreakBeforeFirstCallParameter: 30
PenaltyReturnTypeOnItsOwnLine: 5
PointerAlignment: Left PointerAlignment: Left
ReflowComments: true ReflowComments: true
SortIncludes: true SortIncludes: true
......
...@@ -112,9 +112,9 @@ A SNocs workspace is provided by GitHub user ...@@ -112,9 +112,9 @@ A SNocs workspace is provided by GitHub user
## Supported Compilers ## Supported Compilers
* GCC >= 4.8.3 * GCC >= 7
* Clang >= 3.2 * Clang >= 4
* MSVC >= 2015, update 3 * MSVC >= 2019
## Supported Operating Systems ## Supported Operating Systems
...@@ -127,7 +127,6 @@ A SNocs workspace is provided by GitHub user ...@@ -127,7 +127,6 @@ A SNocs workspace is provided by GitHub user
* Doxygen (for the `doxygen` target) * Doxygen (for the `doxygen` target)
* LaTeX (for the `manual` target) * LaTeX (for the `manual` target)
* Pandoc and Python with pandocfilters (for the `rst` target)
## Scientific Use ## Scientific Use
......
...@@ -10,10 +10,6 @@ ...@@ -10,10 +10,6 @@
# error "No support for 'if constexpr' (__cpp_if_constexpr)" # error "No support for 'if constexpr' (__cpp_if_constexpr)"
#endif #endif
// Unfortunately there's no feature test macro for thread_local. By putting this
// here, at least we'll get a compiler error on unsupported platforms.
[[maybe_unused]] thread_local int foo;
int main(int, char**) { int main(int, char**) {
return 0; return 0;
} }
...@@ -186,6 +186,24 @@ Both event-based approaches send all requests, install a series of one-shot ...@@ -186,6 +186,24 @@ Both event-based approaches send all requests, install a series of one-shot
handlers, and then return from the implementing function. In contrast, the handlers, and then return from the implementing function. In contrast, the
blocking function waits for a response before sending another request. blocking function waits for a response before sending another request.
\subsubsection{Sending Multiple Requests}
Sending the same message to a group of workers is a common work flow in actor
applications. Usually, a manager maintains a set of workers. On request, the
manager fans-out the request to all of its workers and then collects the
results. The function \lstinline`fan_out_request` combined with the merge policy
\lstinline`fan_in_responses` streamlines this exact use case.
In the following snippet, we have a matrix actor (\lstinline`self`) that stores
worker actors for each cell (each simply storing an integer). For computing the
average over a row, we use \lstinline`fan_out_request`. The result handler
passed to \lstinline`then` now gets called only once with a \lstinline`vector`
holding all collected results. Using a response promise \see{promise} further
allows us to delay responding to the client until we have collected all worker
results.
\cppexample[86-98]{message_passing/fan_out_request}
\clearpage \clearpage
\subsubsection{Error Handling in Requests} \subsubsection{Error Handling in Requests}
\label{error-response} \label{error-response}
......
...@@ -24,6 +24,7 @@ add(message_passing cell) ...@@ -24,6 +24,7 @@ add(message_passing cell)
add(message_passing dancing_kirby) add(message_passing dancing_kirby)
add(message_passing delegating) add(message_passing delegating)
add(message_passing divider) add(message_passing divider)
add(message_passing fan_out_request)
add(message_passing fixed_stack) add(message_passing fixed_stack)
add(message_passing promises) add(message_passing promises)
add(message_passing request) add(message_passing request)
......
// This file is partially included in the manual, do not modify
// without updating the references in the *.tex files!
// Manual references: lines 86-98 (MessagePassing.tex)
#include <cassert>
#include <chrono>
#include <cmath>
#include <iomanip>
#include <iostream>
#include <vector>
#include "caf/actor_system.hpp"
#include "caf/actor_system_config.hpp"
#include "caf/event_based_actor.hpp"
#include "caf/exec_main.hpp"
#include "caf/function_view.hpp"
#include "caf/policy/fan_in_responses.hpp"
#include "caf/scoped_actor.hpp"
#include "caf/stateful_actor.hpp"
#include "caf/typed_actor.hpp"
#include "caf/typed_event_based_actor.hpp"
using std::endl;
using std::chrono::seconds;
using namespace caf;
using row_atom = atom_constant<atom("row")>;
using column_atom = atom_constant<atom("column")>;
using average_atom = atom_constant<atom("average")>;
/// A simple actor for storing an integer value.
using cell = typed_actor<
// Writes a new value.
reacts_to<put_atom, int>,
// Reads the value.
replies_to<get_atom>::with<int>>;
/// An for storing a 2-dimensional matrix of integers.
using matrix = typed_actor<
// Writes a new value to given cell (x-coordinate, y-coordinate, new-value).
reacts_to<put_atom, int, int, int>,
// Reads from given cell.
replies_to<get_atom, int, int>::with<int>,
// Computes the average for given row.
replies_to<get_atom, average_atom, row_atom, int>::with<double>,
// Computes the average for given column.
replies_to<get_atom, average_atom, column_atom, int>::with<double>>;
struct cell_state {
int value = 0;
static constexpr const char* name = "cell";
};
cell::behavior_type cell_actor(cell::stateful_pointer<cell_state> self) {
return {
[=](put_atom, int val) { self->state.value = val; },
[=](get_atom) { return self->state.value; },
};
}
struct matrix_state {
using row_type = std::vector<cell>;
std::vector<row_type> rows;
static constexpr const char* name = "matrix";
};
matrix::behavior_type matrix_actor(matrix::stateful_pointer<matrix_state> self,
int rows, int columns) {
// Spawn all cells and return our behavior.
self->state.rows.resize(rows);
for (int row = 0; row < rows; ++row) {
auto& row_vec = self->state.rows[row];
row_vec.resize(columns);
for (int column = 0; column < columns; ++column)
row_vec[column] = self->spawn(cell_actor);
}
return {
[=](put_atom put, int row, int column, int val) {
assert(row < rows && column < columns);
return self->delegate(self->state.rows[row][column], put, val);
},
[=](get_atom get, int row, int column) {
assert(row < rows && column < columns);
return self->delegate(self->state.rows[row][column], get);
},
[=](get_atom get, average_atom, row_atom, int row) {
assert(row < rows);
auto rp = self->make_response_promise<double>();
auto& row_vec = self->state.rows[row];
self->fan_out_request<policy::fan_in_responses>(row_vec, infinite, get)
.then(
[=](std::vector<int> xs) mutable {
assert(xs.size() == static_cast<size_t>(columns));
rp.deliver(std::accumulate(xs.begin(), xs.end(), 0.0) / columns);
},
[=](error& err) mutable { rp.deliver(std::move(err)); });
return rp;
},
[=](get_atom get, average_atom, column_atom, int column) {
assert(column < columns);
std::vector<cell> columns;
columns.reserve(rows);
auto& rows_vec = self->state.rows;
for (int row = 0; row < rows; ++row)
columns.emplace_back(rows_vec[row][column]);
auto rp = self->make_response_promise<double>();
self->fan_out_request<policy::fan_in_responses>(columns, infinite, get)
.then(
[=](std::vector<int> xs) mutable {
assert(xs.size() == static_cast<size_t>(rows));
rp.deliver(std::accumulate(xs.begin(), xs.end(), 0.0) / rows);
},
[=](error& err) mutable { rp.deliver(std::move(err)); });
return rp;
},
};
}
std::ostream& operator<<(std::ostream& out, const expected<int>& x) {
if (x)
return out << *x;
return out << to_string(x.error());
}
void caf_main(actor_system& sys) {
// Spawn our matrix.
static constexpr int rows = 3;
static constexpr int columns = 6;
auto mx = sys.spawn(matrix_actor, rows, columns);
auto f = make_function_view(mx);
// Set cells in our matrix to these values:
// 2 4 8 16 32 64
// 3 9 27 81 243 729
// 4 16 64 256 1024 4096
for (int row = 0; row < rows; ++row)
for (int column = 0; column < columns; ++column)
f(put_atom::value, row, column, (int) pow(row + 2, column + 1));
// Print out matrix.
for (int row = 0; row < rows; ++row) {
for (int column = 0; column < columns; ++column)
std::cout << std::setw(4) << f(get_atom::value, row, column) << ' ';
std::cout << '\n';
}
// Print out AVG for each row and column.
for (int row = 0; row < rows; ++row)
std::cout << "AVG(row " << row << ") = "
<< f(get_atom::value, average_atom::value, row_atom::value, row)
<< '\n';
for (int column = 0; column < columns; ++column)
std::cout << "AVG(column " << column << ") = "
<< f(get_atom::value, average_atom::value, column_atom::value,
column)
<< '\n';
}
CAF_MAIN()
...@@ -57,15 +57,6 @@ ...@@ -57,15 +57,6 @@
#include "caf/mixin/sender.hpp" #include "caf/mixin/sender.hpp"
#include "caf/mixin/subscriber.hpp" #include "caf/mixin/subscriber.hpp"
namespace caf {
namespace mixin {
template <>
struct is_blocking_requester<blocking_actor> : std::true_type { };
} // namespace caf
} // namespace mixin
namespace caf { namespace caf {
/// A thread-mapped or context-switching actor using a blocking /// A thread-mapped or context-switching actor using a blocking
......
...@@ -18,8 +18,9 @@ ...@@ -18,8 +18,9 @@
#pragma once #pragma once
#include <tuple>
#include <chrono> #include <chrono>
#include <tuple>
#include <vector>
#include "caf/actor.hpp" #include "caf/actor.hpp"
#include "caf/check_typed_input.hpp" #include "caf/check_typed_input.hpp"
...@@ -29,15 +30,13 @@ ...@@ -29,15 +30,13 @@
#include "caf/message.hpp" #include "caf/message.hpp"
#include "caf/message_id.hpp" #include "caf/message_id.hpp"
#include "caf/message_priority.hpp" #include "caf/message_priority.hpp"
#include "caf/policy/single_response.hpp"
#include "caf/response_handle.hpp" #include "caf/response_handle.hpp"
#include "caf/response_type.hpp" #include "caf/response_type.hpp"
namespace caf { namespace caf {
namespace mixin { namespace mixin {
template <class T>
struct is_blocking_requester : std::false_type { };
/// A `requester` is an actor that supports /// A `requester` is an actor that supports
/// `self->request(...).{then|await|receive}`. /// `self->request(...).{then|await|receive}`.
template <class Base, class Subtype> template <class Base, class Subtype>
...@@ -60,22 +59,12 @@ public: ...@@ -60,22 +59,12 @@ public:
/// @returns A handle identifying a future-like handle to the response. /// @returns A handle identifying a future-like handle to the response.
/// @warning The returned handle is actor specific and the response to the /// @warning The returned handle is actor specific and the response to the
/// sent message cannot be received by another actor. /// sent message cannot be received by another actor.
template <message_priority P = message_priority::normal, template <message_priority P = message_priority::normal, class Handle = actor,
class Handle = actor, class... Ts> class... Ts>
response_handle<Subtype, auto request(const Handle& dest, const duration& timeout, Ts&&... xs) {
response_type_t< using namespace detail;
typename Handle::signatures,
typename detail::implicit_conversions<
typename std::decay<Ts>::type
>::type...>,
is_blocking_requester<Subtype>::value>
request(const Handle& dest, const duration& timeout, Ts&&... xs) {
static_assert(sizeof...(Ts) > 0, "no message to send"); static_assert(sizeof...(Ts) > 0, "no message to send");
using token = using token = type_list<implicit_conversions_t<decay_t<Ts>>...>;
detail::type_list<
typename detail::implicit_conversions<
typename std::decay<Ts>::type
>::type...>;
static_assert(response_type_unbox<signatures_of_t<Handle>, token>::valid, static_assert(response_type_unbox<signatures_of_t<Handle>, token>::valid,
"receiver does not accept given message"); "receiver does not accept given message");
auto self = static_cast<Subtype*>(this); auto self = static_cast<Subtype*>(this);
...@@ -88,29 +77,89 @@ public: ...@@ -88,29 +77,89 @@ public:
self->eq_impl(req_id.response_id(), self->ctrl(), self->context(), self->eq_impl(req_id.response_id(), self->ctrl(), self->context(),
make_error(sec::invalid_argument)); make_error(sec::invalid_argument));
} }
return {req_id.response_id(), self}; using response_type
} = response_type_t<typename Handle::signatures,
detail::implicit_conversions_t<detail::decay_t<Ts>>...>;
using handle_type
= response_handle<Subtype, policy::single_response<response_type>>;
return handle_type{self, req_id.response_id()};
}
/// Sends `{xs...}` as a synchronous message to `dest` with priority `mp`. /// @copydoc request
/// @returns A handle identifying a future-like handle to the response. template <message_priority P = message_priority::normal, class Rep = int,
/// @warning The returned handle is actor specific and the response to the class Period = std::ratio<1>, class Handle = actor, class... Ts>
/// sent message cannot be received by another actor. auto request(const Handle& dest, std::chrono::duration<Rep, Period> timeout,
template <message_priority P = message_priority::normal, Ts&&... xs) {
class Rep = int, class Period = std::ratio<1>,
class Handle = actor, class... Ts>
response_handle<Subtype,
response_type_t<
typename Handle::signatures,
typename detail::implicit_conversions<
typename std::decay<Ts>::type
>::type...>,
is_blocking_requester<Subtype>::value>
request(const Handle& dest, std::chrono::duration<Rep, Period> timeout,
Ts&&... xs) {
return request(dest, duration{timeout}, std::forward<Ts>(xs)...); return request(dest, duration{timeout}, std::forward<Ts>(xs)...);
} }
/// Sends `{xs...}` to each actor in the range `destinations` as a synchronous
/// message. Response messages get combined into a single result according to
/// the `MergePolicy`.
/// @tparam MergePolicy Configures how individual response messages get
/// combined by the actor. The policy makes sure that the
/// response handler gets invoked at most once. In case of
/// one or more errors, the policy calls the error handler
/// exactly once, with the first error occurred.
/// @tparam Prio Specifies the priority of the synchronous messages.
/// @tparam Container A container type for holding actor handles. Must provide
/// the type alias `value_type` as well as the member
/// functions `begin()`, `end()`, and `size()`.
/// @param destinations A container holding handles to all destination actors.
/// @param timeout Maximum duration before dropping the request. The runtime
/// system will send an error message to the actor in case the
/// receiver does not respond in time.
/// @returns A helper object that takes response handlers via `.await()`,
/// `.then()`, or `.receive()`.
/// @note The returned handle is actor-specific. Only the actor that called
/// `request` can use it for setting response handlers.
template <template <class> class MergePolicy,
message_priority Prio = message_priority::normal, class Container,
class... Ts>
auto fan_out_request(const Container& destinations, const duration& timeout,
Ts&&... xs) {
using handle_type = typename Container::value_type;
using namespace detail;
static_assert(sizeof...(Ts) > 0, "no message to send");
using token = type_list<implicit_conversions_t<decay_t<Ts>>...>;
static_assert(
response_type_unbox<signatures_of_t<handle_type>, token>::valid,
"receiver does not accept given message");
auto dptr = static_cast<Subtype*>(this);
std::vector<message_id> ids;
ids.reserve(destinations.size());
for (const auto& dest : destinations) {
if (!dest)
continue;
auto req_id = dptr->new_request_id(Prio);
dest->eq_impl(req_id, dptr->ctrl(), dptr->context(),
std::forward<Ts>(xs)...);
dptr->request_response_timeout(timeout, req_id);
ids.emplace_back(req_id.response_id());
}
if (ids.empty()) {
auto req_id = dptr->new_request_id(Prio);
dptr->eq_impl(req_id.response_id(), dptr->ctrl(), dptr->context(),
make_error(sec::invalid_argument));
ids.emplace_back(req_id.response_id());
}
using response_type
= response_type_t<typename handle_type::signatures,
detail::implicit_conversions_t<detail::decay_t<Ts>>...>;
using result_type = response_handle<Subtype, MergePolicy<response_type>>;
return result_type{dptr, std::move(ids)};
}
/// @copydoc fan_out_request
template <template <class> class MergePolicy,
message_priority Prio = message_priority::normal, class Rep,
class Period, class Container, class... Ts>
auto fan_out_request(const Container& destinations,
std::chrono::duration<Rep, Period> timeout, Ts&&... xs) {
return request<MergePolicy, Prio>(destinations, duration{timeout},
std::forward<Ts>(xs)...);
}
}; };
} // namespace mixin } // namespace mixin
} // namespace caf } // 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. *
******************************************************************************/
#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/logger.hpp"
#include "caf/message_id.hpp"
namespace caf {
namespace detail {
template <class F, class T>
struct fan_in_responses_helper {
std::vector<T> results;
std::shared_ptr<size_t> pending;
F f;
void operator()(T& x) {
CAF_LOG_TRACE(CAF_ARG2("pending", *pending));
if (*pending > 0) {
results.emplace_back(std::move(x));
if (--*pending == 0)
f(std::move(results));
}
}
template <class Fun>
fan_in_responses_helper(size_t pending, Fun&& f)
: pending(std::make_shared<size_t>(pending)), f(std::forward<Fun>(f)) {
results.reserve(pending);
}
auto wrap() {
return [this](T& x) { (*this)(x); };
}
};
template <class F, class... Ts>
struct fan_in_responses_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) {
CAF_LOG_TRACE(CAF_ARG2("pending", *pending));
if (*pending > 0) {
results.emplace_back(std::move(xs)...);
if (--*pending == 0)
f(std::move(results));
}
}
template <class Fun>
fan_in_responses_tuple_helper(size_t pending, Fun&& f)
: pending(std::make_shared<size_t>(pending)), f(std::forward<Fun>(f)) {
results.reserve(pending);
}
auto wrap() {
return [this](Ts&... xs) { (*this)(xs...); };
}
};
template <class F, class = typename detail::get_callable_trait<F>::arg_types>
struct select_fan_in_responses_helper;
template <class F, class... Ts>
struct select_fan_in_responses_helper<
F, detail::type_list<std::vector<std::tuple<Ts...>>>> {
using type = fan_in_responses_tuple_helper<F, Ts...>;
};
template <class F, class T>
struct select_fan_in_responses_helper<F, detail::type_list<std::vector<T>>> {
using type = fan_in_responses_helper<F, T>;
};
template <class F>
using fan_in_responses_helper_t =
typename select_fan_in_responses_helper<F>::type;
// TODO: Replace with a lambda when switching to C++17 (move g into lambda).
template <class G>
class fan_in_responses_error_handler {
public:
template <class Fun>
fan_in_responses_error_handler(Fun&& fun, std::shared_ptr<size_t> pending)
: handler(std::forward<Fun>(fun)), pending(std::move(pending)) {
// nop
}
void operator()(error& err) {
CAF_LOG_TRACE(CAF_ARG2("pending", *pending));
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_responses {
public:
static constexpr bool is_trivial = false;
using response_type = ResponseType;
using message_id_list = std::vector<message_id>;
template <class Fun>
using type_checker = detail::type_checker<
response_type, detail::fan_in_responses_helper_t<detail::decay_t<Fun>>>;
explicit fan_in_responses(message_id_list ids) : ids_(std::move(ids)) {
CAF_ASSERT(ids_.size()
<= static_cast<size_t>(std::numeric_limits<int>::max()));
}
fan_in_responses(fan_in_responses&&) noexcept = default;
fan_in_responses& operator=(fan_in_responses&&) noexcept = default;
template <class Self, class F, class OnError>
void await(Self* self, F&& f, OnError&& g) const {
CAF_LOG_TRACE(CAF_ARG(ids_));
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) const {
CAF_LOG_TRACE(CAF_ARG(ids_));
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 G>
void receive(Self* self, F&& f, G&& g) const {
CAF_LOG_TRACE(CAF_ARG(ids_));
using helper_type = detail::fan_in_responses_helper_t<detail::decay_t<F>>;
helper_type helper{ids_.size(), std::forward<F>(f)};
auto error_handler = [&](error& err) {
if (*helper.pending > 0) {
*helper.pending = 0;
helper.results.clear();
g(err);
}
};
for (auto id : ids_) {
typename Self::accept_one_cond rc;
self->varargs_receive(rc, id, helper.wrap(), error_handler);
}
}
const message_id_list& ids() const noexcept {
return ids_;
}
private:
template <class F, class OnError>
behavior make_behavior(F&& f, OnError&& g) const {
using namespace detail;
using helper_type = fan_in_responses_helper_t<decay_t<F>>;
using error_handler_type = fan_in_responses_error_handler<decay_t<OnError>>;
helper_type helper{ids_.size(), std::move(f)};
error_handler_type err_helper{std::forward<OnError>(g), helper.pending};
return {
std::move(helper),
std::move(err_helper),
};
}
message_id_list 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. *
******************************************************************************/
#pragma once
#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 policy {
/// Trivial policy for handling a single result in a `response_handler`.
/// @relates mixin::requester
/// @relates response_handle
template <class ResponseType>
class single_response {
public:
static constexpr bool is_trivial = true;
using response_type = ResponseType;
template <class Fun>
using type_checker = detail::type_checker<response_type, Fun>;
explicit single_response(message_id mid) noexcept : mid_(mid) {
// nop
}
single_response(single_response&&) noexcept = default;
single_response& operator=(single_response&&) noexcept = default;
template <class Self, class F, class OnError>
void await(Self* self, F&& f, OnError&& g) const {
behavior bhvr{std::forward<F>(f), std::forward<OnError>(g)};
self->add_awaited_response_handler(mid_, std::move(bhvr));
}
template <class Self, class F, class OnError>
void then(Self* self, F&& f, OnError&& g) const {
behavior bhvr{std::forward<F>(f), std::forward<OnError>(g)};
self->add_multiplexed_response_handler(mid_, std::move(bhvr));
}
template <class Self, class F, class OnError>
void receive(Self* self, F&& f, OnError&& g) const {
typename Self::accept_one_cond rc;
self->varargs_receive(rc, mid_, std::forward<F>(f),
std::forward<OnError>(g));
}
message_id id() const noexcept {
return mid_;
}
private:
message_id mid_;
};
} // namespace policy
} // namespace caf
...@@ -20,8 +20,10 @@ ...@@ -20,8 +20,10 @@
#include <type_traits> #include <type_traits>
#include "caf/actor_traits.hpp"
#include "caf/catch_all.hpp" #include "caf/catch_all.hpp"
#include "caf/message_id.hpp" #include "caf/message_id.hpp"
#include "caf/none.hpp"
#include "caf/sec.hpp" #include "caf/sec.hpp"
#include "caf/system_messages.hpp" #include "caf/system_messages.hpp"
#include "caf/typed_behavior.hpp" #include "caf/typed_behavior.hpp"
...@@ -33,145 +35,137 @@ namespace caf { ...@@ -33,145 +35,137 @@ namespace caf {
/// This helper class identifies an expected response message and enables /// This helper class identifies an expected response message and enables
/// `request(...).then(...)`. /// `request(...).then(...)`.
template <class Self, class Output, bool IsBlocking> template <class ActorType, class Policy>
class response_handle; class response_handle {
/******************************************************************************
* nonblocking *
******************************************************************************/
template <class Self, class Output>
class response_handle<Self, Output, false> {
public: public:
response_handle() = delete; // -- member types -----------------------------------------------------------
response_handle(const response_handle&) = default;
response_handle& operator=(const response_handle&) = default;
response_handle(message_id mid, Self* self) : mid_(mid), self_(self) { using actor_type = ActorType;
// nop
}
template <class F, class E = detail::is_callable_t<F>> using traits = actor_traits<actor_type>;
void await(F f) const {
await_impl(f);
}
template <class F, class OnError, class E1 = detail::is_callable_t<F>, using policy_type = Policy;
class E2 = detail::is_handler_for_ef<OnError, error>>
void await(F f, OnError e) const {
await_impl(f, e);
}
template <class F, class E = detail::is_callable_t<F>> using response_type = typename policy_type::response_type;
void then(F f) const {
then_impl(f);
}
template <class F, class OnError, class E1 = detail::is_callable_t<F>, // -- constructors, destructors, and assignment operators --------------------
class E2 = detail::is_handler_for_ef<OnError, error>>
void then(F f, OnError e) const {
then_impl(f, e);
}
message_id id() const noexcept { response_handle() = delete;
return mid_;
}
Self* self() noexcept { response_handle(const response_handle&) = delete;
return self_;
response_handle& operator=(const response_handle&) = delete;
response_handle(response_handle&&) noexcept = default;
response_handle& operator=(response_handle&&) noexcept = delete;
template <class... Ts>
explicit response_handle(actor_type* self, Ts&&... xs)
: self_(self), policy_(std::forward<Ts>(xs)...) {
// nop
} }
private: // -- non-blocking API -------------------------------------------------------
template <class F, class OnError>
void await_impl(F& f, OnError& g) const { template <class T = traits, class F = none_t, class OnError = none_t,
class = detail::enable_if_t<T::is_non_blocking>>
void await(F f, OnError g) const {
static_assert(detail::is_callable<F>::value, "F must provide a single, "
"non-template operator()");
static_assert(detail::is_callable_with<OnError, error&>::value,
"OnError must provide an operator() that takes a caf::error");
using result_type = typename detail::get_callable_trait<F>::result_type; using result_type = typename detail::get_callable_trait<F>::result_type;
static_assert(std::is_same<void, result_type>::value, static_assert(std::is_same<void, result_type>::value,
"response handlers are not allowed to have a return " "response handlers are not allowed to have a return "
"type other than void"); "type other than void");
detail::type_checker<Output, F>::check(); policy_type::template type_checker<F>::check();
self_->add_awaited_response_handler(mid_, policy_.await(self_, std::move(f), std::move(g));
behavior{std::move(f), std::move(g)});
} }
template <class F> template <class T = traits, class F = none_t,
void await_impl(F& f) const { class = detail::enable_if_t<T::is_non_blocking>>
void await(F f) const {
auto self = self_; auto self = self_;
auto on_error = [self](error& err) { self->call_error_handler(err); }; await(std::move(f), [self](error& err) { self->call_error_handler(err); });
return await_impl(f, on_error);
} }
template <class F, class OnError> template <class T = traits, class F = none_t, class OnError = none_t,
void then_impl(F& f, OnError& g) const { class = detail::enable_if_t<T::is_non_blocking>>
void then(F f, OnError g) const {
static_assert(detail::is_callable<F>::value, "F must provide a single, "
"non-template operator()");
static_assert(detail::is_callable_with<OnError, error&>::value,
"OnError must provide an operator() that takes a caf::error");
using result_type = typename detail::get_callable_trait<F>::result_type; using result_type = typename detail::get_callable_trait<F>::result_type;
static_assert(std::is_same<void, result_type>::value, static_assert(std::is_same<void, result_type>::value,
"response handlers are not allowed to have a return " "response handlers are not allowed to have a return "
"type other than void"); "type other than void");
detail::type_checker<Output, F>::check(); policy_type::template type_checker<F>::check();
self_->add_multiplexed_response_handler(mid_, behavior{std::move(f), policy_.then(self_, std::move(f), std::move(g));
std::move(g)});
} }
template <class F> template <class T = traits, class F = none_t,
void then_impl(F& f) const { class = detail::enable_if_t<T::is_non_blocking>>
void then(F f) const {
auto self = self_; auto self = self_;
auto on_error = [self](error& err) { self->call_error_handler(err); }; then(std::move(f), [self](error& err) { self->call_error_handler(err); });
return then_impl(f, on_error);
}
message_id mid_;
Self* self_;
};
/******************************************************************************
* blocking *
******************************************************************************/
template <class Self, class Output>
class response_handle<Self, Output, true> {
public:
response_handle() = delete;
response_handle(const response_handle&) = default;
response_handle& operator=(const response_handle&) = default;
response_handle(message_id mid, Self* self) : mid_(mid), self_(self) {
// nop
} }
using error_handler = std::function<void(error&)>; // -- blocking API -----------------------------------------------------------
template <class F, class OnError, template <class T = traits, class F = none_t, class OnError = none_t,
class E = detail::is_handler_for_ef<OnError, error>> class = detail::enable_if_t<T::is_blocking>>
detail::is_callable_t<F> receive(F f, OnError g) { detail::is_handler_for_ef<OnError, error> receive(F f, OnError g) {
static_assert(detail::is_callable<F>::value, "F must provide a single, "
"non-template operator()");
static_assert(detail::is_callable_with<OnError, error&>::value,
"OnError must provide an operator() that takes a caf::error");
using result_type = typename detail::get_callable_trait<F>::result_type; using result_type = typename detail::get_callable_trait<F>::result_type;
static_assert(std::is_same<void, result_type>::value, static_assert(std::is_same<void, result_type>::value,
"response handlers are not allowed to have a return " "response handlers are not allowed to have a return "
"type other than void"); "type other than void");
detail::type_checker<Output, F>::check(); policy_type::template type_checker<F>::check();
typename Self::accept_one_cond rc; policy_.receive(self_, std::move(f), std::move(g));
self_->varargs_receive(rc, mid_, std::move(f), std::move(g));
} }
template <class OnError, class F, class E = detail::is_callable_t<F>> template <class T = traits, class OnError = none_t, class F = none_t,
class = detail::enable_if_t<T::is_blocking>>
detail::is_handler_for_ef<OnError, error> receive(OnError g, F f) { detail::is_handler_for_ef<OnError, error> receive(OnError g, F f) {
// TODO: allowing blocking actors to pass the error handler in first may be
// more flexible, but it makes the API asymmetric. Consider
// deprecating / removing this member function.
receive(std::move(f), std::move(g)); receive(std::move(f), std::move(g));
} }
template <class OnError, class F, template <class T = policy_type, class OnError = none_t, class F = none_t,
class E = detail::is_handler_for_ef<OnError, error>> class E = detail::is_handler_for_ef<OnError, error>,
class = detail::enable_if_t<T::is_trivial>>
void receive(OnError g, catch_all<F> f) { void receive(OnError g, catch_all<F> f) {
typename Self::accept_one_cond rc; // TODO: this bypasses the policy. Either we deprecate `catch_all` or *all*
self_->varargs_receive(rc, mid_, std::move(g), std::move(f)); // policies must support it. Currently, we only enable this member
// function on the trivial policy for backwards compatibility.
typename actor_type::accept_one_cond rc;
self_->varargs_receive(rc, id(), std::move(g), std::move(f));
} }
// -- properties -------------------------------------------------------------
template <class T = policy_type, class = detail::enable_if_t<T::is_trivial>>
message_id id() const noexcept { message_id id() const noexcept {
return mid_; return policy_.id();
} }
Self* self() noexcept { actor_type* self() noexcept {
return self_; return self_;
} }
private: private:
message_id mid_; /// Points to the parent actor.
Self* self_; actor_type* self_;
/// Configures how the actor wants to process an incoming response.
policy_type policy_;
}; };
} // namespace caf } // namespace caf
...@@ -144,9 +144,9 @@ class typed_actor : detail::comparable<typed_actor<Sigs...>>, ...@@ -144,9 +144,9 @@ class typed_actor : detail::comparable<typed_actor<Sigs...>>,
template <class T, template <class T,
class = detail::enable_if_t<actor_traits<T>::is_statically_typed>> class = detail::enable_if_t<actor_traits<T>::is_statically_typed>>
typed_actor(T* ptr) : ptr_(ptr->ctrl()) { typed_actor(T* ptr) : ptr_(ptr->ctrl()) {
static_assert(detail::tl_subset_of<signatures, static_assert(
typename T::signatures>::value, detail::tl_subset_of<signatures, typename T::signatures>::value,
"Cannot assign T* to incompatible handle type"); "Cannot assign T* to incompatible handle type");
CAF_ASSERT(ptr != nullptr); CAF_ASSERT(ptr != nullptr);
} }
......
...@@ -23,6 +23,7 @@ ...@@ -23,6 +23,7 @@
#include "caf/test/dsl.hpp" #include "caf/test/dsl.hpp"
#include "caf/event_based_actor.hpp" #include "caf/event_based_actor.hpp"
#include "caf/policy/fan_in_responses.hpp"
using namespace caf; using namespace caf;
...@@ -150,4 +151,31 @@ CAF_TEST(delegated request with integer result) { ...@@ -150,4 +151,31 @@ CAF_TEST(delegated request with integer result) {
CAF_CHECK_EQUAL(*result, 3); CAF_CHECK_EQUAL(*result, 3);
} }
CAF_TEST(requesters support fan_out_request) {
using policy::fan_in_responses;
std::vector<adding_server_type> workers{
make_server([](int x, int y) { return x + y; }),
make_server([](int x, int y) { return x + y; }),
make_server([](int x, int y) { return x + y; }),
};
run();
auto sum = std::make_shared<int>(0);
auto client = sys.spawn([=](event_based_actor* self) {
self->fan_out_request<fan_in_responses>(workers, infinite, 1, 2)
.then([=](std::vector<int> results) {
for (auto result : results)
CAF_CHECK_EQUAL(result, 3);
*sum = std::accumulate(results.begin(), results.end(), 0);
});
});
run_once();
expect((int, int), from(client).to(workers[0]).with(1, 2));
expect((int), from(workers[0]).to(client).with(3));
expect((int, int), from(client).to(workers[1]).with(1, 2));
expect((int), from(workers[1]).to(client).with(3));
expect((int, int), from(client).to(workers[2]).with(1, 2));
expect((int), from(workers[2]).to(client).with(3));
CAF_CHECK_EQUAL(*sum, 9);
}
CAF_TEST_FIXTURE_SCOPE_END() CAF_TEST_FIXTURE_SCOPE_END()
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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_responses
#include "caf/policy/fan_in_responses.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_responses;
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_responses_tests, fixture)
CAF_TEST(fan_in_responses 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_responses<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_responses<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_responses<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_responses<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_responses 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_responses<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_responses<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_responses<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()
...@@ -645,9 +645,9 @@ protected: ...@@ -645,9 +645,9 @@ protected:
template <class... Ts> template <class... Ts>
struct test_coordinator_fixture_fetch_helper { struct test_coordinator_fixture_fetch_helper {
template <class Self, class Interface> template <class Self, template <class> class Policy, class Interface>
std::tuple<Ts...> std::tuple<Ts...>
operator()(caf::response_handle<Self, Interface, true>& from) const { operator()(caf::response_handle<Self, Policy<Interface>>& from) const {
std::tuple<Ts...> result; std::tuple<Ts...> result;
from.receive([&](Ts&... xs) { result = std::make_tuple(std::move(xs)...); }, from.receive([&](Ts&... xs) { result = std::make_tuple(std::move(xs)...); },
[&](caf::error& err) { [&](caf::error& err) {
...@@ -659,8 +659,8 @@ struct test_coordinator_fixture_fetch_helper { ...@@ -659,8 +659,8 @@ struct test_coordinator_fixture_fetch_helper {
template <class T> template <class T>
struct test_coordinator_fixture_fetch_helper<T> { struct test_coordinator_fixture_fetch_helper<T> {
template <class Self, class Interface> template <class Self, template <class> class Policy, class Interface>
T operator()(caf::response_handle<Self, Interface, true>& from) const { T operator()(caf::response_handle<Self, Policy<Interface>>& from) const {
T result; T result;
from.receive([&](T& x) { result = std::move(x); }, from.receive([&](T& x) { result = std::move(x); },
[&](caf::error& err) { [&](caf::error& err) {
......
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