Commit c1a10f43 authored by Dominik Charousset's avatar Dominik Charousset

Enforce timeouts when using requests, relates #442

parent dc7e3f4f
...@@ -26,8 +26,8 @@ behavior mirror(event_based_actor* self) { ...@@ -26,8 +26,8 @@ behavior mirror(event_based_actor* self) {
void hello_world(event_based_actor* self, const actor& buddy) { void hello_world(event_based_actor* self, const actor& buddy) {
// send "Hello World!" to our buddy ... // send "Hello World!" to our buddy ...
self->request(buddy, "Hello World!").then( self->request(buddy, std::chrono::seconds(10), "Hello World!").then(
// ... wait for a response ... // ... wait up to 10s for a response ...
[=](const string& what) { [=](const string& what) {
// ... and print it // ... and print it
aout(self) << what << endl; aout(self) << what << endl;
......
...@@ -115,11 +115,11 @@ template <class Handle, class... Ts> ...@@ -115,11 +115,11 @@ template <class Handle, class... Ts>
void tester(scoped_actor& self, const Handle& hdl, int x, int y, Ts&&... xs) { void tester(scoped_actor& self, const Handle& hdl, int x, int y, Ts&&... xs) {
self->monitor(hdl); self->monitor(hdl);
// first test: x + y = z // first test: x + y = z
self->request(hdl, add_atom::value, x, y).receive( self->request(hdl, indefinite, add_atom::value, x, y).receive(
[&](int res1) { [&](int res1) {
aout(self) << x << " + " << y << " = " << res1 << endl; aout(self) << x << " + " << y << " = " << res1 << endl;
// second test: x - y = z // second test: x - y = z
self->request(hdl, sub_atom::value, x, y).receive( self->request(hdl, indefinite, sub_atom::value, x, y).receive(
[&](int res2) { [&](int res2) {
aout(self) << x << " - " << y << " = " << res2 << endl; aout(self) << x << " - " << y << " = " << res2 << endl;
self->send_exit(hdl, exit_reason::user_shutdown); self->send_exit(hdl, exit_reason::user_shutdown);
......
...@@ -66,7 +66,7 @@ public: ...@@ -66,7 +66,7 @@ public:
private: private:
void request_task(atom_value op, int lhs, int rhs) { void request_task(atom_value op, int lhs, int rhs) {
request(server_, op, lhs, rhs).then( request(server_, indefinite, op, lhs, rhs).then(
[=](result_atom, int result) { [=](result_atom, int result) {
aout(this) << lhs << (op == plus_atom::value ? " + " : " - ") aout(this) << lhs << (op == plus_atom::value ? " + " : " - ")
<< rhs << " = " << result << endl; << rhs << " = " << result << endl;
......
...@@ -201,6 +201,9 @@ ...@@ -201,6 +201,9 @@
} static_cast<void>(0) } static_cast<void>(0)
#endif #endif
// Convenience macros.
#define CAF_IGNORE_UNUSED(x) static_cast<void>(x);
#define CAF_CRITICAL(error) \ #define CAF_CRITICAL(error) \
printf("%s:%u: critical error: '%s'\n", __FILE__, __LINE__, error); \ printf("%s:%u: critical error: '%s'\n", __FILE__, __LINE__, error); \
abort() abort()
......
...@@ -64,7 +64,6 @@ using sorted_builtin_types = ...@@ -64,7 +64,6 @@ using sorted_builtin_types =
strmap, // @strmap strmap, // @strmap
std::set<std::string>, // @strset std::set<std::string>, // @strset
std::vector<std::string>, // @strvec std::vector<std::string>, // @strvec
sync_timeout_msg, // @sync_timeout
timeout_msg, // @timeout timeout_msg, // @timeout
uint16_t, // @u16 uint16_t, // @u16
std::u16string, // @u16_str std::u16string, // @u16_str
......
...@@ -640,6 +640,15 @@ public: ...@@ -640,6 +640,15 @@ public:
static constexpr bool value = false; static constexpr bool value = false;
}; };
/// Checks whether T is convertible to either `std::function<void (T&)>`
/// or `std::function<void (const T&)>`.
template <class F, class T>
struct is_handler_for {
static constexpr bool value =
std::is_convertible<F, std::function<void (T&)>>::value
|| std::is_convertible<F, std::function<void (const T&)>>::value;
};
template <class T> template <class T>
struct value_type_of { struct value_type_of {
using type = typename T::value_type; using type = typename T::value_type;
...@@ -665,6 +674,13 @@ struct deconst_kvp<std::pair<K, V>> { ...@@ -665,6 +674,13 @@ struct deconst_kvp<std::pair<K, V>> {
typename std::remove_const<V>::type>; typename std::remove_const<V>::type>;
}; };
template <class T>
using is_callable_t = typename std::enable_if<is_callable<T>::value>::type;
template <class F, class T>
using is_handler_for_ef =
typename std::enable_if<is_handler_for<F, T>::value>::type;
} // namespace detail } // namespace detail
} // namespace caf } // namespace caf
......
...@@ -75,6 +75,15 @@ constexpr time_unit get_time_unit_from_period() { ...@@ -75,6 +75,15 @@ constexpr time_unit get_time_unit_from_period() {
return ratio_to_time_unit_helper<Period::num, Period::den>::value; return ratio_to_time_unit_helper<Period::num, Period::den>::value;
} }
/// Represents an infinite amount of timeout for specifying "invalid" timeouts.
struct indefinite_t {
constexpr indefinite_t() {
// nop
}
};
static constexpr indefinite_t indefinite = indefinite_t{};
/// Time duration consisting of a `time_unit` and a 64 bit unsigned integer. /// Time duration consisting of a `time_unit` and a 64 bit unsigned integer.
class duration { class duration {
public: public:
...@@ -86,6 +95,10 @@ public: ...@@ -86,6 +95,10 @@ public:
// nop // nop
} }
constexpr duration(const indefinite_t&) : unit(time_unit::invalid), count(0) {
// nop
}
/// Creates a new instance from an STL duration. /// Creates a new instance from an STL duration.
/// @throws std::invalid_argument Thrown if `d.count() is negative. /// @throws std::invalid_argument Thrown if `d.count() is negative.
template <class Rep, class Period> template <class Rep, class Period>
...@@ -118,13 +131,14 @@ private: ...@@ -118,13 +131,14 @@ private:
template <class Rep, intmax_t Num, intmax_t D> template <class Rep, intmax_t Num, intmax_t D>
static uint64_t rd(const std::chrono::duration<Rep, std::ratio<Num, D>>& d) { static uint64_t rd(const std::chrono::duration<Rep, std::ratio<Num, D>>& d) {
// assertion (via ctors): Num == 1 || (Num == 60 && D == 1) // assertion (via ctors): Num == 1 || (Num == 60 && D == 1)
if (d.count() < 0) { if (d.count() < 0)
throw std::invalid_argument("negative durations are not supported"); throw std::invalid_argument("negative durations are not supported");
}
return static_cast<uint64_t>(d.count()) * static_cast<uint64_t>(Num); return static_cast<uint64_t>(d.count()) * static_cast<uint64_t>(Num);
} }
}; };
std::string to_string(const duration& x);
/// @relates duration /// @relates duration
template <class Processor> template <class Processor>
void serialize(Processor& proc, duration& x, const unsigned int) { void serialize(Processor& proc, duration& x, const unsigned int) {
......
...@@ -35,7 +35,7 @@ enum class exit_reason : uint8_t { ...@@ -35,7 +35,7 @@ enum class exit_reason : uint8_t {
unhandled_exception = 0x02, unhandled_exception = 0x02,
/// Indicates that the actor received an unexpected synchronous reply message. /// Indicates that the actor received an unexpected synchronous reply message.
unhandled_sync_failure = 0x04, unhandled_request_error = 0x04,
/// Indicates that the exit reason for this actor is unknown, i.e., /// Indicates that the exit reason for this actor is unknown, i.e.,
/// the actor has been terminated and no longer exists. /// the actor has been terminated and no longer exists.
......
...@@ -136,7 +136,7 @@ public: ...@@ -136,7 +136,7 @@ public:
R result; R result;
function_view_storage<R> h{result}; function_view_storage<R> h{result};
try { try {
self_->request(impl_, std::forward<Ts>(xs)...).receive(h); self_->request(impl_, indefinite, std::forward<Ts>(xs)...).receive(h);
} }
catch (std::exception&) { catch (std::exception&) {
assign(invalid_actor); assign(invalid_actor);
......
...@@ -81,7 +81,6 @@ struct down_msg; ...@@ -81,7 +81,6 @@ struct down_msg;
struct timeout_msg; struct timeout_msg;
struct group_down_msg; struct group_down_msg;
struct invalid_actor_t; struct invalid_actor_t;
struct sync_timeout_msg;
struct invalid_actor_addr_t; struct invalid_actor_addr_t;
struct illegal_message_element; struct illegal_message_element;
struct prohibit_top_level_spawn_marker; struct prohibit_top_level_spawn_marker;
......
...@@ -467,12 +467,14 @@ public: ...@@ -467,12 +467,14 @@ public:
return current_element_; return current_element_;
} }
template <class Handle, class... Ts> template <class ActorHandle, class... Ts>
message_id request_impl(message_priority mp, const Handle& dh, Ts&&... xs) { message_id request_impl(message_priority mp, const ActorHandle& dh,
const duration& timeout, Ts&&... xs) {
if (! dh) if (! dh)
throw std::invalid_argument("cannot request to invalid_actor"); throw std::invalid_argument("cannot send requests to invalid actors");
auto req_id = new_request_id(mp); auto req_id = new_request_id(mp);
send_impl(req_id, actor_cast<abstract_actor*>(dh), std::forward<Ts>(xs)...); send_impl(req_id, actor_cast<abstract_actor*>(dh), std::forward<Ts>(xs)...);
request_sync_timeout_msg(timeout, req_id);
return req_id.response_id(); return req_id.response_id();
} }
...@@ -600,8 +602,7 @@ public: ...@@ -600,8 +602,7 @@ public:
using error_handler = std::function<void (error&)>; using error_handler = std::function<void (error&)>;
using pending_response = using pending_response = std::pair<const message_id, behavior>;
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);
...@@ -611,10 +612,9 @@ public: ...@@ -611,10 +612,9 @@ public:
bool awaits(message_id mid) const; bool awaits(message_id mid) const;
maybe<pending_response&> find_awaited_response(message_id mid); pending_response* find_awaited_response(message_id mid);
void set_awaited_response_handler(message_id response_id, behavior bhvr, void set_awaited_response_handler(message_id response_id, behavior bhvr);
error_handler f = nullptr);
behavior& awaited_response_handler(); behavior& awaited_response_handler();
...@@ -624,10 +624,9 @@ public: ...@@ -624,10 +624,9 @@ public:
bool multiplexes(message_id mid) const; bool multiplexes(message_id mid) const;
maybe<pending_response&> find_multiplexed_response(message_id mid); pending_response* find_multiplexed_response(message_id mid);
void set_multiplexed_response_handler(message_id response_id, behavior bhvr, 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
...@@ -663,10 +662,7 @@ protected: ...@@ -663,10 +662,7 @@ protected:
std::forward_list<pending_response> awaited_responses_; std::forward_list<pending_response> awaited_responses_;
// identifies all IDs of async messages waiting for a response // identifies all IDs of async messages waiting for a response
std::unordered_map< std::unordered_map<message_id, behavior> multiplexed_responses_;
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
......
...@@ -21,6 +21,7 @@ ...@@ -21,6 +21,7 @@
#define CAF_MIXIN_SYNC_SENDER_HPP #define CAF_MIXIN_SYNC_SENDER_HPP
#include <tuple> #include <tuple>
#include <chrono>
#include "caf/actor.hpp" #include "caf/actor.hpp"
#include "caf/message.hpp" #include "caf/message.hpp"
...@@ -53,9 +54,9 @@ public: ...@@ -53,9 +54,9 @@ public:
/// @throws std::invalid_argument if `dest == invalid_actor` /// @throws std::invalid_argument if `dest == invalid_actor`
template <class... Ts> template <class... Ts>
response_handle_type request(message_priority mp, const actor& dest, response_handle_type request(message_priority mp, const actor& dest,
Ts&&... xs) { const duration& timeout, Ts&&... xs) {
static_assert(sizeof...(Ts) > 0, "no message to send"); static_assert(sizeof...(Ts) > 0, "no message to send");
return {dptr()->request_impl(mp, dest, std::forward<Ts>(xs)...), return {dptr()->request_impl(mp, dest, timeout, std::forward<Ts>(xs)...),
dptr()}; dptr()};
} }
...@@ -65,8 +66,10 @@ public: ...@@ -65,8 +66,10 @@ public:
/// sent message cannot be received by another actor. /// sent message cannot be received by another actor.
/// @throws std::invalid_argument if `dest == invalid_actor` /// @throws std::invalid_argument if `dest == invalid_actor`
template <class... Ts> template <class... Ts>
response_handle_type request(const actor& dest, Ts&&... xs) { response_handle_type request(const actor& dest,
return request(message_priority::normal, dest, std::forward<Ts>(xs)...); const duration& timeout, Ts&&... xs) {
return request(message_priority::normal, dest,
timeout, std::forward<Ts>(xs)...);
} }
/// Sends `{xs...}` as a synchronous message to `dest` with priority `mp`. /// Sends `{xs...}` as a synchronous message to `dest` with priority `mp`.
...@@ -84,7 +87,8 @@ public: ...@@ -84,7 +87,8 @@ public:
>::type...> >::type...>
>::type, >::type,
HandleTag> HandleTag>
request(message_priority mp, const typed_actor<Sigs...>& dest, Ts&&... xs) { request(message_priority mp, const typed_actor<Sigs...>& 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 =
detail::type_list< detail::type_list<
...@@ -93,7 +97,7 @@ public: ...@@ -93,7 +97,7 @@ public:
>::type...>; >::type...>;
token tk; token tk;
check_typed_input(dest, tk); check_typed_input(dest, tk);
return {dptr()->request_impl(mp, dest, std::forward<Ts>(xs)...), return {dptr()->request_impl(mp, dest, timeout, std::forward<Ts>(xs)...),
dptr()}; dptr()};
} }
...@@ -112,7 +116,8 @@ public: ...@@ -112,7 +116,8 @@ public:
>::type...> >::type...>
>::type, >::type,
HandleTag> HandleTag>
request(const typed_actor<Sigs...>& dest, Ts&&... xs) { request(const typed_actor<Sigs...>& 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 =
detail::type_list< detail::type_list<
...@@ -121,98 +126,11 @@ public: ...@@ -121,98 +126,11 @@ public:
>::type...>; >::type...>;
token tk; token tk;
check_typed_input(dest, tk); check_typed_input(dest, tk);
return {dptr()->request_impl(message_priority::normal, return {dptr()->request_impl(message_priority::normal, dest,
dest, std::forward<Ts>(xs)...), timeout, std::forward<Ts>(xs)...),
dptr()}; dptr()};
} }
/****************************************************************************
* timed_request(...) *
****************************************************************************/
/// Sends `{xs...}` as a synchronous message to `dest` with priority `mp`
/// and relative timeout `rtime`.
/// @returns A handle identifying a future-like handle to the response.
/// @warning The returned handle is actor specific and the response to the
/// sent message cannot be received by another actor.
/// @throws std::invalid_argument if `dest == invalid_actor`
template <class... Ts>
response_handle_type timed_request(message_priority mp, const actor& dest,
const duration& rtime, Ts&&... xs) {
static_assert(sizeof...(Ts) > 0, "no message to send");
return {dptr()->timed_request_impl(mp, dest, rtime,
std::forward<Ts>(xs)...),
dptr()};
}
/// Sends `{xs...}` as a synchronous message to `dest` with
/// relative timeout `rtime`.
/// @returns A handle identifying a future-like handle to the response.
/// @warning The returned handle is actor specific and the response to the
/// sent message cannot be received by another actor.
/// @throws std::invalid_argument if `dest == invalid_actor`
template <class... Ts>
response_handle_type timed_request(const actor& dest, const duration& rtime,
Ts&&... xs) {
return timed_request(message_priority::normal, dest, rtime,
std::forward<Ts>(xs)...);
}
/// Sends `{xs...}` as a synchronous message to `dest` with priority `mp`
/// and relative timeout `rtime`.
/// @returns A handle identifying a future-like handle to the response.
/// @warning The returned handle is actor specific and the response to the
/// sent message cannot be received by another actor.
/// @throws std::invalid_argument if `dest == invalid_actor`
template <class... Sigs, class... Ts>
response_handle<Subtype,
typename detail::deduce_output_type<
detail::type_list<Sigs...>,
typename detail::implicit_conversions<
typename std::decay<Ts>::type
>::type...
>::type,
HandleTag>
timed_request(message_priority mp, const typed_actor<Sigs...>& dest,
const duration& rtime, Ts&&... xs) {
static_assert(sizeof...(Ts) > 0, "no message to send");
using token =
detail::type_list<
typename detail::implicit_conversions<
typename std::decay<Ts>::type
>::type...>;
token tk;
check_typed_input(dest, tk);
return {dptr()->timed_request_impl(mp, dest, rtime,
std::forward<Ts>(xs)...),
dptr()};
}
/// Sends `{xs...}` as a synchronous message to `dest` with
/// relative timeout `rtime`.
/// @returns A handle identifying a future-like handle to the response.
/// @warning The returned handle is actor specific and the response to the
/// sent message cannot be received by another actor.
/// @throws std::invalid_argument if `dest == invalid_actor`
template <class... Sigs, class... Ts>
response_handle<Subtype,
typename detail::deduce_output_type<
detail::type_list<Sigs...>,
typename detail::implicit_conversions<
typename std::decay<Ts>::type
>::type...
>::type,
HandleTag>
timed_request(const typed_actor<Sigs...>& dest, const duration& rtime,
Ts&&... xs) {
return timed_request(message_priority::normal, dest, rtime,
std::forward<Ts>(xs)...);
}
/****************************************************************************
* deprecated member functions *
****************************************************************************/
private: private:
Subtype* dptr() { Subtype* dptr() {
return static_cast<Subtype*>(this); return static_cast<Subtype*>(this);
......
This diff is collapsed.
...@@ -35,6 +35,8 @@ enum class sec : uint8_t { ...@@ -35,6 +35,8 @@ enum class sec : uint8_t {
unexpected_response, unexpected_response,
/// Indicates that the receiver of a request is no longer alive. /// Indicates that the receiver of a request is no longer alive.
request_receiver_down, request_receiver_down,
/// Indicates that a request message timed out.
request_timeout,
/// Unpublishing failed because the actor is `invalid_actor`. /// Unpublishing failed because the actor is `invalid_actor`.
no_actor_to_unpublish, no_actor_to_unpublish,
/// Unpublishing failed because the actor is not bound to given port. /// Unpublishing failed because the actor is not bound to given port.
......
...@@ -247,6 +247,7 @@ void actor_registry::start() { ...@@ -247,6 +247,7 @@ void actor_registry::start() {
unsubscribe_all(actor_cast<actor>(dm.source)); unsubscribe_all(actor_cast<actor>(dm.source));
}, },
others >> [=](const message& msg) -> error { others >> [=](const message& msg) -> error {
CAF_IGNORE_UNUSED(msg);
CAF_LOG_WARNING("unexpected:" << CAF_ARG(msg)); CAF_LOG_WARNING("unexpected:" << CAF_ARG(msg));
return sec::unexpected_message; return sec::unexpected_message;
} }
...@@ -265,6 +266,7 @@ void actor_registry::start() { ...@@ -265,6 +266,7 @@ void actor_registry::start() {
return {ok_atom::value, res.first, res.second}; return {ok_atom::value, res.first, res.second};
}, },
others >> [=](const message& msg) { others >> [=](const message& msg) {
CAF_IGNORE_UNUSED(msg);
CAF_LOG_WARNING("unexpected:" << CAF_ARG(msg)); CAF_LOG_WARNING("unexpected:" << CAF_ARG(msg));
} }
}; };
......
...@@ -52,32 +52,26 @@ void blocking_actor::dequeue(behavior& bhvr, message_id mid) { ...@@ -52,32 +52,26 @@ 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` // push an empty sync response handler for `blocking_actor`
if (mid != invalid_message_id && ! find_awaited_response(mid)) if (mid != invalid_message_id && ! find_awaited_response(mid))
awaited_responses_.emplace_front(mid, std::make_pair(behavior{}, nullptr)); awaited_responses_.emplace_front(mid, behavior{});
// 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;
}
// requesting an invalid timeout will reset our active timeout // requesting an invalid timeout will reset our active timeout
uint32_t timeout_id = 0; uint32_t timeout_id = 0;
if (mid == invalid_message_id) { if (mid == invalid_message_id)
timeout_id = request_timeout(bhvr.timeout()); timeout_id = request_timeout(bhvr.timeout());
} else {
request_sync_timeout_msg(bhvr.timeout(), mid);
}
// read incoming messages // read incoming messages
for (;;) { for (;;) {
await_data(); await_data();
auto msg = next_message(); auto msg = next_message();
switch (invoke_message(msg, bhvr, mid)) { switch (invoke_message(msg, bhvr, mid)) {
case im_success: case im_success:
if (mid == invalid_message_id) { if (mid == invalid_message_id)
reset_timeout(timeout_id); reset_timeout(timeout_id);
}
return; return;
case im_skipped: case im_skipped:
if (msg) { if (msg)
push_to_cache(std::move(msg)); push_to_cache(std::move(msg));
}
break; break;
default: default:
// delete msg // delete msg
......
...@@ -36,6 +36,24 @@ std::string to_string(const time_unit& x) { ...@@ -36,6 +36,24 @@ std::string to_string(const time_unit& x) {
} }
} }
std::string to_string(const duration& x) {
auto result = std::to_string(x.count);
switch (x.unit) {
case time_unit::seconds:
result += "s";
break;
case time_unit::milliseconds:
result += "ms";
break;
case time_unit::microseconds:
result += "us";
break;
default:
return "indefinite";
}
return result;
}
bool operator==(const duration& lhs, const duration& rhs) { bool operator==(const duration& lhs, const duration& rhs) {
return lhs.unit == rhs.unit && lhs.count == rhs.count; return lhs.unit == rhs.unit && lhs.count == rhs.count;
} }
......
...@@ -29,8 +29,8 @@ const char* to_string(exit_reason x) { ...@@ -29,8 +29,8 @@ const char* to_string(exit_reason x) {
return "normal"; return "normal";
case exit_reason::unhandled_exception: case exit_reason::unhandled_exception:
return "unhandled_exception"; return "unhandled_exception";
case exit_reason::unhandled_sync_failure: case exit_reason::unhandled_request_error:
return "unhandled_sync_failure"; return "unhandled_request_error";
case exit_reason::unknown: case exit_reason::unknown:
return "unknown"; return "unknown";
case exit_reason::out_of_workers: case exit_reason::out_of_workers:
......
...@@ -148,20 +148,19 @@ uint32_t local_actor::request_timeout(const duration& d) { ...@@ -148,20 +148,19 @@ uint32_t local_actor::request_timeout(const duration& d) {
} }
void local_actor::request_sync_timeout_msg(const duration& d, message_id mid) { void local_actor::request_sync_timeout_msg(const duration& d, message_id mid) {
if (! d.valid()) { CAF_LOG_TRACE(CAF_ARG(d) << CAF_ARG(mid));
if (! d.valid())
return; return;
} delayed_send_impl(mid.response_id(), this, d,
delayed_send_impl(mid, this, d, make_message(sync_timeout_msg{})); make_message(sec::request_timeout));
} }
void local_actor::handle_timeout(behavior& bhvr, uint32_t timeout_id) { void local_actor::handle_timeout(behavior& bhvr, uint32_t timeout_id) {
if (! is_active_timeout(timeout_id)) { if (! is_active_timeout(timeout_id))
return; return;
}
bhvr.handle_timeout(); bhvr.handle_timeout();
if (bhvr_stack_.empty() || bhvr_stack_.back() != bhvr) { if (bhvr_stack_.empty() || bhvr_stack_.back() != bhvr)
return; return;
}
// auto-remove behavior for blocking actors // auto-remove behavior for blocking actors
if (is_blocking()) { if (is_blocking()) {
CAF_ASSERT(bhvr_stack_.back() == bhvr); CAF_ASSERT(bhvr_stack_.back() == bhvr);
...@@ -203,6 +202,7 @@ msg_type filter_msg(local_actor* self, mailbox_element& node) { ...@@ -203,6 +202,7 @@ msg_type filter_msg(local_actor* self, mailbox_element& node) {
if (msg.size() > 1 && msg.match_element<sys_atom>(0) && node.sender) { if (msg.size() > 1 && msg.match_element<sys_atom>(0) && node.sender) {
bool mismatch = false; bool mismatch = false;
msg.apply({ msg.apply({
/*
[&](sys_atom, migrate_atom, const actor& mm) { [&](sys_atom, migrate_atom, const actor& mm) {
// migrate this actor to `target` // migrate this actor to `target`
if (! self->is_serializable()) { if (! self->is_serializable()) {
...@@ -245,6 +245,7 @@ msg_type filter_msg(local_actor* self, mailbox_element& node) { ...@@ -245,6 +245,7 @@ msg_type filter_msg(local_actor* self, mailbox_element& node) {
} }
}); });
}, },
*/
[&](sys_atom, migrate_atom, std::vector<char>& buf) { [&](sys_atom, migrate_atom, std::vector<char>& buf) {
// "replace" this actor with the content of `buf` // "replace" this actor with the content of `buf`
if (! self->is_serializable()) { if (! self->is_serializable()) {
...@@ -389,61 +390,6 @@ private: ...@@ -389,61 +390,6 @@ private:
local_actor* self_; local_actor* self_;
}; };
/*
response_promise fetch_response_promise(local_actor* self, int) {
return self->make_response_promise();
}
response_promise fetch_response_promise(local_actor*, response_promise& hdl) {
return std::move(hdl);
}
// enables `return request(...).then(...)`
bool handle_message_id_res(local_actor* self, message& res,
response_promise hdl) {
CAF_ASSERT(hdl.pending());
CAF_LOG_TRACE(CAF_ARG(res));
if (res.match_elements<atom_value, uint64_t>()
&& res.get_as<atom_value>(0) == atom("MESSAGE_ID")) {
CAF_LOG_DEBUG("message handler returned a message id wrapper");
}
return false;
}
// - extracts response message from handler
// - returns true if fun was successfully invoked
template <class Handle = int>
bool post_process_invoke_res(local_actor* self, bool is_sync_request,
maybe<message>&& res, Handle hdl = Handle{}) {
CAF_LOG_TRACE(CAF_ARG(is_sync_request) << CAF_ARG(res));
// an empty response means self has skipped the message
if (res.empty())
return false;
// get a response promise for the original request
auto rp = fetch_response_promise(self, hdl);
// return true if self has answered to the original request,
// e.g., by forwarding or delegating it
if (! rp.pending())
return res.valid();
// fulfill the promise
if (res) {
CAF_LOG_DEBUG("respond via response_promise");
// deliver empty messages only for sync responses
if (! handle_message_id_res(self, *res, rp)
&& (! res->empty() || is_sync_request))
rp.deliver(std::move(*res));
return true;
} else if (is_sync_request) {
CAF_LOG_DEBUG("report error back to sync caller");
if (res.empty())
res = sec::unexpected_response;
rp.deliver(make_message(res.error()));
}
return false;
}
*/
invoke_message_result local_actor::invoke_message(mailbox_element_ptr& ptr, invoke_message_result local_actor::invoke_message(mailbox_element_ptr& ptr,
behavior& fun, behavior& fun,
message_id awaited_id) { message_id awaited_id) {
...@@ -483,7 +429,7 @@ invoke_message_result local_actor::invoke_message(mailbox_element_ptr& ptr, ...@@ -483,7 +429,7 @@ invoke_message_result local_actor::invoke_message(mailbox_element_ptr& ptr,
CAF_LOG_DEBUG("handle as multiplexed response:" << CAF_ARG(ptr->msg) CAF_LOG_DEBUG("handle as multiplexed response:" << CAF_ARG(ptr->msg)
<< CAF_ARG(mid) << CAF_ARG(awaited_id)); << CAF_ARG(mid) << CAF_ARG(awaited_id));
if (! awaited_id.valid()) { if (! awaited_id.valid()) {
auto& ref_fun = ref_opt->second.first; auto& ref_fun = ref_opt->second;
bool is_sync_tout = ptr->msg.match_elements<sync_timeout_msg>(); bool is_sync_tout = ptr->msg.match_elements<sync_timeout_msg>();
ptr.swap(current_element_); ptr.swap(current_element_);
if (is_sync_tout) { if (is_sync_tout) {
...@@ -494,7 +440,7 @@ invoke_message_result local_actor::invoke_message(mailbox_element_ptr& ptr, ...@@ -494,7 +440,7 @@ invoke_message_result local_actor::invoke_message(mailbox_element_ptr& ptr,
//} else if (! post_process_invoke_res(this, false, //} else if (! post_process_invoke_res(this, false,
// ref_fun(current_element_->msg))) { // ref_fun(current_element_->msg))) {
CAF_LOG_WARNING("multiplexed response failure occured:" << CAF_ARG(id())); CAF_LOG_WARNING("multiplexed response failure occured:" << CAF_ARG(id()));
quit(exit_reason::unhandled_sync_failure); quit(exit_reason::unhandled_request_error);
} }
ptr.swap(current_element_); ptr.swap(current_element_);
mark_multiplexed_arrived(mid); mark_multiplexed_arrived(mid);
...@@ -515,7 +461,7 @@ invoke_message_result local_actor::invoke_message(mailbox_element_ptr& ptr, ...@@ -515,7 +461,7 @@ invoke_message_result local_actor::invoke_message(mailbox_element_ptr& ptr,
//if (! post_process_invoke_res(this, false, //if (! post_process_invoke_res(this, false,
// fun(current_element_->msg))) { // fun(current_element_->msg))) {
CAF_LOG_WARNING("sync response failure occured:" << CAF_ARG(id())); CAF_LOG_WARNING("sync response failure occured:" << CAF_ARG(id()));
quit(exit_reason::unhandled_sync_failure); quit(exit_reason::unhandled_request_error);
} }
} }
ptr.swap(current_element_); ptr.swap(current_element_);
...@@ -589,33 +535,26 @@ bool local_actor::awaits(message_id mid) const { ...@@ -589,33 +535,26 @@ bool local_actor::awaits(message_id mid) const {
predicate); predicate);
} }
maybe<local_actor::pending_response&> local_actor::pending_response*
local_actor::find_awaited_response(message_id mid) { local_actor::find_awaited_response(message_id mid) {
awaited_response_predicate predicate{mid}; awaited_response_predicate predicate{mid};
auto last = awaited_responses_.end(); auto last = awaited_responses_.end();
auto i = std::find_if(awaited_responses_.begin(), last, predicate); auto i = std::find_if(awaited_responses_.begin(), last, predicate);
if (i != last) if (i != last)
return *i; return &(*i);
return none; return nullptr;
} }
void local_actor::set_awaited_response_handler(message_id response_id, behavior bhvr, void local_actor::set_awaited_response_handler(message_id response_id, behavior bhvr) {
error_handler f) {
if (bhvr.timeout().valid()) {
request_sync_timeout_msg(bhvr.timeout(), response_id);
}
auto opt_ref = find_awaited_response(response_id); auto opt_ref = find_awaited_response(response_id);
if (opt_ref) { if (opt_ref)
opt_ref->second.first = std::move(bhvr); opt_ref->second = std::move(bhvr);
opt_ref->second.second = std::move(f); else
} else { awaited_responses_.emplace_front(response_id, std::move(bhvr));
awaited_responses_.emplace_front(response_id,
std::make_pair(std::move(bhvr), std::move(f)));
}
} }
behavior& local_actor::awaited_response_handler() { behavior& local_actor::awaited_response_handler() {
return awaited_responses_.front().second.first; return awaited_responses_.front().second;
} }
message_id local_actor::awaited_response_id() { message_id local_actor::awaited_response_id() {
...@@ -635,28 +574,23 @@ bool local_actor::multiplexes(message_id mid) const { ...@@ -635,28 +574,23 @@ bool local_actor::multiplexes(message_id mid) const {
return it != multiplexed_responses_.end(); return it != multiplexed_responses_.end();
} }
maybe<local_actor::pending_response&> local_actor::pending_response*
local_actor::find_multiplexed_response(message_id mid) { local_actor::find_multiplexed_response(message_id mid) {
auto it = multiplexed_responses_.find(mid); auto it = multiplexed_responses_.find(mid);
if (it != multiplexed_responses_.end()) { if (it != multiplexed_responses_.end())
return *it; return &(*it);
} return nullptr;
return none;
} }
void local_actor::set_multiplexed_response_handler(message_id response_id, behavior bhvr, void local_actor::set_multiplexed_response_handler(message_id response_id, behavior bhvr) {
error_handler f) {
if (bhvr.timeout().valid()) { if (bhvr.timeout().valid()) {
request_sync_timeout_msg(bhvr.timeout(), response_id); request_sync_timeout_msg(bhvr.timeout(), response_id);
} }
auto opt_ref = find_multiplexed_response(response_id); auto opt_ref = find_multiplexed_response(response_id);
if (opt_ref) { if (opt_ref)
opt_ref->second.first = std::move(bhvr); opt_ref->second = std::move(bhvr);
opt_ref->second.second = std::move(f); else
} else { multiplexed_responses_.emplace(response_id, std::move(bhvr));
multiplexed_responses_.emplace(response_id,
std::make_pair(std::move(bhvr), std::move(f)));
}
} }
void local_actor::launch(execution_unit* eu, bool lazy, bool hide) { void local_actor::launch(execution_unit* eu, bool lazy, bool hide) {
......
...@@ -47,7 +47,9 @@ behavior fan_out_fan_in(stateful_actor<splitter_state>* self, ...@@ -47,7 +47,9 @@ behavior fan_out_fan_in(stateful_actor<splitter_state>* self,
self->state.pending = workers.size(); self->state.pending = workers.size();
// request().await() has LIFO ordering // request().await() has LIFO ordering
for (auto i = workers.rbegin(); i != workers.rend(); ++i) for (auto i = workers.rbegin(); i != workers.rend(); ++i)
self->request(actor_cast<actor>(*i), msg).generic_await( // TODO: maybe infer some useful timeout or use config parameter?
self->request(actor_cast<actor>(*i), indefinite, msg)
.generic_await(
[=](const message& tmp) { [=](const message& tmp) {
self->state.result += tmp; self->state.result += tmp;
if (--self->state.pending == 0) if (--self->state.pending == 0)
......
...@@ -79,7 +79,6 @@ const char* numbered_type_names[] = { ...@@ -79,7 +79,6 @@ const char* numbered_type_names[] = {
"@strmap", "@strmap",
"@strset", "@strset",
"@strvec", "@strvec",
"@sync_timeout",
"@timeout", "@timeout",
"@u16", "@u16",
"@u16str", "@u16str",
......
...@@ -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).receive( self->request(w, indefinite, 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).receive( self->request(w, indefinite, 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).receive( self->request(w, indefinite, 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,12 +177,9 @@ CAF_TEST(random_actor_pool) { ...@@ -177,12 +177,9 @@ 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).receive( self->request(w, std::chrono::milliseconds(250), 1, 2).receive(
[&](int res) { [&](int res) {
CAF_CHECK_EQUAL(res, 3); CAF_CHECK_EQUAL(res, 3);
},
after(std::chrono::milliseconds(250)) >> [] {
CAF_ERROR("didn't receive a down message");
} }
); );
} }
...@@ -212,12 +209,12 @@ CAF_TEST(split_join_actor_pool) { ...@@ -212,12 +209,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}).receive( self->request(w, indefinite, 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}).receive( self->request(w, indefinite, std::vector<int>{6, 7, 8, 9, 10}).receive(
[&](int res) { [&](int res) {
CAF_CHECK_EQUAL(res, 40); CAF_CHECK_EQUAL(res, 40);
} }
......
...@@ -113,7 +113,7 @@ CAF_TEST(lifetime_3) { ...@@ -113,7 +113,7 @@ CAF_TEST(lifetime_3) {
em_sender->link_to(bound->address()); em_sender->link_to(bound->address());
anon_send_exit(em_sender, exit_reason::kill); anon_send_exit(em_sender, exit_reason::kill);
wait_until_exited(); wait_until_exited();
self->request(dbl, 1).receive( self->request(dbl, indefinite, 1).receive(
[](int v) { [](int v) {
CAF_CHECK(v == 2); CAF_CHECK(v == 2);
}, },
...@@ -129,7 +129,7 @@ CAF_TEST(request_response_promise) { ...@@ -129,7 +129,7 @@ CAF_TEST(request_response_promise) {
auto bound = dbl.bind(1); auto bound = dbl.bind(1);
anon_send_exit(bound, exit_reason::kill); anon_send_exit(bound, exit_reason::kill);
CAF_CHECK(exited(bound)); CAF_CHECK(exited(bound));
self->request(bound, message{}).receive( self->request(bound, indefinite, message{}).receive(
[](int) { [](int) {
CAF_CHECK(false); CAF_CHECK(false);
}, },
...@@ -159,12 +159,12 @@ CAF_TEST(partial_currying) { ...@@ -159,12 +159,12 @@ CAF_TEST(partial_currying) {
CAF_CHECK(aut.node() == bound.node()); CAF_CHECK(aut.node() == bound.node());
CAF_CHECK(aut != bound); CAF_CHECK(aut != bound);
CAF_CHECK(system.registry().running() == 1); CAF_CHECK(system.registry().running() == 1);
self->request(bound, 2.0).receive( self->request(bound, indefinite, 2.0).receive(
[](double y) { [](double y) {
CAF_CHECK(y == 2.0); CAF_CHECK(y == 2.0);
} }
); );
self->request(bound, 10).receive( self->request(bound, indefinite, 10).receive(
[](int y) { [](int y) {
CAF_CHECK(y == 10); CAF_CHECK(y == 10);
} }
...@@ -175,7 +175,7 @@ CAF_TEST(partial_currying) { ...@@ -175,7 +175,7 @@ CAF_TEST(partial_currying) {
CAF_TEST(full_currying) { CAF_TEST(full_currying) {
auto dbl_actor = system.spawn(testee); auto dbl_actor = system.spawn(testee);
auto bound = dbl_actor.bind(1); auto bound = dbl_actor.bind(1);
self->request(bound, message{}).receive( self->request(bound, indefinite, message{}).receive(
[](int v) { [](int v) {
CAF_CHECK(v == 2); CAF_CHECK(v == 2);
}, },
...@@ -210,12 +210,12 @@ CAF_TEST(type_safe_currying) { ...@@ -210,12 +210,12 @@ CAF_TEST(type_safe_currying) {
CAF_CHECK(system.registry().running() == 1); CAF_CHECK(system.registry().running() == 1);
static_assert(std::is_same<decltype(bound), curried_signature>::value, static_assert(std::is_same<decltype(bound), curried_signature>::value,
"bind returned wrong actor handle"); "bind returned wrong actor handle");
self->request(bound, 2.0).receive( self->request(bound, indefinite, 2.0).receive(
[](double y) { [](double y) {
CAF_CHECK(y == 2.0); CAF_CHECK(y == 2.0);
} }
); );
self->request(bound, 10).receive( self->request(bound, indefinite, 10).receive(
[](int y) { [](int y) {
CAF_CHECK(y == 10); CAF_CHECK(y == 10);
} }
...@@ -237,7 +237,7 @@ CAF_TEST(reordering) { ...@@ -237,7 +237,7 @@ CAF_TEST(reordering) {
auto bound = aut.bind(_2, _1); auto bound = aut.bind(_2, _1);
CAF_CHECK(aut != bound); CAF_CHECK(aut != bound);
CAF_CHECK(system.registry().running() == 1); CAF_CHECK(system.registry().running() == 1);
self->request(bound, 2.0, 10).receive( self->request(bound, indefinite, 2.0, 10).receive(
[](double y) { [](double y) {
CAF_CHECK(y == 20.0); CAF_CHECK(y == 20.0);
} }
...@@ -263,7 +263,7 @@ CAF_TEST(type_safe_reordering) { ...@@ -263,7 +263,7 @@ CAF_TEST(type_safe_reordering) {
CAF_CHECK(system.registry().running() == 1); CAF_CHECK(system.registry().running() == 1);
static_assert(std::is_same<decltype(bound), swapped_signature>::value, static_assert(std::is_same<decltype(bound), swapped_signature>::value,
"bind returned wrong actor handle"); "bind returned wrong actor handle");
self->request(bound, 2.0, 10).receive( self->request(bound, indefinite, 2.0, 10).receive(
[](double y) { [](double y) {
CAF_CHECK(y == 20.0); CAF_CHECK(y == 20.0);
} }
......
...@@ -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).receive( self->request(tst, indefinite, abc_atom::value).receive(
[](int i) { [](int i) {
CAF_CHECK_EQUAL(i, 42); CAF_CHECK_EQUAL(i, 42);
} }
......
...@@ -140,7 +140,7 @@ CAF_TEST(composable_behaviors) { ...@@ -140,7 +140,7 @@ CAF_TEST(composable_behaviors) {
//auto x1 = sys.spawn<stateful_impl<foo_actor_state>>(); //auto x1 = sys.spawn<stateful_impl<foo_actor_state>>();
auto x1 = sys.spawn<foo_actor_state>(); auto x1 = sys.spawn<foo_actor_state>();
scoped_actor self{sys}; scoped_actor self{sys};
self->request(x1, 1, 2, 4).receive( self->request(x1, indefinite, 1, 2, 4).receive(
[](int y) { [](int y) {
CAF_CHECK(y == 7); CAF_CHECK(y == 7);
} }
...@@ -148,12 +148,12 @@ CAF_TEST(composable_behaviors) { ...@@ -148,12 +148,12 @@ CAF_TEST(composable_behaviors) {
self->send_exit(x1, exit_reason::kill); self->send_exit(x1, exit_reason::kill);
//auto x2 = sys.spawn<stateful_impl<composed_behavior<i3_actor_state, d_actor_state>>>(); //auto x2 = sys.spawn<stateful_impl<composed_behavior<i3_actor_state, d_actor_state>>>();
auto x2 = sys.spawn<composed_behavior<i3_actor_state, d_actor_state>>(); auto x2 = sys.spawn<composed_behavior<i3_actor_state, d_actor_state>>();
self->request(x2, 1, 2, 4).receive( self->request(x2, indefinite, 1, 2, 4).receive(
[](int y) { [](int y) {
CAF_CHECK(y == 7); CAF_CHECK(y == 7);
} }
); );
self->request(x2, 1.0).receive( self->request(x2, indefinite, 1.0).receive(
[](double y1, double y2) { [](double y1, double y2) {
CAF_CHECK(y1 == 1.0); CAF_CHECK(y1 == 1.0);
CAF_CHECK(y1 == y2); CAF_CHECK(y1 == y2);
...@@ -162,7 +162,7 @@ CAF_TEST(composable_behaviors) { ...@@ -162,7 +162,7 @@ CAF_TEST(composable_behaviors) {
self->send_exit(x2, exit_reason::kill); self->send_exit(x2, exit_reason::kill);
//auto x3 = sys.spawn<stateful_impl<foo_actor_state2>>(); //auto x3 = sys.spawn<stateful_impl<foo_actor_state2>>();
auto x3 = sys.spawn<foo_actor_state2>(); auto x3 = sys.spawn<foo_actor_state2>();
self->request(x3, 1, 2, 4).receive( self->request(x3, indefinite, 1, 2, 4).receive(
[](int y) { [](int y) {
CAF_CHECK(y == -5); CAF_CHECK(y == -5);
} }
...@@ -170,7 +170,7 @@ CAF_TEST(composable_behaviors) { ...@@ -170,7 +170,7 @@ CAF_TEST(composable_behaviors) {
self->send_exit(x3, exit_reason::kill); self->send_exit(x3, exit_reason::kill);
//auto x4 = sys.spawn<stateful_impl<dict_calc_state>>(); //auto x4 = sys.spawn<stateful_impl<dict_calc_state>>();
auto x4 = sys.spawn<dict_calc_state>(); auto x4 = sys.spawn<dict_calc_state>();
self->request(x4, add_atom::value, 10, 20).receive( self->request(x4, indefinite, add_atom::value, 10, 20).receive(
[](int y) { [](int y) {
CAF_CHECK(y == 30); CAF_CHECK(y == 30);
} }
......
...@@ -514,15 +514,11 @@ CAF_TEST(requests) { ...@@ -514,15 +514,11 @@ CAF_TEST(requests) {
auto sync_testee = system.spawn([](blocking_actor* s) { auto sync_testee = system.spawn([](blocking_actor* s) {
s->receive ( s->receive (
on("hi", arg_match) >> [&](actor from) { on("hi", arg_match) >> [&](actor from) {
s->request(from, "whassup?", s).receive( s->request(from, chrono::minutes(1), "whassup?", s).receive(
[&](const string& str) { [&](const string& str) {
CAF_CHECK(s->current_sender() != nullptr); CAF_CHECK(s->current_sender() != nullptr);
CAF_CHECK_EQUAL(str, "nothing"); CAF_CHECK_EQUAL(str, "nothing");
s->send(from, "goodbye!"); s->send(from, "goodbye!");
},
after(chrono::minutes(1)) >> [] {
CAF_ERROR("Error in unit test.");
abort();
} }
); );
}, },
...@@ -552,7 +548,7 @@ CAF_TEST(requests) { ...@@ -552,7 +548,7 @@ CAF_TEST(requests) {
} }
); );
self->await_all_other_actors_done(); self->await_all_other_actors_done();
self->request(sync_testee, "!?").receive( self->request(sync_testee, chrono::microseconds(1), "!?").receive(
[] { [] {
CAF_ERROR("Unexpected empty message"); CAF_ERROR("Unexpected empty message");
}, },
...@@ -560,11 +556,7 @@ CAF_TEST(requests) { ...@@ -560,11 +556,7 @@ CAF_TEST(requests) {
if (err == sec::request_receiver_down) if (err == sec::request_receiver_down)
CAF_MESSAGE("received `request_receiver_down`"); CAF_MESSAGE("received `request_receiver_down`");
else else
CAF_ERROR("received unexpected error: " CAF_ERROR("received unexpected error: " << self->system().render(err));
<< self->system().render(err));
},
after(chrono::microseconds(1)) >> [] {
CAF_ERROR("Unexpected timeout");
} }
); );
} }
...@@ -611,7 +603,7 @@ typed_testee::behavior_type testee() { ...@@ -611,7 +603,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).receive( self->request(x, indefinite, abc_atom::value).receive(
[](const std::string& str) { [](const std::string& str) {
CAF_CHECK_EQUAL(str, "abc"); CAF_CHECK_EQUAL(str, "abc");
} }
...@@ -790,7 +782,7 @@ CAF_TEST(move_only_argument) { ...@@ -790,7 +782,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).receive( self->request(testee, indefinite, 1.f).receive(
[](int i) { [](int i) {
CAF_CHECK(i == 42); CAF_CHECK(i == 42);
} }
......
...@@ -22,6 +22,8 @@ ...@@ -22,6 +22,8 @@
#define CAF_SUITE local_migration #define CAF_SUITE local_migration
#include "caf/test/unit_test.hpp" #include "caf/test/unit_test.hpp"
/* --- "disabled" (see #199) ---
#include "caf/all.hpp" #include "caf/all.hpp"
#include "caf/actor_registry.hpp" #include "caf/actor_registry.hpp"
...@@ -83,12 +85,13 @@ CAF_TEST(migrate_locally) { ...@@ -83,12 +85,13 @@ 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).receive( self->request(a, indefinite, 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).receive( self->request(a, indefinite, 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 +100,13 @@ CAF_TEST(migrate_locally) { ...@@ -97,12 +100,13 @@ 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).receive( self->request(b, indefinite, 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).receive( self->request(b, indefinite, 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());
...@@ -115,3 +119,8 @@ CAF_TEST(migrate_locally) { ...@@ -115,3 +119,8 @@ CAF_TEST(migrate_locally) {
self->await_all_other_actors_done(); self->await_all_other_actors_done();
} }
} }
*/
CAF_TEST(migrate_locally) {
// nop
}
...@@ -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).receive([](int i) { self->request(testee, indefinite, a_atom::value).receive([](int i) {
CAF_CHECK_EQUAL(i, 1); CAF_CHECK_EQUAL(i, 1);
}); });
self->request(testee, b_atom::value).receive([](int i) { self->request(testee, indefinite, b_atom::value).receive([](int i) {
CAF_CHECK_EQUAL(i, 2); CAF_CHECK_EQUAL(i, 2);
}); });
self->request(testee, c_atom::value).receive([](int i) { self->request(testee, indefinite, 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, "*").receive( self->request(config_server, indefinite, 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).receive( self->request(config_server, indefinite, 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, "*").receive( self->request(config_server, indefinite, 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())
......
...@@ -120,7 +120,7 @@ public: ...@@ -120,7 +120,7 @@ public:
behavior make_behavior() override { behavior make_behavior() override {
return { return {
[=](go_atom, const actor& next) { [=](go_atom, const actor& next) {
request(next, gogo_atom::value).then( request(next, indefinite, gogo_atom::value).then(
[=](atom_value) { [=](atom_value) {
CAF_MESSAGE("send 'ok' to buddy"); CAF_MESSAGE("send 'ok' to buddy");
send(buddy(), ok_atom::value); send(buddy(), ok_atom::value);
...@@ -196,7 +196,7 @@ public: ...@@ -196,7 +196,7 @@ public:
return { return {
others >> [=](message& msg) -> response_promise { others >> [=](message& msg) -> response_promise {
auto rp = make_response_promise(); auto rp = make_response_promise();
request(buddy(), std::move(msg)).then( request(buddy(), indefinite, std::move(msg)).then(
[=](gogogo_atom x) mutable { [=](gogogo_atom x) mutable {
rp.deliver(x); rp.deliver(x);
quit(); quit();
...@@ -270,7 +270,7 @@ CAF_TEST(test_void_res) { ...@@ -270,7 +270,7 @@ CAF_TEST(test_void_res) {
}; };
}); });
scoped_actor self{system}; scoped_actor self{system};
self->request(buddy, 1, 2).receive( self->request(buddy, indefinite, 1, 2).receive(
[] { [] {
CAF_MESSAGE("received void res"); CAF_MESSAGE("received void res");
} }
...@@ -287,7 +287,7 @@ CAF_TEST(pending_quit) { ...@@ -287,7 +287,7 @@ CAF_TEST(pending_quit) {
}; };
}); });
system.spawn([mirror](event_based_actor* self) { system.spawn([mirror](event_based_actor* self) {
self->request(mirror, 42).then( self->request(mirror, indefinite, 42).then(
[](int) { [](int) {
CAF_ERROR("received result, should've been terminated already"); CAF_ERROR("received result, should've been terminated already");
}, },
...@@ -310,7 +310,7 @@ CAF_TEST(request) { ...@@ -310,7 +310,7 @@ CAF_TEST(request) {
CAF_CHECK_EQUAL(i, 0); CAF_CHECK_EQUAL(i, 0);
} }
); );
s->request(foi, i_atom::value).receive( s->request(foi, indefinite, i_atom::value).receive(
[&](int i) { [&](int i) {
CAF_CHECK_EQUAL(i, 0); CAF_CHECK_EQUAL(i, 0);
++invocations; ++invocations;
...@@ -319,7 +319,7 @@ CAF_TEST(request) { ...@@ -319,7 +319,7 @@ CAF_TEST(request) {
CAF_ERROR("Error: " << s->system().render(err)); CAF_ERROR("Error: " << s->system().render(err));
} }
); );
s->request(foi, f_atom::value).receive( s->request(foi, indefinite, f_atom::value).receive(
[&](float f) { [&](float f) {
CAF_CHECK_EQUAL(f, 0.f); CAF_CHECK_EQUAL(f, 0.f);
++invocations; ++invocations;
...@@ -333,11 +333,14 @@ CAF_TEST(request) { ...@@ -333,11 +333,14 @@ 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).receive( s->request(foi, indefinite, f_atom::value).receive(
[&](int) { [&](int) {
printf("******* %s %d\n", __FILE__, __LINE__);
CAF_ERROR("int handler called");
int_handler_called = true; int_handler_called = true;
}, },
[&](const error&) { [&](const error&) {
printf("******* %s %d\n", __FILE__, __LINE__);
CAF_MESSAGE("error received"); CAF_MESSAGE("error received");
error_handler_called = true; error_handler_called = true;
} }
...@@ -356,7 +359,7 @@ CAF_TEST(request) { ...@@ -356,7 +359,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).receive([&](int value) { self->request(mirror, indefinite, 42).receive([&](int value) {
continuation_called = true; continuation_called = true;
CAF_CHECK_EQUAL(value, 42); CAF_CHECK_EQUAL(value, 42);
}); });
...@@ -395,12 +398,13 @@ CAF_TEST(request) { ...@@ -395,12 +398,13 @@ 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).receive( self->request(self, milliseconds(50), no_way_atom::value).receive(
[&](int) { [&](int) {
CAF_ERROR("unexpected message of type int"); CAF_ERROR("unexpected message of type int");
}, },
after(milliseconds(50)) >> [] { [&](const error& err) {
CAF_MESSAGE("got timeout"); CAF_MESSAGE("err = " << system.render(err));
CAF_REQUIRE(err == sec::request_timeout);
} }
); );
CAF_MESSAGE("expect two DOWN messages and one 'NoWay'"); CAF_MESSAGE("expect two DOWN messages and one 'NoWay'");
...@@ -439,15 +443,13 @@ CAF_TEST(request) { ...@@ -439,15 +443,13 @@ CAF_TEST(request) {
}, },
[&](const error& err) { [&](const error& err) {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
CAF_ERROR("Error: " << self->system().render(err)); CAF_REQUIRE(err == sec::request_timeout);
},
after(milliseconds(500)) >> [&] {
CAF_MESSAGE("timeout occured"); CAF_MESSAGE("timeout occured");
timeout_occured = true; timeout_occured = true;
} }
); );
CAF_CHECK_EQUAL(timeout_occured, true); CAF_CHECK_EQUAL(timeout_occured, true);
self->request(c, gogo_atom::value).receive( self->request(c, indefinite, gogo_atom::value).receive(
[](gogogo_atom) { [](gogogo_atom) {
CAF_MESSAGE("received `gogogo_atom`"); CAF_MESSAGE("received `gogogo_atom`");
}, },
...@@ -472,7 +474,7 @@ CAF_TEST(request) { ...@@ -472,7 +474,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).receive( s->request(serv, indefinite, request_atom::value).receive(
[&](response_atom) { [&](response_atom) {
CAF_MESSAGE("received 'response'"); CAF_MESSAGE("received 'response'");
CAF_CHECK(s->current_sender() == work); CAF_CHECK(s->current_sender() == work);
...@@ -482,7 +484,7 @@ CAF_TEST(request) { ...@@ -482,7 +484,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, indefinite, request_atom::value);
send_as(work, serv, idle_atom::value, work); send_as(work, serv, idle_atom::value, work);
handle.receive( handle.receive(
[&](response_atom) { [&](response_atom) {
...@@ -514,7 +516,7 @@ behavior snyc_send_no_then_A(event_based_actor * self) { ...@@ -514,7 +516,7 @@ behavior snyc_send_no_then_A(event_based_actor * self) {
behavior snyc_send_no_then_B(event_based_actor * self) { behavior snyc_send_no_then_B(event_based_actor * self) {
return { return {
[=](int number) { [=](int number) {
self->request(self->spawn(snyc_send_no_then_A), number); self->request(self->spawn(snyc_send_no_then_A), indefinite, number);
self->quit(); self->quit();
} }
}; };
...@@ -533,7 +535,7 @@ CAF_TEST(async_request) { ...@@ -533,7 +535,7 @@ CAF_TEST(async_request) {
} }
}; };
}); });
self->request(receiver, 1).then( self->request(receiver, indefinite, 1).then(
[=](int) {} [=](int) {}
); );
return { return {
......
...@@ -19,7 +19,7 @@ ...@@ -19,7 +19,7 @@
#include "caf/config.hpp" #include "caf/config.hpp"
#define CAF_SUITE sync_timeout #define CAF_SUITE request_timeout
#include "caf/test/unit_test.hpp" #include "caf/test/unit_test.hpp"
#include <thread> #include <thread>
...@@ -49,13 +49,13 @@ behavior ping1(event_based_actor* self, const actor& pong_actor) { ...@@ -49,13 +49,13 @@ behavior ping1(event_based_actor* self, const actor& pong_actor) {
self->send(self, send_ping_atom::value); self->send(self, send_ping_atom::value);
return { return {
[=](send_ping_atom) { [=](send_ping_atom) {
self->request(pong_actor, ping_atom::value).then( self->request(pong_actor, std::chrono::milliseconds(100), ping_atom::value).then(
[=](pong_atom) { [=](pong_atom) {
CAF_ERROR("received pong atom"); CAF_ERROR("received pong atom");
self->quit(exit_reason::user_shutdown); self->quit(exit_reason::user_shutdown);
}, },
after(std::chrono::milliseconds(100)) >> [=] { [=](const error& err) {
CAF_MESSAGE("sync timeout: check"); CAF_REQUIRE(err == sec::request_timeout);
self->quit(exit_reason::user_shutdown); self->quit(exit_reason::user_shutdown);
} }
); );
...@@ -69,12 +69,13 @@ behavior ping2(event_based_actor* self, const actor& pong_actor) { ...@@ -69,12 +69,13 @@ behavior ping2(event_based_actor* self, const actor& pong_actor) {
auto received_inner = std::make_shared<bool>(false); auto received_inner = std::make_shared<bool>(false);
return { return {
[=](send_ping_atom) { [=](send_ping_atom) {
self->request(pong_actor, ping_atom::value).then( self->request(pong_actor, std::chrono::milliseconds(100), ping_atom::value).then(
[=](pong_atom) { [=](pong_atom) {
CAF_ERROR("received pong atom"); CAF_ERROR("received pong atom");
self->quit(exit_reason::user_shutdown); self->quit(exit_reason::user_shutdown);
}, },
after(std::chrono::milliseconds(100)) >> [=] { [=](const error& err) {
CAF_REQUIRE(err == sec::request_timeout);
CAF_MESSAGE("inner timeout: check"); CAF_MESSAGE("inner timeout: check");
*received_inner = true; *received_inner = true;
} }
...@@ -92,12 +93,14 @@ behavior ping3(event_based_actor* self, const actor& pong_actor) { ...@@ -92,12 +93,14 @@ behavior ping3(event_based_actor* self, const actor& pong_actor) {
self->send(self, send_ping_atom::value); self->send(self, send_ping_atom::value);
return { return {
[=](send_ping_atom) { [=](send_ping_atom) {
self->request(pong_actor, ping_atom::value).then( self->request(pong_actor, std::chrono::milliseconds(100),
ping_atom::value).then(
[=](pong_atom) { [=](pong_atom) {
CAF_ERROR("received pong atom"); CAF_ERROR("received pong atom");
self->quit(exit_reason::user_shutdown); self->quit(exit_reason::user_shutdown);
}, },
after(std::chrono::milliseconds(100)) >> [=] { [=](const error& err) {
CAF_REQUIRE(err == sec::request_timeout);
CAF_MESSAGE("async timeout: check"); CAF_MESSAGE("async timeout: check");
self->quit(exit_reason::user_shutdown); self->quit(exit_reason::user_shutdown);
} }
...@@ -112,12 +115,14 @@ behavior ping4(event_based_actor* self, const actor& pong_actor) { ...@@ -112,12 +115,14 @@ behavior ping4(event_based_actor* self, const actor& pong_actor) {
auto received_outer = std::make_shared<bool>(false); auto received_outer = std::make_shared<bool>(false);
return { return {
[=](send_ping_atom) { [=](send_ping_atom) {
self->request(pong_actor, ping_atom::value).then( self->request(pong_actor, std::chrono::milliseconds(100),
ping_atom::value).then(
[=](pong_atom) { [=](pong_atom) {
CAF_ERROR("received pong atom"); CAF_ERROR("received pong atom");
self->quit(exit_reason::user_shutdown); self->quit(exit_reason::user_shutdown);
}, },
after(std::chrono::milliseconds(100)) >> [=] { [=](const error& err) {
CAF_REQUIRE(err == sec::request_timeout);
CAF_CHECK_EQUAL(*received_outer, true); CAF_CHECK_EQUAL(*received_outer, true);
self->quit(exit_reason::user_shutdown); self->quit(exit_reason::user_shutdown);
} }
...@@ -132,28 +137,27 @@ behavior ping4(event_based_actor* self, const actor& pong_actor) { ...@@ -132,28 +137,27 @@ behavior ping4(event_based_actor* self, const actor& pong_actor) {
void ping5(event_based_actor* self, const actor& pong_actor) { void ping5(event_based_actor* self, const actor& pong_actor) {
self->link_to(pong_actor); self->link_to(pong_actor);
auto flag = std::make_shared<int>(0); auto timeouts = std::make_shared<int>(0);
self->request(pong_actor, ping_atom::value).then( self->request(pong_actor, std::chrono::milliseconds(100),
ping_atom::value).then(
[=](pong_atom) { [=](pong_atom) {
CAF_ERROR("received pong atom"); CAF_ERROR("received pong atom");
*flag = 1;
}, },
after(std::chrono::milliseconds(100)) >> [=] { [=](const error& err) {
CAF_MESSAGE("multiplexed response timeout: check"); CAF_REQUIRE(err == sec::request_timeout);
CAF_CHECK_EQUAL(*flag, 4); if (++*timeouts == 2)
*flag = 2; self->quit();
self->quit(exit_reason::user_shutdown);
} }
); );
self->request(pong_actor, ping_atom::value).await( self->request(pong_actor, std::chrono::milliseconds(100),
ping_atom::value).await(
[=](pong_atom) { [=](pong_atom) {
CAF_ERROR("received pong atom"); CAF_ERROR("received pong atom");
*flag = 3;
}, },
after(std::chrono::milliseconds(100)) >> [=] { [=](const error& err) {
CAF_MESSAGE("awaited response timeout: check"); CAF_REQUIRE(err == sec::request_timeout);
CAF_CHECK_EQUAL(*flag, 0); if (++*timeouts == 2)
*flag = 4; self->quit();
} }
); );
} }
......
...@@ -158,7 +158,7 @@ CAF_TEST(lifetime_3) { ...@@ -158,7 +158,7 @@ CAF_TEST(lifetime_3) {
em_sender->link_to(h.address()); em_sender->link_to(h.address());
anon_send_exit(em_sender, exit_reason::kill); anon_send_exit(em_sender, exit_reason::kill);
wait_until_exited(); wait_until_exited();
self->request(f, 1).receive( self->request(f, indefinite, 1).receive(
[](int v) { [](int v) {
CAF_CHECK(v == 2); CAF_CHECK(v == 2);
}, },
...@@ -166,7 +166,7 @@ CAF_TEST(lifetime_3) { ...@@ -166,7 +166,7 @@ CAF_TEST(lifetime_3) {
CAF_CHECK(false); CAF_CHECK(false);
} }
); );
self->request(g, 1).receive( self->request(g, indefinite, 1).receive(
[](int v) { [](int v) {
CAF_CHECK(v == 2); CAF_CHECK(v == 2);
}, },
...@@ -184,7 +184,7 @@ CAF_TEST(request_response_promise) { ...@@ -184,7 +184,7 @@ CAF_TEST(request_response_promise) {
auto h = f * g; auto h = f * g;
anon_send_exit(h, exit_reason::kill); anon_send_exit(h, exit_reason::kill);
CAF_CHECK(exited(h)); CAF_CHECK(exited(h));
self->request(h, 1).receive( self->request(h, indefinite, 1).receive(
[](int) { [](int) {
CAF_CHECK(false); CAF_CHECK(false);
}, },
...@@ -201,7 +201,7 @@ CAF_TEST(dot_composition_1) { ...@@ -201,7 +201,7 @@ CAF_TEST(dot_composition_1) {
auto first = system.spawn(typed_first_stage); auto first = system.spawn(typed_first_stage);
auto second = system.spawn(typed_second_stage); auto second = system.spawn(typed_second_stage);
auto first_then_second = second * first; auto first_then_second = second * first;
self->request(first_then_second, 42).receive( self->request(first_then_second, indefinite, 42).receive(
[](double res) { [](double res) {
CAF_CHECK(res == (42 * 2.0) * (42 * 4.0)); CAF_CHECK(res == (42 * 2.0) * (42 * 4.0));
} }
...@@ -215,7 +215,7 @@ CAF_TEST(dot_composition_2) { ...@@ -215,7 +215,7 @@ CAF_TEST(dot_composition_2) {
auto dbl_actor = system.spawn(testee); auto dbl_actor = system.spawn(testee);
auto dbl_x4_actor = dbl_actor * dbl_actor auto dbl_x4_actor = dbl_actor * dbl_actor
* dbl_actor * dbl_actor; * dbl_actor * dbl_actor;
self->request(dbl_x4_actor, 1).receive( self->request(dbl_x4_actor, indefinite, 1).receive(
[](int v) { [](int v) {
CAF_CHECK(v == 16); CAF_CHECK(v == 16);
}, },
......
...@@ -59,19 +59,19 @@ CAF_TEST(test_serial_reply) { ...@@ -59,19 +59,19 @@ CAF_TEST(test_serial_reply) {
[=](hi_atom) mutable { [=](hi_atom) mutable {
auto rp = self->make_response_promise(); auto rp = self->make_response_promise();
CAF_MESSAGE("received 'hi there'"); CAF_MESSAGE("received 'hi there'");
self->request(c0, sub0_atom::value).then( self->request(c0, indefinite, sub0_atom::value).then(
[=](sub0_atom) mutable { [=](sub0_atom) mutable {
CAF_MESSAGE("received 'sub0'"); CAF_MESSAGE("received 'sub0'");
self->request(c1, sub1_atom::value).then( self->request(c1, indefinite, sub1_atom::value).then(
[=](sub1_atom) mutable { [=](sub1_atom) mutable {
CAF_MESSAGE("received 'sub1'"); CAF_MESSAGE("received 'sub1'");
self->request(c2, sub2_atom::value).then( self->request(c2, indefinite, sub2_atom::value).then(
[=](sub2_atom) mutable { [=](sub2_atom) mutable {
CAF_MESSAGE("received 'sub2'"); CAF_MESSAGE("received 'sub2'");
self->request(c3, sub3_atom::value).then( self->request(c3, indefinite, sub3_atom::value).then(
[=](sub3_atom) mutable { [=](sub3_atom) mutable {
CAF_MESSAGE("received 'sub3'"); CAF_MESSAGE("received 'sub3'");
self->request(c4, sub4_atom::value).then( self->request(c4, indefinite, sub4_atom::value).then(
[=](sub4_atom) mutable { [=](sub4_atom) mutable {
CAF_MESSAGE("received 'sub4'"); CAF_MESSAGE("received 'sub4'");
rp.deliver(ho_atom::value); rp.deliver(ho_atom::value);
...@@ -90,7 +90,7 @@ CAF_TEST(test_serial_reply) { ...@@ -90,7 +90,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).receive( self->request(master, indefinite, hi_atom::value).receive(
[](ho_atom) { [](ho_atom) {
CAF_MESSAGE("received 'ho'"); CAF_MESSAGE("received 'ho'");
}, },
......
...@@ -108,7 +108,7 @@ CAF_TEST(kill_second) { ...@@ -108,7 +108,7 @@ CAF_TEST(kill_second) {
CAF_TEST(untyped_splicing) { CAF_TEST(untyped_splicing) {
init_untyped(); init_untyped();
self->request(first_and_second, 42.0).receive( self->request(first_and_second, indefinite, 42.0).receive(
[](double x, double y, double z) { [](double x, double y, double z) {
CAF_CHECK(x == (42.0 * 2.0)); CAF_CHECK(x == (42.0 * 2.0));
CAF_CHECK(y == (42.0 * 4.0)); CAF_CHECK(y == (42.0 * 4.0));
...@@ -126,7 +126,7 @@ CAF_TEST(typed_splicing) { ...@@ -126,7 +126,7 @@ CAF_TEST(typed_splicing) {
::with<double, double, double>>; ::with<double, double, double>>;
static_assert(std::is_same<decltype(x_and_y), expected_type>::value, static_assert(std::is_same<decltype(x_and_y), expected_type>::value,
"splice() did not compute the correct result"); "splice() did not compute the correct result");
self->request(x_and_y, 42.0).receive( self->request(x_and_y, indefinite, 42.0).receive(
[](double x, double y, double z) { [](double x, double y, double z) {
CAF_CHECK(x == (42.0 * 2.0)); CAF_CHECK(x == (42.0 * 2.0));
CAF_CHECK(y == (42.0 * 4.0)); CAF_CHECK(y == (42.0 * 4.0));
......
...@@ -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).receive( self->request(aut, indefinite, 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).receive( self->request(aut, indefinite, get_atom::value).receive(
[&](const string& str) { [&](const string& str) {
CAF_CHECK_EQUAL(str, expected); CAF_CHECK_EQUAL(str, expected);
} }
......
...@@ -147,18 +147,18 @@ CAF_TEST_FIXTURE_SCOPE(typed_spawn_tests, fixture) ...@@ -147,18 +147,18 @@ CAF_TEST_FIXTURE_SCOPE(typed_spawn_tests, fixture)
CAF_TEST(typed_response_promise) { CAF_TEST(typed_response_promise) {
typed_response_promise<int> resp; typed_response_promise<int> resp;
resp.deliver(1); // delivers on an invalid promise has no effect resp.deliver(1); // delivers on an invalid promise has no effect
self->request(foo, get_atom::value, 42).receive( self->request(foo, indefinite, 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).receive( self->request(foo, indefinite, 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);
} }
); );
self->request(foo, get_atom::value, 3.14, 3.14).receive( self->request(foo, indefinite, get_atom::value, 3.14, 3.14).receive(
[](double x, double y) { [](double x, double y) {
CAF_CHECK_EQUAL(x, 3.14 * 2); CAF_CHECK_EQUAL(x, 3.14 * 2);
CAF_CHECK_EQUAL(y, 3.14 * 2); CAF_CHECK_EQUAL(y, 3.14 * 2);
...@@ -172,7 +172,7 @@ CAF_TEST(typed_response_promise) { ...@@ -172,7 +172,7 @@ CAF_TEST(typed_response_promise) {
CAF_TEST(typed_response_promise_chained) { CAF_TEST(typed_response_promise_chained) {
auto composed = foo * foo * foo; auto composed = foo * foo * foo;
self->request(composed, 1).receive( self->request(composed, indefinite, 1).receive(
[](int v) { [](int v) {
CAF_CHECK_EQUAL(v, 8); CAF_CHECK_EQUAL(v, 8);
}, },
...@@ -185,7 +185,7 @@ CAF_TEST(typed_response_promise_chained) { ...@@ -185,7 +185,7 @@ CAF_TEST(typed_response_promise_chained) {
// verify that only requests get an error response message // verify that only requests get an error response message
CAF_TEST(error_response_message) { CAF_TEST(error_response_message) {
self->request(foo, get_atom::value, 3.14).receive( self->request(foo, indefinite, get_atom::value, 3.14).receive(
[](double) { [](double) {
CAF_ERROR("unexpected ordinary response message received"); CAF_ERROR("unexpected ordinary response message received");
}, },
......
...@@ -118,10 +118,10 @@ public: ...@@ -118,10 +118,10 @@ public:
}; };
void client(event_based_actor* self, actor parent, server_type serv) { void client(event_based_actor* self, actor parent, server_type serv) {
self->request(serv, my_request{0, 0}).then( self->request(serv, indefinite, my_request{0, 0}).then(
[=](bool val1) { [=](bool val1) {
CAF_CHECK_EQUAL(val1, true); CAF_CHECK_EQUAL(val1, true);
self->request(serv, my_request{10, 20}).then( self->request(serv, indefinite, my_request{10, 20}).then(
[=](bool val2) { [=](bool val2) {
CAF_CHECK_EQUAL(val2, false); CAF_CHECK_EQUAL(val2, false);
self->send(parent, passed_atom::value); self->send(parent, passed_atom::value);
...@@ -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}).receive( self->request(ts, indefinite, 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}).receive( self->request(ts, indefinite, my_request{0, 0}).receive(
[](bool value) { [](bool value) {
CAF_CHECK_EQUAL(value, true); CAF_CHECK_EQUAL(value, true);
} }
...@@ -370,7 +370,6 @@ CAF_TEST(event_testee_series) { ...@@ -370,7 +370,6 @@ CAF_TEST(event_testee_series) {
self->send(et, "hello again event testee!"); self->send(et, "hello again event testee!");
self->send(et, "goodbye event testee!"); self->send(et, "goodbye event testee!");
typed_actor<replies_to<get_state_msg>::with<string>> sub_et = et; typed_actor<replies_to<get_state_msg>::with<string>> sub_et = et;
// $:: is the anonymous namespace
set<string> iface{"caf::replies_to<get_state_msg>::with<@str>", set<string> iface{"caf::replies_to<get_state_msg>::with<@str>",
"caf::replies_to<@str>::with<void>", "caf::replies_to<@str>::with<void>",
"caf::replies_to<float>::with<void>", "caf::replies_to<float>::with<void>",
...@@ -403,7 +402,7 @@ CAF_TEST(string_delegator_chain) { ...@@ -403,7 +402,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!").receive( self->request(aut, indefinite, "Hello World!").receive(
[](const string& answer) { [](const string& answer) {
CAF_CHECK_EQUAL(answer, "!dlroW olleH"); CAF_CHECK_EQUAL(answer, "!dlroW olleH");
} }
...@@ -417,7 +416,7 @@ CAF_TEST(maybe_string_delegator_chain) { ...@@ -417,7 +416,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, "").receive( self->request(aut, indefinite, "").receive(
[](ok_atom, const string&) { [](ok_atom, const string&) {
throw std::logic_error("unexpected result!"); throw std::logic_error("unexpected result!");
}, },
...@@ -428,7 +427,7 @@ CAF_TEST(maybe_string_delegator_chain) { ...@@ -428,7 +427,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").receive( self->request(aut, indefinite, "abcd").receive(
[](ok_atom, const string& str) { [](ok_atom, const string& str) {
CAF_CHECK_EQUAL(str, "dcba"); CAF_CHECK_EQUAL(str, "dcba");
}, },
......
...@@ -121,7 +121,7 @@ uint16_t middleman::publish(const actor_addr& whom, std::set<std::string> sigs, ...@@ -121,7 +121,7 @@ uint16_t middleman::publish(const actor_addr& whom, std::set<std::string> sigs,
uint16_t result; uint16_t result;
std::string error_msg; std::string error_msg;
try { try {
self->request(mm, publish_atom::value, port, self->request(mm, indefinite, publish_atom::value, port,
std::move(whom), std::move(sigs), str, ru).receive( std::move(whom), std::move(sigs), str, ru).receive(
[&](ok_atom, uint16_t res) { [&](ok_atom, uint16_t res) {
result = res; result = res;
...@@ -168,7 +168,8 @@ maybe<uint16_t> middleman::publish_local_groups(uint16_t port, const char* in) { ...@@ -168,7 +168,8 @@ 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).receive( self->request(actor_handle(), indefinite,
unpublish_atom::value, whom, port).receive(
[] { [] {
// ok, basp_broker is done // ok, basp_broker is done
}, },
...@@ -184,7 +185,8 @@ actor_addr middleman::remote_actor(std::set<std::string> ifs, ...@@ -184,7 +185,8 @@ 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).receive( self->request(mm, indefinite, 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)
......
...@@ -112,7 +112,7 @@ public: ...@@ -112,7 +112,7 @@ public:
} }
std::vector<response_promise> tmp{std::move(rp)}; std::vector<response_promise> tmp{std::move(rp)};
pending_.emplace(key, std::move(tmp)); pending_.emplace(key, std::move(tmp));
request(broker_, connect_atom::value, hdl, port).then( request(broker_, indefinite, connect_atom::value, hdl, port).then(
[=](ok_atom, node_id& nid, actor_addr& addr, mpi_set& sigs) { [=](ok_atom, node_id& nid, actor_addr& addr, mpi_set& sigs) {
auto i = pending_.find(key); auto i = pending_.find(key);
if (i == pending_.end()) if (i == pending_.end())
......
...@@ -560,8 +560,8 @@ CAF_TEST(remote_actor_and_send) { ...@@ -560,8 +560,8 @@ CAF_TEST(remote_actor_and_send) {
CAF_REQUIRE(mpx()->pending_scribes().count(make_pair(lo, 4242)) == 1); CAF_REQUIRE(mpx()->pending_scribes().count(make_pair(lo, 4242)) == 1);
auto mm1 = system.middleman().actor_handle(); auto mm1 = system.middleman().actor_handle();
actor result; actor result;
auto f = self()->request(mm1, connect_atom::value, auto f = self()->request(mm1, indefinite,
lo, uint16_t{4242}); connect_atom::value, lo, uint16_t{4242});
// wait until BASP broker has received and processed the connect message // wait until BASP broker has received and processed the connect message
while (! aut()->valid(remote_hdl(0))) while (! aut()->valid(remote_hdl(0)))
mpx()->exec_runnable(); mpx()->exec_runnable();
......
...@@ -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).receive( self->request(serv, indefinite, 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); });
......
...@@ -87,7 +87,7 @@ void make_client_behavior(event_based_actor* self, ...@@ -87,7 +87,7 @@ void make_client_behavior(event_based_actor* self,
actor server, group grp) { actor server, group grp) {
self->spawn_in_group(grp, make_reflector_behavior); self->spawn_in_group(grp, make_reflector_behavior);
self->spawn_in_group(grp, make_reflector_behavior); self->spawn_in_group(grp, make_reflector_behavior);
self->request(server, spawn_atom::value, grp).then( self->request(server, indefinite, spawn_atom::value, grp).then(
[=](const std::vector<actor>& vec) { [=](const std::vector<actor>& vec) {
auto is_remote = [=](actor actor) { auto is_remote = [=](actor actor) {
return actor->node() != self->node(); return actor->node() != self->node();
...@@ -140,7 +140,7 @@ CAF_TEST(server_side_group_comm) { ...@@ -140,7 +140,7 @@ CAF_TEST(server_side_group_comm) {
CAF_REQUIRE(server); CAF_REQUIRE(server);
scoped_actor group_resolver(client_side, true); scoped_actor group_resolver(client_side, true);
group grp; group grp;
group_resolver->request(server, get_group_atom::value).receive( group_resolver->request(server, indefinite, get_group_atom::value).receive(
[&](const group& x) { [&](const group& x) {
grp = x; grp = x;
} }
......
...@@ -68,7 +68,7 @@ behavior server(stateful_actor<server_state>* self) { ...@@ -68,7 +68,7 @@ behavior server(stateful_actor<server_state>* self) {
CAF_REQUIRE(self->node() != s.node()); CAF_REQUIRE(self->node() != s.node());
self->state.client = actor_cast<actor>(s); self->state.client = actor_cast<actor>(s);
auto mm = self->system().middleman().actor_handle(); auto mm = self->system().middleman().actor_handle();
self->request(mm, spawn_atom::value, self->request(mm, indefinite, spawn_atom::value,
s.node(), "mirror", make_message()).then( s.node(), "mirror", make_message()).then(
[=](ok_atom, const actor_addr& addr, const std::set<std::string>& ifs) { [=](ok_atom, const actor_addr& addr, const std::set<std::string>& ifs) {
CAF_LOG_TRACE(CAF_ARG(addr) << CAF_ARG(ifs)); CAF_LOG_TRACE(CAF_ARG(addr) << CAF_ARG(ifs));
......
...@@ -187,7 +187,7 @@ void run_server(int argc, char** argv) { ...@@ -187,7 +187,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).receive( self->request(serv, indefinite, 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([=] {
......
...@@ -94,7 +94,7 @@ void run_client(int argc, char** argv, uint16_t port) { ...@@ -94,7 +94,7 @@ void run_client(int argc, char** argv, uint16_t port) {
port); port);
CAF_REQUIRE(serv); CAF_REQUIRE(serv);
scoped_actor self{system}; scoped_actor self{system};
self->request(serv, ping{42}) self->request(serv, indefinite, ping{42})
.receive([](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);
......
...@@ -72,7 +72,7 @@ struct fixture { ...@@ -72,7 +72,7 @@ struct fixture {
maybe<actor> remote_actor(const char* hostname, uint16_t port) { maybe<actor> remote_actor(const char* hostname, uint16_t port) {
maybe<actor> result; maybe<actor> result;
scoped_actor self{system, true}; scoped_actor self{system, true};
self->request(system.middleman().actor_handle(), self->request(system.middleman().actor_handle(), indefinite,
connect_atom::value, hostname, port).receive( connect_atom::value, hostname, port).receive(
[&](ok_atom, node_id&, actor_addr& res, std::set<std::string>& xs) { [&](ok_atom, node_id&, actor_addr& res, std::set<std::string>& xs) {
CAF_REQUIRE(xs.empty()); CAF_REQUIRE(xs.empty());
......
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