Commit d17d2b03 authored by ufownl's avatar ufownl Committed by Dominik Charousset

Requests multiplexing in event-based actors

parent f5559262
...@@ -25,6 +25,7 @@ ...@@ -25,6 +25,7 @@
#include <exception> #include <exception>
#include <functional> #include <functional>
#include <forward_list> #include <forward_list>
#include <unordered_map>
#include "caf/fwd.hpp" #include "caf/fwd.hpp"
...@@ -514,11 +515,11 @@ public: ...@@ -514,11 +515,11 @@ public:
} }
inline bool has_behavior() const { inline bool has_behavior() const {
return ! bhvr_stack_.empty() || ! pending_responses_.empty(); return ! bhvr_stack_.empty()
|| ! awaited_responses_.empty()
|| ! multiplexed_responses_.empty();
} }
behavior& get_behavior();
virtual void initialize() = 0; virtual void initialize() = 0;
// clear behavior stack and call cleanup if actor either has no // clear behavior stack and call cleanup if actor either has no
...@@ -548,25 +549,35 @@ public: ...@@ -548,25 +549,35 @@ public:
using error_handler = std::function<void (error&)>; using error_handler = std::function<void (error&)>;
using pending_response = std::tuple<message_id, behavior, error_handler>; using pending_response =
std::pair<const message_id, std::pair<behavior, error_handler>>;
message_id new_request_id(message_priority mp); message_id new_request_id(message_priority mp);
void mark_arrived(message_id response_id); void mark_awaited_arrived(message_id mid);
bool awaits_response() const; bool awaits_response() const;
bool awaits(message_id response_id) const; bool awaits(message_id mid) const;
maybe<pending_response&> find_pending_response(message_id mid); maybe<pending_response&> find_awaited_response(message_id mid);
void set_response_handler(message_id response_id, behavior bhvr, void set_awaited_response_handler(message_id response_id, behavior bhvr,
error_handler f = nullptr); error_handler f = nullptr);
behavior& awaited_response_handler(); behavior& awaited_response_handler();
message_id awaited_response_id(); message_id awaited_response_id();
void mark_multiplexed_arrived(message_id mid);
bool multiplexes(message_id mid) const;
maybe<pending_response&> find_multiplexed_response(message_id mid);
void set_multiplexed_response_handler(message_id response_id, behavior bhvr,
error_handler f = nullptr);
// these functions are dispatched via the actor policies table // these functions are dispatched via the actor policies table
void launch(execution_unit* eu, bool lazy, bool hide); void launch(execution_unit* eu, bool lazy, bool hide);
...@@ -598,7 +609,13 @@ protected: ...@@ -598,7 +609,13 @@ protected:
message_id last_request_id_; message_id last_request_id_;
// identifies all IDs of sync messages waiting for a response // identifies all IDs of sync messages waiting for a response
std::forward_list<pending_response> pending_responses_; std::forward_list<pending_response> awaited_responses_;
// identifies all IDs of async messages waiting for a response
std::unordered_map<
message_id,
std::pair<behavior, error_handler>
> multiplexed_responses_;
// points to dummy_node_ if no callback is currently invoked, // points to dummy_node_ if no callback is currently invoked,
// points to the node under processing otherwise // points to the node under processing otherwise
......
...@@ -79,6 +79,31 @@ public: ...@@ -79,6 +79,31 @@ public:
using error_handler = std::function<void (error&)>; using error_handler = std::function<void (error&)>;
template <class F, class T>
typename get_continue_helper<Output, F>::type
await(F f, error_handler ef, timeout_definition<T> tdef) const {
return await_impl(f, ef, std::move(tdef));
}
template <class F>
typename get_continue_helper<Output, F>::type
await(F f, error_handler ef = nullptr) const {
return await_impl(f, ef);
}
template <class F, class T>
typename get_continue_helper<Output, F>::type
await(F f, timeout_definition<T> tdef) const {
return await(std::move(f), nullptr, std::move(tdef));
}
void generic_await(std::function<void (message&)> f, error_handler ef) {
behavior tmp{
others >> f
};
self_->set_awaited_response_handler(mid_, behavior{std::move(tmp)}, std::move(ef));
}
template <class F, class T> template <class F, class T>
typename get_continue_helper<Output, F>::type typename get_continue_helper<Output, F>::type
then(F f, error_handler ef, timeout_definition<T> tdef) const { then(F f, error_handler ef, timeout_definition<T> tdef) const {
...@@ -101,23 +126,35 @@ public: ...@@ -101,23 +126,35 @@ public:
behavior tmp{ behavior tmp{
others >> f others >> f
}; };
self_->set_response_handler(mid_, behavior{std::move(tmp)}, std::move(ef)); self_->set_multiplexed_response_handler(mid_, behavior{std::move(tmp)}, std::move(ef));
} }
private: private:
template <class F, class... Ts> template <class F, class... Ts>
typename get_continue_helper<Output, F>::type typename get_continue_helper<Output, F>::type
then_impl(F& f, error_handler& ef, Ts&&... xs) const { await_impl(F& f, error_handler& ef, Ts&&... xs) const {
static_assert(detail::is_callable<F>::value, "argument is not callable"); static_assert(detail::is_callable<F>::value, "argument is not callable");
static_assert(! std::is_base_of<match_case, F>::value, static_assert(! std::is_base_of<match_case, F>::value,
"match cases are not allowed in this context"); "match cases are not allowed in this context");
detail::type_checker<Output, F>::check(); detail::type_checker<Output, F>::check();
self_->set_response_handler(mid_, self_->set_awaited_response_handler(mid_,
behavior{std::move(f), std::forward<Ts>(xs)...}, behavior{std::move(f), std::forward<Ts>(xs)...},
std::move(ef)); std::move(ef));
return {mid_}; return {mid_};
} }
template <class F, class... Ts>
typename get_continue_helper<Output, F>::type
then_impl(F& f, error_handler& ef, Ts&&... xs) const {
static_assert(detail::is_callable<F>::value, "argument is not callable");
static_assert(! std::is_base_of<match_case, F>::value,
"match cases are not allowed in this context");
detail::type_checker<Output, F>::check();
self_->set_multiplexed_response_handler(mid_,
behavior{std::move(f), std::forward<Ts>(xs)...},
std::move(ef));
return {mid_};
}
message_id mid_; message_id mid_;
Self* self_; Self* self_;
...@@ -140,23 +177,23 @@ public: ...@@ -140,23 +177,23 @@ public:
using error_handler = std::function<void (error&)>; using error_handler = std::function<void (error&)>;
template <class F, class T> template <class F, class T>
void await(F f, error_handler ef, timeout_definition<T> tdef) { void receive(F f, error_handler ef, timeout_definition<T> tdef) {
await_impl(f, ef, std::move(tdef)); receive_impl(f, ef, std::move(tdef));
} }
template <class F> template <class F>
void await(F f, error_handler ef = nullptr) { void receive(F f, error_handler ef = nullptr) {
await_impl(f, ef); receive_impl(f, ef);
} }
template <class F, class T> template <class F, class T>
void await(F f, timeout_definition<T> tdef) { void receive(F f, timeout_definition<T> tdef) {
await(std::move(f), nullptr, std::move(tdef)); receive(std::move(f), nullptr, std::move(tdef));
} }
private: private:
template <class F, class... Ts> template <class F, class... Ts>
void await_impl(F& f, error_handler& ef, Ts&&... xs) { void receive_impl(F& f, error_handler& ef, Ts&&... xs) {
static_assert(detail::is_callable<F>::value, "argument is not callable"); static_assert(detail::is_callable<F>::value, "argument is not callable");
static_assert(! std::is_base_of<match_case, F>::value, static_assert(! std::is_base_of<match_case, F>::value,
"match cases are not allowed in this context"); "match cases are not allowed in this context");
......
...@@ -50,6 +50,9 @@ void blocking_actor::initialize() { ...@@ -50,6 +50,9 @@ void blocking_actor::initialize() {
void blocking_actor::dequeue(behavior& bhvr, message_id mid) { void blocking_actor::dequeue(behavior& bhvr, message_id mid) {
CAF_LOG_TRACE(CAF_ARG(mid)); CAF_LOG_TRACE(CAF_ARG(mid));
// push an empty sync response handler for `blocking_actor`
if (mid != invalid_message_id && ! find_awaited_response(mid))
awaited_responses_.emplace_front(mid, std::make_pair(behavior{}, nullptr));
// try to dequeue from cache first // try to dequeue from cache first
if (invoke_from_cache(bhvr, mid)) { if (invoke_from_cache(bhvr, mid)) {
return; return;
......
This diff is collapsed.
...@@ -84,7 +84,7 @@ CAF_TEST(round_robin_actor_pool) { ...@@ -84,7 +84,7 @@ CAF_TEST(round_robin_actor_pool) {
self->send(w, sys_atom::value, put_atom::value, spawn_worker()); self->send(w, sys_atom::value, put_atom::value, spawn_worker());
std::vector<actor_addr> workers; std::vector<actor_addr> workers;
for (int i = 0; i < 6; ++i) { for (int i = 0; i < 6; ++i) {
self->request(w, i, i).await( self->request(w, i, i).receive(
[&](int res) { [&](int res) {
CAF_CHECK_EQUAL(res, i + i); CAF_CHECK_EQUAL(res, i + i);
auto sender = self->current_sender(); auto sender = self->current_sender();
...@@ -99,7 +99,7 @@ CAF_TEST(round_robin_actor_pool) { ...@@ -99,7 +99,7 @@ CAF_TEST(round_robin_actor_pool) {
return addr == invalid_actor_addr; return addr == invalid_actor_addr;
}; };
CAF_CHECK(std::none_of(workers.begin(), workers.end(), is_invalid)); CAF_CHECK(std::none_of(workers.begin(), workers.end(), is_invalid));
self->request(w, sys_atom::value, get_atom::value).await( self->request(w, sys_atom::value, get_atom::value).receive(
[&](std::vector<actor>& ws) { [&](std::vector<actor>& ws) {
std::sort(workers.begin(), workers.end()); std::sort(workers.begin(), workers.end());
std::sort(ws.begin(), ws.end()); std::sort(ws.begin(), ws.end());
...@@ -118,7 +118,7 @@ CAF_TEST(round_robin_actor_pool) { ...@@ -118,7 +118,7 @@ CAF_TEST(round_robin_actor_pool) {
CAF_CHECK(dm.source == workers.back()); CAF_CHECK(dm.source == workers.back());
workers.pop_back(); workers.pop_back();
// check whether actor pool removed failed worker // check whether actor pool removed failed worker
self->request(w, sys_atom::value, get_atom::value).await( self->request(w, sys_atom::value, get_atom::value).receive(
[&](std::vector<actor>& ws) { [&](std::vector<actor>& ws) {
std::sort(ws.begin(), ws.end()); std::sort(ws.begin(), ws.end());
CAF_CHECK(workers.size() == ws.size() CAF_CHECK(workers.size() == ws.size()
...@@ -177,7 +177,7 @@ CAF_TEST(random_actor_pool) { ...@@ -177,7 +177,7 @@ CAF_TEST(random_actor_pool) {
scoped_actor self{system}; scoped_actor self{system};
auto w = actor_pool::make(&context, 5, spawn_worker, actor_pool::random()); auto w = actor_pool::make(&context, 5, spawn_worker, actor_pool::random());
for (int i = 0; i < 5; ++i) { for (int i = 0; i < 5; ++i) {
self->request(w, 1, 2).await( self->request(w, 1, 2).receive(
[&](int res) { [&](int res) {
CAF_CHECK_EQUAL(res, 3); CAF_CHECK_EQUAL(res, 3);
}, },
...@@ -212,12 +212,12 @@ CAF_TEST(split_join_actor_pool) { ...@@ -212,12 +212,12 @@ CAF_TEST(split_join_actor_pool) {
scoped_actor self{system}; scoped_actor self{system};
auto w = actor_pool::make(&context, 5, spawn_split_worker, auto w = actor_pool::make(&context, 5, spawn_split_worker,
actor_pool::split_join<int>(join_fun, split_fun)); actor_pool::split_join<int>(join_fun, split_fun));
self->request(w, std::vector<int>{1, 2, 3, 4, 5}).await( self->request(w, std::vector<int>{1, 2, 3, 4, 5}).receive(
[&](int res) { [&](int res) {
CAF_CHECK_EQUAL(res, 15); CAF_CHECK_EQUAL(res, 15);
} }
); );
self->request(w, std::vector<int>{6, 7, 8, 9, 10}).await( self->request(w, std::vector<int>{6, 7, 8, 9, 10}).receive(
[&](int res) { [&](int res) {
CAF_CHECK_EQUAL(res, 40); CAF_CHECK_EQUAL(res, 40);
} }
......
...@@ -133,7 +133,7 @@ testee::behavior_type testee_impl(testee::pointer self) { ...@@ -133,7 +133,7 @@ testee::behavior_type testee_impl(testee::pointer self) {
CAF_TEST(request_atom_constants) { CAF_TEST(request_atom_constants) {
scoped_actor self{system}; scoped_actor self{system};
auto tst = system.spawn(testee_impl); auto tst = system.spawn(testee_impl);
self->request(tst, abc_atom::value).await( self->request(tst, abc_atom::value).receive(
[](int i) { [](int i) {
CAF_CHECK_EQUAL(i, 42); CAF_CHECK_EQUAL(i, 42);
} }
......
...@@ -514,7 +514,7 @@ CAF_TEST(requests) { ...@@ -514,7 +514,7 @@ CAF_TEST(requests) {
auto sync_testee = system.spawn<blocking_api>([](blocking_actor* s) { auto sync_testee = system.spawn<blocking_api>([](blocking_actor* s) {
s->receive ( s->receive (
on("hi", arg_match) >> [&](actor from) { on("hi", arg_match) >> [&](actor from) {
s->request(from, "whassup?", s).await( s->request(from, "whassup?", s).receive(
[&](const string& str) -> string { [&](const string& str) -> string {
CAF_CHECK(s->current_sender() != nullptr); CAF_CHECK(s->current_sender() != nullptr);
CAF_CHECK_EQUAL(str, "nothing"); CAF_CHECK_EQUAL(str, "nothing");
...@@ -552,7 +552,7 @@ CAF_TEST(requests) { ...@@ -552,7 +552,7 @@ CAF_TEST(requests) {
} }
); );
self->await_all_other_actors_done(); self->await_all_other_actors_done();
self->request(sync_testee, "!?").await( self->request(sync_testee, "!?").receive(
[] { [] {
CAF_TEST_ERROR("Unexpected empty message"); CAF_TEST_ERROR("Unexpected empty message");
}, },
...@@ -611,7 +611,7 @@ typed_testee::behavior_type testee() { ...@@ -611,7 +611,7 @@ typed_testee::behavior_type testee() {
CAF_TEST(typed_await) { CAF_TEST(typed_await) {
scoped_actor self{system}; scoped_actor self{system};
auto x = system.spawn(testee); auto x = system.spawn(testee);
self->request(x, abc_atom::value).await( self->request(x, abc_atom::value).receive(
[](const std::string& str) { [](const std::string& str) {
CAF_CHECK_EQUAL(str, "abc"); CAF_CHECK_EQUAL(str, "abc");
} }
...@@ -790,7 +790,7 @@ CAF_TEST(move_only_argument) { ...@@ -790,7 +790,7 @@ CAF_TEST(move_only_argument) {
}; };
auto testee = system.spawn(f, std::move(uptr)); auto testee = system.spawn(f, std::move(uptr));
scoped_actor self{system}; scoped_actor self{system};
self->request(testee, 1.f).await( self->request(testee, 1.f).receive(
[](int i) { [](int i) {
CAF_CHECK(i == 42); CAF_CHECK(i == 42);
} }
......
...@@ -83,12 +83,12 @@ CAF_TEST(migrate_locally) { ...@@ -83,12 +83,12 @@ CAF_TEST(migrate_locally) {
scoped_actor self{system}; scoped_actor self{system};
self->send(a, put_atom::value, 42); self->send(a, put_atom::value, 42);
// migrate from a to b // migrate from a to b
self->request(a, sys_atom::value, migrate_atom::value, mm1).await( self->request(a, sys_atom::value, migrate_atom::value, mm1).receive(
[&](ok_atom, const actor_addr& dest) { [&](ok_atom, const actor_addr& dest) {
CAF_CHECK(dest == b); CAF_CHECK(dest == b);
} }
); );
self->request(a, get_atom::value).await( self->request(a, get_atom::value).receive(
[&](int result) { [&](int result) {
CAF_CHECK(result == 42); CAF_CHECK(result == 42);
CAF_CHECK(self->current_sender() == b.address()); CAF_CHECK(self->current_sender() == b.address());
...@@ -97,12 +97,12 @@ CAF_TEST(migrate_locally) { ...@@ -97,12 +97,12 @@ CAF_TEST(migrate_locally) {
auto mm2 = system.spawn(pseudo_mm, a); auto mm2 = system.spawn(pseudo_mm, a);
self->send(b, put_atom::value, 23); self->send(b, put_atom::value, 23);
// migrate back from b to a // migrate back from b to a
self->request(b, sys_atom::value, migrate_atom::value, mm2).await( self->request(b, sys_atom::value, migrate_atom::value, mm2).receive(
[&](ok_atom, const actor_addr& dest) { [&](ok_atom, const actor_addr& dest) {
CAF_CHECK(dest == a); CAF_CHECK(dest == a);
} }
); );
self->request(b, get_atom::value).await( self->request(b, get_atom::value).receive(
[&](int result) { [&](int result) {
CAF_CHECK(result == 23); CAF_CHECK(result == 23);
CAF_CHECK(self->current_sender() == a.address()); CAF_CHECK(self->current_sender() == a.address());
......
...@@ -49,13 +49,13 @@ struct fixture { ...@@ -49,13 +49,13 @@ struct fixture {
void run_testee(actor testee) { void run_testee(actor testee) {
scoped_actor self{system}; scoped_actor self{system};
self->request(testee, a_atom::value).await([](int i) { self->request(testee, a_atom::value).receive([](int i) {
CAF_CHECK_EQUAL(i, 1); CAF_CHECK_EQUAL(i, 1);
}); });
self->request(testee, b_atom::value).await([](int i) { self->request(testee, b_atom::value).receive([](int i) {
CAF_CHECK_EQUAL(i, 2); CAF_CHECK_EQUAL(i, 2);
}); });
self->request(testee, c_atom::value).await([](int i) { self->request(testee, c_atom::value).receive([](int i) {
CAF_CHECK_EQUAL(i, 3); CAF_CHECK_EQUAL(i, 3);
}); });
self->send_exit(testee, exit_reason::user_shutdown); self->send_exit(testee, exit_reason::user_shutdown);
......
...@@ -111,7 +111,7 @@ struct fixture { ...@@ -111,7 +111,7 @@ struct fixture {
CAF_REQUIRE(config_server != invalid_actor); CAF_REQUIRE(config_server != invalid_actor);
// clear config // clear config
scoped_actor self{system}; scoped_actor self{system};
self->request(config_server, get_atom::value, "*").await( self->request(config_server, get_atom::value, "*").receive(
[&](ok_atom, std::vector<std::pair<std::string, message>>& msgs) { [&](ok_atom, std::vector<std::pair<std::string, message>>& msgs) {
for (auto& kvp : msgs) for (auto& kvp : msgs)
self->send(config_server, put_atom::value, kvp.first, message{}); self->send(config_server, put_atom::value, kvp.first, message{});
...@@ -151,7 +151,7 @@ struct fixture { ...@@ -151,7 +151,7 @@ struct fixture {
>::type; >::type;
bool result = false; bool result = false;
scoped_actor self{system}; scoped_actor self{system};
self->request(config_server, get_atom::value, key).await( self->request(config_server, get_atom::value, key).receive(
[&](ok_atom, std::string&, message& msg) { [&](ok_atom, std::string&, message& msg) {
msg.apply( msg.apply(
[&](type& val) { [&](type& val) {
...@@ -186,7 +186,7 @@ struct fixture { ...@@ -186,7 +186,7 @@ struct fixture {
if (config_server != invalid_actor) { if (config_server != invalid_actor) {
size_t result = 0; size_t result = 0;
scoped_actor self{system}; scoped_actor self{system};
self->request(config_server, get_atom::value, "*").await( self->request(config_server, get_atom::value, "*").receive(
[&](ok_atom, std::vector<std::pair<std::string, message>>& msgs) { [&](ok_atom, std::vector<std::pair<std::string, message>>& msgs) {
for (auto& kvp : msgs) for (auto& kvp : msgs)
if (! kvp.second.empty()) if (! kvp.second.empty())
......
...@@ -268,7 +268,7 @@ CAF_TEST(test_void_res) { ...@@ -268,7 +268,7 @@ CAF_TEST(test_void_res) {
}; };
}); });
scoped_actor self{system}; scoped_actor self{system};
self->request(buddy, 1, 2).await( self->request(buddy, 1, 2).receive(
[] { [] {
CAF_MESSAGE("received void res"); CAF_MESSAGE("received void res");
} }
...@@ -309,7 +309,7 @@ CAF_TEST(request) { ...@@ -309,7 +309,7 @@ CAF_TEST(request) {
CAF_CHECK_EQUAL(i, 0); CAF_CHECK_EQUAL(i, 0);
} }
); );
s->request(foi, i_atom::value).await( s->request(foi, i_atom::value).receive(
[&](int i) { [&](int i) {
CAF_CHECK_EQUAL(i, 0); CAF_CHECK_EQUAL(i, 0);
++invocations; ++invocations;
...@@ -318,7 +318,7 @@ CAF_TEST(request) { ...@@ -318,7 +318,7 @@ CAF_TEST(request) {
CAF_TEST_ERROR("Error: " << s->system().render(err)); CAF_TEST_ERROR("Error: " << s->system().render(err));
} }
); );
s->request(foi, f_atom::value).await( s->request(foi, f_atom::value).receive(
[&](float f) { [&](float f) {
CAF_CHECK_EQUAL(f, 0.f); CAF_CHECK_EQUAL(f, 0.f);
++invocations; ++invocations;
...@@ -332,7 +332,7 @@ CAF_TEST(request) { ...@@ -332,7 +332,7 @@ CAF_TEST(request) {
// provoke invocation of s->handle_sync_failure() // provoke invocation of s->handle_sync_failure()
bool error_handler_called = false; bool error_handler_called = false;
bool int_handler_called = false; bool int_handler_called = false;
s->request(foi, f_atom::value).await( s->request(foi, f_atom::value).receive(
[&](int) { [&](int) {
int_handler_called = true; int_handler_called = true;
}, },
...@@ -356,7 +356,7 @@ CAF_TEST(request) { ...@@ -356,7 +356,7 @@ CAF_TEST(request) {
); );
auto mirror = system.spawn<sync_mirror>(); auto mirror = system.spawn<sync_mirror>();
bool continuation_called = false; bool continuation_called = false;
self->request(mirror, 42).await([&](int value) { self->request(mirror, 42).receive([&](int value) {
continuation_called = true; continuation_called = true;
CAF_CHECK_EQUAL(value, 42); CAF_CHECK_EQUAL(value, 42);
}); });
...@@ -396,7 +396,7 @@ CAF_TEST(request) { ...@@ -396,7 +396,7 @@ CAF_TEST(request) {
CAF_MESSAGE("block on `await_all_other_actors_done`"); CAF_MESSAGE("block on `await_all_other_actors_done`");
self->await_all_other_actors_done(); self->await_all_other_actors_done();
CAF_MESSAGE("`await_all_other_actors_done` finished"); CAF_MESSAGE("`await_all_other_actors_done` finished");
self->request(self, no_way_atom::value).await( self->request(self, no_way_atom::value).receive(
[&](int) { [&](int) {
CAF_TEST_ERROR("Unexpected message"); CAF_TEST_ERROR("Unexpected message");
}, },
...@@ -437,7 +437,7 @@ CAF_TEST(request) { ...@@ -437,7 +437,7 @@ CAF_TEST(request) {
auto c = self->spawn<C>(); // replies only to 'gogo' messages auto c = self->spawn<C>(); // replies only to 'gogo' messages
// first test: sync error must occur, continuation must not be called // first test: sync error must occur, continuation must not be called
bool timeout_occured = false; bool timeout_occured = false;
self->request(c, milliseconds(500), hi_there_atom::value).await( self->request(c, milliseconds(500), hi_there_atom::value).receive(
[&](hi_there_atom) { [&](hi_there_atom) {
CAF_TEST_ERROR("C did reply to 'HiThere'"); CAF_TEST_ERROR("C did reply to 'HiThere'");
}, },
...@@ -451,7 +451,7 @@ CAF_TEST(request) { ...@@ -451,7 +451,7 @@ CAF_TEST(request) {
} }
); );
CAF_CHECK_EQUAL(timeout_occured, true); CAF_CHECK_EQUAL(timeout_occured, true);
self->request(c, gogo_atom::value).await( self->request(c, gogo_atom::value).receive(
[](gogogo_atom) { [](gogogo_atom) {
CAF_MESSAGE("received `gogogo_atom`"); CAF_MESSAGE("received `gogogo_atom`");
}, },
...@@ -476,7 +476,7 @@ CAF_TEST(request) { ...@@ -476,7 +476,7 @@ CAF_TEST(request) {
}); });
// first 'idle', then 'request' // first 'idle', then 'request'
anon_send(serv, idle_atom::value, work); anon_send(serv, idle_atom::value, work);
s->request(serv, request_atom::value).await( s->request(serv, request_atom::value).receive(
[&](response_atom) { [&](response_atom) {
CAF_MESSAGE("received `response_atom`"); CAF_MESSAGE("received `response_atom`");
CAF_CHECK(s->current_sender() == work); CAF_CHECK(s->current_sender() == work);
...@@ -488,7 +488,7 @@ CAF_TEST(request) { ...@@ -488,7 +488,7 @@ CAF_TEST(request) {
// first 'request', then 'idle' // first 'request', then 'idle'
auto handle = s->request(serv, request_atom::value); auto handle = s->request(serv, request_atom::value);
send_as(work, serv, idle_atom::value, work); send_as(work, serv, idle_atom::value, work);
handle.await( handle.receive(
[&](response_atom) { [&](response_atom) {
CAF_CHECK(s->current_sender() == work); CAF_CHECK(s->current_sender() == work);
}, },
...@@ -529,4 +529,26 @@ CAF_TEST(request_no_then) { ...@@ -529,4 +529,26 @@ CAF_TEST(request_no_then) {
anon_send(system.spawn(snyc_send_no_then_B), 8); anon_send(system.spawn(snyc_send_no_then_B), 8);
} }
CAF_TEST(async_request) {
auto foo = system.spawn([](event_based_actor* self) -> behavior {
auto receiver = self->spawn<linked>([](event_based_actor* self) -> behavior{
return {
[=](int) {
return self->make_response_promise();
}
};
});
self->request(receiver, 1).then(
[=](int) {}
);
return {
[=](int) {
CAF_MESSAGE("int received");
self->quit(exit_reason::user_shutdown);
}
};
});
anon_send(foo, 1);
}
CAF_TEST_FIXTURE_SCOPE_END() CAF_TEST_FIXTURE_SCOPE_END()
...@@ -87,6 +87,77 @@ behavior ping2(event_based_actor* self, const actor& pong_actor) { ...@@ -87,6 +87,77 @@ behavior ping2(event_based_actor* self, const actor& pong_actor) {
}; };
} }
behavior ping3(event_based_actor* self, const actor& pong_actor) {
self->link_to(pong_actor);
self->send(self, send_ping_atom::value);
return {
[=](send_ping_atom) {
self->request(pong_actor, ping_atom::value).then(
[=](pong_atom) {
CAF_TEST_ERROR("received pong atom");
self->quit(exit_reason::user_shutdown);
},
after(std::chrono::milliseconds(100)) >> [=] {
CAF_MESSAGE("async timeout: check");
self->quit(exit_reason::user_shutdown);
}
);
}
};
}
behavior ping4(event_based_actor* self, const actor& pong_actor) {
self->link_to(pong_actor);
self->send(self, send_ping_atom::value);
auto received_outer = std::make_shared<bool>(false);
return {
[=](send_ping_atom) {
self->request(pong_actor, ping_atom::value).then(
[=](pong_atom) {
CAF_TEST_ERROR("received pong atom");
self->quit(exit_reason::user_shutdown);
},
after(std::chrono::milliseconds(100)) >> [=] {
CAF_CHECK_EQUAL(*received_outer, true);
self->quit(exit_reason::user_shutdown);
}
);
},
after(std::chrono::milliseconds(50)) >> [=] {
CAF_MESSAGE("outer timeout: check");
*received_outer = true;
}
};
}
void ping5(event_based_actor* self, const actor& pong_actor) {
self->link_to(pong_actor);
auto flag = std::make_shared<int>(0);
self->request(pong_actor, ping_atom::value).then(
[=](pong_atom) {
CAF_TEST_ERROR("received pong atom");
*flag = 1;
},
after(std::chrono::milliseconds(100)) >> [=] {
CAF_MESSAGE("multiplexed response timeout: check");
CAF_CHECK_EQUAL(*flag, 4);
*flag = 2;
self->quit(exit_reason::user_shutdown);
}
);
self->request(pong_actor, ping_atom::value).await(
[=](pong_atom) {
CAF_TEST_ERROR("received pong atom");
*flag = 3;
},
after(std::chrono::milliseconds(100)) >> [=] {
CAF_MESSAGE("awaited response timeout: check");
CAF_CHECK_EQUAL(*flag, 0);
*flag = 4;
}
);
}
struct fixture { struct fixture {
actor_system system; actor_system system;
}; };
...@@ -97,10 +168,16 @@ CAF_TEST_FIXTURE_SCOPE(atom_tests, fixture) ...@@ -97,10 +168,16 @@ CAF_TEST_FIXTURE_SCOPE(atom_tests, fixture)
CAF_TEST(single_timeout) { CAF_TEST(single_timeout) {
system.spawn(ping1, system.spawn(pong)); system.spawn(ping1, system.spawn(pong));
system.spawn(ping3, system.spawn(pong));
} }
CAF_TEST(scoped_timeout) { CAF_TEST(scoped_timeout) {
system.spawn(ping2, system.spawn(pong)); system.spawn(ping2, system.spawn(pong));
system.spawn(ping4, system.spawn(pong));
}
CAF_TEST(awaited_multiplexed_timeout) {
system.spawn(ping5, system.spawn(pong));
} }
CAF_TEST_FIXTURE_SCOPE_END() CAF_TEST_FIXTURE_SCOPE_END()
...@@ -89,7 +89,7 @@ CAF_TEST(test_serial_reply) { ...@@ -89,7 +89,7 @@ CAF_TEST(test_serial_reply) {
}); });
scoped_actor self{system}; scoped_actor self{system};
CAF_MESSAGE("ID of main: " << self->id()); CAF_MESSAGE("ID of main: " << self->id());
self->request(master, hi_atom::value).await( self->request(master, hi_atom::value).receive(
[](ho_atom) { [](ho_atom) {
CAF_MESSAGE("received 'ho'"); CAF_MESSAGE("received 'ho'");
}, },
......
...@@ -93,7 +93,7 @@ struct fixture { ...@@ -93,7 +93,7 @@ struct fixture {
self->send(aut, add_atom::value, 7); self->send(aut, add_atom::value, 7);
self->send(aut, add_atom::value, 4); self->send(aut, add_atom::value, 4);
self->send(aut, add_atom::value, 9); self->send(aut, add_atom::value, 9);
self->request(aut, get_atom::value).await( self->request(aut, get_atom::value).receive(
[](int x) { [](int x) {
CAF_CHECK_EQUAL(x, 20); CAF_CHECK_EQUAL(x, 20);
} }
...@@ -110,7 +110,7 @@ struct fixture { ...@@ -110,7 +110,7 @@ struct fixture {
}; };
}); });
scoped_actor self{system}; scoped_actor self{system};
self->request(aut, get_atom::value).await( self->request(aut, get_atom::value).receive(
[&](const string& str) { [&](const string& str) {
CAF_CHECK_EQUAL(str, expected); CAF_CHECK_EQUAL(str, expected);
} }
......
...@@ -108,12 +108,12 @@ CAF_TEST_FIXTURE_SCOPE(typed_spawn_tests, fixture) ...@@ -108,12 +108,12 @@ CAF_TEST_FIXTURE_SCOPE(typed_spawn_tests, fixture)
CAF_TEST(typed_response_promise) { CAF_TEST(typed_response_promise) {
scoped_actor self{system}; scoped_actor self{system};
auto foo = self->spawn<foo_actor_impl>(); auto foo = self->spawn<foo_actor_impl>();
self->request(foo, get_atom::value, 42).await( self->request(foo, get_atom::value, 42).receive(
[](int x) { [](int x) {
CAF_CHECK_EQUAL(x, 84); CAF_CHECK_EQUAL(x, 84);
} }
); );
self->request(foo, get_atom::value, 42, 52).await( self->request(foo, get_atom::value, 42, 52).receive(
[](int x, int y) { [](int x, int y) {
CAF_CHECK_EQUAL(x, 84); CAF_CHECK_EQUAL(x, 84);
CAF_CHECK_EQUAL(y, 104); CAF_CHECK_EQUAL(y, 104);
......
...@@ -146,12 +146,12 @@ void test_typed_spawn(server_type ts) { ...@@ -146,12 +146,12 @@ void test_typed_spawn(server_type ts) {
CAF_CHECK_EQUAL(value, true); CAF_CHECK_EQUAL(value, true);
} }
); );
self->request(ts, my_request{10, 20}).await( self->request(ts, my_request{10, 20}).receive(
[](bool value) { [](bool value) {
CAF_CHECK_EQUAL(value, false); CAF_CHECK_EQUAL(value, false);
} }
); );
self->request(ts, my_request{0, 0}).await( self->request(ts, my_request{0, 0}).receive(
[](bool value) { [](bool value) {
CAF_CHECK_EQUAL(value, true); CAF_CHECK_EQUAL(value, true);
} }
...@@ -417,7 +417,7 @@ CAF_TEST(reverter_relay_chain) { ...@@ -417,7 +417,7 @@ CAF_TEST(reverter_relay_chain) {
true); true);
set<string> iface{"caf::replies_to<@str>::with<@str>"}; set<string> iface{"caf::replies_to<@str>::with<@str>"};
CAF_CHECK(aut->message_types() == iface); CAF_CHECK(aut->message_types() == iface);
self->request(aut, "Hello World!").await( self->request(aut, "Hello World!").receive(
[](const string& answer) { [](const string& answer) {
CAF_CHECK_EQUAL(answer, "!dlroW olleH"); CAF_CHECK_EQUAL(answer, "!dlroW olleH");
} }
...@@ -434,7 +434,7 @@ CAF_TEST(string_delegator_chain) { ...@@ -434,7 +434,7 @@ CAF_TEST(string_delegator_chain) {
true); true);
set<string> iface{"caf::replies_to<@str>::with<@str>"}; set<string> iface{"caf::replies_to<@str>::with<@str>"};
CAF_CHECK(aut->message_types() == iface); CAF_CHECK(aut->message_types() == iface);
self->request(aut, "Hello World!").await( self->request(aut, "Hello World!").receive(
[](const string& answer) { [](const string& answer) {
CAF_CHECK_EQUAL(answer, "!dlroW olleH"); CAF_CHECK_EQUAL(answer, "!dlroW olleH");
} }
...@@ -448,7 +448,7 @@ CAF_TEST(maybe_string_delegator_chain) { ...@@ -448,7 +448,7 @@ CAF_TEST(maybe_string_delegator_chain) {
auto aut = system.spawn(maybe_string_delegator, auto aut = system.spawn(maybe_string_delegator,
system.spawn(maybe_string_reverter)); system.spawn(maybe_string_reverter));
CAF_MESSAGE("send empty string, expect error"); CAF_MESSAGE("send empty string, expect error");
self->request(aut, "").await( self->request(aut, "").receive(
[](ok_atom, const string&) { [](ok_atom, const string&) {
throw std::logic_error("unexpected result!"); throw std::logic_error("unexpected result!");
}, },
...@@ -459,7 +459,7 @@ CAF_TEST(maybe_string_delegator_chain) { ...@@ -459,7 +459,7 @@ CAF_TEST(maybe_string_delegator_chain) {
} }
); );
CAF_MESSAGE("send abcd string, expect dcba"); CAF_MESSAGE("send abcd string, expect dcba");
self->request(aut, "abcd").await( self->request(aut, "abcd").receive(
[](ok_atom, const string& str) { [](ok_atom, const string& str) {
CAF_CHECK_EQUAL(str, "dcba"); CAF_CHECK_EQUAL(str, "dcba");
}, },
...@@ -546,7 +546,7 @@ CAF_TEST(dot_composition) { ...@@ -546,7 +546,7 @@ CAF_TEST(dot_composition) {
auto second = system.spawn(second_stage_impl); auto second = system.spawn(second_stage_impl);
auto first_then_second = first * second; auto first_then_second = first * second;
scoped_actor self{system}; scoped_actor self{system};
self->request(first_then_second, 42).await( self->request(first_then_second, 42).receive(
[](double res) { [](double res) {
CAF_CHECK(res == (42 * 2.0) * (42 * 4.0)); CAF_CHECK(res == (42 * 2.0) * (42 * 4.0));
} }
...@@ -577,12 +577,12 @@ CAF_TEST(currying) { ...@@ -577,12 +577,12 @@ CAF_TEST(currying) {
CAF_CHECK(system.registry().running() == 1); CAF_CHECK(system.registry().running() == 1);
scoped_actor self{system}; scoped_actor self{system};
CAF_CHECK(system.registry().running() == 2); CAF_CHECK(system.registry().running() == 2);
self->request(bound, 2.0).await( self->request(bound, 2.0).receive(
[](double y) { [](double y) {
CAF_CHECK(y == 2.0); CAF_CHECK(y == 2.0);
} }
); );
self->request(bound, 10).await( self->request(bound, 10).receive(
[](int y) { [](int y) {
CAF_CHECK(y == 10); CAF_CHECK(y == 10);
} }
...@@ -618,12 +618,12 @@ CAF_TEST(type_safe_currying) { ...@@ -618,12 +618,12 @@ CAF_TEST(type_safe_currying) {
"bind returned wrong actor handle"); "bind returned wrong actor handle");
scoped_actor self{system}; scoped_actor self{system};
CAF_CHECK(system.registry().running() == 2); CAF_CHECK(system.registry().running() == 2);
self->request(bound, 2.0).await( self->request(bound, 2.0).receive(
[](double y) { [](double y) {
CAF_CHECK(y == 2.0); CAF_CHECK(y == 2.0);
} }
); );
self->request(bound, 10).await( self->request(bound, 10).receive(
[](int y) { [](int y) {
CAF_CHECK(y == 10); CAF_CHECK(y == 10);
} }
...@@ -649,7 +649,7 @@ CAF_TEST(reordering) { ...@@ -649,7 +649,7 @@ CAF_TEST(reordering) {
CAF_CHECK(system.registry().running() == 1); CAF_CHECK(system.registry().running() == 1);
scoped_actor self{system}; scoped_actor self{system};
CAF_CHECK(system.registry().running() == 2); CAF_CHECK(system.registry().running() == 2);
self->request(bound, 2.0, 10).await( self->request(bound, 2.0, 10).receive(
[](double y) { [](double y) {
CAF_CHECK(y == 20.0); CAF_CHECK(y == 20.0);
} }
...@@ -679,7 +679,7 @@ CAF_TEST(type_safe_reordering) { ...@@ -679,7 +679,7 @@ CAF_TEST(type_safe_reordering) {
"bind returned wrong actor handle"); "bind returned wrong actor handle");
scoped_actor self{system}; scoped_actor self{system};
CAF_CHECK(system.registry().running() == 2); CAF_CHECK(system.registry().running() == 2);
self->request(bound, 2.0, 10).await( self->request(bound, 2.0, 10).receive(
[](double y) { [](double y) {
CAF_CHECK(y == 20.0); CAF_CHECK(y == 20.0);
} }
......
...@@ -223,7 +223,7 @@ uint16_t middleman::publish(const actor_addr& whom, std::set<std::string> sigs, ...@@ -223,7 +223,7 @@ uint16_t middleman::publish(const actor_addr& whom, std::set<std::string> sigs,
std::string error_msg; std::string error_msg;
try { try {
self->request(mm, publish_atom::value, port, self->request(mm, publish_atom::value, port,
std::move(whom), std::move(sigs), str, ru).await( std::move(whom), std::move(sigs), str, ru).receive(
[&](ok_atom, uint16_t res) { [&](ok_atom, uint16_t res) {
result = res; result = res;
}, },
...@@ -269,7 +269,7 @@ maybe<uint16_t> middleman::publish_local_groups(uint16_t port, const char* in) { ...@@ -269,7 +269,7 @@ maybe<uint16_t> middleman::publish_local_groups(uint16_t port, const char* in) {
void middleman::unpublish(const actor_addr& whom, uint16_t port) { void middleman::unpublish(const actor_addr& whom, uint16_t port) {
CAF_LOG_TRACE(CAF_ARG(whom) << CAF_ARG(port)); CAF_LOG_TRACE(CAF_ARG(whom) << CAF_ARG(port));
scoped_actor self{system(), true}; scoped_actor self{system(), true};
self->request(actor_handle(), unpublish_atom::value, whom, port).await( self->request(actor_handle(), unpublish_atom::value, whom, port).receive(
[] { [] {
// ok, basp_broker is done // ok, basp_broker is done
}, },
...@@ -285,7 +285,7 @@ actor_addr middleman::remote_actor(std::set<std::string> ifs, ...@@ -285,7 +285,7 @@ actor_addr middleman::remote_actor(std::set<std::string> ifs,
auto mm = actor_handle(); auto mm = actor_handle();
actor_addr result; actor_addr result;
scoped_actor self{system(), true}; scoped_actor self{system(), true};
self->request(mm, connect_atom::value, std::move(host), port).await( self->request(mm, connect_atom::value, std::move(host), port).receive(
[&](ok_atom, const node_id&, actor_addr res, std::set<std::string>& xs) { [&](ok_atom, const node_id&, actor_addr res, std::set<std::string>& xs) {
CAF_LOG_TRACE(CAF_ARG(res) << CAF_ARG(xs)); CAF_LOG_TRACE(CAF_ARG(res) << CAF_ARG(xs));
if (!res) if (!res)
......
...@@ -154,13 +154,13 @@ void run_earth(bool use_asio, bool as_server, uint16_t pub_port) { ...@@ -154,13 +154,13 @@ void run_earth(bool use_asio, bool as_server, uint16_t pub_port) {
self->receive_while([&] { return mars_addr == invalid_actor_addr; })( self->receive_while([&] { return mars_addr == invalid_actor_addr; })(
[&](put_atom, const actor_addr& addr) { [&](put_atom, const actor_addr& addr) {
auto hdl = actor_cast<actor>(addr); auto hdl = actor_cast<actor>(addr);
self->request(hdl, sys_atom::value, get_atom::value, "info").await( self->request(hdl, sys_atom::value, get_atom::value, "info").then(
[&](ok_atom, const string&, const actor_addr&, const string& name) { [&](ok_atom, const string&, const actor_addr&, const string& name) {
if (name != "testee") if (name != "testee")
return; return;
mars_addr = addr; mars_addr = addr;
CAF_MESSAGE(CAF_ARG(mars_addr)); CAF_MESSAGE(CAF_ARG(mars_addr));
self->request(actor_cast<actor>(mars_addr), get_atom::value).await( self->request(actor_cast<actor>(mars_addr), get_atom::value).then(
[&](uint16_t mp) { [&](uint16_t mp) {
CAF_MESSAGE("mars published its actor at port " << mp); CAF_MESSAGE("mars published its actor at port " << mp);
mars_port = mp; mars_port = mp;
...@@ -185,7 +185,7 @@ void run_earth(bool use_asio, bool as_server, uint16_t pub_port) { ...@@ -185,7 +185,7 @@ void run_earth(bool use_asio, bool as_server, uint16_t pub_port) {
self->receive_while([&] { return jupiter_addr == invalid_actor_addr; })( self->receive_while([&] { return jupiter_addr == invalid_actor_addr; })(
[&](put_atom, const actor_addr& addr) { [&](put_atom, const actor_addr& addr) {
auto hdl = actor_cast<actor>(addr); auto hdl = actor_cast<actor>(addr);
self->request(hdl, sys_atom::value, get_atom::value, "info").await( self->request(hdl, sys_atom::value, get_atom::value, "info").then(
[&](ok_atom, const string&, const actor_addr&, const string& name) { [&](ok_atom, const string&, const actor_addr&, const string& name) {
if (name != "testee") if (name != "testee")
return; return;
......
...@@ -574,7 +574,7 @@ CAF_TEST(remote_actor_and_send) { ...@@ -574,7 +574,7 @@ CAF_TEST(remote_actor_and_send) {
this_node(), remote_node(0), this_node(), remote_node(0),
invalid_actor_id, pseudo_remote(0)->id()); invalid_actor_id, pseudo_remote(0)->id());
CAF_MESSAGE("BASP broker should've send the proxy"); CAF_MESSAGE("BASP broker should've send the proxy");
f.await( f.receive(
[&](ok_atom, node_id nid, actor_addr res, std::set<std::string> ifs) { [&](ok_atom, node_id nid, actor_addr res, std::set<std::string> ifs) {
auto aptr = actor_cast<abstract_actor_ptr>(res); auto aptr = actor_cast<abstract_actor_ptr>(res);
CAF_REQUIRE(aptr.downcast<forwarding_actor_proxy>() != nullptr); CAF_REQUIRE(aptr.downcast<forwarding_actor_proxy>() != nullptr);
......
...@@ -180,7 +180,7 @@ void run_server(int argc, char** argv) { ...@@ -180,7 +180,7 @@ void run_server(int argc, char** argv) {
auto serv = system.middleman().spawn_broker(peer_acceptor_fun, auto serv = system.middleman().spawn_broker(peer_acceptor_fun,
system.spawn(pong)); system.spawn(pong));
std::thread child; std::thread child;
self->request(serv, publish_atom::value).await( self->request(serv, publish_atom::value).receive(
[&](uint16_t port) { [&](uint16_t port) {
CAF_MESSAGE("server is running on port " << port); CAF_MESSAGE("server is running on port " << port);
child = std::thread([=] { run_client(argc, argv, port); }); child = std::thread([=] { run_client(argc, argv, port); });
......
...@@ -196,7 +196,7 @@ void run_server(int argc, char** argv) { ...@@ -196,7 +196,7 @@ void run_server(int argc, char** argv) {
scoped_actor self{system}; scoped_actor self{system};
auto serv = system.middleman().spawn_broker(acceptor_fun, system.spawn(pong)); auto serv = system.middleman().spawn_broker(acceptor_fun, system.spawn(pong));
std::thread child; std::thread child;
self->request(serv, publish_atom::value).await( self->request(serv, publish_atom::value).receive(
[&](uint16_t port) { [&](uint16_t port) {
CAF_MESSAGE("server is running on port " << port); CAF_MESSAGE("server is running on port " << port);
child = std::thread([=] { child = std::thread([=] {
......
...@@ -95,7 +95,7 @@ void run_client(int argc, char** argv, uint16_t port) { ...@@ -95,7 +95,7 @@ void run_client(int argc, char** argv, uint16_t port) {
CAF_REQUIRE(serv); CAF_REQUIRE(serv);
scoped_actor self{system}; scoped_actor self{system};
self->request(serv, ping{42}) self->request(serv, ping{42})
.await([](const pong& p) { CAF_CHECK_EQUAL(p.value, 42); }); .receive([](const pong& p) { CAF_CHECK_EQUAL(p.value, 42); });
anon_send_exit(serv, exit_reason::user_shutdown); anon_send_exit(serv, exit_reason::user_shutdown);
self->monitor(serv); self->monitor(serv);
self->receive([&](const down_msg& dm) { self->receive([&](const down_msg& dm) {
......
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