Commit 4fe0a7b2 authored by Dominik Charousset's avatar Dominik Charousset

Handle void results in fan_out_request

(cherry picked from commit cb478b5c)
parent bfcef3f9
...@@ -12,6 +12,8 @@ is based on [Keep a Changelog](https://keepachangelog.com). ...@@ -12,6 +12,8 @@ is based on [Keep a Changelog](https://keepachangelog.com).
- The JSON reader now automatically widens integers to doubles as necessary. - The JSON reader now automatically widens integers to doubles as necessary.
- Parsing deeply nested JSON inputs no longer produces a stack overflow. - Parsing deeply nested JSON inputs no longer produces a stack overflow.
Instead, the parser rejects any JSON with too many nesting levels. Instead, the parser rejects any JSON with too many nesting levels.
- The `fan_out_request` request now properly deals with actor handles that
respond with `void` (#1369).
## [0.18.6] ## [0.18.6]
......
...@@ -37,8 +37,11 @@ using select_all_helper_value_t = ...@@ -37,8 +37,11 @@ using select_all_helper_value_t =
typename select_all_helper_value_oracle<Ts...>::type; typename select_all_helper_value_oracle<Ts...>::type;
template <class F, class... Ts> template <class F, class... Ts>
struct select_all_helper { struct select_all_helper;
using value_type = select_all_helper_value_t<Ts...>;
template <class F, class T, class... Ts>
struct select_all_helper<F, T, Ts...> {
using value_type = select_all_helper_value_t<T, Ts...>;
std::vector<value_type> results; std::vector<value_type> results;
std::shared_ptr<size_t> pending; std::shared_ptr<size_t> pending;
disposable timeouts; disposable timeouts;
...@@ -52,10 +55,10 @@ struct select_all_helper { ...@@ -52,10 +55,10 @@ struct select_all_helper {
results.reserve(pending); results.reserve(pending);
} }
void operator()(Ts&... xs) { void operator()(T& x, Ts&... xs) {
CAF_LOG_TRACE(CAF_ARG2("pending", *pending)); CAF_LOG_TRACE(CAF_ARG2("pending", *pending));
if (*pending > 0) { if (*pending > 0) {
results.emplace_back(std::move(xs)...); results.emplace_back(std::move(x), std::move(xs)...);
if (--*pending == 0) { if (--*pending == 0) {
timeouts.dispose(); timeouts.dispose();
f(std::move(results)); f(std::move(results));
...@@ -64,7 +67,34 @@ struct select_all_helper { ...@@ -64,7 +67,34 @@ struct select_all_helper {
} }
auto wrap() { auto wrap() {
return [this](Ts&... xs) { (*this)(xs...); }; return [this](T& x, Ts&... xs) { (*this)(x, xs...); };
}
};
template <class F>
struct select_all_helper<F> {
std::shared_ptr<size_t> pending;
disposable timeouts;
F f;
template <class Fun>
select_all_helper(size_t pending, disposable timeouts, Fun&& f)
: pending(std::make_shared<size_t>(pending)),
timeouts(std::move(timeouts)),
f(std::forward<Fun>(f)) {
// nop
}
void operator()() {
CAF_LOG_TRACE(CAF_ARG2("pending", *pending));
if (*pending > 0 && --*pending == 0) {
timeouts.dispose();
f();
}
}
auto wrap() {
return [this] { (*this)(); };
} }
}; };
...@@ -82,6 +112,11 @@ struct select_select_all_helper<F, detail::type_list<std::vector<T>>> { ...@@ -82,6 +112,11 @@ struct select_select_all_helper<F, detail::type_list<std::vector<T>>> {
using type = select_all_helper<F, T>; using type = select_all_helper<F, T>;
}; };
template <class F>
struct select_select_all_helper<F, detail::type_list<>> {
using type = select_all_helper<F>;
};
template <class F> template <class F>
using select_all_helper_t = typename select_select_all_helper<F>::type; using select_all_helper_t = typename select_select_all_helper<F>::type;
......
...@@ -19,7 +19,8 @@ namespace { ...@@ -19,7 +19,8 @@ namespace {
using discarding_server_type = typed_actor<replies_to<int, int>::with<void>>; using discarding_server_type = typed_actor<replies_to<int, int>::with<void>>;
using adding_server_type = typed_actor<replies_to<int, int>::with<int>>; using adding_server_type = typed_actor<result<int>(int, int)>;
using no_op_server_type = typed_actor<result<void>(int, int)>;
using result_type = variant<none_t, unit_t, int>; using result_type = variant<none_t, unit_t, int>;
...@@ -166,6 +167,30 @@ CAF_TEST(requesters support fan_out_request) { ...@@ -166,6 +167,30 @@ CAF_TEST(requesters support fan_out_request) {
CHECK_EQ(*sum, 9); CHECK_EQ(*sum, 9);
} }
CAF_TEST(requesters support fan_out_request with result<void>) {
using policy::select_all;
std::vector<no_op_server_type> workers{
make_server([](int, int) {}),
make_server([](int, int) {}),
make_server([](int, int) {}),
};
run();
auto ran = std::make_shared<bool>(false);
auto client = sys.spawn([=](event_based_actor* self) {
self->fan_out_request<select_all>(workers, infinite, 1, 2).then([=]() {
*ran = true;
});
});
run_once();
expect((int, int), from(client).to(workers[0]).with(1, 2));
expect((), from(workers[0]).to(client));
expect((int, int), from(client).to(workers[1]).with(1, 2));
expect((), from(workers[1]).to(client));
expect((int, int), from(client).to(workers[2]).with(1, 2));
expect((), from(workers[2]).to(client));
CHECK_EQ(*ran, true);
}
#ifdef CAF_ENABLE_EXCEPTIONS #ifdef CAF_ENABLE_EXCEPTIONS
CAF_TEST(exceptions while processing requests trigger error messages) { CAF_TEST(exceptions while processing requests trigger error messages) {
......
...@@ -250,18 +250,29 @@ caf::optional<std::tuple<T, Ts...>> try_extract(caf_handle x) { ...@@ -250,18 +250,29 @@ caf::optional<std::tuple<T, Ts...>> try_extract(caf_handle x) {
/// Returns the content of the next mailbox element without taking it out of /// Returns the content of the next mailbox element without taking it out of
/// the mailbox. Fails on an empty mailbox or if the content of the next /// the mailbox. Fails on an empty mailbox or if the content of the next
/// element does not match `<T, Ts...>`. /// element does not match `<T, Ts...>`.
template <class T, class... Ts> template <class... Ts>
std::tuple<T, Ts...> extract(caf_handle x, int src_line) { std::tuple<Ts...> extract(caf_handle x, int src_line) {
if (auto result = try_extract<T, Ts...>(x); result != caf::none) { if constexpr (sizeof...(Ts) > 0) {
return std::move(*result); if (auto result = try_extract<Ts...>(x)) {
return std::move(*result);
} else {
auto ptr = x->peek_at_next_mailbox_element();
if (ptr == nullptr)
CAF_FAIL("cannot peek at the next message: mailbox is empty", src_line);
else
CAF_FAIL("message does not match expected types: "
<< to_string(ptr->content()),
src_line);
}
} else { } else {
auto ptr = x->peek_at_next_mailbox_element(); auto ptr = x->peek_at_next_mailbox_element();
if (ptr == nullptr) if (ptr == nullptr)
CAF_FAIL("cannot peek at the next message: mailbox is empty", src_line); CAF_FAIL("cannot peek at the next message: mailbox is empty", src_line);
else else if (!ptr->content().empty())
CAF_FAIL("message does not match expected types: " CAF_FAIL("message does not match (expected an empty message): "
<< to_string(ptr->content()), << to_string(ptr->content()),
src_line); src_line);
return {};
} }
} }
......
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