Commit 2e2eb1c5 authored by Dominik Charousset's avatar Dominik Charousset

Add error, exit, and down handlers

parent 23e33b4f
...@@ -106,6 +106,13 @@ behavior broker_impl(broker* self, connection_handle hdl, const actor& buddy) { ...@@ -106,6 +106,13 @@ behavior broker_impl(broker* self, connection_handle hdl, const actor& buddy) {
assert(self->num_connections() == 1); assert(self->num_connections() == 1);
// monitor buddy to quit broker if buddy is done // monitor buddy to quit broker if buddy is done
self->monitor(buddy); self->monitor(buddy);
self->set_down_handler([=](down_msg& dm) {
if (dm.source == buddy) {
aout(self) << "our buddy is down" << endl;
// quit for same reason
self->quit(dm.reason);
}
});
// setup: we are exchanging only messages consisting of an atom // setup: we are exchanging only messages consisting of an atom
// (as uint64_t) and an integer value (int32_t) // (as uint64_t) and an integer value (int32_t)
self->configure_read(hdl, receive_policy::exactly(sizeof(uint64_t) + self->configure_read(hdl, receive_policy::exactly(sizeof(uint64_t) +
...@@ -130,13 +137,6 @@ behavior broker_impl(broker* self, connection_handle hdl, const actor& buddy) { ...@@ -130,13 +137,6 @@ behavior broker_impl(broker* self, connection_handle hdl, const actor& buddy) {
write_int(self, hdl, static_cast<uint64_t>(av)); write_int(self, hdl, static_cast<uint64_t>(av));
write_int(self, hdl, i); write_int(self, hdl, i);
}, },
[=](const down_msg& dm) {
if (dm.source == buddy) {
aout(self) << "our buddy is down" << endl;
// quit for same reason
self->quit(dm.reason);
}
},
[=](const new_data_msg& msg) { [=](const new_data_msg& msg) {
// read the atom value as uint64_t from buffer // read the atom value as uint64_t from buffer
uint64_t atm_val; uint64_t atm_val;
......
...@@ -46,6 +46,9 @@ behavior connection_worker(broker* self, connection_handle hdl) { ...@@ -46,6 +46,9 @@ behavior connection_worker(broker* self, connection_handle hdl) {
behavior server(broker* self) { behavior server(broker* self) {
auto counter = std::make_shared<int>(0); auto counter = std::make_shared<int>(0);
self->set_down_handler([=](down_msg&) {
++*counter;
});
self->delayed_send(self, std::chrono::seconds(1), tick_atom::value); self->delayed_send(self, std::chrono::seconds(1), tick_atom::value);
return { return {
[=](const new_connection_msg& ncm) { [=](const new_connection_msg& ncm) {
...@@ -53,9 +56,6 @@ behavior server(broker* self) { ...@@ -53,9 +56,6 @@ behavior server(broker* self) {
self->monitor(worker); self->monitor(worker);
self->link_to(worker); self->link_to(worker);
}, },
[=](const down_msg&) {
++*counter;
},
[=](tick_atom) { [=](tick_atom) {
aout(self) << "Finished " << *counter << " requests per second." << endl; aout(self) << "Finished " << *counter << " requests per second." << endl;
*counter = 0; *counter = 0;
......
...@@ -5,7 +5,7 @@ ...@@ -5,7 +5,7 @@
// This example is partially included in the manual, do not modify // This example is partially included in the manual, do not modify
// without updating the references in the *.tex files! // without updating the references in the *.tex files!
// Manual references: lines 17-21, 31-65, 67-101, and 135-140 (Actor.tex) // Manual references: lines 17-21, 31-65, 67-101, and 134-139 (Actor.tex)
#include <iostream> #include <iostream>
...@@ -107,7 +107,6 @@ void tester(scoped_actor&) { ...@@ -107,7 +107,6 @@ void tester(scoped_actor&) {
// tests a calculator instance // tests a calculator instance
template <class Handle, class... Ts> 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);
// first test: x + y = z // first test: x + y = z
self->request(hdl, infinite, add_atom::value, x, y).receive( self->request(hdl, infinite, add_atom::value, x, y).receive(
[&](int res1) { [&](int res1) {
...@@ -116,7 +115,6 @@ void tester(scoped_actor& self, const Handle& hdl, int x, int y, Ts&&... xs) { ...@@ -116,7 +115,6 @@ void tester(scoped_actor& self, const Handle& hdl, int x, int y, Ts&&... xs) {
self->request(hdl, infinite, sub_atom::value, x, y).receive( self->request(hdl, infinite, 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);
} }
); );
}, },
...@@ -126,7 +124,6 @@ void tester(scoped_actor& self, const Handle& hdl, int x, int y, Ts&&... xs) { ...@@ -126,7 +124,6 @@ void tester(scoped_actor& self, const Handle& hdl, int x, int y, Ts&&... xs) {
self->quit(exit_reason::user_shutdown); self->quit(exit_reason::user_shutdown);
} }
); );
self->receive([](const down_msg&) {});
tester(self, std::forward<Ts>(xs)...); tester(self, std::forward<Ts>(xs)...);
} }
......
...@@ -64,11 +64,11 @@ int main() { ...@@ -64,11 +64,11 @@ int main() {
cells.emplace_back(system.spawn(cell_impl, i * i)); cells.emplace_back(system.spawn(cell_impl, i * i));
scoped_actor self{system}; scoped_actor self{system};
aout(self) << "waiting_testee" << endl; aout(self) << "waiting_testee" << endl;
self->spawn<monitored>(waiting_testee, cells); auto x1 = self->spawn(waiting_testee, cells);
self->receive([](const down_msg&) {}); self->wait_for(x1);
aout(self) << "multiplexed_testee" << endl; aout(self) << "multiplexed_testee" << endl;
self->spawn<monitored>(multiplexed_testee, cells); auto x2 = self->spawn(multiplexed_testee, cells);
self->receive([](const down_msg&) {}); self->wait_for(x2);
aout(self) << "blocking_testee" << endl; aout(self) << "blocking_testee" << endl;
system.spawn(blocking_testee, cells); system.spawn(blocking_testee, cells);
} }
...@@ -21,7 +21,6 @@ ...@@ -21,7 +21,6 @@
#define CAF_ABSTRACT_ACTOR_HPP #define CAF_ABSTRACT_ACTOR_HPP
#include <set> #include <set>
#include <mutex>
#include <atomic> #include <atomic>
#include <memory> #include <memory>
#include <string> #include <string>
...@@ -29,7 +28,6 @@ ...@@ -29,7 +28,6 @@
#include <cstdint> #include <cstdint>
#include <exception> #include <exception>
#include <type_traits> #include <type_traits>
#include <condition_variable>
#include "caf/fwd.hpp" #include "caf/fwd.hpp"
#include "caf/node_id.hpp" #include "caf/node_id.hpp"
...@@ -134,8 +132,6 @@ public: ...@@ -134,8 +132,6 @@ public:
}; };
// flags storing runtime information used by ... // flags storing runtime information used by ...
static constexpr int trap_exit_flag = 0x0001; // local_actor
static constexpr int trap_error_flag = 0x0002; // local_actor
static constexpr int has_timeout_flag = 0x0004; // single_timeout static constexpr int has_timeout_flag = 0x0004; // single_timeout
static constexpr int is_registered_flag = 0x0008; // (several actors) static constexpr int is_registered_flag = 0x0008; // (several actors)
static constexpr int is_initialized_flag = 0x0010; // event-based actors static constexpr int is_initialized_flag = 0x0010; // event-based actors
......
...@@ -23,8 +23,8 @@ ...@@ -23,8 +23,8 @@
#include <mutex> #include <mutex>
#include <condition_variable> #include <condition_variable>
#include "caf/send.hpp"
#include "caf/none.hpp" #include "caf/none.hpp"
#include "caf/extend.hpp" #include "caf/extend.hpp"
#include "caf/behavior.hpp" #include "caf/behavior.hpp"
#include "caf/local_actor.hpp" #include "caf/local_actor.hpp"
...@@ -57,6 +57,8 @@ public: ...@@ -57,6 +57,8 @@ public:
~blocking_actor(); ~blocking_actor();
void enqueue(mailbox_element_ptr, execution_unit*) override;
/************************************************************************** /**************************************************************************
* utility stuff and receive() member function family * * utility stuff and receive() member function family *
**************************************************************************/ **************************************************************************/
...@@ -72,7 +74,8 @@ public: ...@@ -72,7 +74,8 @@ public:
static_assert(sizeof...(Ts) > 0, static_assert(sizeof...(Ts) > 0,
"operator() requires at least one argument"); "operator() requires at least one argument");
behavior bhvr{std::forward<Ts>(xs)...}; behavior bhvr{std::forward<Ts>(xs)...};
while (stmt_()) dq_(bhvr); while (stmt_())
dq_(bhvr);
} }
}; };
...@@ -185,6 +188,26 @@ public: ...@@ -185,6 +188,26 @@ public:
/// Implements the actor's behavior. /// Implements the actor's behavior.
virtual void act(); virtual void act();
/// Blocks this actor until all `xs...` have terminated.
template <class... Ts>
void wait_for(Ts&&... xs) {
using wait_for_atom = atom_constant<atom("waitFor")>;
auto old = default_handler_;
default_handler_ = skip;
size_t expected = 0;
size_t i = 0;
size_t attach_results[] = {attach_functor(xs)...};
for (auto res : attach_results)
expected += res;
receive_for(i, expected)(
[](wait_for_atom) {
// nop
}
);
// restore custom default handler
default_handler_ = old;
}
/// @cond PRIVATE /// @cond PRIVATE
inline const error& fail_state() { inline const error& fail_state() {
...@@ -195,7 +218,7 @@ public: ...@@ -195,7 +218,7 @@ public:
void dequeue(behavior& bhvr, message_id mid = invalid_message_id); void dequeue(behavior& bhvr, message_id mid = invalid_message_id);
using local_actor::await_data; void await_data();
/// @endcond /// @endcond
...@@ -204,6 +227,26 @@ protected: ...@@ -204,6 +227,26 @@ protected:
std::function<void(behavior&)> make_dequeue_callback() { std::function<void(behavior&)> make_dequeue_callback() {
return [=](behavior& bhvr) { dequeue(bhvr); }; return [=](behavior& bhvr) { dequeue(bhvr); };
} }
private:
size_t attach_functor(const actor&);
size_t attach_functor(const actor_addr&);
size_t attach_functor(const strong_actor_ptr&);
template <class... Ts>
size_t attach_functor(const typed_actor<Ts...>& x) {
return attach_functor(actor_cast<strong_actor_ptr>(x));
}
template <class Container>
size_t attach_functor(const Container& xs) {
size_t res = 0;
for (auto& x : xs)
res += attach_functor(x);
return res;
}
}; };
} // namespace caf } // namespace caf
......
...@@ -64,6 +64,7 @@ using sorted_builtin_types = ...@@ -64,6 +64,7 @@ using sorted_builtin_types =
strong_actor_ptr, // @strong_actor_ptr strong_actor_ptr, // @strong_actor_ptr
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_msg
timeout_msg, // @timeout timeout_msg, // @timeout
uint16_t, // @u16 uint16_t, // @u16
std::u16string, // @u16_str std::u16string, // @u16_str
......
...@@ -61,17 +61,6 @@ struct deduce_rhs_result { ...@@ -61,17 +61,6 @@ struct deduce_rhs_result {
using type = type_list<>; using type = type_list<>;
}; };
template <class T>
struct is_hidden_msg_handler : std::false_type { };
template <>
struct is_hidden_msg_handler<typed_mpi<type_list<exit_msg>,
type_list<void>>> : std::true_type { };
template <>
struct is_hidden_msg_handler<typed_mpi<type_list<down_msg>,
type_list<void>>> : std::true_type { };
template <class T> template <class T>
struct deduce_mpi { struct deduce_mpi {
using result = typename implicit_conversions<typename T::result_type>::type; using result = typename implicit_conversions<typename T::result_type>::type;
......
...@@ -74,6 +74,7 @@ class proxy_registry; ...@@ -74,6 +74,7 @@ class proxy_registry;
class continue_helper; class continue_helper;
class mailbox_element; class mailbox_element;
class message_handler; class message_handler;
class sync_timeout_msg;
class response_promise; class response_promise;
class event_based_actor; class event_based_actor;
class type_erased_tuple; class type_erased_tuple;
......
...@@ -114,9 +114,19 @@ public: ...@@ -114,9 +114,19 @@ public:
using mailbox_type = detail::single_reader_queue<mailbox_element, using mailbox_type = detail::single_reader_queue<mailbox_element,
detail::disposer>; detail::disposer>;
using unexpected_handler /// Function object for handling unmatched messages.
= std::function<result<message>(local_actor* self, using default_handler
const type_erased_tuple*)>; = std::function<result<message> (local_actor* self,
const type_erased_tuple*)>;
/// Function object for handling error messages.
using error_handler = std::function<void (local_actor* self, error&)>;
/// Function object for handling down messages.
using down_handler = std::function<void (local_actor* self, down_msg&)>;
/// Function object for handling exit messages.
using exit_handler = std::function<void (local_actor* self, exit_msg&)>;
// -- constructors, destructors, and assignment operators -------------------- // -- constructors, destructors, and assignment operators --------------------
...@@ -246,12 +256,12 @@ public: ...@@ -246,12 +256,12 @@ public:
} }
/// Sends an exit message to `dest`. /// Sends an exit message to `dest`.
void send_exit(const actor_addr& dest, exit_reason reason); void send_exit(const actor_addr& dest, error reason);
/// Sends an exit message to `dest`. /// Sends an exit message to `dest`.
template <class ActorHandle> template <class ActorHandle>
void send_exit(const ActorHandle& dest, exit_reason reason) { void send_exit(const ActorHandle& dest, error reason) {
send_exit(dest.address(), reason); send_exit(dest.address(), std::move(reason));
} }
/// Sends a message to `dest` that is delayed by `rel_time` /// Sends a message to `dest` that is delayed by `rel_time`
...@@ -311,8 +321,41 @@ public: ...@@ -311,8 +321,41 @@ public:
// -- miscellaneous actor operations ----------------------------------------- // -- miscellaneous actor operations -----------------------------------------
/// Sets a custom handler for unexpected messages. /// Sets a custom handler for unexpected messages.
inline void set_default_handler(unexpected_handler fun) { inline void set_default_handler(default_handler fun) {
unexpected_handler_ = std::move(fun); default_handler_ = std::move(fun);
}
/// Sets a custom handler for error messages.
inline void set_error_handler(error_handler fun) {
error_handler_ = std::move(fun);
}
/// Sets a custom handler for error messages.
template <class T>
auto set_error_handler(T fun) -> decltype(fun(std::declval<error&>())) {
set_error_handler([fun](local_actor*, error& x) { fun(x); });
}
/// Sets a custom handler for down messages.
inline void set_down_handler(down_handler fun) {
down_handler_ = std::move(fun);
}
/// Sets a custom handler for down messages.
template <class T>
auto set_down_handler(T fun) -> decltype(fun(std::declval<down_msg&>())) {
set_down_handler([fun](local_actor*, down_msg& x) { fun(x); });
}
/// Sets a custom handler for error messages.
inline void set_exit_handler(exit_handler fun) {
exit_handler_ = std::move(fun);
}
/// Sets a custom handler for exit messages.
template <class T>
auto set_exit_handler(T fun) -> decltype(fun(std::declval<exit_msg&>())) {
set_exit_handler([fun](local_actor*, exit_msg& x) { fun(x); });
} }
/// Returns the execution unit currently used by this actor. /// Returns the execution unit currently used by this actor.
...@@ -357,26 +400,6 @@ public: ...@@ -357,26 +400,6 @@ public:
/// blocking API calls such as {@link receive()}. /// blocking API calls such as {@link receive()}.
void quit(error reason = error{}); void quit(error reason = error{});
/// Checks whether this actor traps exit messages.
inline bool trap_exit() const {
return get_flag(trap_exit_flag);
}
/// Enables or disables trapping of exit messages.
inline void trap_exit(bool value) {
set_flag(value, trap_exit_flag);
}
/// Checks whether this actor traps error messages.
inline bool trap_error() const {
return get_flag(trap_error_flag);
}
/// Enables or disables trapping of error messages.
inline void trap_error(bool value) {
set_flag(value, trap_error_flag);
}
/// @cond PRIVATE /// @cond PRIVATE
void monitor(abstract_actor* whom); void monitor(abstract_actor* whom);
...@@ -635,8 +658,6 @@ public: ...@@ -635,8 +658,6 @@ public:
behavior& fun, behavior& fun,
message_id awaited_response); message_id awaited_response);
using error_handler = std::function<void (error&)>;
using pending_response = std::pair<const message_id, behavior>; using pending_response = std::pair<const message_id, behavior>;
message_id new_request_id(message_priority mp); message_id new_request_id(message_priority mp);
...@@ -684,9 +705,6 @@ public: ...@@ -684,9 +705,6 @@ public:
void do_become(behavior bhvr, bool discard_old); void do_become(behavior bhvr, bool discard_old);
protected: protected:
// used only in thread-mapped actors
void await_data();
// used by both event-based and blocking actors // used by both event-based and blocking actors
mailbox_type mailbox_; mailbox_type mailbox_;
...@@ -719,7 +737,16 @@ protected: ...@@ -719,7 +737,16 @@ protected:
std::set<group> subscriptions_; std::set<group> subscriptions_;
// used for setting custom functions for handling unexpected messages // used for setting custom functions for handling unexpected messages
unexpected_handler unexpected_handler_; default_handler default_handler_;
// used for setting custom error handlers
error_handler error_handler_;
// used for setting custom down message handlers
down_handler down_handler_;
// used for setting custom exit message handlers
exit_handler exit_handler_;
/// @endcond /// @endcond
...@@ -741,6 +768,22 @@ private: ...@@ -741,6 +768,22 @@ private:
void delayed_send_impl(message_id mid, strong_actor_ptr whom, void delayed_send_impl(message_id mid, strong_actor_ptr whom,
const duration& rtime, message data); const duration& rtime, message data);
enum class msg_type {
expired_timeout, // an 'old & obsolete' timeout
timeout, // triggers currently active timeout
ordinary, // an asynchronous message or sync. request
response, // a response
sys_message // a system message, e.g., exit_msg or down_msg
};
msg_type filter_msg(mailbox_element& node);
void handle_response(mailbox_element_ptr&, local_actor::pending_response&);
class private_thread;
private_thread* private_thread_;
}; };
/// A smart pointer to a {@link local_actor} instance. /// A smart pointer to a {@link local_actor} instance.
......
...@@ -158,6 +158,14 @@ public: ...@@ -158,6 +158,14 @@ public:
param_decay param_decay
>::type; >::type;
static_assert(! std::is_same<pattern, detail::type_list<exit_msg>>::value,
"exit_msg not allowed in message handlers, "
"did you mean to use set_exit_handler()?");
static_assert(! std::is_same<pattern, detail::type_list<down_msg>>::value,
"down_msg not allowed in message handlers, "
"did you mean to use set_down_handler()?");
using intermediate_tuple = using intermediate_tuple =
typename detail::tl_apply< typename detail::tl_apply<
typename detail::tl_map< typename detail::tl_map<
......
...@@ -34,9 +34,8 @@ ...@@ -34,9 +34,8 @@
namespace caf { namespace caf {
/// Sent to all links when an actor is terminated. /// Sent to all links when an actor is terminated.
/// @note This message can be handled manually by calling /// @note Actors can override the default handler by calling
/// `local_actor::trap_exit(true)` and is otherwise handled /// `self->set_exit_handler(...)`.
/// implicitly by the runtime system.
struct exit_msg { struct exit_msg {
/// The source of this message, i.e., the terminated actor. /// The source of this message, i.e., the terminated actor.
actor_addr source; actor_addr source;
...@@ -48,6 +47,12 @@ inline std::string to_string(const exit_msg& x) { ...@@ -48,6 +47,12 @@ inline std::string to_string(const exit_msg& x) {
return "exit" + deep_to_string(std::tie(x.source, x.reason)); return "exit" + deep_to_string(std::tie(x.source, x.reason));
} }
template <class Processor>
void serialize(Processor& proc, exit_msg& x, const unsigned int) {
proc & x.source;
proc & x.reason;
}
/// Sent to all actors monitoring an actor when it is terminated. /// Sent to all actors monitoring an actor when it is terminated.
struct down_msg { struct down_msg {
/// The source of this message, i.e., the terminated actor. /// The source of this message, i.e., the terminated actor.
...@@ -60,41 +65,8 @@ inline std::string to_string(const down_msg& x) { ...@@ -60,41 +65,8 @@ inline std::string to_string(const down_msg& x) {
return "down" + deep_to_string(std::tie(x.source, x.reason)); return "down" + deep_to_string(std::tie(x.source, x.reason));
} }
template <class T> template <class Processor>
typename std::enable_if< void serialize(Processor& proc, down_msg& x, const unsigned int) {
detail::tl_exists<
detail::type_list<exit_msg, down_msg>,
detail::tbind<std::is_same, T>::template type
>::value,
bool
>::type
operator==(const T& lhs, const T& rhs) {
return lhs.source == rhs.source && lhs.reason == rhs.reason;
}
template <class T>
typename std::enable_if<
detail::tl_exists<
detail::type_list<exit_msg, down_msg>,
detail::tbind<std::is_same, T>::template type
>::value,
bool
>::type
operator!=(const T& lhs, const T& rhs) {
return !(lhs == rhs);
}
template <class Processor, class T>
typename std::enable_if<
detail::tl_exists<
detail::type_list<exit_msg, down_msg>,
detail::tbind<
std::is_same,
typename std::remove_const<T>::type
>::template type
>::value
>::type
serialize(Processor& proc, T& x, const unsigned int) {
proc & x.source; proc & x.source;
proc & x.reason; proc & x.reason;
} }
...@@ -109,16 +81,6 @@ inline std::string to_string(const group_down_msg& x) { ...@@ -109,16 +81,6 @@ inline std::string to_string(const group_down_msg& x) {
return "group_down" + deep_to_string(std::tie(x.source)); return "group_down" + deep_to_string(std::tie(x.source));
} }
/// @relates group_down_msg
inline bool operator==(const group_down_msg& lhs, const group_down_msg& rhs) {
return lhs.source == rhs.source;
}
/// @relates group_down_msg
inline bool operator!=(const group_down_msg& lhs, const group_down_msg& rhs) {
return !(lhs == rhs);
}
/// @relates group_down_msg /// @relates group_down_msg
template <class Processor> template <class Processor>
void serialize(Processor& proc, group_down_msg& x, const unsigned int) { void serialize(Processor& proc, group_down_msg& x, const unsigned int) {
...@@ -128,20 +90,16 @@ void serialize(Processor& proc, group_down_msg& x, const unsigned int) { ...@@ -128,20 +90,16 @@ void serialize(Processor& proc, group_down_msg& x, const unsigned int) {
/// Sent whenever a timeout occurs during a synchronous send. /// Sent whenever a timeout occurs during a synchronous send.
/// This system message does not have any fields, because the message ID /// This system message does not have any fields, because the message ID
/// sent alongside this message identifies the matching request that timed out. /// sent alongside this message identifies the matching request that timed out.
struct sync_timeout_msg { }; class sync_timeout_msg { };
inline std::string to_string(const sync_timeout_msg&) { inline std::string to_string(const sync_timeout_msg&) {
return "sync_timeout"; return "sync_timeout";
} }
/// @relates sync_timeout_msg /// @relates group_down_msg
inline bool operator==(const sync_timeout_msg&, const sync_timeout_msg&) { template <class Processor>
return true; void serialize(Processor&, sync_timeout_msg&, const unsigned int) {
} // nop
/// @relates sync_timeout_msg
inline bool operator!=(const sync_timeout_msg&, const sync_timeout_msg&) {
return false;
} }
/// Signalizes a timeout event. /// Signalizes a timeout event.
...@@ -155,16 +113,6 @@ inline std::string to_string(const timeout_msg& x) { ...@@ -155,16 +113,6 @@ inline std::string to_string(const timeout_msg& x) {
return "timeout" + deep_to_string(std::tie(x.timeout_id)); return "timeout" + deep_to_string(std::tie(x.timeout_id));
} }
/// @relates timeout_msg
inline bool operator==(const timeout_msg& lhs, const timeout_msg& rhs) {
return lhs.timeout_id == rhs.timeout_id;
}
/// @relates timeout_msg
inline bool operator!=(const timeout_msg& lhs, const timeout_msg& rhs) {
return !(lhs == rhs);
}
/// @relates timeout_msg /// @relates timeout_msg
template <class Processor> template <class Processor>
void serialize(Processor& proc, timeout_msg& x, const unsigned int) { void serialize(Processor& proc, timeout_msg& x, const unsigned int) {
......
...@@ -209,11 +209,7 @@ private: ...@@ -209,11 +209,7 @@ private:
template <class... Ts> template <class... Ts>
void set(intrusive_ptr<detail::default_behavior_impl<std::tuple<Ts...>>> bp) { void set(intrusive_ptr<detail::default_behavior_impl<std::tuple<Ts...>>> bp) {
using mpi = using mpi = detail::type_list<typename detail::deduce_mpi<Ts>::type...>;
typename detail::tl_filter_not<
detail::type_list<typename detail::deduce_mpi<Ts>::type...>,
detail::is_hidden_msg_handler
>::type;
static_assert(detail::tl_is_distinct<mpi>::value, static_assert(detail::tl_is_distinct<mpi>::value,
"multiple handler defintions found"); "multiple handler defintions found");
detail::static_asserter<signatures, mpi, detail::ctm>::verify_match(); detail::static_asserter<signatures, mpi, detail::ctm>::verify_match();
......
...@@ -78,30 +78,25 @@ public: ...@@ -78,30 +78,25 @@ public:
// nop // nop
} }
inline mailbox_element_ptr dequeue() {
blocking_actor::await_data();
return next_message();
}
bool await_data(const hrc::time_point& tp) { bool await_data(const hrc::time_point& tp) {
if (has_next_message()) { if (has_next_message())
return true; return true;
}
return mailbox().synchronized_await(mtx_, cv_, tp); return mailbox().synchronized_await(mtx_, cv_, tp);
} }
mailbox_element_ptr try_dequeue() {
blocking_actor::await_data();
return next_message();
}
mailbox_element_ptr try_dequeue(const hrc::time_point& tp) { mailbox_element_ptr try_dequeue(const hrc::time_point& tp) {
if (await_data(tp)) { if (await_data(tp))
return next_message(); return next_message();
}
return mailbox_element_ptr{}; return mailbox_element_ptr{};
} }
void act() override { void act() override {
trap_exit(true);
// setup & local variables // setup & local variables
bool received_exit = false;
mailbox_element_ptr msg_ptr;
std::multimap<hrc::time_point, delayed_msg> messages; std::multimap<hrc::time_point, delayed_msg> messages;
// message handling rules // message handling rules
message_handler mfun{ message_handler mfun{
...@@ -109,17 +104,14 @@ public: ...@@ -109,17 +104,14 @@ public:
message_id mid, message& msg) { message_id mid, message& msg) {
insert_dmsg(messages, d, std::move(from), insert_dmsg(messages, d, std::move(from),
std::move(to), mid, std::move(msg)); std::move(to), mid, std::move(msg));
},
[&](const exit_msg&) {
received_exit = true;
} }
}; };
// loop mailbox_element_ptr msg_ptr;
while (! received_exit) { for (;;) {
while (! msg_ptr) { while (! msg_ptr) {
if (messages.empty()) if (messages.empty()) {
msg_ptr = dequeue(); msg_ptr = try_dequeue();
else { } else {
auto tout = hrc::now(); auto tout = hrc::now();
// handle timeouts (send messages) // handle timeouts (send messages)
auto it = messages.begin(); auto it = messages.begin();
...@@ -128,11 +120,15 @@ public: ...@@ -128,11 +120,15 @@ public:
it = messages.erase(it); it = messages.erase(it);
} }
// wait for next message or next timeout // wait for next message or next timeout
if (it != messages.end()) { if (it != messages.end())
msg_ptr = try_dequeue(it->first); msg_ptr = try_dequeue(it->first);
}
} }
} }
if (msg_ptr->msg.match_element<exit_msg>(0)) {
auto& em = msg_ptr->msg.get_as<exit_msg>(0);
if (em.reason)
quit(em.reason);
}
mfun(msg_ptr->msg); mfun(msg_ptr->msg);
msg_ptr.reset(); msg_ptr.reset();
} }
...@@ -268,9 +264,14 @@ void printer_loop(blocking_actor* self) { ...@@ -268,9 +264,14 @@ void printer_loop(blocking_actor* self) {
std::cout << line << std::flush; std::cout << line << std::flush;
line.clear(); line.clear();
}; };
self->trap_exit(true); /*
bool running = true; bool running = true;
self->set_exit_handler([&](exit_msg&) {
running = false;
});
self->receive_while([&] { return running; })( self->receive_while([&] { return running; })(
*/
self->receive_loop(
[&](add_atom, actor_id aid, std::string& str) { [&](add_atom, actor_id aid, std::string& str) {
if (str.empty() || aid == invalid_actor_id) if (str.empty() || aid == invalid_actor_id)
return; return;
...@@ -297,9 +298,6 @@ void printer_loop(blocking_actor* self) { ...@@ -297,9 +298,6 @@ void printer_loop(blocking_actor* self) {
auto d = get_data(aid, true); auto d = get_data(aid, true);
if (d) if (d)
d->redirect = get_sink_handle(self->system(), fcache, fn, flag); d->redirect = get_sink_handle(self->system(), fcache, fn, flag);
},
[&](const exit_msg&) {
running = false;
} }
); );
} }
...@@ -333,16 +331,9 @@ void* abstract_coordinator::subtype_ptr() { ...@@ -333,16 +331,9 @@ void* abstract_coordinator::subtype_ptr() {
void abstract_coordinator::stop_actors() { void abstract_coordinator::stop_actors() {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
scoped_actor self{system_, true}; scoped_actor self{system_, true};
self->monitor(timer_);
self->monitor(printer_);
anon_send_exit(timer_, exit_reason::user_shutdown); anon_send_exit(timer_, exit_reason::user_shutdown);
anon_send_exit(printer_, exit_reason::user_shutdown); anon_send_exit(printer_, exit_reason::user_shutdown);
int i = 0; self->wait_for(timer_, printer_);
self->receive_for(i, 2)(
[](const down_msg&) {
// nop
}
);
} }
abstract_coordinator::abstract_coordinator(actor_system& sys) abstract_coordinator::abstract_coordinator(actor_system& sys)
......
...@@ -182,6 +182,10 @@ void actor_registry::start() { ...@@ -182,6 +182,10 @@ void actor_registry::start() {
self->state.data[key].second.erase(subscriber); self->state.data[key].second.erase(subscriber);
subscribers.erase(i); subscribers.erase(i);
}; };
self->set_down_handler([=](down_msg& dm) {
CAF_LOG_TRACE(CAF_ARG(dm));
unsubscribe_all(actor_cast<actor>(dm.source));
});
return { return {
[=](put_atom, const std::string& key, message& msg) { [=](put_atom, const std::string& key, message& msg) {
CAF_LOG_TRACE(CAF_ARG(key) << CAF_ARG(msg)); CAF_LOG_TRACE(CAF_ARG(key) << CAF_ARG(msg));
...@@ -237,10 +241,6 @@ void actor_registry::start() { ...@@ -237,10 +241,6 @@ void actor_registry::start() {
} }
self->state.subscribers[subscriber].erase(key); self->state.subscribers[subscriber].erase(key);
self->state.data[key].second.erase(subscriber); self->state.data[key].second.erase(subscriber);
},
[=](const down_msg& dm) {
CAF_LOG_TRACE(CAF_ARG(dm));
unsubscribe_all(actor_cast<actor>(dm.source));
} }
}; };
}; };
......
...@@ -17,13 +17,15 @@ ...@@ -17,13 +17,15 @@
* http://www.boost.org/LICENSE_1_0.txt. * * http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/ ******************************************************************************/
#include "caf/exception.hpp"
#include "caf/actor_system.hpp"
#include "caf/blocking_actor.hpp" #include "caf/blocking_actor.hpp"
#include "caf/logger.hpp" #include "caf/logger.hpp"
#include "caf/exception.hpp"
#include "caf/actor_system.hpp"
#include "caf/actor_registry.hpp" #include "caf/actor_registry.hpp"
#include "caf/detail/sync_request_bouncer.hpp"
namespace caf { namespace caf {
blocking_actor::blocking_actor(actor_config& sys) blocking_actor::blocking_actor(actor_config& sys)
...@@ -35,6 +37,18 @@ blocking_actor::~blocking_actor() { ...@@ -35,6 +37,18 @@ blocking_actor::~blocking_actor() {
// avoid weak-vtables warning // avoid weak-vtables warning
} }
void blocking_actor::enqueue(mailbox_element_ptr ptr, execution_unit*) {
auto mid = ptr->mid;
auto sender = ptr->sender;
// returns false if mailbox has been closed
if (! mailbox().synchronized_enqueue(mtx_, cv_, ptr.release())) {
if (mid.is_request()) {
detail::sync_request_bouncer srb{exit_reason()};
srb(sender, mid);
}
}
}
void blocking_actor::await_all_other_actors_done() { void blocking_actor::await_all_other_actors_done() {
system().registry().await_running_count_equal(is_registered() ? 1 : 0); system().registry().await_running_count_equal(is_registered() ? 1 : 0);
} }
...@@ -51,16 +65,16 @@ void blocking_actor::initialize() { ...@@ -51,16 +65,16 @@ 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, 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
uint32_t timeout_id = 0; uint32_t timeout_id = 0;
if (mid == invalid_message_id) if (mid != invalid_message_id)
awaited_responses_.emplace_front(mid, bhvr);
else
timeout_id = request_timeout(bhvr.timeout()); timeout_id = request_timeout(bhvr.timeout());
if (mid != invalid_message_id && ! find_awaited_response(mid))
awaited_responses_.emplace_front(mid, behavior{});
// read incoming messages // read incoming messages
for (;;) { for (;;) {
await_data(); await_data();
...@@ -81,4 +95,28 @@ void blocking_actor::dequeue(behavior& bhvr, message_id mid) { ...@@ -81,4 +95,28 @@ void blocking_actor::dequeue(behavior& bhvr, message_id mid) {
} }
} }
void blocking_actor::await_data() {
if (! has_next_message())
mailbox().synchronized_await(mtx_, cv_);
}
size_t blocking_actor::attach_functor(const actor& x) {
return attach_functor(actor_cast<strong_actor_ptr>(x));
}
size_t blocking_actor::attach_functor(const actor_addr& x) {
return attach_functor(actor_cast<strong_actor_ptr>(x));
}
size_t blocking_actor::attach_functor(const strong_actor_ptr& ptr) {
using wait_for_atom = atom_constant<atom("waitFor")>;
if (! ptr)
return 0;
auto self = actor_cast<actor>(this);
ptr->get()->attach_functor([=](const error&) {
anon_send(self, wait_for_atom::value);
});
return 1;
}
} // namespace caf } // namespace caf
...@@ -48,23 +48,14 @@ class local_group_module; ...@@ -48,23 +48,14 @@ class local_group_module;
void await_all_locals_down(actor_system& sys, std::initializer_list<actor> xs) { void await_all_locals_down(actor_system& sys, std::initializer_list<actor> xs) {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
size_t awaited_down_msgs = 0;
scoped_actor self{sys, true}; scoped_actor self{sys, true};
std::vector<actor> ys;
for (auto& x : xs) for (auto& x : xs)
if (x != invalid_actor && x.node() == sys.node()) { if (x.node() == sys.node()) {
self->monitor(x);
self->send_exit(x, exit_reason::kill); self->send_exit(x, exit_reason::kill);
++awaited_down_msgs; ys.push_back(x);
} }
size_t i = 0; self->wait_for(ys);
self->receive_for(i, awaited_down_msgs) (
[](const down_msg&) {
// nop
},
after(std::chrono::seconds(1)) >> [&] {
throw std::logic_error("at least one actor did not quit within 1s");
}
);
} }
class local_group : public abstract_group { class local_group : public abstract_group {
...@@ -170,6 +161,18 @@ public: ...@@ -170,6 +161,18 @@ public:
return message{}; return message{};
}; };
set_default_handler(fwd); set_default_handler(fwd);
set_down_handler([=](down_msg& dm) {
CAF_LOG_TRACE(CAF_ARG(dm));
if (dm.source) {
auto first = acquaintances_.begin();
auto last = acquaintances_.end();
auto i = std::find_if(first, last, [&](const actor& a) {
return a == dm.source;
});
if (i != last)
acquaintances_.erase(i);
}
});
// return behavior // return behavior
return { return {
[=](join_atom, const actor& other) { [=](join_atom, const actor& other) {
...@@ -190,18 +193,6 @@ public: ...@@ -190,18 +193,6 @@ public:
group_->send_all_subscribers(current_element_->sender, what, context()); group_->send_all_subscribers(current_element_->sender, what, context());
// forward to all acquaintances // forward to all acquaintances
send_to_acquaintances(what); send_to_acquaintances(what);
},
[=](const down_msg& dm) {
CAF_LOG_TRACE(CAF_ARG(dm));
if (dm.source) {
auto first = acquaintances_.begin();
auto last = acquaintances_.end();
auto i = std::find_if(first, last, [&](const actor& a) {
return a == dm.source;
});
if (i != last)
acquaintances_.erase(i);
}
} }
}; };
} }
...@@ -305,13 +296,16 @@ private: ...@@ -305,13 +296,16 @@ private:
local_group_proxy* grp) { local_group_proxy* grp) {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
self->monitor(grp->broker_); self->monitor(grp->broker_);
self->set_down_handler([=](down_msg& down) {
CAF_LOG_TRACE(CAF_ARG(down));
auto msg = make_message(group_down_msg{group(grp)});
grp->send_all_subscribers(self->ctrl(), std::move(msg),
self->context());
self->quit(down.reason);
});
return { return {
[=](const down_msg& down) { [] {
CAF_LOG_TRACE(CAF_ARG(down)); // nop
auto msg = make_message(group_down_msg{group(grp)});
grp->send_all_subscribers(self->ctrl(), std::move(msg),
self->context());
self->quit(down.reason);
} }
}; };
} }
...@@ -332,7 +326,7 @@ behavior proxy_broker::make_behavior() { ...@@ -332,7 +326,7 @@ behavior proxy_broker::make_behavior() {
set_default_handler(fwd); set_default_handler(fwd);
// return dummy behavior // return dummy behavior
return { return {
[](const down_msg&) { [] {
// nop // nop
} }
}; };
......
...@@ -18,6 +18,7 @@ ...@@ -18,6 +18,7 @@
******************************************************************************/ ******************************************************************************/
#include <string> #include <string>
#include <condition_variable>
#include "caf/sec.hpp" #include "caf/sec.hpp"
#include "caf/atom.hpp" #include "caf/atom.hpp"
...@@ -39,6 +40,70 @@ ...@@ -39,6 +40,70 @@
namespace caf { namespace caf {
class local_actor::private_thread {
public:
private_thread(local_actor* self) : self_(self) {
intrusive_ptr_add_ref(self->ctrl());
scheduler::inc_detached_threads();
}
void run() {
CAF_SET_LOGGER_SYS(&self_->system());
CAF_PUSH_AID(self_->id());
CAF_LOG_TRACE("");
auto job = const_cast<local_actor*>(self_);
scoped_execution_unit ctx{&job->system()};
auto max_throughput = std::numeric_limits<size_t>::max();
for (;;) {
bool resume_later;
do {
resume_later = false;
switch (job->resume(&ctx, max_throughput)) {
case resumable::resume_later:
resume_later = true;
break;
case resumable::done:
intrusive_ptr_release(job->ctrl());
return;
case resumable::awaiting_message:
intrusive_ptr_release(job->ctrl());
break;
case resumable::shutdown_execution_unit:
return;
}
} while (resume_later);
// wait until actor becomes ready again or was destroyed
std::unique_lock<std::mutex> guard(mtx_);
cv_.wait(guard);
job = const_cast<local_actor*>(self_);
if (! job)
return;
}
}
void resume() {
std::unique_lock<std::mutex> guard(mtx_);
cv_.notify_one();
}
void shutdown() {
std::unique_lock<std::mutex> guard(mtx_);
self_ = nullptr;
cv_.notify_one();
}
static void exec(private_thread* this_ptr) {
this_ptr->run();
delete this_ptr;
scheduler::dec_detached_threads();
}
private:
std::mutex mtx_;
std::condition_variable cv_;
volatile local_actor* self_;
};
result<message> reflect(local_actor*, const type_erased_tuple* x) { result<message> reflect(local_actor*, const type_erased_tuple* x) {
return message::from(x); return message::from(x);
} }
...@@ -60,6 +125,21 @@ result<message> drop(local_actor*, const type_erased_tuple*) { ...@@ -60,6 +125,21 @@ result<message> drop(local_actor*, const type_erased_tuple*) {
return sec::unexpected_message; return sec::unexpected_message;
} }
void default_error_handler(local_actor* ptr, error& x) {
ptr->quit(std::move(x));
}
void default_down_handler(local_actor* ptr, down_msg& x) {
aout(ptr) << "*** unhandled down message [id: " << ptr->id()
<< ", name: " << ptr->name() << "]: " << to_string(x)
<< std::endl;
}
void default_exit_handler(local_actor* ptr, exit_msg& x) {
if (x.reason)
ptr->quit(std::move(x.reason));
}
// local actors are created with a reference count of one that is adjusted // local actors are created with a reference count of one that is adjusted
// later on in spawn(); this prevents subtle bugs that lead to segfaults, // later on in spawn(); this prevents subtle bugs that lead to segfaults,
// e.g., when calling address() in the ctor of a derived class // e.g., when calling address() in the ctor of a derived class
...@@ -68,7 +148,11 @@ local_actor::local_actor(actor_config& cfg) ...@@ -68,7 +148,11 @@ local_actor::local_actor(actor_config& cfg)
context_(cfg.host), context_(cfg.host),
timeout_id_(0), timeout_id_(0),
initial_behavior_fac_(std::move(cfg.init_fun)), initial_behavior_fac_(std::move(cfg.init_fun)),
unexpected_handler_(print_and_drop) { default_handler_(print_and_drop),
error_handler_(default_error_handler),
down_handler_(default_down_handler),
exit_handler_(default_exit_handler),
private_thread_(nullptr) {
if (cfg.groups != nullptr) if (cfg.groups != nullptr)
for (auto& grp : *cfg.groups) for (auto& grp : *cfg.groups)
join(grp); join(grp);
...@@ -189,140 +273,65 @@ uint32_t local_actor::active_timeout_id() const { ...@@ -189,140 +273,65 @@ uint32_t local_actor::active_timeout_id() const {
return timeout_id_; return timeout_id_;
} }
enum class msg_type { local_actor::msg_type local_actor::filter_msg(mailbox_element& node) {
normal_exit, // an exit message with normal exit reason
non_normal_exit, // an exit message with abnormal exit reason
expired_timeout, // an 'old & obsolete' timeout
timeout, // triggers currently active timeout
ordinary, // an asynchronous message or sync. request
response, // a response
sys_message // a system message, e.g., signalizing migration
};
msg_type filter_msg(local_actor* self, mailbox_element& node) {
message& msg = node.msg; message& msg = node.msg;
auto mid = node.mid; auto mid = node.mid;
if (mid.is_response()) if (mid.is_response())
return msg_type::response; return msg_type::response;
// intercept system messages, e.g., signalizing migration switch (msg.type_token()) {
if (msg.size() > 1 && msg.match_element<sys_atom>(0) && node.sender) { case detail::make_type_token<atom_value, atom_value, std::string>():
bool mismatch = true; if (msg.get_as<atom_value>(0) == sys_atom::value
msg.apply({ && msg.get_as<atom_value>(1) == get_atom::value) {
/* auto& what = msg.get_as<std::string>(2);
[&](sys_atom, migrate_atom, const actor& mm) {
// migrate this actor to `target`
if (! self->is_serializable()) {
node.sender->enqueue(
mailbox_element::make_joint(self->address(), node.mid.response_id(),
{}, sec::state_not_serializable),
self->context());
return;
}
std::vector<char> buf;
binary_serializer bs{std::back_inserter(buf), self->context()};
self->save_state(bs, 0);
auto sender = node.sender;
// request(...)
auto req = self->request_impl(message_priority::normal, mm,
migrate_atom::value, self->name(),
std::move(buf));
self->set_awaited_response_handler(req, behavior{
[=](ok_atom, const actor_addr& dest) {
// respond to original message with {'OK', dest}
sender->enqueue(mailbox_element::make_joint(self->address(),
mid.response_id(), {},
ok_atom::value, dest),
self->context());
// "decay" into a proxy for `dest`
auto dest_hdl = actor_cast<actor>(dest);
self->do_become(behavior{
others >> [=] {
self->forward_current_message(dest_hdl);
}
}, false);
self->is_migrated_from(true);
},
[=](error& err) {
// respond to original message with {'ERROR', errmsg}
sender->enqueue(mailbox_element::make_joint(self->address(),
mid.response_id(), {},
std::move(err)),
self->context());
}
});
},
[&](sys_atom, migrate_atom, std::vector<char>& buf) {
// "replace" this actor with the content of `buf`
if (! self->is_serializable()) {
node.sender->enqueue(mailbox_element::make_joint(
self->address(), node.mid.response_id(), {},
sec::state_not_serializable),
self->context());
return;
}
if (self->is_migrated_from()) {
// undo the `do_become` we did when migrating away from this object
self->bhvr_stack().pop_back();
self->is_migrated_from(false);
}
binary_deserializer bd{buf.data(), buf.size(), self->context()};
self->load_state(bd, 0);
node.sender->enqueue(
mailbox_element::make_joint(self->address(), node.mid.response_id(),
{}, ok_atom::value, self->address()),
self->context());
},
*/
[&](sys_atom, get_atom, std::string& what) {
CAF_LOG_TRACE(CAF_ARG(what));
mismatch = false;
if (what == "info") { if (what == "info") {
CAF_LOG_DEBUG("reply to 'info' message"); CAF_LOG_DEBUG("reply to 'info' message");
node.sender->enqueue( node.sender->enqueue(
mailbox_element::make_joint(self->ctrl(), mailbox_element::make_joint(ctrl(),
node.mid.response_id(), node.mid.response_id(),
{}, ok_atom::value, std::move(what), {}, ok_atom::value, std::move(what),
self->address(), self->name()), address(), name()),
self->context()); context());
return; } else {
node.sender->enqueue(
mailbox_element::make_joint(ctrl(),
node.mid.response_id(),
{}, sec::unsupported_sys_key),
context());
} }
node.sender->enqueue( return msg_type::sys_message;
mailbox_element::make_joint(self->ctrl(),
node.mid.response_id(),
{}, sec::unsupported_sys_key),
self->context());
} }
}); return msg_type::ordinary;
return mismatch ? msg_type::ordinary : msg_type::sys_message; case detail::make_type_token<timeout_msg>(): {
} auto& tm = msg.get_as<timeout_msg>(0);
// all other system messages always consist of one element auto tid = tm.timeout_id;
if (msg.size() != 1) CAF_ASSERT(! mid.valid());
return msg_type::ordinary; return is_active_timeout(tid) ? msg_type::timeout
if (msg.match_element<timeout_msg>(0)) { : msg_type::expired_timeout;
auto& tm = msg.get_as<timeout_msg>(0);
auto tid = tm.timeout_id;
CAF_ASSERT(! mid.valid());
return self->is_active_timeout(tid) ? msg_type::timeout
: msg_type::expired_timeout;
}
if (msg.match_element<exit_msg>(0)) {
auto& em = msg.get_as<exit_msg>(0);
CAF_ASSERT(! mid.valid());
// make sure to get rid of attachables if they're no longer needed
self->unlink_from(em.source);
if (em.reason == exit_reason::kill) {
self->quit(em.reason);
return msg_type::non_normal_exit;
} }
if (self->trap_exit() == false) { case detail::make_type_token<exit_msg>(): {
if (em.reason != exit_reason::normal) { auto& em = msg.get_as_mutable<exit_msg>(0);
self->quit(em.reason); // make sure to get rid of attachables if they're no longer needed
return msg_type::non_normal_exit; unlink_from(em.source);
} // exit_reason::kill is always fatal
return msg_type::normal_exit; if (em.reason == exit_reason::kill)
quit(std::move(em.reason));
else
exit_handler_(this, em);
return msg_type::sys_message;
} }
case detail::make_type_token<down_msg>(): {
auto& dm = msg.get_as_mutable<down_msg>(0);
down_handler_(this, dm);
return msg_type::sys_message;
}
case detail::make_type_token<error>(): {
auto& err = msg.get_as_mutable<error>(0);
error_handler_(this, err);
return msg_type::sys_message;
}
default:
return msg_type::ordinary;
} }
return msg_type::ordinary;
} }
class invoke_result_visitor_helper { class invoke_result_visitor_helper {
...@@ -396,27 +405,52 @@ private: ...@@ -396,27 +405,52 @@ private:
local_actor* self_; local_actor* self_;
}; };
void local_actor::handle_response(mailbox_element_ptr& ptr,
local_actor::pending_response& pr) {
CAF_ASSERT(ptr != nullptr);
CAF_LOG_TRACE(CAF_ARG(*ptr) << CAF_ARG(awaited_id));
auto& ref_fun = pr.second;
ptr.swap(current_element_);
auto& msg = current_element_->msg;
auto guard = detail::make_scope_guard([&] {
ptr.swap(current_element_);
});
local_actor_invoke_result_visitor visitor{this};
auto invoke_error = [&](error err) {
auto tmp = make_message(err);
if (ref_fun(visitor, tmp) == match_case::no_match) {
CAF_LOG_WARNING("multiplexed response failure occured:"
<< CAF_ARG(id()));
error_handler_(this, err);
}
};
if (msg.type_token() == detail::make_type_token<sync_timeout_msg>()) {
// TODO: check if condition can ever be true
if (ref_fun.timeout().valid())
ref_fun.handle_timeout();
invoke_error(sec::request_timeout);
} else if (ref_fun(visitor, msg) == match_case::no_match) {
if (msg.type_token() == detail::make_type_token<error>()) {
error_handler_(this, msg.get_as_mutable<error>(0));
} else {
// wrap unhandled message into an error object and try invoke again
invoke_error(make_error(sec::unexpected_response, current_element_->msg));
}
}
}
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) {
CAF_ASSERT(ptr != nullptr); CAF_ASSERT(ptr != nullptr);
CAF_LOG_TRACE(CAF_ARG(*ptr) << CAF_ARG(awaited_id)); CAF_LOG_TRACE(CAF_ARG(*ptr) << CAF_ARG(awaited_id));
local_actor_invoke_result_visitor visitor{this}; switch (filter_msg(*ptr)) {
switch (filter_msg(this, *ptr)) {
case msg_type::normal_exit:
CAF_LOG_DEBUG("dropped normal exit signal");
return im_dropped;
case msg_type::expired_timeout: case msg_type::expired_timeout:
CAF_LOG_DEBUG("dropped expired timeout message"); CAF_LOG_DEBUG("dropped expired timeout message");
return im_dropped; return im_dropped;
case msg_type::sys_message: case msg_type::sys_message:
CAF_LOG_DEBUG("handled system message"); CAF_LOG_DEBUG("handled system message");
return im_dropped; return im_dropped;
case msg_type::non_normal_exit:
CAF_LOG_DEBUG("handled non-normal exit signal");
// this message was handled
// by calling quit(...)
return im_success;
case msg_type::timeout: { case msg_type::timeout: {
if (awaited_id == invalid_message_id) { if (awaited_id == invalid_message_id) {
CAF_LOG_DEBUG("handle timeout message"); CAF_LOG_DEBUG("handle timeout message");
...@@ -430,54 +464,23 @@ invoke_message_result local_actor::invoke_message(mailbox_element_ptr& ptr, ...@@ -430,54 +464,23 @@ invoke_message_result local_actor::invoke_message(mailbox_element_ptr& ptr,
} }
case msg_type::response: { case msg_type::response: {
auto mid = ptr->mid; auto mid = ptr->mid;
auto ref_opt = find_multiplexed_response(mid); auto response_handler = find_multiplexed_response(mid);
if (ref_opt) { if (response_handler) {
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; handle_response(ptr, *response_handler);
bool is_sync_tout = ptr->msg.match_elements<sync_timeout_msg>();
ptr.swap(current_element_);
if (is_sync_tout) {
if (ref_fun.timeout().valid()) {
ref_fun.handle_timeout();
}
} else if (ref_fun(visitor, current_element_->msg) == match_case::no_match) {
// wrap unexpected message into an error object and try invoke again
auto tmp = make_message(make_error(sec::unexpected_response,
current_element_->msg));
if (ref_fun(visitor, tmp) == match_case::no_match) {
CAF_LOG_WARNING("multiplexed response failure occured:"
<< CAF_ARG(id()));
quit(exit_reason::unhandled_request_error);
}
}
ptr.swap(current_element_);
mark_multiplexed_arrived(mid); mark_multiplexed_arrived(mid);
return im_success; return im_success;
} }
CAF_LOG_DEBUG("skipped multiplexed response:" << CAF_ARG(awaited_id)); CAF_LOG_DEBUG("skipped multiplexed response:" << CAF_ARG(awaited_id));
return im_skipped; return im_skipped;
} else if (awaits(mid)) { }
response_handler = find_awaited_response(mid);
if (response_handler) {
if (awaited_id.valid() && mid == awaited_id) { if (awaited_id.valid() && mid == awaited_id) {
bool is_sync_tout = ptr->msg.match_elements<sync_timeout_msg>(); handle_response(ptr, *response_handler);
ptr.swap(current_element_); mark_awaited_arrived(mid);
if (is_sync_tout) {
if (fun.timeout().valid()) {
fun.handle_timeout();
}
} else if (fun(visitor, current_element_->msg) == match_case::no_match) {
// wrap unexpected message into an error object and try invoke again
auto tmp = make_message(make_error(sec::unexpected_response,
current_element_->msg));
if (fun(visitor, tmp) == match_case::no_match) {
CAF_LOG_WARNING("multiplexed response failure occured:"
<< CAF_ARG(id()));
quit(exit_reason::unhandled_request_error);
}
}
ptr.swap(current_element_);
mark_awaited_arrived(awaited_id);
return im_success; return im_success;
} }
return im_skipped; return im_skipped;
...@@ -485,7 +488,8 @@ invoke_message_result local_actor::invoke_message(mailbox_element_ptr& ptr, ...@@ -485,7 +488,8 @@ invoke_message_result local_actor::invoke_message(mailbox_element_ptr& ptr,
CAF_LOG_DEBUG("dropped expired response"); CAF_LOG_DEBUG("dropped expired response");
return im_dropped; return im_dropped;
} }
case msg_type::ordinary: case msg_type::ordinary: {
local_actor_invoke_result_visitor visitor{this};
if (awaited_id.valid()) { if (awaited_id.valid()) {
CAF_LOG_DEBUG("skipped asynchronous message:" << CAF_ARG(awaited_id)); CAF_LOG_DEBUG("skipped asynchronous message:" << CAF_ARG(awaited_id));
return im_skipped; return im_skipped;
...@@ -504,7 +508,7 @@ invoke_message_result local_actor::invoke_message(mailbox_element_ptr& ptr, ...@@ -504,7 +508,7 @@ invoke_message_result local_actor::invoke_message(mailbox_element_ptr& ptr,
case match_case::no_match: { case match_case::no_match: {
if (had_timeout) if (had_timeout)
has_timeout(true); has_timeout(true);
auto sres = unexpected_handler_(this, auto sres = default_handler_(this,
current_element_->msg.cvals().get()); current_element_->msg.cvals().get());
if (sres.flag != rt_skip) if (sres.flag != rt_skip)
visitor.visit(sres); visitor.visit(sres);
...@@ -519,6 +523,7 @@ invoke_message_result local_actor::invoke_message(mailbox_element_ptr& ptr, ...@@ -519,6 +523,7 @@ invoke_message_result local_actor::invoke_message(mailbox_element_ptr& ptr,
return im_skipped; return im_skipped;
} }
return im_success; return im_success;
}
} }
// should be unreachable // should be unreachable
CAF_CRITICAL("invalid message type"); CAF_CRITICAL("invalid message type");
...@@ -622,31 +627,41 @@ void local_actor::launch(execution_unit* eu, bool lazy, bool hide) { ...@@ -622,31 +627,41 @@ void local_actor::launch(execution_unit* eu, bool lazy, bool hide) {
CAF_LOG_TRACE(CAF_ARG(lazy) << CAF_ARG(hide)); CAF_LOG_TRACE(CAF_ARG(lazy) << CAF_ARG(hide));
is_registered(! hide); is_registered(! hide);
if (is_detached()) { if (is_detached()) {
// actor lives in its own thread if (is_blocking()) {
CAF_PUSH_AID(id()); std::thread([](strong_actor_ptr ptr) {
CAF_LOG_TRACE(CAF_ARG(lazy) << CAF_ARG(hide)); // actor lives in its own thread
scheduler::inc_detached_threads(); auto this_ptr = ptr->get();
//intrusive_ptr<local_actor> mself{this}; CAF_ASSERT(dynamic_cast<blocking_actor*>(this_ptr) != 0);
actor_system* sys = &eu->system(); auto self = static_cast<blocking_actor*>(this_ptr);
std::thread{[hide, sys](strong_actor_ptr mself) { error rsn;
CAF_SET_LOGGER_SYS(sys); std::exception_ptr eptr = nullptr;
// this extra scope makes sure that the trace logger is try {
// destructed before dec_detached_threads() is called self->act();
{
CAF_PUSH_AID(mself->id());
CAF_LOG_TRACE("");
scoped_execution_unit ctx{sys};
auto max_throughput = std::numeric_limits<size_t>::max();
auto ptr = static_cast<local_actor*>(actor_cast<abstract_actor*>(mself));
while (ptr->resume(&ctx, max_throughput) != resumable::done) {
// await new data before resuming actor
ptr->await_data();
CAF_ASSERT(ptr->mailbox().blocked() == false);
} }
mself.reset(); catch (actor_exited& e) {
} rsn = e.reason();
scheduler::dec_detached_threads(); }
}, strong_actor_ptr{ctrl()}}.detach(); catch (...) {
rsn = exit_reason::unhandled_exception;
eptr = std::current_exception();
}
if (eptr) {
auto opt_reason = self->handle(eptr);
rsn = opt_reason ? *opt_reason
: exit_reason::unhandled_exception;
}
try {
self->on_exit();
}
catch (...) {
// simply ignore exception
}
self->cleanup(std::move(rsn), self->context());
}, ctrl()).detach();
return;
}
private_thread_ = new private_thread(this);
std::thread(private_thread::exec, private_thread_).detach();
return; return;
} }
CAF_ASSERT(eu != nullptr); CAF_ASSERT(eu != nullptr);
...@@ -663,32 +678,24 @@ void local_actor::launch(execution_unit* eu, bool lazy, bool hide) { ...@@ -663,32 +678,24 @@ void local_actor::launch(execution_unit* eu, bool lazy, bool hide) {
void local_actor::enqueue(mailbox_element_ptr ptr, execution_unit* eu) { void local_actor::enqueue(mailbox_element_ptr ptr, execution_unit* eu) {
CAF_PUSH_AID(id()); CAF_PUSH_AID(id());
CAF_ASSERT(ptr);
CAF_LOG_TRACE(CAF_ARG(*ptr)); CAF_LOG_TRACE(CAF_ARG(*ptr));
if (is_detached()) { CAF_ASSERT(ptr != nullptr);
// actor lives in its own thread CAF_ASSERT(! is_blocking());
auto mid = ptr->mid;
auto sender = ptr->sender;
// returns false if mailbox has been closed
if (! mailbox().synchronized_enqueue(mtx_, cv_, ptr.release())) {
if (mid.is_request()) {
detail::sync_request_bouncer srb{exit_reason()};
srb(sender, mid);
}
}
return;
}
// actor is cooperatively scheduled
auto mid = ptr->mid; auto mid = ptr->mid;
auto sender = ptr->sender; auto sender = ptr->sender;
switch (mailbox().enqueue(ptr.release())) { switch (mailbox().enqueue(ptr.release())) {
case detail::enqueue_result::unblocked_reader: { case detail::enqueue_result::unblocked_reader: {
// add a reference count to this actor and re-schedule it // add a reference count to this actor and re-schedule it
intrusive_ptr_add_ref(ctrl()); intrusive_ptr_add_ref(ctrl());
if (eu) if (is_detached()) {
eu->exec_later(this); CAF_ASSERT(private_thread_ != nullptr);
else private_thread_->resume();
home_system().scheduler().enqueue(this); } else {
if (eu)
eu->exec_later(this);
else
home_system().scheduler().enqueue(this);
}
break; break;
} }
case detail::enqueue_result::queue_closed: { case detail::enqueue_result::queue_closed: {
...@@ -712,39 +719,9 @@ resumable::resume_result local_actor::resume(execution_unit* eu, ...@@ -712,39 +719,9 @@ resumable::resume_result local_actor::resume(execution_unit* eu,
size_t max_throughput) { size_t max_throughput) {
CAF_PUSH_AID(id()); CAF_PUSH_AID(id());
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
CAF_ASSERT(eu); CAF_ASSERT(eu != nullptr);
CAF_ASSERT(! is_blocking());
context(eu); context(eu);
if (is_blocking()) {
// actor lives in its own thread
CAF_ASSERT(dynamic_cast<blocking_actor*>(this) != 0);
auto self = static_cast<blocking_actor*>(this);
error rsn;
std::exception_ptr eptr = nullptr;
try {
self->act();
}
catch (actor_exited& e) {
rsn = e.reason();
}
catch (...) {
rsn = exit_reason::unhandled_exception;
eptr = std::current_exception();
}
if (eptr) {
auto opt_reason = self->handle(eptr);
rsn = opt_reason ? *opt_reason
: exit_reason::unhandled_exception;
}
try {
self->on_exit();
}
catch (...) {
// simply ignore exception
}
// exit reason might have been changed by on_exit()
self->cleanup(std::move(rsn), context());
return resumable::done;
}
if (is_initialized() && (! has_behavior() || is_terminated())) { if (is_initialized() && (! has_behavior() || is_terminated())) {
CAF_LOG_DEBUG_IF(! has_behavior(), CAF_LOG_DEBUG_IF(! has_behavior(),
"resume called on an actor without behavior"); "resume called on an actor without behavior");
...@@ -856,7 +833,13 @@ local_actor::exec_event(mailbox_element_ptr& ptr) { ...@@ -856,7 +833,13 @@ local_actor::exec_event(mailbox_element_ptr& ptr) {
push_to_cache(std::move(ptr)); push_to_cache(std::move(ptr));
break; break;
case im_dropped: case im_dropped:
// destroy msg // system messages are reported as dropped but might
// still terminate the actor
bhvr_stack().cleanup();
if (finished()) {
CAF_LOG_DEBUG("actor exited");
return {resumable::resume_result::done, res};
}
break; break;
} }
return {resumable::resume_result::resume_later, res}; return {resumable::resume_result::resume_later, res};
...@@ -976,13 +959,6 @@ void local_actor::do_become(behavior bhvr, bool discard_old) { ...@@ -976,13 +959,6 @@ void local_actor::do_become(behavior bhvr, bool discard_old) {
bhvr_stack_.push_back(std::move(bhvr)); bhvr_stack_.push_back(std::move(bhvr));
} }
void local_actor::await_data() {
if (has_next_message()) {
return;
}
mailbox().synchronized_await(mtx_, cv_);
}
void local_actor::send_impl(message_id mid, abstract_channel* dest, void local_actor::send_impl(message_id mid, abstract_channel* dest,
message what) const { message what) const {
if (! dest) if (! dest)
...@@ -990,7 +966,7 @@ void local_actor::send_impl(message_id mid, abstract_channel* dest, ...@@ -990,7 +966,7 @@ void local_actor::send_impl(message_id mid, abstract_channel* dest,
dest->enqueue(ctrl(), mid, std::move(what), context()); dest->enqueue(ctrl(), mid, std::move(what), context());
} }
void local_actor::send_exit(const actor_addr& whom, exit_reason reason) { void local_actor::send_exit(const actor_addr& whom, error reason) {
send(message_priority::high, actor_cast<actor>(whom), send(message_priority::high, actor_cast<actor>(whom),
exit_msg{address(), reason}); exit_msg{address(), reason});
} }
...@@ -1026,6 +1002,10 @@ bool local_actor::finished() { ...@@ -1026,6 +1002,10 @@ bool local_actor::finished() {
bool local_actor::cleanup(error&& fail_state, execution_unit* host) { bool local_actor::cleanup(error&& fail_state, execution_unit* host) {
CAF_LOG_TRACE(CAF_ARG(fail_state)); CAF_LOG_TRACE(CAF_ARG(fail_state));
if (is_detached() && ! is_blocking()) {
CAF_ASSERT(private_thread_ != nullptr);
private_thread_->shutdown();
}
current_mailbox_element().reset(); current_mailbox_element().reset();
if (! mailbox_.closed()) { if (! mailbox_.closed()) {
detail::sync_request_bouncer f{fail_state}; detail::sync_request_bouncer f{fail_state};
...@@ -1047,9 +1027,8 @@ void local_actor::quit(error x) { ...@@ -1047,9 +1027,8 @@ void local_actor::quit(error x) {
CAF_LOG_TRACE(CAF_ARG(x)); CAF_LOG_TRACE(CAF_ARG(x));
fail_state_ = std::move(x); fail_state_ = std::move(x);
is_terminated(true); is_terminated(true);
if (is_blocking()) { if (is_blocking())
throw actor_exited(fail_state_); throw actor_exited(fail_state_);
}
} }
} // namespace caf } // namespace caf
...@@ -78,6 +78,7 @@ const char* numbered_type_names[] = { ...@@ -78,6 +78,7 @@ const char* numbered_type_names[] = {
"@strong_actor_ptr", "@strong_actor_ptr",
"@strset", "@strset",
"@strvec", "@strvec",
"@sync_timeout_msg",
"@timeout", "@timeout",
"@u16", "@u16",
"@u16str", "@u16str",
......
...@@ -32,6 +32,8 @@ using std::endl; ...@@ -32,6 +32,8 @@ using std::endl;
namespace { namespace {
using down_atom = atom_constant<atom("down")>;
struct fixture { struct fixture {
actor_system_config cfg; actor_system_config cfg;
...@@ -52,12 +54,8 @@ struct fixture { ...@@ -52,12 +54,8 @@ struct fixture {
CAF_CHECK(ifs.empty()); CAF_CHECK(ifs.empty());
auto aut = actor_cast<actor>(res); auto aut = actor_cast<actor>(res);
CAF_REQUIRE(aut != invalid_actor); CAF_REQUIRE(aut != invalid_actor);
self->monitor(aut); self->wait_for(aut);
self->receive( CAF_MESSAGE("aut done");
[](const down_msg& dm) {
CAF_CHECK_EQUAL(dm.reason, exit_reason::normal);
}
);
} }
}; };
......
...@@ -66,14 +66,20 @@ public: ...@@ -66,14 +66,20 @@ public:
template <class ExitMsgType> template <class ExitMsgType>
behavior tester(event_based_actor* self, const actor& aut) { behavior tester(event_based_actor* self, const actor& aut) {
if (std::is_same<ExitMsgType, exit_msg>::value) { if (std::is_same<ExitMsgType, exit_msg>::value) {
self->trap_exit(true); self->set_exit_handler([self](exit_msg& msg) {
// must be still alive at this point
CAF_CHECK_EQUAL(s_testees.load(), 1);
CAF_CHECK_EQUAL(msg.reason, exit_reason::user_shutdown);
// testee might be still running its cleanup code in
// another worker thread; by waiting some milliseconds, we make sure
// testee had enough time to return control to the scheduler
// which in turn destroys it by dropping the last remaining reference
self->delayed_send(self, std::chrono::milliseconds(30),
check_atom::value);
});
self->link_to(aut); self->link_to(aut);
} else { } else {
self->monitor(aut); self->set_down_handler([self](down_msg& msg) {
}
anon_send_exit(aut, exit_reason::user_shutdown);
return {
[self](const ExitMsgType& msg) {
// must be still alive at this point // must be still alive at this point
CAF_CHECK_EQUAL(s_testees.load(), 1); CAF_CHECK_EQUAL(s_testees.load(), 1);
CAF_CHECK_EQUAL(msg.reason, exit_reason::user_shutdown); CAF_CHECK_EQUAL(msg.reason, exit_reason::user_shutdown);
...@@ -83,7 +89,11 @@ behavior tester(event_based_actor* self, const actor& aut) { ...@@ -83,7 +89,11 @@ behavior tester(event_based_actor* self, const actor& aut) {
// which in turn destroys it by dropping the last remaining reference // which in turn destroys it by dropping the last remaining reference
self->delayed_send(self, std::chrono::milliseconds(30), self->delayed_send(self, std::chrono::milliseconds(30),
check_atom::value); check_atom::value);
}, });
self->monitor(aut);
}
anon_send_exit(aut, exit_reason::user_shutdown);
return {
[self](check_atom) { [self](check_atom) {
// make sure aut's dtor and on_exit() have been called // make sure aut's dtor and on_exit() have been called
CAF_CHECK_EQUAL(s_testees.load(), 0); CAF_CHECK_EQUAL(s_testees.load(), 0);
......
...@@ -108,50 +108,29 @@ CAF_TEST(round_robin_actor_pool) { ...@@ -108,50 +108,29 @@ CAF_TEST(round_robin_actor_pool) {
} }
); );
anon_send_exit(workers.back(), exit_reason::user_shutdown); anon_send_exit(workers.back(), exit_reason::user_shutdown);
self->receive( self->wait_for(workers.back());
[&](const down_msg& dm) { workers.pop_back();
CAF_CHECK_EQUAL(dm.source, workers.back().address()); // poll actor pool up to 10 times or until it removes the failed worker
workers.pop_back(); bool success = false;
// poll actor pool up to 10 times or until it removes the failed worker size_t i = 0;
bool success = false; while (! success && ++i <= 10) {
size_t i = 0; self->request(w, infinite, sys_atom::value, get_atom::value).receive(
while (! success && ++i <= 10) { [&](std::vector<actor>& ws) {
self->request(w, infinite, sys_atom::value, get_atom::value).receive( success = workers.size() == ws.size();
[&](std::vector<actor>& ws) { if (success) {
success = workers.size() == ws.size(); std::sort(ws.begin(), ws.end());
if (success) { CAF_CHECK_EQUAL(workers, ws);
std::sort(ws.begin(), ws.end()); } else {
CAF_CHECK_EQUAL(workers, ws); // wait a bit until polling again
} else { std::this_thread::sleep_for(std::chrono::milliseconds(5));
// wait a bit until polling again }
std::this_thread::sleep_for(std::chrono::milliseconds(5));
}
}
);
}
CAF_REQUIRE(success);
},
after(std::chrono::milliseconds(250)) >> [] {
CAF_ERROR("didn't receive a down message");
}
);
CAF_MESSAGE("about to send exit to workers");
self->send_exit(w, exit_reason::user_shutdown);
for (int i = 0; i < 6; ++i) {
self->receive(
[&](const down_msg& dm) {
auto last = workers.end();
auto src = dm.source;
CAF_CHECK_NOT_EQUAL(src, invalid_actor_addr);
auto pos = std::find(workers.begin(), last, src);
if (pos != last)
workers.erase(pos);
},
after(std::chrono::milliseconds(250)) >> [] {
CAF_ERROR("didn't receive a down message");
} }
); );
} }
CAF_REQUIRE(success);
CAF_MESSAGE("about to send exit to workers");
self->send_exit(w, exit_reason::user_shutdown);
self->wait_for(workers);
} }
CAF_TEST(broadcast_actor_pool) { CAF_TEST(broadcast_actor_pool) {
......
...@@ -43,7 +43,9 @@ struct fixture { ...@@ -43,7 +43,9 @@ struct fixture {
actor testee; actor testee;
fixture() : self(system), mirror(system.spawn(mirror_impl)) { fixture() : self(system), mirror(system.spawn(mirror_impl)) {
// nop self->set_down_handler([](local_actor*, down_msg& dm) {
CAF_CHECK_EQUAL(dm.reason, exit_reason::normal);
});
} }
template <class... Ts> template <class... Ts>
...@@ -52,11 +54,7 @@ struct fixture { ...@@ -52,11 +54,7 @@ struct fixture {
} }
~fixture() { ~fixture() {
self->receive( self->wait_for(testee);
[](const down_msg& dm) {
CAF_CHECK_EQUAL(dm.reason, exit_reason::normal);
}
);
} }
}; };
......
...@@ -40,14 +40,6 @@ behavior testee(event_based_actor* self) { ...@@ -40,14 +40,6 @@ behavior testee(event_based_actor* self) {
} }
struct fixture { struct fixture {
void wait_until_is_terminated() {
self->receive(
[](const down_msg&) {
// nop
}
);
}
template <class Actor> template <class Actor>
static bool exited(const Actor& handle) { static bool exited(const Actor& handle) {
auto ptr = actor_cast<abstract_actor*>(handle); auto ptr = actor_cast<abstract_actor*>(handle);
...@@ -85,7 +77,7 @@ CAF_TEST(lifetime_1) { ...@@ -85,7 +77,7 @@ CAF_TEST(lifetime_1) {
auto dbl = system.spawn(testee); auto dbl = system.spawn(testee);
self->monitor(dbl); self->monitor(dbl);
anon_send_exit(dbl, exit_reason::kill); anon_send_exit(dbl, exit_reason::kill);
wait_until_is_terminated(); self->wait_for(dbl);
auto bound = dbl.bind(1); auto bound = dbl.bind(1);
CAF_CHECK(exited(bound)); CAF_CHECK(exited(bound));
} }
...@@ -96,7 +88,7 @@ CAF_TEST(lifetime_2) { ...@@ -96,7 +88,7 @@ CAF_TEST(lifetime_2) {
auto bound = dbl.bind(1); auto bound = dbl.bind(1);
self->monitor(bound); self->monitor(bound);
anon_send(dbl, message{}); anon_send(dbl, message{});
wait_until_is_terminated(); self->wait_for(dbl);
} }
CAF_TEST(request_response_promise) { CAF_TEST(request_response_promise) {
......
...@@ -134,6 +134,10 @@ bool operator==(const counting_string& x, const char* y) { ...@@ -134,6 +134,10 @@ bool operator==(const counting_string& x, const char* y) {
return x.str() == y; return x.str() == y;
} }
std::string to_string(const counting_string& ref) {
return ref.str();
}
} // namespace <anonymous> } // namespace <anonymous>
namespace std { namespace std {
......
...@@ -53,19 +53,17 @@ CAF_TEST(constructor_attach) { ...@@ -53,19 +53,17 @@ CAF_TEST(constructor_attach) {
class spawner : public event_based_actor { class spawner : public event_based_actor {
public: public:
spawner(actor_config& cfg) : event_based_actor(cfg), downs_(0) { spawner(actor_config& cfg) : event_based_actor(cfg), downs_(0) {
// nop set_down_handler([=](down_msg& msg) {
CAF_CHECK_EQUAL(msg.reason, exit_reason::user_shutdown);
CAF_CHECK_EQUAL(msg.source, testee_.address());
if (++downs_ == 2)
quit(msg.reason);
});
} }
behavior make_behavior() { behavior make_behavior() {
testee_ = spawn<testee, monitored>(this); testee_ = spawn<testee, monitored>(this);
return { return {
[=](const down_msg& msg) {
CAF_CHECK_EQUAL(msg.reason, exit_reason::user_shutdown);
CAF_CHECK_EQUAL(msg.source, testee_.address());
if (++downs_ == 2) {
quit(msg.reason);
}
},
[=](done_atom, const error& reason) { [=](done_atom, const error& reason) {
CAF_CHECK_EQUAL(reason, exit_reason::user_shutdown); CAF_CHECK_EQUAL(reason, exit_reason::user_shutdown);
if (++downs_ == 2) { if (++downs_ == 2) {
......
...@@ -73,21 +73,19 @@ CAF_TEST(test_custom_exception_handler) { ...@@ -73,21 +73,19 @@ CAF_TEST(test_custom_exception_handler) {
auto testee3 = self->spawn<exception_testee, monitored>(); auto testee3 = self->spawn<exception_testee, monitored>();
self->send(testee3, "foo"); self->send(testee3, "foo");
// receive all down messages // receive all down messages
auto i = 0; self->set_down_handler([&](down_msg& dm) {
self->receive_for(i, 3)( if (dm.source == testee1) {
[&](const down_msg& dm) { CAF_CHECK_EQUAL(dm.reason, exit_reason::normal);
if (dm.source == testee1) {
CAF_CHECK_EQUAL(dm.reason, exit_reason::normal);
}
else if (dm.source == testee2) {
CAF_CHECK_EQUAL(dm.reason, exit_reason::unhandled_exception);
}
else if (dm.source == testee3) {
CAF_CHECK_EQUAL(dm.reason, exit_reason::remote_link_unreachable);
}
else {
throw std::runtime_error("received message from unexpected source");
}
} }
); else if (dm.source == testee2) {
CAF_CHECK_EQUAL(dm.reason, exit_reason::unhandled_exception);
}
else if (dm.source == testee3) {
CAF_CHECK_EQUAL(dm.reason, exit_reason::remote_link_unreachable);
}
else {
throw std::runtime_error("received message from unexpected source");
}
});
self->wait_for(testee1, testee2, testee3);
} }
...@@ -278,11 +278,13 @@ behavior master(event_based_actor* self) { ...@@ -278,11 +278,13 @@ behavior master(event_based_actor* self) {
behavior slave(event_based_actor* self, actor master) { behavior slave(event_based_actor* self, actor master) {
self->link_to(master); self->link_to(master);
self->trap_exit(true); self->set_exit_handler([=](exit_msg& msg) {
CAF_MESSAGE("slave: received exit message");
self->quit(msg.reason);
});
return { return {
[=](const exit_msg& msg) { [] {
CAF_MESSAGE("slave: received exit message"); // nop
self->quit(msg.reason);
} }
}; };
} }
...@@ -358,47 +360,29 @@ CAF_TEST(self_receive_with_zero_timeout) { ...@@ -358,47 +360,29 @@ CAF_TEST(self_receive_with_zero_timeout) {
CAF_TEST(mirror) { CAF_TEST(mirror) {
scoped_actor self{system}; scoped_actor self{system};
auto mirror = self->spawn<simple_mirror, monitored>(); auto mirror = self->spawn<simple_mirror>();
self->send(mirror, "hello mirror"); self->send(mirror, "hello mirror");
self->receive ( self->receive (
[](const std::string& msg) { [](const std::string& msg) {
CAF_CHECK_EQUAL(msg, "hello mirror"); CAF_CHECK_EQUAL(msg, "hello mirror");
} }
); );
self->send_exit(mirror, exit_reason::user_shutdown);
self->receive (
[&](const down_msg& dm) {
if (dm.reason == exit_reason::user_shutdown)
CAF_MESSAGE("received `down_msg`");
else
CAF_ERROR("Unexpected message");
}
);
} }
CAF_TEST(detached_mirror) { CAF_TEST(detached_mirror) {
scoped_actor self{system}; scoped_actor self{system};
auto mirror = self->spawn<simple_mirror, monitored+detached>(); auto mirror = self->spawn<simple_mirror, detached>();
self->send(mirror, "hello mirror"); self->send(mirror, "hello mirror");
self->receive ( self->receive (
[](const std::string& msg) { [](const std::string& msg) {
CAF_CHECK_EQUAL(msg, "hello mirror"); CAF_CHECK_EQUAL(msg, "hello mirror");
} }
); );
self->send_exit(mirror, exit_reason::user_shutdown);
self->receive (
[&](const down_msg& dm) {
if (dm.reason == exit_reason::user_shutdown)
CAF_MESSAGE("received `down_msg`");
else
CAF_ERROR("Unexpected message");
}
);
} }
CAF_TEST(priority_aware_mirror) { CAF_TEST(priority_aware_mirror) {
scoped_actor self{system}; scoped_actor self{system};
auto mirror = self->spawn<simple_mirror, monitored + priority_aware>(); auto mirror = self->spawn<simple_mirror, priority_aware>();
CAF_MESSAGE("spawned mirror"); CAF_MESSAGE("spawned mirror");
self->send(mirror, "hello mirror"); self->send(mirror, "hello mirror");
self->receive ( self->receive (
...@@ -406,15 +390,6 @@ CAF_TEST(priority_aware_mirror) { ...@@ -406,15 +390,6 @@ CAF_TEST(priority_aware_mirror) {
CAF_CHECK_EQUAL(msg, "hello mirror"); CAF_CHECK_EQUAL(msg, "hello mirror");
} }
); );
self->send_exit(mirror, exit_reason::user_shutdown);
self->receive (
[&](const down_msg& dm) {
if (dm.reason == exit_reason::user_shutdown)
CAF_MESSAGE("received `down_msg`");
else
CAF_ERROR("Unexpected message");
}
);
} }
CAF_TEST(send_to_self) { CAF_TEST(send_to_self) {
...@@ -550,25 +525,22 @@ CAF_TEST(constructor_attach) { ...@@ -550,25 +525,22 @@ CAF_TEST(constructor_attach) {
class spawner : public event_based_actor { class spawner : public event_based_actor {
public: public:
spawner(actor_config& cfg) : event_based_actor(cfg), downs_(0) { spawner(actor_config& cfg) : event_based_actor(cfg), downs_(0) {
// nop set_down_handler([=](down_msg& msg) {
CAF_CHECK_EQUAL(msg.reason, exit_reason::user_shutdown);
if (++downs_ == 2)
quit(msg.reason);
});
set_exit_handler([=](exit_msg& msg) {
send_exit(testee_, std::move(msg.reason));
});
} }
behavior make_behavior() { behavior make_behavior() {
trap_exit(true);
testee_ = spawn<testee, monitored>(this); testee_ = spawn<testee, monitored>(this);
return { return {
[=](const down_msg& msg) {
CAF_CHECK_EQUAL(msg.reason, exit_reason::user_shutdown);
if (++downs_ == 2) {
quit(msg.reason);
}
},
[=](ok_atom, const error& reason) { [=](ok_atom, const error& reason) {
CAF_CHECK_EQUAL(reason, exit_reason::user_shutdown); CAF_CHECK_EQUAL(reason, exit_reason::user_shutdown);
if (++downs_ == 2) if (++downs_ == 2)
quit(reason); quit(reason);
},
[=](exit_msg& msg) {
delegate(testee_, std::move(msg));
} }
}; };
} }
...@@ -630,28 +602,36 @@ CAF_TEST(custom_exception_handler) { ...@@ -630,28 +602,36 @@ CAF_TEST(custom_exception_handler) {
auto testee3 = self->spawn<exception_testee, monitored>(); auto testee3 = self->spawn<exception_testee, monitored>();
self->send(testee3, "foo"); self->send(testee3, "foo");
// receive all down messages // receive all down messages
auto i = 0; int downs_received = 0;
self->receive_for(i, 3)( self->set_down_handler([&](down_msg& dm) {
[&](const down_msg& dm) { if (dm.source == testee1) {
if (dm.source == testee1) { ++downs_received;
CAF_CHECK_EQUAL(dm.reason, exit_reason::unhandled_exception); CAF_CHECK_EQUAL(dm.reason, exit_reason::unhandled_exception);
}
else if (dm.source == testee2) {
CAF_CHECK_EQUAL(dm.reason, exit_reason::unknown);
}
else if (dm.source == testee3) {
CAF_CHECK_EQUAL(dm.reason, exit_reason::unhandled_exception);
}
else {
throw std::runtime_error("received message from unexpected source");
}
} }
); else if (dm.source == testee2) {
CAF_CHECK_EQUAL(dm.reason, exit_reason::unknown);
++downs_received;
}
else if (dm.source == testee3) {
CAF_CHECK_EQUAL(dm.reason, exit_reason::unhandled_exception);
++downs_received;
}
else {
throw std::runtime_error("received message from unexpected source");
}
if (downs_received == 3)
self->send(self, message{});
});
// this exploits the fact that down messages are counted by dequeue
behavior dummy{[] {}};
self->dequeue(dummy);
} }
CAF_TEST(kill_the_immortal) { CAF_TEST(kill_the_immortal) {
auto wannabe_immortal = system.spawn([](event_based_actor* self) -> behavior { auto wannabe_immortal = system.spawn([](event_based_actor* self) -> behavior {
self->trap_exit(true); self->set_exit_handler([](local_actor*, exit_msg&) {
// nop
});
return { return {
[] { [] {
// nop // nop
...@@ -659,14 +639,8 @@ CAF_TEST(kill_the_immortal) { ...@@ -659,14 +639,8 @@ CAF_TEST(kill_the_immortal) {
}; };
}); });
scoped_actor self{system}; scoped_actor self{system};
self->monitor(wannabe_immortal);
self->send_exit(wannabe_immortal, exit_reason::kill); self->send_exit(wannabe_immortal, exit_reason::kill);
self->receive( self->wait_for(wannabe_immortal);
[&](const down_msg& dm) {
CAF_CHECK_EQUAL(dm.reason, exit_reason::kill);
CAF_CHECK_EQUAL(dm.source, wannabe_immortal.address());
}
);
} }
CAF_TEST(move_only_argument) { CAF_TEST(move_only_argument) {
......
...@@ -58,7 +58,11 @@ public: ...@@ -58,7 +58,11 @@ public:
: event_based_actor(cfg), : event_based_actor(cfg),
aut_(std::move(aut)), aut_(std::move(aut)),
msg_(make_message(1, 2, 3)) { msg_(make_message(1, 2, 3)) {
// nop set_down_handler([=](down_msg& dm) {
CAF_CHECK_EQUAL(dm.reason, exit_reason::normal);
CAF_CHECK_EQUAL(dm.source, aut_.address());
quit();
});
} }
behavior make_behavior() override { behavior make_behavior() override {
...@@ -69,11 +73,6 @@ public: ...@@ -69,11 +73,6 @@ public:
CAF_CHECK_EQUAL(a, 1); CAF_CHECK_EQUAL(a, 1);
CAF_CHECK_EQUAL(b, 2); CAF_CHECK_EQUAL(b, 2);
CAF_CHECK_EQUAL(c, 3); CAF_CHECK_EQUAL(c, 3);
},
[=](const down_msg& dm) {
CAF_CHECK_EQUAL(dm.reason, exit_reason::normal);
CAF_CHECK_EQUAL(dm.source, aut_.address());
quit();
} }
}; };
} }
......
...@@ -55,14 +55,6 @@ second_stage::behavior_type typed_second_stage() { ...@@ -55,14 +55,6 @@ second_stage::behavior_type typed_second_stage() {
} }
struct fixture { struct fixture {
void wait_until_is_terminated() {
self->receive(
[](const down_msg&) {
CAF_CHECK(true);
}
);
}
template <class Actor> template <class Actor>
static bool exited(const Actor& handle) { static bool exited(const Actor& handle) {
auto ptr = actor_cast<abstract_actor*>(handle); auto ptr = actor_cast<abstract_actor*>(handle);
...@@ -98,9 +90,8 @@ CAF_TEST(identity) { ...@@ -98,9 +90,8 @@ CAF_TEST(identity) {
CAF_TEST(lifetime_1a) { CAF_TEST(lifetime_1a) {
auto g = system.spawn(testee); auto g = system.spawn(testee);
auto f = system.spawn(testee); auto f = system.spawn(testee);
self->monitor(g);
anon_send_exit(g, exit_reason::kill); anon_send_exit(g, exit_reason::kill);
wait_until_is_terminated(); self->wait_for(g);
auto h = f * g; auto h = f * g;
CAF_CHECK(exited(h)); CAF_CHECK(exited(h));
} }
...@@ -109,9 +100,8 @@ CAF_TEST(lifetime_1a) { ...@@ -109,9 +100,8 @@ CAF_TEST(lifetime_1a) {
CAF_TEST(lifetime_1b) { CAF_TEST(lifetime_1b) {
auto g = system.spawn(testee); auto g = system.spawn(testee);
auto f = system.spawn(testee); auto f = system.spawn(testee);
self->monitor(f);
anon_send_exit(f, exit_reason::kill); anon_send_exit(f, exit_reason::kill);
wait_until_is_terminated(); self->wait_for(f);
auto h = f * g; auto h = f * g;
CAF_CHECK(exited(h)); CAF_CHECK(exited(h));
} }
......
...@@ -70,17 +70,6 @@ struct fixture { ...@@ -70,17 +70,6 @@ struct fixture {
second = system.spawn(untyped_second_stage); second = system.spawn(untyped_second_stage);
first_and_second = splice(first, second.bind(23.0, _1)); first_and_second = splice(first, second.bind(23.0, _1));
} }
void await_down(const actor& x) {
self->monitor(x);
self->receive(
[&](const down_msg& dm) -> optional<skip_t> {
if (dm.source != x)
return skip();
return none;
}
);
}
}; };
} // namespace <anonymous> } // namespace <anonymous>
...@@ -97,13 +86,13 @@ CAF_TEST(identity) { ...@@ -97,13 +86,13 @@ CAF_TEST(identity) {
CAF_TEST(kill_first) { CAF_TEST(kill_first) {
init_untyped(); init_untyped();
anon_send_exit(first, exit_reason::kill); anon_send_exit(first, exit_reason::kill);
await_down(first_and_second); self->wait_for(first_and_second);
} }
CAF_TEST(kill_second) { CAF_TEST(kill_second) {
init_untyped(); init_untyped();
anon_send_exit(second, exit_reason::kill); anon_send_exit(second, exit_reason::kill);
await_down(first_and_second); self->wait_for(first_and_second);
} }
CAF_TEST(untyped_splicing) { CAF_TEST(untyped_splicing) {
......
...@@ -186,11 +186,10 @@ CAF_TEST(error_response_message) { ...@@ -186,11 +186,10 @@ CAF_TEST(error_response_message) {
[](double) { [](double) {
CAF_ERROR("unexpected ordinary response message received"); CAF_ERROR("unexpected ordinary response message received");
}, },
[](const error& err) { [](error& err) {
CAF_CHECK_EQUAL(err.code(), static_cast<uint8_t>(sec::unexpected_message)); CAF_CHECK_EQUAL(err.code(), static_cast<uint8_t>(sec::unexpected_message));
} }
); );
self->send(foo, get_atom::value, 3.14);
self->send(foo, get_atom::value, 42); self->send(foo, get_atom::value, 42);
self->receive( self->receive(
[](int x) { [](int x) {
...@@ -200,6 +199,12 @@ CAF_TEST(error_response_message) { ...@@ -200,6 +199,12 @@ CAF_TEST(error_response_message) {
CAF_ERROR("unexpected ordinary response message received: " << x); CAF_ERROR("unexpected ordinary response message received: " << x);
} }
); );
self->set_error_handler([&](error& err) {
CAF_CHECK_EQUAL(err.code(), static_cast<uint8_t>(sec::unexpected_message));
self->send(self, message{});
});
self->send(foo, get_atom::value, 3.14);
self->receive([] {});
} }
// verify that delivering to a satisfied promise has no effect // verify that delivering to a satisfied promise has no effect
......
...@@ -274,19 +274,15 @@ behavior foo(event_based_actor* self) { ...@@ -274,19 +274,15 @@ behavior foo(event_based_actor* self) {
} }
int_actor::behavior_type int_fun2(int_actor::pointer self) { int_actor::behavior_type int_fun2(int_actor::pointer self) {
self->trap_exit(true); self->set_down_handler([=](down_msg& dm) {
CAF_CHECK_EQUAL(dm.reason, exit_reason::normal);
self->quit();
});
return { return {
[=](int i) { [=](int i) {
self->monitor(self->current_sender()); self->monitor(self->current_sender());
return i * i; return i * i;
}, },
[=](const down_msg& dm) {
CAF_CHECK_EQUAL(dm.reason, exit_reason::normal);
self->quit();
},
[=](const exit_msg&) {
CAF_ERROR("Unexpected message");
}
}; };
} }
...@@ -333,17 +329,13 @@ struct fixture { ...@@ -333,17 +329,13 @@ struct fixture {
} }
); );
CAF_CHECK_EQUAL(system.registry().running(), 2u); CAF_CHECK_EQUAL(system.registry().running(), 2u);
self->spawn<monitored>(client, self, ts); auto c1 = self->spawn(client, self, ts);
self->receive( self->receive(
[](passed_atom) { [](passed_atom) {
CAF_MESSAGE("received `passed_atom`"); CAF_MESSAGE("received `passed_atom`");
} }
); );
self->receive( self->wait_for(c1);
[](const down_msg& dm) {
CAF_CHECK(dm.reason == exit_reason::normal);
}
);
CAF_CHECK_EQUAL(system.registry().running(), 2u); CAF_CHECK_EQUAL(system.registry().running(), 2u);
} }
}; };
...@@ -485,12 +477,8 @@ CAF_TEST(check_signature) { ...@@ -485,12 +477,8 @@ CAF_TEST(check_signature) {
} }
}; };
}; };
self->spawn<monitored>(bar_action); auto x = self->spawn(bar_action);
self->receive( self->wait_for(x);
[](const down_msg& dm) {
CAF_CHECK_EQUAL(dm.reason, exit_reason::normal);
}
);
} }
CAF_TEST_FIXTURE_SCOPE_END() CAF_TEST_FIXTURE_SCOPE_END()
......
...@@ -264,6 +264,11 @@ void basp_broker_state::learned_new_node(const node_id& nid) { ...@@ -264,6 +264,11 @@ void basp_broker_state::learned_new_node(const node_id& nid) {
auto& tmp = spawn_servers[nid]; auto& tmp = spawn_servers[nid];
tmp = system().spawn<hidden>([=](event_based_actor* this_actor) -> behavior { tmp = system().spawn<hidden>([=](event_based_actor* this_actor) -> behavior {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
// terminate when receiving a down message
this_actor->set_down_handler([=](down_msg& dm) {
CAF_LOG_TRACE(CAF_ARG(dm));
this_actor->quit(std::move(dm.reason));
});
// skip messages until we receive the initial ok_atom // skip messages until we receive the initial ok_atom
this_actor->set_default_handler(skip); this_actor->set_default_handler(skip);
return { return {
...@@ -281,10 +286,6 @@ void basp_broker_state::learned_new_node(const node_id& nid) { ...@@ -281,10 +286,6 @@ void basp_broker_state::learned_new_node(const node_id& nid) {
this_actor->delegate(config_serv, get_atom::value, this_actor->delegate(config_serv, get_atom::value,
std::move(type), std::move(args)); std::move(type), std::move(args));
return {}; return {};
},
[=](const down_msg& dm) {
CAF_LOG_TRACE(CAF_ARG(dm));
this_actor->quit(dm.reason);
} }
); );
}, },
...@@ -339,6 +340,10 @@ void basp_broker_state::learned_new_node_indirectly(const node_id& nid) { ...@@ -339,6 +340,10 @@ void basp_broker_state::learned_new_node_indirectly(const node_id& nid) {
auto connection_helper = [=](event_based_actor* helper, actor s) -> behavior { auto connection_helper = [=](event_based_actor* helper, actor s) -> behavior {
CAF_LOG_TRACE(CAF_ARG(s)); CAF_LOG_TRACE(CAF_ARG(s));
helper->monitor(s); helper->monitor(s);
helper->set_down_handler([=](down_msg& dm) {
CAF_LOG_TRACE(CAF_ARG(dm));
helper->quit(std::move(dm.reason));
});
return { return {
// this config is send from the remote `ConfigServ` // this config is send from the remote `ConfigServ`
[=](ok_atom, const std::string&, message& msg) { [=](ok_atom, const std::string&, message& msg) {
...@@ -368,10 +373,6 @@ void basp_broker_state::learned_new_node_indirectly(const node_id& nid) { ...@@ -368,10 +373,6 @@ void basp_broker_state::learned_new_node_indirectly(const node_id& nid) {
} }
}); });
}, },
[=](const down_msg& dm) {
CAF_LOG_TRACE(CAF_ARG(dm));
helper->quit(dm.reason);
},
after(std::chrono::minutes(10)) >> [=] { after(std::chrono::minutes(10)) >> [=] {
CAF_LOG_TRACE(CAF_ARG("")); CAF_LOG_TRACE(CAF_ARG(""));
// nothing heard in about 10 minutes... just a call it a day, then // nothing heard in about 10 minutes... just a call it a day, then
......
...@@ -310,13 +310,8 @@ void middleman::stop() { ...@@ -310,13 +310,8 @@ void middleman::stop() {
hooks_.reset(); hooks_.reset();
named_brokers_.clear(); named_brokers_.clear();
scoped_actor self{system(), true}; scoped_actor self{system(), true};
self->monitor(manager_);
self->send_exit(manager_, exit_reason::user_shutdown); self->send_exit(manager_, exit_reason::user_shutdown);
self->receive( self->wait_for(manager_);
[](const down_msg&) {
// nop
}
);
} }
void middleman::init(actor_system_config& cfg) { void middleman::init(actor_system_config& cfg) {
......
...@@ -46,7 +46,16 @@ public: ...@@ -46,7 +46,16 @@ public:
middleman_actor_impl(actor_config& cfg, actor default_broker) middleman_actor_impl(actor_config& cfg, actor default_broker)
: middleman_actor::base(cfg), : middleman_actor::base(cfg),
broker_(default_broker) { broker_(default_broker) {
// nop set_down_handler([=](down_msg& dm) {
auto i = cached_.begin();
auto e = cached_.end();
while (i != e) {
if (get<1>(i->second) == dm.source)
i = cached_.erase(i);
else
++i;
}
});
} }
void on_exit() override { void on_exit() override {
...@@ -158,16 +167,6 @@ public: ...@@ -158,16 +167,6 @@ public:
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
delegate(broker_, atm, std::move(nid)); delegate(broker_, atm, std::move(nid));
return {}; return {};
},
[=](const down_msg& dm) {
auto i = cached_.begin();
auto e = cached_.end();
while (i != e) {
if (get<1>(i->second) == dm.source)
i = cached_.erase(i);
else
++i;
}
} }
}; };
} }
......
...@@ -64,6 +64,10 @@ void ping(event_based_actor* self, size_t num_pings) { ...@@ -64,6 +64,10 @@ void ping(event_based_actor* self, size_t num_pings) {
void pong(event_based_actor* self) { void pong(event_based_actor* self) {
CAF_MESSAGE("pong actor started"); CAF_MESSAGE("pong actor started");
self->set_down_handler([=](down_msg& dm) {
CAF_MESSAGE("received down_msg{" << to_string(dm.reason) << "}");
self->quit(dm.reason);
});
self->become( self->become(
[=](ping_atom, int value) -> std::tuple<atom_value, int> { [=](ping_atom, int value) -> std::tuple<atom_value, int> {
CAF_MESSAGE("received `ping_atom`"); CAF_MESSAGE("received `ping_atom`");
...@@ -72,10 +76,6 @@ void pong(event_based_actor* self) { ...@@ -72,10 +76,6 @@ void pong(event_based_actor* self) {
self->become( self->become(
[](ping_atom, int val) { [](ping_atom, int val) {
return std::make_tuple(pong_atom::value, val); return std::make_tuple(pong_atom::value, val);
},
[=](const down_msg& dm) {
CAF_MESSAGE("received down_msg{" << to_string(dm.reason) << "}");
self->quit(dm.reason);
} }
); );
// reply to 'ping' // reply to 'ping'
...@@ -102,6 +102,11 @@ void peer_fun(broker* self, connection_handle hdl, const actor& buddy) { ...@@ -102,6 +102,11 @@ void peer_fun(broker* self, connection_handle hdl, const actor& buddy) {
buf.insert(buf.end(), first, first + sizeof(int)); buf.insert(buf.end(), first, first + sizeof(int));
self->flush(hdl); self->flush(hdl);
}; };
self->set_down_handler([=](down_msg& dm) {
CAF_MESSAGE("received: " << to_string(dm));
if (dm.source == buddy)
self->quit(dm.reason);
});
self->become( self->become(
[=](const connection_closed_msg&) { [=](const connection_closed_msg&) {
CAF_MESSAGE("received connection_closed_msg"); CAF_MESSAGE("received connection_closed_msg");
...@@ -122,11 +127,6 @@ void peer_fun(broker* self, connection_handle hdl, const actor& buddy) { ...@@ -122,11 +127,6 @@ void peer_fun(broker* self, connection_handle hdl, const actor& buddy) {
[=](pong_atom, int value) { [=](pong_atom, int value) {
CAF_MESSAGE("received: pong " << value); CAF_MESSAGE("received: pong " << value);
write(pong_atom::value, value); write(pong_atom::value, value);
},
[=](const down_msg& dm) {
CAF_MESSAGE("received: " << to_string(dm));
if (dm.source == buddy)
self->quit(dm.reason);
} }
); );
} }
......
...@@ -60,25 +60,23 @@ behavior make_reflector_behavior(event_based_actor* self) { ...@@ -60,25 +60,23 @@ behavior make_reflector_behavior(event_based_actor* self) {
using spawn_atom = atom_constant<atom("Spawn")>; using spawn_atom = atom_constant<atom("Spawn")>;
using get_group_atom = atom_constant<atom("GetGroup")>; using get_group_atom = atom_constant<atom("GetGroup")>;
struct await_reflector_down_behavior {
event_based_actor* self;
int cnt;
void operator()(const down_msg&) {
if (++cnt == 5)
self->quit();
}
};
struct await_reflector_reply_behavior { struct await_reflector_reply_behavior {
event_based_actor* self; event_based_actor* self;
int cnt; int cnt;
int downs;
std::vector<actor> vec;
void operator()(const std::string& str, double val) { void operator()(const std::string& str, double val) {
CAF_CHECK_EQUAL(str, "Hello reflector!"); CAF_CHECK_EQUAL(str, "Hello reflector!");
CAF_CHECK_EQUAL(val, 5.0); CAF_CHECK_EQUAL(val, 5.0);
if (++cnt == 7) if (++cnt == 7) {
self->become(await_reflector_down_behavior{self, 0}); for (auto actor : vec)
self->monitor(actor);
self->set_down_handler([=](down_msg&) {
if (++downs == 5)
self->quit();
});
}
} }
}; };
...@@ -95,10 +93,7 @@ void make_client_behavior(event_based_actor* self, ...@@ -95,10 +93,7 @@ void make_client_behavior(event_based_actor* self,
}; };
CAF_CHECK(std::all_of(vec.begin(), vec.end(), is_remote)); CAF_CHECK(std::all_of(vec.begin(), vec.end(), is_remote));
self->send(grp, "Hello reflector!", 5.0); self->send(grp, "Hello reflector!", 5.0);
for (auto actor : vec) { self->become(await_reflector_reply_behavior{self, 0, 0, vec});
self->monitor(actor);
}
self->become(await_reflector_reply_behavior{self, 0});
} }
); );
} }
......
...@@ -70,6 +70,10 @@ behavior ping(event_based_actor* self, size_t num_pings) { ...@@ -70,6 +70,10 @@ behavior ping(event_based_actor* self, size_t num_pings) {
behavior pong(event_based_actor* self) { behavior pong(event_based_actor* self) {
CAF_MESSAGE("pong actor started"); CAF_MESSAGE("pong actor started");
self->set_down_handler([=](down_msg& dm) {
CAF_MESSAGE("received down_msg{" << to_string(dm.reason) << "}");
self->quit(dm.reason);
});
return { return {
[=](ping_atom, int value) -> std::tuple<atom_value, int> { [=](ping_atom, int value) -> std::tuple<atom_value, int> {
CAF_MESSAGE("received `ping_atom`"); CAF_MESSAGE("received `ping_atom`");
...@@ -78,10 +82,6 @@ behavior pong(event_based_actor* self) { ...@@ -78,10 +82,6 @@ behavior pong(event_based_actor* self) {
self->become( self->become(
[](ping_atom, int val) { [](ping_atom, int val) {
return std::make_tuple(pong_atom::value, val); return std::make_tuple(pong_atom::value, val);
},
[=](const down_msg& dm) {
CAF_MESSAGE("received down_msg{" << to_string(dm.reason) << "}");
self->quit(dm.reason);
} }
); );
// reply to 'ping' // reply to 'ping'
...@@ -112,6 +112,11 @@ peer::behavior_type peer_fun(peer::broker_pointer self, connection_handle hdl, ...@@ -112,6 +112,11 @@ peer::behavior_type peer_fun(peer::broker_pointer self, connection_handle hdl,
buf.insert(buf.end(), first, first + sizeof(int)); buf.insert(buf.end(), first, first + sizeof(int));
self->flush(hdl); self->flush(hdl);
}; };
self->set_down_handler([=](down_msg& dm) {
CAF_MESSAGE("received down_msg");
if (dm.source == buddy)
self->quit(std::move(dm.reason));
});
return { return {
[=](const connection_closed_msg&) { [=](const connection_closed_msg&) {
CAF_MESSAGE("received connection_closed_msg"); CAF_MESSAGE("received connection_closed_msg");
...@@ -132,12 +137,6 @@ peer::behavior_type peer_fun(peer::broker_pointer self, connection_handle hdl, ...@@ -132,12 +137,6 @@ peer::behavior_type peer_fun(peer::broker_pointer self, connection_handle hdl,
[=](pong_atom, int value) { [=](pong_atom, int value) {
CAF_MESSAGE("received pong{" << value << "}"); CAF_MESSAGE("received pong{" << value << "}");
write(pong_atom::value, value); write(pong_atom::value, value);
},
[=](const down_msg& dm) {
CAF_MESSAGE("received down_msg");
if (dm.source == buddy) {
self->quit(dm.reason);
}
} }
}; };
} }
......
...@@ -94,14 +94,13 @@ void run_client(int argc, char** argv, uint16_t port) { ...@@ -94,14 +94,13 @@ 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, infinite, ping{42}) self->request(serv, infinite, ping{42}).receive(
.receive([](const pong& p) { CAF_CHECK_EQUAL(p.value, 42); }); [](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->wait_for(serv);
self->receive([&](const down_msg& dm) {
CAF_CHECK_EQUAL(dm.reason, exit_reason::user_shutdown);
CAF_CHECK_EQUAL(dm.source, serv.address());
});
} }
void run_server(int argc, char** argv) { void run_server(int argc, char** argv) {
......
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