Commit da9346e1 authored by Dominik Charousset's avatar Dominik Charousset

Add new choose_response policy

The choose policy simply picks the first valid result for a fan-out
request and ignores all other responses.
parent 35c5d3e1
...@@ -251,6 +251,7 @@ set(CAF_CORE_TEST_SOURCES ...@@ -251,6 +251,7 @@ set(CAF_CORE_TEST_SOURCES
test/or_else.cpp test/or_else.cpp
test/pipeline_streaming.cpp test/pipeline_streaming.cpp
test/policy/categorized.cpp test/policy/categorized.cpp
test/policy/choose_response.cpp
test/policy/fan_in_responses.cpp test/policy/fan_in_responses.cpp
test/request_timeout.cpp test/request_timeout.cpp
test/result.cpp test/result.cpp
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2020 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 <memory>
#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/logger.hpp"
#include "caf/sec.hpp"
namespace caf::detail {
template <class F, class = typename get_callable_trait<F>::arg_types>
struct choose_response_factory;
template <class F, class... Ts>
struct choose_response_factory<F, type_list<Ts...>> {
template <class Fun>
static auto make(std::shared_ptr<size_t> pending, Fun&& fun) {
return [pending, f{std::forward<Fun>(fun)}](Ts... xs) mutable {
CAF_LOG_TRACE(CAF_ARG2("pending", *pending));
if (*pending > 0) {
f(xs...);
*pending = 0;
}
};
}
};
} // namespace caf::detail
namespace caf::policy {
/// Enables a `response_handle` to choose one response for any number of
/// response messages.
/// @relates mixin::requester
/// @relates response_handle
template <class ResponseType>
class choose_response {
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::decay_t<Fun>>;
explicit choose_response(message_id_list 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) 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 factory = detail::choose_response_factory<std::decay_t<F>>;
auto pending = std::make_shared<size_t>(ids_.size());
auto fw = factory::make(pending, std::forward<F>(f));
auto gw = make_error_handler(std::move(pending), std::forward<G>(g));
for (auto id : ids_) {
typename Self::accept_one_cond rc;
auto fcopy = fw;
auto gcopy = gw;
self->varargs_receive(rc, id, fcopy, gcopy);
}
}
const message_id_list& ids() const noexcept {
return ids_;
}
private:
template <class OnError>
auto make_error_handler(std::shared_ptr<size_t> p, OnError&& g) const {
return [p{std::move(p)}, g{std::forward<OnError>(g)}](error&) mutable {
if (*p == 0) {
// nop
} else if (*p == 1) {
auto err = make_error(sec::all_requests_failed);
g(err);
} else {
--*p;
}
};
}
template <class F, class OnError>
behavior make_behavior(F&& f, OnError&& g) const {
using factory = detail::choose_response_factory<std::decay_t<F>>;
auto pending = std::make_shared<size_t>(ids_.size());
auto result_handler = factory::make(pending, std::forward<F>(f));
return {
std::move(result_handler),
make_error_handler(std::move(pending), std::forward<OnError>(g)),
};
}
message_id_list ids_;
};
} // namespace caf::policy
...@@ -130,6 +130,8 @@ enum class sec : uint8_t { ...@@ -130,6 +130,8 @@ enum class sec : uint8_t {
remote_lookup_failed, remote_lookup_failed,
/// Serialization failed because actor_system::tracing_context is null. /// Serialization failed because actor_system::tracing_context is null.
no_tracing_context, no_tracing_context,
/// No request produced a valid result.
all_requests_failed,
}; };
/// @relates sec /// @relates sec
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright 2011-2020 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.choose_response
#include "caf/policy/choose_response.hpp"
#include "caf/test/dsl.hpp"
#include "caf/actor_system.hpp"
#include "caf/event_based_actor.hpp"
#include "caf/sec.hpp"
using caf::policy::choose_response;
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);
}
auto make_error_handler() {
return [this](const error& err) {
CAF_FAIL("unexpected error: " << sys.render(err));
};
}
auto 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(choose_response_tests, fixture)
CAF_TEST(choose_response picks the first arriving integer) {
auto f = [](int x, int y) { return x + y; };
auto server1 = make_server(f);
auto server2 = make_server(f);
SUBTEST("request.receive") {
SUBTEST("single integer") {
auto r1 = self->request(server1, infinite, 1, 2);
auto r2 = self->request(server2, infinite, 2, 3);
choose_response<detail::type_list<int>> choose{{r1.id(), r2.id()}};
run();
choose.receive(
self.ptr(), [](int result) { CAF_CHECK_EQUAL(result, 3); },
make_error_handler());
}
}
SUBTEST("request.then") {
int result = 0;
auto client = sys.spawn([=, &result](event_based_actor* client_ptr) {
auto r1 = client_ptr->request(server1, infinite, 1, 2);
auto r2 = client_ptr->request(server2, infinite, 2, 3);
choose_response<detail::type_list<int>> choose{{r1.id(), r2.id()}};
choose.then(
client_ptr, [&result](int x) { result = x; }, 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 picks the first arriving result");
CAF_CHECK_EQUAL(result, 3);
}
SUBTEST("request.await") {
int result = 0;
auto client = sys.spawn([=, &result](event_based_actor* client_ptr) {
auto r1 = client_ptr->request(server1, infinite, 1, 2);
auto r2 = client_ptr->request(server2, infinite, 2, 3);
choose_response<detail::type_list<int>> choose{{r1.id(), r2.id()}};
choose.await(
client_ptr, [&result](int x) { result = x; }, 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(result, 5);
}
}
CAF_TEST(choose_response calls the error handler at most once) {
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);
choose_response<detail::type_list<int>> choose{{r1.id(), r2.id()}};
run();
size_t errors = 0;
choose.receive(
self.ptr(),
[](int) { 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);
choose_response<detail::type_list<int>> choose{{r1.id(), r2.id()}};
choose.then(
client_ptr,
[](int) { 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);
choose_response<detail::type_list<int>> choose{{r1.id(), r2.id()}};
choose.await(
client_ptr,
[](int) { 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