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

Propagate actor fail states via `error`

parent fa6707df
...@@ -40,8 +40,8 @@ using kickoff_atom = atom_constant<atom("kickoff")>; ...@@ -40,8 +40,8 @@ using kickoff_atom = atom_constant<atom("kickoff")>;
// utility function to print an exit message with custom name // utility function to print an exit message with custom name
void print_on_exit(const actor& hdl, const std::string& name) { void print_on_exit(const actor& hdl, const std::string& name) {
hdl->attach_functor([=](exit_reason reason) { hdl->attach_functor([=](const error& reason) {
cout << name << " exited with reason " << to_string(reason) << endl; cout << name << " exited: " << to_string(reason) << endl;
}); });
} }
......
...@@ -95,38 +95,16 @@ public: ...@@ -95,38 +95,16 @@ public:
/// Detaches the first attached object that matches `what`. /// Detaches the first attached object that matches `what`.
virtual size_t detach(const attachable::token& what) = 0; virtual size_t detach(const attachable::token& what) = 0;
/// Links this actor to `whom`. /// Establishes a link relation between this actor and `x`
inline void link_to(const actor_addr& whom) {
link_impl(establish_link_op, whom);
}
/// Links this actor to `whom`.
template <class ActorHandle>
void link_to(const ActorHandle& whom) {
link_to(whom.address());
}
/// Unlinks this actor from `whom`.
inline void unlink_from(const actor_addr& other) {
link_impl(remove_link_op, other);
}
/// Unlinks this actor from `whom`.
template <class ActorHandle>
void unlink_from(const ActorHandle& other) {
unlink_from(other.address());
}
/// Establishes a link relation between this actor and `other`
/// and returns whether the operation succeeded. /// and returns whether the operation succeeded.
inline bool establish_backlink(const actor_addr& other) { inline bool establish_backlink(abstract_actor* x) {
return link_impl(establish_backlink_op, other); return link_impl(establish_backlink_op, x);
} }
/// Removes the link relation between this actor and `other` /// Removes the link relation between this actor and `x`
/// and returns whether the operation succeeded. /// and returns whether the operation succeeded.
inline bool remove_backlink(const actor_addr& other) { inline bool remove_backlink(abstract_actor* x) {
return link_impl(remove_backlink_op, other); return link_impl(remove_backlink_op, x);
} }
/// Returns the set of accepted messages types as strings or /// Returns the set of accepted messages types as strings or
...@@ -157,15 +135,18 @@ public: ...@@ -157,15 +135,18 @@ 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_exit_flag = 0x0001; // local_actor
static constexpr int has_timeout_flag = 0x0002; // single_timeout static constexpr int trap_error_flag = 0x0002; // local_actor
static constexpr int is_registered_flag = 0x0004; // (several actors) static constexpr int has_timeout_flag = 0x0004; // single_timeout
static constexpr int is_initialized_flag = 0x0008; // event-based actors static constexpr int is_registered_flag = 0x0008; // (several actors)
static constexpr int is_blocking_flag = 0x0010; // blocking_actor static constexpr int is_initialized_flag = 0x0010; // event-based actors
static constexpr int is_detached_flag = 0x0020; // local_actor static constexpr int is_blocking_flag = 0x0020; // blocking_actor
static constexpr int is_priority_aware_flag = 0x0040; // local_actor static constexpr int is_detached_flag = 0x0040; // local_actor
static constexpr int is_serializable_flag = 0x0040; // local_actor static constexpr int is_priority_aware_flag = 0x0080; // local_actor
static constexpr int is_migrated_from_flag = 0x0080; // local_actor static constexpr int is_serializable_flag = 0x0100; // local_actor
static constexpr int has_used_aout_flag = 0x0100; // local_actor static constexpr int is_migrated_from_flag = 0x0200; // local_actor
static constexpr int has_used_aout_flag = 0x0400; // local_actor
static constexpr int is_terminated_flag = 0x0800; // local_actor
static constexpr int is_cleaned_up_flag = 0x1000; // monitorable_actor
inline void set_flag(bool enable_flag, int mask) { inline void set_flag(bool enable_flag, int mask) {
auto x = flags(); auto x = flags();
...@@ -242,7 +223,23 @@ public: ...@@ -242,7 +223,23 @@ public:
return static_cast<bool>(flags() & is_actor_decorator_mask); return static_cast<bool>(flags() & is_actor_decorator_mask);
} }
virtual bool link_impl(linking_operation op, const actor_addr& other) = 0; inline bool is_terminated() const {
return get_flag(is_terminated_flag);
}
inline void is_terminated(bool value) {
set_flag(value, is_terminated_flag);
}
inline bool is_cleaned_up() const {
return get_flag(is_cleaned_up_flag);
}
inline void is_cleaned_up(bool value) {
set_flag(value, is_cleaned_up_flag);
}
virtual bool link_impl(linking_operation op, abstract_actor* other) = 0;
/// @endcond /// @endcond
......
...@@ -105,6 +105,9 @@ public: ...@@ -105,6 +105,9 @@ public:
actor_pool(actor_config& cfg); actor_pool(actor_config& cfg);
protected:
void on_cleanup() override;
private: private:
bool filter(upgrade_lock<detail::shared_spinlock>&, bool filter(upgrade_lock<detail::shared_spinlock>&,
const strong_actor_ptr& sender, message_id mid, const strong_actor_ptr& sender, message_id mid,
......
...@@ -38,15 +38,15 @@ public: ...@@ -38,15 +38,15 @@ public:
~actor_proxy(); ~actor_proxy();
/// Establishes a local link state that's not synchronized back /// Establishes a local link state that's
/// to the remote instance. /// not synchronized back to the remote instance.
virtual void local_link_to(const actor_addr& other) = 0; virtual void local_link_to(abstract_actor* other) = 0;
/// Removes a local link state. /// Removes a local link state.
virtual void local_unlink_from(const actor_addr& other) = 0; virtual void local_unlink_from(abstract_actor* other) = 0;
/// Invokes cleanup code. /// Invokes cleanup code.
virtual void kill_proxy(execution_unit* ctx, exit_reason reason) = 0; virtual void kill_proxy(execution_unit* ctx, error reason) = 0;
}; };
} // namespace caf } // namespace caf
......
...@@ -49,20 +49,15 @@ public: ...@@ -49,20 +49,15 @@ public:
~actor_registry(); ~actor_registry();
/// A registry entry consists of a pointer to the actor and an
/// exit reason. An entry with a nullptr means the actor has finished
/// execution for given reason.
using id_entry = std::pair<strong_actor_ptr, exit_reason>;
/// Returns the the local actor associated to `key`. /// Returns the the local actor associated to `key`.
id_entry get(actor_id key) const; strong_actor_ptr get(actor_id key) const;
/// Associates a local actor with its ID. /// Associates a local actor with its ID.
void put(actor_id key, const strong_actor_ptr& value); void put(actor_id key, const strong_actor_ptr& value);
/// Removes an actor from this registry, /// Removes an actor from this registry,
/// leaving `reason` for future reference. /// leaving `reason` for future reference.
void erase(actor_id key, exit_reason reason); void erase(actor_id key);
/// Increases running-actors-count by one. /// Increases running-actors-count by one.
void inc_running(); void inc_running();
...@@ -103,7 +98,7 @@ private: ...@@ -103,7 +98,7 @@ private:
// Stops this component. // Stops this component.
void stop(); void stop();
using entries = std::unordered_map<actor_id, id_entry>; using entries = std::unordered_map<actor_id, strong_actor_ptr>;
actor_registry(actor_system& sys); actor_registry(actor_system& sys);
......
...@@ -25,6 +25,7 @@ ...@@ -25,6 +25,7 @@
#include <typeinfo> #include <typeinfo>
#include <exception> #include <exception>
#include "caf/error.hpp"
#include "caf/optional.hpp" #include "caf/optional.hpp"
#include "caf/exit_reason.hpp" #include "caf/exit_reason.hpp"
#include "caf/execution_unit.hpp" #include "caf/execution_unit.hpp"
...@@ -77,7 +78,7 @@ public: ...@@ -77,7 +78,7 @@ public:
/// Executed if the actor finished execution with given `reason`. /// Executed if the actor finished execution with given `reason`.
/// The default implementation does nothing. /// The default implementation does nothing.
/// @warning `host` can be `nullptr` /// @warning `host` can be `nullptr`
virtual void actor_exited(exit_reason reason, execution_unit* host); virtual void actor_exited(const error& fail_state, execution_unit* host);
/// Returns `true` if `what` selects this instance, otherwise `false`. /// Returns `true` if `what` selects this instance, otherwise `false`.
virtual bool matches(const token& what); virtual bool matches(const token& what);
......
...@@ -187,6 +187,10 @@ public: ...@@ -187,6 +187,10 @@ public:
/// @cond PRIVATE /// @cond PRIVATE
inline const error& fail_state() {
return fail_state_;
}
void initialize() override; void initialize() override;
void dequeue(behavior& bhvr, message_id mid = invalid_message_id); void dequeue(behavior& bhvr, message_id mid = invalid_message_id);
......
...@@ -43,6 +43,9 @@ public: ...@@ -43,6 +43,9 @@ public:
// in either case, the processing is done synchronously // in either case, the processing is done synchronously
void enqueue(mailbox_element_ptr what, execution_unit* host) override; void enqueue(mailbox_element_ptr what, execution_unit* host) override;
protected:
void on_cleanup() override;
private: private:
strong_actor_ptr decorated_; strong_actor_ptr decorated_;
message merger_; message merger_;
......
...@@ -48,6 +48,9 @@ public: ...@@ -48,6 +48,9 @@ public:
message_types_set message_types() const override; message_types_set message_types() const override;
protected:
void on_cleanup() override;
private: private:
strong_actor_ptr f_; strong_actor_ptr f_;
strong_actor_ptr g_; strong_actor_ptr g_;
......
...@@ -50,6 +50,7 @@ public: ...@@ -50,6 +50,7 @@ public:
message_types_set message_types() const override; message_types_set message_types() const override;
private: private:
const size_t num_workers;
std::vector<strong_actor_ptr> workers_; std::vector<strong_actor_ptr> workers_;
message_types_set msg_types_; message_types_set msg_types_;
}; };
......
...@@ -38,7 +38,7 @@ public: ...@@ -38,7 +38,7 @@ public:
static constexpr size_t token_type = attachable::token::observer; static constexpr size_t token_type = attachable::token::observer;
}; };
void actor_exited(exit_reason reason, execution_unit* host) override; void actor_exited(const error& fail_state, execution_unit* host) override;
bool matches(const token& what) override; bool matches(const token& what) override;
......
...@@ -36,8 +36,8 @@ struct functor_attachable : attachable { ...@@ -36,8 +36,8 @@ struct functor_attachable : attachable {
functor_attachable(F arg) : functor_(std::move(arg)) { functor_attachable(F arg) : functor_(std::move(arg)) {
// nop // nop
} }
void actor_exited(exit_reason reason, execution_unit*) override { void actor_exited(const error& fail_state, execution_unit*) override {
functor_(reason); functor_(fail_state);
} }
static constexpr size_t token_type = attachable::token::anonymous; static constexpr size_t token_type = attachable::token::anonymous;
}; };
...@@ -48,7 +48,7 @@ struct functor_attachable<F, 0> : attachable { ...@@ -48,7 +48,7 @@ struct functor_attachable<F, 0> : attachable {
functor_attachable(F arg) : functor_(std::move(arg)) { functor_attachable(F arg) : functor_(std::move(arg)) {
// nop // nop
} }
void actor_exited(exit_reason, execution_unit*) override { void actor_exited(const error&, execution_unit*) override {
functor_(); functor_();
} }
}; };
......
...@@ -38,11 +38,10 @@ namespace caf { ...@@ -38,11 +38,10 @@ namespace caf {
namespace detail { namespace detail {
struct sync_request_bouncer { struct sync_request_bouncer {
exit_reason rsn; error rsn;
explicit sync_request_bouncer(exit_reason r); explicit sync_request_bouncer(error r);
void operator()(const strong_actor_ptr& sender, const message_id& mid) const; void operator()(const strong_actor_ptr& sender, const message_id& mid) const;
void operator()(const mailbox_element& e) const; void operator()(const mailbox_element& e) const;
}; };
} // namespace detail } // namespace detail
......
...@@ -47,7 +47,7 @@ namespace caf { ...@@ -47,7 +47,7 @@ namespace caf {
/// ///
/// # Why not `std::error_code` or `std::error_condition`? /// # Why not `std::error_code` or `std::error_condition`?
/// ///
/// First, the standard does///not* define the values for `std::errc`. /// First, the standard does *not* define the values for `std::errc`.
/// This means serializing error conditions (which are meant to be portable) /// This means serializing error conditions (which are meant to be portable)
/// is not safe in a distributed setting unless all machines are running the /// is not safe in a distributed setting unless all machines are running the
/// same operating system and version of the C++ standard library. /// same operating system and version of the C++ standard library.
...@@ -100,7 +100,14 @@ public: ...@@ -100,7 +100,14 @@ public:
const message& context() const; const message& context() const;
/// Returns `code() != 0`. /// Returns `code() != 0`.
explicit operator bool() const; inline explicit operator bool() const noexcept {
return code_ != 0;
}
/// Returns `code() == 0`.
inline bool operator!() const noexcept {
return code_ == 0;
}
/// Sets the error code to 0. /// Sets the error code to 0.
void clear(); void clear();
......
...@@ -53,17 +53,17 @@ private: ...@@ -53,17 +53,17 @@ private:
class actor_exited : public caf_exception { class actor_exited : public caf_exception {
public: public:
~actor_exited() noexcept; ~actor_exited() noexcept;
explicit actor_exited(exit_reason exit_reason); explicit actor_exited(error exit_reason);
actor_exited(const actor_exited&) = default; actor_exited(const actor_exited&) = default;
actor_exited& operator=(const actor_exited&) = default; actor_exited& operator=(const actor_exited&) = default;
/// Returns the exit reason. /// Returns the exit reason.
inline exit_reason reason() const noexcept { inline error reason() const noexcept {
return reason_; return reason_;
} }
private: private:
exit_reason reason_; error reason_;
}; };
/// Thrown to indicate that either an actor publishing failed or /// Thrown to indicate that either an actor publishing failed or
......
...@@ -20,16 +20,13 @@ ...@@ -20,16 +20,13 @@
#ifndef CAF_EXIT_REASON_HPP #ifndef CAF_EXIT_REASON_HPP
#define CAF_EXIT_REASON_HPP #define CAF_EXIT_REASON_HPP
#include <cstdint> #include "caf/error.hpp"
namespace caf { namespace caf {
enum class exit_reason : uint8_t { enum class exit_reason : uint8_t {
/// Indicates that the actor is still alive. /// Indicates that an actor finished execution without error.
not_exited = 0x00, normal = 0x00,
/// Indicates that an actor finished execution.
normal = 0x01,
/// Indicates that an actor finished execution because of an unhandled exception. /// Indicates that an actor finished execution because of an unhandled exception.
unhandled_exception = 0x02, unhandled_exception = 0x02,
...@@ -55,12 +52,21 @@ enum class exit_reason : uint8_t { ...@@ -55,12 +52,21 @@ enum class exit_reason : uint8_t {
remote_link_unreachable = 0x21, remote_link_unreachable = 0x21,
/// Indicates that the actor was killed because it became unreachable. /// Indicates that the actor was killed because it became unreachable.
unreachable = 0x40 unreachable = 0x40,
/// Indicates that the actor was killed after receiving an error message.
unhandled_error = 0x41
}; };
/// Returns a string representation of given exit reason. /// Returns a string representation of given exit reason.
const char* to_string(exit_reason x); const char* to_string(exit_reason x);
/// @relates exit_reason
error make_error(exit_reason);
/// @relates exit_reason
error make_error(exit_reason, message context);
} // namespace caf } // namespace caf
#endif // CAF_EXIT_REASON_HPP #endif // CAF_EXIT_REASON_HPP
...@@ -38,13 +38,13 @@ public: ...@@ -38,13 +38,13 @@ public:
void enqueue(mailbox_element_ptr what, execution_unit* host) override; void enqueue(mailbox_element_ptr what, execution_unit* host) override;
bool link_impl(linking_operation op, const actor_addr& other) override; bool link_impl(linking_operation op, abstract_actor* other) override;
void local_link_to(const actor_addr& other) override; void local_link_to(abstract_actor* other) override;
void local_unlink_from(const actor_addr& other) override; void local_unlink_from(abstract_actor* other) override;
void kill_proxy(execution_unit* ctx, exit_reason reason) override; void kill_proxy(execution_unit* ctx, error reason) override;
actor manager() const; actor manager() const;
......
...@@ -32,6 +32,7 @@ ...@@ -32,6 +32,7 @@
#include "caf/fwd.hpp" #include "caf/fwd.hpp"
#include "caf/actor.hpp" #include "caf/actor.hpp"
#include "caf/error.hpp"
#include "caf/extend.hpp" #include "caf/extend.hpp"
#include "caf/logger.hpp" #include "caf/logger.hpp"
#include "caf/message.hpp" #include "caf/message.hpp"
...@@ -354,7 +355,7 @@ public: ...@@ -354,7 +355,7 @@ public:
/// @warning This member function throws immediately in thread-based actors /// @warning This member function throws immediately in thread-based actors
/// that do not use the behavior stack, i.e., actors that use /// that do not use the behavior stack, i.e., actors that use
/// blocking API calls such as {@link receive()}. /// blocking API calls such as {@link receive()}.
void quit(exit_reason reason = exit_reason::normal); void quit(error reason = error{});
/// Checks whether this actor traps exit messages. /// Checks whether this actor traps exit messages.
inline bool trap_exit() const { inline bool trap_exit() const {
...@@ -366,6 +367,16 @@ public: ...@@ -366,6 +367,16 @@ public:
set_flag(value, trap_exit_flag); 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);
...@@ -583,14 +594,6 @@ public: ...@@ -583,14 +594,6 @@ public:
return delegate(message_priority::normal, dest, std::forward<Ts>(xs)...); return delegate(message_priority::normal, dest, std::forward<Ts>(xs)...);
} }
inline exit_reason planned_exit_reason() const {
return planned_exit_reason_;
}
inline void planned_exit_reason(exit_reason value) {
planned_exit_reason_ = value;
}
inline detail::behavior_stack& bhvr_stack() { inline detail::behavior_stack& bhvr_stack() {
return bhvr_stack_; return bhvr_stack_;
} }
...@@ -611,7 +614,7 @@ public: ...@@ -611,7 +614,7 @@ public:
// valid behavior left or has set a planned exit reason // valid behavior left or has set a planned exit reason
bool finished(); bool finished();
void cleanup(exit_reason reason, execution_unit* host) override; bool cleanup(error&& reason, execution_unit* host) override;
// an actor can have multiple pending timeouts, but only // an actor can have multiple pending timeouts, but only
// the latest one is active (i.e. the pending_timeouts_.back()) // the latest one is active (i.e. the pending_timeouts_.back())
...@@ -703,9 +706,6 @@ protected: ...@@ -703,9 +706,6 @@ protected:
// points to the node under processing otherwise // points to the node under processing otherwise
mailbox_element_ptr current_element_; mailbox_element_ptr current_element_;
// set by quit
exit_reason planned_exit_reason_;
// identifies the timeout messages we are currently waiting for // identifies the timeout messages we are currently waiting for
uint32_t timeout_id_; uint32_t timeout_id_;
......
...@@ -31,6 +31,8 @@ ...@@ -31,6 +31,8 @@
#include <type_traits> #include <type_traits>
#include <condition_variable> #include <condition_variable>
#include "caf/actor_addr.hpp"
#include "caf/actor_cast.hpp"
#include "caf/abstract_actor.hpp" #include "caf/abstract_actor.hpp"
#include "caf/mailbox_element.hpp" #include "caf/mailbox_element.hpp"
...@@ -51,10 +53,40 @@ public: ...@@ -51,10 +53,40 @@ public:
size_t detach(const attachable::token& what) override; size_t detach(const attachable::token& what) override;
/// Returns the exit reason of this actor or // -- linking and monitoring -------------------------------------------------
/// `exit_reason::not_exited` if it is still alive.
inline exit_reason get_exit_reason() const { /// Links this actor to `x`.
return exit_reason_.load(std::memory_order_relaxed); void link_to(const actor_addr& x) {
auto ptr = actor_cast<strong_actor_ptr>(x);
if (! ptr || ptr->get() == this)
return;
link_impl(establish_link_op, ptr->get());
}
/// Links this actor to `x`.
template <class ActorHandle>
void link_to(const ActorHandle& x) {
auto ptr = actor_cast<abstract_actor*>(x);
if (! ptr || ptr == this)
return;
link_impl(establish_link_op, ptr);
}
/// Unlinks this actor from `x`.
void unlink_from(const actor_addr& x) {
auto ptr = actor_cast<strong_actor_ptr>(x);
if (! ptr || ptr->get() == this)
return;
link_impl(remove_link_op, ptr->get());
}
/// Links this actor to `x`.
template <class ActorHandle>
void unlink_from(const ActorHandle& x) {
auto ptr = actor_cast<abstract_actor*>(x);
if (! ptr || ptr == this)
return;
link_impl(remove_link_op, ptr);
} }
/// @cond PRIVATE /// @cond PRIVATE
...@@ -62,19 +94,24 @@ public: ...@@ -62,19 +94,24 @@ public:
/// Called by the runtime system to perform cleanup actions for this actor. /// Called by the runtime system to perform cleanup actions for this actor.
/// Subtypes should always call this member function when overriding it. /// Subtypes should always call this member function when overriding it.
/// This member function is thread-safe, and if the actor has already exited /// This member function is thread-safe, and if the actor has already exited
/// upon invocation, nothing is done. /// upon invocation, nothing is done. The return value of this member
virtual void cleanup(exit_reason reason, execution_unit* host); /// function is ignored by scheduled actors.
virtual bool cleanup(error&& reason, execution_unit* host);
// returns `exit_reason_ != exit_reason::not_exited`;
// it's possible for user code to call this method
// without acquiring `mtx_`
inline bool exited() const {
return get_exit_reason() != exit_reason::not_exited;
}
/// @endcond /// @endcond
protected: protected:
/// Allows subclasses to add additional cleanup code to the
/// critical secion in `cleanup`. This member function is
/// called inside of a critical section.
virtual void on_cleanup();
/// Sends a response message if `what` is a request.
void bounce(mailbox_element_ptr& what);
/// Sends a response message if `what` is a request.
void bounce(mailbox_element_ptr& what, const error& err);
/// Creates a new actor instance. /// Creates a new actor instance.
explicit monitorable_actor(actor_config& cfg); explicit monitorable_actor(actor_config& cfg);
...@@ -82,15 +119,15 @@ protected: ...@@ -82,15 +119,15 @@ protected:
* here be dragons: end of public interface * * here be dragons: end of public interface *
****************************************************************************/ ****************************************************************************/
bool link_impl(linking_operation op, const actor_addr& other) override; bool link_impl(linking_operation op, abstract_actor* x) override;
bool establish_link_impl(const actor_addr& other); bool establish_link_impl(abstract_actor* other);
bool remove_link_impl(const actor_addr& other); bool remove_link_impl(abstract_actor* other);
bool establish_backlink_impl(const actor_addr& other); bool establish_backlink_impl(abstract_actor* other);
bool remove_backlink_impl(const actor_addr& other); bool remove_backlink_impl(abstract_actor* other);
// tries to run a custom exception handler for `eptr` // tries to run a custom exception handler for `eptr`
optional<exit_reason> handle(const std::exception_ptr& eptr); optional<exit_reason> handle(const std::exception_ptr& eptr);
...@@ -119,18 +156,30 @@ protected: ...@@ -119,18 +156,30 @@ protected:
bool trap_exit, F& down_msg_handler) { bool trap_exit, F& down_msg_handler) {
auto& msg = node.msg; auto& msg = node.msg;
if (msg.size() == 1 && msg.match_element<down_msg>(0)) { if (msg.size() == 1 && msg.match_element<down_msg>(0)) {
down_msg_handler(msg.get_as<down_msg>(0)); down_msg_handler(msg.get_as_mutable<down_msg>(0));
return true; return true;
} }
return handle_system_message(node, context, trap_exit); return handle_system_message(node, context, trap_exit);
} }
// initially set to exit_reason::not_exited; // Calls `fun` with exclusive access to an actor's state.
// can only be stored with `mtx_` acquired; template <class F>
// may be read without acquiring `mtx_` auto exclusive_critical_section(F fun) -> decltype(fun()) {
std::atomic<exit_reason> exit_reason_; std::unique_lock<std::mutex> guard{mtx_};
return fun();
}
template <class F>
auto shared_critical_section(F fun) -> decltype(fun()) {
std::unique_lock<std::mutex> guard{mtx_};
return fun();
}
// is protected by `mtx_` in actors that are not scheduled, but
// can be accessed without lock for event-based and blocking actors
error fail_state_;
// guards access to exit_reason_, attachables_, links_, // guards access to exit_state_, attachables_, links_,
// and enqueue operations if actor is thread-mapped // and enqueue operations if actor is thread-mapped
mutable std::mutex mtx_; mutable std::mutex mtx_;
......
...@@ -90,7 +90,7 @@ public: ...@@ -90,7 +90,7 @@ public:
/// Deletes the proxy with id `aid` for `node`. /// Deletes the proxy with id `aid` for `node`.
void erase(const key_type& node, actor_id aid, void erase(const key_type& node, actor_id aid,
exit_reason rsn = exit_reason::remote_link_unreachable); error rsn = exit_reason::remote_link_unreachable);
/// Queries whether there are any proxies left. /// Queries whether there are any proxies left.
bool empty() const; bool empty() const;
...@@ -103,7 +103,7 @@ public: ...@@ -103,7 +103,7 @@ public:
} }
private: private:
void kill_proxy(strong_actor_ptr&, exit_reason); void kill_proxy(strong_actor_ptr&, error);
actor_system& system_; actor_system& system_;
backend& backend_; backend& backend_;
......
...@@ -43,8 +43,8 @@ enum class sec : uint8_t { ...@@ -43,8 +43,8 @@ enum class sec : uint8_t {
no_actor_published_at_port, no_actor_published_at_port,
/// Migration failed because the state of an actor is not serializable. /// Migration failed because the state of an actor is not serializable.
state_not_serializable, state_not_serializable,
/// An actor received an invalid key for `('sys', 'get', key)` messages. /// An actor received an unsupported key for `('sys', 'get', key)` messages.
invalid_sys_key, unsupported_sys_key,
/// An actor received an unsupported system message. /// An actor received an unsupported system message.
unsupported_sys_message, unsupported_sys_message,
/// A remote node disconnected during CAF handshake. /// A remote node disconnected during CAF handshake.
......
...@@ -41,7 +41,7 @@ struct exit_msg { ...@@ -41,7 +41,7 @@ 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;
/// The exit reason of the terminated actor. /// The exit reason of the terminated actor.
exit_reason reason; error reason;
}; };
inline std::string to_string(const exit_msg& x) { inline std::string to_string(const exit_msg& x) {
...@@ -53,7 +53,7 @@ struct down_msg { ...@@ -53,7 +53,7 @@ struct down_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;
/// The exit reason of the terminated actor. /// The exit reason of the terminated actor.
exit_reason reason; error reason;
}; };
inline std::string to_string(const down_msg& x) { inline std::string to_string(const down_msg& x) {
......
...@@ -43,7 +43,7 @@ ...@@ -43,7 +43,7 @@
namespace caf { namespace caf {
// exit_reason_ is guaranteed to be set to 0, i.e., exit_reason::not_exited, // exit_state_ is guaranteed to be set to 0, i.e., exit_reason::not_exited,
// by std::atomic<> constructor // by std::atomic<> constructor
actor_control_block* abstract_actor::ctrl() const { actor_control_block* abstract_actor::ctrl() const {
......
...@@ -37,14 +37,14 @@ void actor_companion::on_enqueue(enqueue_handler handler) { ...@@ -37,14 +37,14 @@ void actor_companion::on_enqueue(enqueue_handler handler) {
} }
void actor_companion::enqueue(mailbox_element_ptr ptr, execution_unit*) { void actor_companion::enqueue(mailbox_element_ptr ptr, execution_unit*) {
CAF_ASSERT(ptr);
shared_lock<lock_type> guard(lock_); shared_lock<lock_type> guard(lock_);
if (on_enqueue_) { if (on_enqueue_)
on_enqueue_(std::move(ptr)); on_enqueue_(std::move(ptr));
}
} }
void actor_companion::enqueue(strong_actor_ptr sender, message_id mid, message content, void actor_companion::enqueue(strong_actor_ptr sender, message_id mid,
execution_unit* eu) { message content, execution_unit* eu) {
using detail::memory; using detail::memory;
auto ptr = mailbox_element::make(sender, mid, {}, std::move(content)); auto ptr = mailbox_element::make(sender, mid, {}, std::move(content));
enqueue(std::move(ptr), eu); enqueue(std::move(ptr), eu);
......
...@@ -96,7 +96,7 @@ void load_actor(deserializer& source, T& storage) { ...@@ -96,7 +96,7 @@ void load_actor(deserializer& source, T& storage) {
CAF_LOG_DEBUG(CAF_ARG(aid) << CAF_ARG(nid)); CAF_LOG_DEBUG(CAF_ARG(aid) << CAF_ARG(nid));
// deal with local actors // deal with local actors
if (sys.node() == nid) { if (sys.node() == nid) {
storage = actor_cast<T>(sys.registry().get(aid).first); storage = actor_cast<T>(sys.registry().get(aid));
CAF_LOG_DEBUG("fetch actor handle from local actor registry: " CAF_LOG_DEBUG("fetch actor handle from local actor registry: "
<< (storage ? "found" : "not found")); << (storage ? "found" : "not found"));
return; return;
......
...@@ -139,36 +139,33 @@ void actor_pool::enqueue(mailbox_element_ptr what, execution_unit* eu) { ...@@ -139,36 +139,33 @@ void actor_pool::enqueue(mailbox_element_ptr what, execution_unit* eu) {
policy_(home_system(), guard, workers_, what, eu); policy_(home_system(), guard, workers_, what, eu);
} }
actor_pool::actor_pool(actor_config& cfg) actor_pool::actor_pool(actor_config& cfg) : monitorable_actor(cfg) {
: monitorable_actor(cfg),
planned_reason_(exit_reason::not_exited) {
is_registered(true); is_registered(true);
} }
void actor_pool::on_cleanup() {
// nop
}
bool actor_pool::filter(upgrade_lock<detail::shared_spinlock>& guard, bool actor_pool::filter(upgrade_lock<detail::shared_spinlock>& guard,
const strong_actor_ptr& sender, message_id mid, const strong_actor_ptr& sender, message_id mid,
const message& msg, execution_unit* eu) { const message& msg, execution_unit* eu) {
CAF_LOG_TRACE(CAF_ARG(mid) << CAF_ARG(msg)); CAF_LOG_TRACE(CAF_ARG(mid) << CAF_ARG(msg));
auto rsn = planned_reason_;
if (rsn != caf::exit_reason::not_exited) {
guard.unlock();
if (mid.valid()) {
detail::sync_request_bouncer srq{rsn};
srq(sender, mid);
}
return true;
}
if (msg.match_elements<exit_msg>()) { if (msg.match_elements<exit_msg>()) {
// acquire second mutex as well
std::vector<actor> workers; std::vector<actor> workers;
// send exit messages *always* to all workers and clear vector afterwards auto tmp = msg.get_as<exit_msg>(0).reason;
// but first swap workers_ out of the critical section if (cleanup(std::move(tmp), eu)) {
upgrade_to_unique_lock<detail::shared_spinlock> unique_guard{guard}; // send exit messages *always* to all workers and clear vector afterwards
workers_.swap(workers); // but first swap workers_ out of the critical section
planned_reason_ = msg.get_as<exit_msg>(0).reason; upgrade_to_unique_lock<detail::shared_spinlock> unique_guard{guard};
unique_guard.unlock(); workers_.swap(workers);
for (auto& w : workers) unique_guard.unlock();
anon_send(w, msg); for (auto& w : workers)
quit(eu); anon_send(w, msg);
quit(eu);
is_registered(false);
}
return true; return true;
} }
if (msg.match_elements<down_msg>()) { if (msg.match_elements<down_msg>()) {
......
...@@ -59,13 +59,13 @@ actor_registry::actor_registry(actor_system& sys) : running_(0), system_(sys) { ...@@ -59,13 +59,13 @@ actor_registry::actor_registry(actor_system& sys) : running_(0), system_(sys) {
// nop // nop
} }
actor_registry::id_entry actor_registry::get(actor_id key) const { strong_actor_ptr actor_registry::get(actor_id key) const {
shared_guard guard(instances_mtx_); shared_guard guard(instances_mtx_);
auto i = entries_.find(key); auto i = entries_.find(key);
if (i != entries_.end()) if (i != entries_.end())
return i->second; return i->second;
CAF_LOG_DEBUG("key invalid, assume actor no longer exists:" << CAF_ARG(key)); CAF_LOG_DEBUG("key invalid, assume actor no longer exists:" << CAF_ARG(key));
return {nullptr, exit_reason::unknown}; return nullptr;
} }
void actor_registry::put(actor_id key, actor_control_block* val) { void actor_registry::put(actor_id key, actor_control_block* val) {
...@@ -76,18 +76,14 @@ void actor_registry::put(actor_id key, actor_control_block* val) { ...@@ -76,18 +76,14 @@ void actor_registry::put(actor_id key, actor_control_block* val) {
return; return;
{ // lifetime scope of guard { // lifetime scope of guard
exclusive_guard guard(instances_mtx_); exclusive_guard guard(instances_mtx_);
if (! entries_.emplace(key, if (! entries_.emplace(key, actor_cast<strong_actor_ptr>(val)).second)
id_entry{actor_cast<strong_actor_ptr>(val),
exit_reason::not_exited}).second) {
// already defined
return; return;
}
} }
// attach functor without lock // attach functor without lock
CAF_LOG_INFO("added actor:" << CAF_ARG(key)); CAF_LOG_INFO("added actor:" << CAF_ARG(key));
actor_registry* reg = this; actor_registry* reg = this;
strong_val->attach_functor([key, reg](exit_reason reason) { strong_val->attach_functor([key, reg]() {
reg->erase(key, reason); reg->erase(key);
}); });
} }
...@@ -95,15 +91,9 @@ void actor_registry::put(actor_id key, const strong_actor_ptr& val) { ...@@ -95,15 +91,9 @@ void actor_registry::put(actor_id key, const strong_actor_ptr& val) {
put(key, actor_cast<actor_control_block*>(val)); put(key, actor_cast<actor_control_block*>(val));
} }
void actor_registry::erase(actor_id key, exit_reason reason) { void actor_registry::erase(actor_id key) {
exclusive_guard guard(instances_mtx_); exclusive_guard guard(instances_mtx_);
auto i = entries_.find(key); entries_.erase(key);
if (i != entries_.end()) {
auto& entry = i->second;
CAF_LOG_INFO("erased actor:" << CAF_ARG(key) << CAF_ARG(reason));
entry.first = nullptr;
entry.second = reason;
}
} }
void actor_registry::inc_running() { void actor_registry::inc_running() {
......
...@@ -44,38 +44,34 @@ adapter::adapter(strong_actor_ptr decorated, message msg) ...@@ -44,38 +44,34 @@ adapter::adapter(strong_actor_ptr decorated, message msg)
default_attachable::make_monitor(decorated_->get()->address(), address())); default_attachable::make_monitor(decorated_->get()->address(), address()));
} }
void adapter::enqueue(mailbox_element_ptr what, execution_unit* host) { void adapter::enqueue(mailbox_element_ptr what, execution_unit* context) {
if (! what) CAF_ASSERT(what);
return; // not even an empty message auto down_msg_handler = [&](down_msg& dm) {
auto reason = get_exit_reason(); cleanup(std::move(dm.reason), context);
if (reason != exit_reason::not_exited) {
// actor has exited
auto& mid = what->mid;
if (mid.is_request()) {
// make sure that a request always gets a response;
// the exit reason reflects the first actor on the
// forwarding chain that is out of service
detail::sync_request_bouncer rb{reason};
rb(what->sender, mid);
}
return;
}
auto down_handler = [&](const down_msg& dm) {
if (dm.source == decorated_)
cleanup(dm.reason, host);
}; };
// handle and consume the system message; if (handle_system_message(*what, context, false, down_msg_handler))
// the only effect that MAY result from handling a system message return;
// is to exit the actor if it hasn't exited already; strong_actor_ptr decorated;
// `handle_system_message()` is thread-safe, and if the actor message merger;
// has already exited upon the invocation, nothing is done error fail_state;
if (handle_system_message(*what, host, false, down_handler)) shared_critical_section([&] {
decorated = decorated_;
merger = merger_;
fail_state = fail_state_;
});
if (! decorated) {
bounce(what, fail_state);
return; return;
// process and forward non-system messages }
auto& msg = what->msg; auto& msg = what->msg;
message tmp{detail::merged_tuple::make(merger_, std::move(msg))}; message tmp{detail::merged_tuple::make(merger, std::move(msg))};
msg.swap(tmp); msg.swap(tmp);
decorated_->enqueue(std::move(what), host); decorated->enqueue(std::move(what), context);
}
void adapter::on_cleanup() {
decorated_.reset();
merger_.reset();
} }
} // namespace decorator } // namespace decorator
......
...@@ -34,7 +34,7 @@ optional<exit_reason> attachable::handle_exception(const std::exception_ptr&) { ...@@ -34,7 +34,7 @@ optional<exit_reason> attachable::handle_exception(const std::exception_ptr&) {
return none; return none;
} }
void attachable::actor_exited(exit_reason, execution_unit*) { void attachable::actor_exited(const error&, execution_unit*) {
// nop // nop
} }
......
...@@ -29,13 +29,13 @@ namespace caf { ...@@ -29,13 +29,13 @@ namespace caf {
namespace { namespace {
template <class MsgType> template <class MsgType>
message make(abstract_actor* self, exit_reason reason) { message make(abstract_actor* self, const error& reason) {
return make_message(MsgType{self->address(), reason}); return make_message(MsgType{self->address(), reason});
} }
} // namespace <anonymous> } // namespace <anonymous>
void default_attachable::actor_exited(exit_reason rsn, execution_unit* host) { void default_attachable::actor_exited(const error& rsn, execution_unit* host) {
CAF_ASSERT(observed_ != observer_); CAF_ASSERT(observed_ != observer_);
auto factory = type_ == monitor ? &make<down_msg> : &make<exit_msg>; auto factory = type_ == monitor ? &make<down_msg> : &make<exit_msg>;
auto observer = actor_cast<strong_actor_ptr>(observer_); auto observer = actor_cast<strong_actor_ptr>(observer_);
......
...@@ -52,10 +52,6 @@ const message& error::context() const { ...@@ -52,10 +52,6 @@ const message& error::context() const {
return context_; return context_;
} }
error::operator bool() const {
return code_ != 0;
}
void error::clear() { void error::clear() {
code_ = 0; code_ = 0;
} }
...@@ -98,6 +94,9 @@ void serialize(deserializer& source, error& x, const unsigned int) { ...@@ -98,6 +94,9 @@ void serialize(deserializer& source, error& x, const unsigned int) {
} }
int error::compare(uint8_t x, atom_value y) const { int error::compare(uint8_t x, atom_value y) const {
// exception: all errors with default value are considered no error -> equal
if (code_ == 0 && x == 0)
return 0;
if (category_ < y) if (category_ < y)
return -1; return -1;
if (category_ > y) if (category_ > y)
...@@ -110,6 +109,8 @@ int error::compare(const error& x) const { ...@@ -110,6 +109,8 @@ int error::compare(const error& x) const {
} }
std::string to_string(const error& x) { std::string to_string(const error& x) {
if (! x)
return "no-error";
std::string result = "error("; std::string result = "error(";
result += to_string(x.category()); result += to_string(x.category());
result += ", "; result += ", ";
......
...@@ -33,9 +33,9 @@ ...@@ -33,9 +33,9 @@
namespace { namespace {
std::string ae_what(caf::exit_reason reason) { std::string ae_what(caf::error reason) {
std::ostringstream oss; std::ostringstream oss;
oss << "actor exited with reason " << to_string(reason); oss << "actor exited with reason: " << to_string(reason);
return oss.str(); return oss.str();
} }
...@@ -59,7 +59,7 @@ actor_exited::~actor_exited() noexcept { ...@@ -59,7 +59,7 @@ actor_exited::~actor_exited() noexcept {
// nop // nop
} }
actor_exited::actor_exited(exit_reason x) : caf_exception(ae_what(x)) { actor_exited::actor_exited(error x) : caf_exception(ae_what(x)) {
reason_ = x; reason_ = x;
} }
......
...@@ -23,8 +23,6 @@ namespace caf { ...@@ -23,8 +23,6 @@ namespace caf {
const char* to_string(exit_reason x) { const char* to_string(exit_reason x) {
switch (x) { switch (x) {
case exit_reason::not_exited:
return "not_exited";
case exit_reason::normal: case exit_reason::normal:
return "normal"; return "normal";
case exit_reason::unhandled_exception: case exit_reason::unhandled_exception:
...@@ -41,9 +39,19 @@ const char* to_string(exit_reason x) { ...@@ -41,9 +39,19 @@ const char* to_string(exit_reason x) {
return "kill"; return "kill";
case exit_reason::remote_link_unreachable: case exit_reason::remote_link_unreachable:
return "remote_link_unreachable"; return "remote_link_unreachable";
case exit_reason::unhandled_error:
return "unhandled_error";
default: default:
return "-invalid-"; return "-invalid-";
} }
} }
error make_error(exit_reason x) {
return {static_cast<uint8_t>(x), atom("exit")};
}
error make_error(exit_reason x, message context) {
return {static_cast<uint8_t>(x), atom("exit"), std::move(context)};
}
} // namespace caf } // namespace caf
...@@ -69,22 +69,21 @@ void forwarding_actor_proxy::forward_msg(strong_actor_ptr sender, ...@@ -69,22 +69,21 @@ void forwarding_actor_proxy::forward_msg(strong_actor_ptr sender,
void forwarding_actor_proxy::enqueue(mailbox_element_ptr what, void forwarding_actor_proxy::enqueue(mailbox_element_ptr what,
execution_unit*) { execution_unit*) {
CAF_PUSH_AID(0); CAF_PUSH_AID(0);
if (! what) CAF_ASSERT(what);
return;
forward_msg(std::move(what->sender), what->mid, forward_msg(std::move(what->sender), what->mid,
std::move(what->msg), &what->stages); std::move(what->msg), &what->stages);
} }
bool forwarding_actor_proxy::link_impl(linking_operation op, bool forwarding_actor_proxy::link_impl(linking_operation op,
const actor_addr& other) { abstract_actor* other) {
switch (op) { switch (op) {
case establish_link_op: case establish_link_op:
if (establish_link_impl(other)) { if (establish_link_impl(other)) {
// causes remote actor to link to (proxy of) other // causes remote actor to link to (proxy of) other
// receiving peer will call: this->local_link_to(other) // receiving peer will call: this->local_link_to(other)
forward_msg(ctrl(), invalid_message_id, forward_msg(ctrl(), invalid_message_id,
make_message(link_atom::value, other)); make_message(link_atom::value, other->address()));
return true; return true;
} }
break; break;
...@@ -92,7 +91,7 @@ bool forwarding_actor_proxy::link_impl(linking_operation op, ...@@ -92,7 +91,7 @@ bool forwarding_actor_proxy::link_impl(linking_operation op,
if (remove_link_impl(other)) { if (remove_link_impl(other)) {
// causes remote actor to unlink from (proxy of) other // causes remote actor to unlink from (proxy of) other
forward_msg(ctrl(), invalid_message_id, forward_msg(ctrl(), invalid_message_id,
make_message(unlink_atom::value, other)); make_message(unlink_atom::value, other->address()));
return true; return true;
} }
break; break;
...@@ -100,7 +99,7 @@ bool forwarding_actor_proxy::link_impl(linking_operation op, ...@@ -100,7 +99,7 @@ bool forwarding_actor_proxy::link_impl(linking_operation op,
if (establish_backlink_impl(other)) { if (establish_backlink_impl(other)) {
// causes remote actor to unlink from (proxy of) other // causes remote actor to unlink from (proxy of) other
forward_msg(ctrl(), invalid_message_id, forward_msg(ctrl(), invalid_message_id,
make_message(link_atom::value, other)); make_message(link_atom::value, other->address()));
return true; return true;
} }
break; break;
...@@ -108,7 +107,7 @@ bool forwarding_actor_proxy::link_impl(linking_operation op, ...@@ -108,7 +107,7 @@ bool forwarding_actor_proxy::link_impl(linking_operation op,
if (remove_backlink_impl(other)) { if (remove_backlink_impl(other)) {
// causes remote actor to unlink from (proxy of) other // causes remote actor to unlink from (proxy of) other
forward_msg(ctrl(), invalid_message_id, forward_msg(ctrl(), invalid_message_id,
make_message(unlink_atom::value, other)); make_message(unlink_atom::value, other->address()));
return true; return true;
} }
break; break;
...@@ -116,18 +115,18 @@ bool forwarding_actor_proxy::link_impl(linking_operation op, ...@@ -116,18 +115,18 @@ bool forwarding_actor_proxy::link_impl(linking_operation op,
return false; return false;
} }
void forwarding_actor_proxy::local_link_to(const actor_addr& other) { void forwarding_actor_proxy::local_link_to(abstract_actor* other) {
establish_link_impl(other); establish_link_impl(other);
} }
void forwarding_actor_proxy::local_unlink_from(const actor_addr& other) { void forwarding_actor_proxy::local_unlink_from(abstract_actor* other) {
remove_link_impl(other); remove_link_impl(other);
} }
void forwarding_actor_proxy::kill_proxy(execution_unit* ctx, exit_reason rsn) { void forwarding_actor_proxy::kill_proxy(execution_unit* ctx, error rsn) {
CAF_ASSERT(ctx != nullptr); CAF_ASSERT(ctx != nullptr);
manager(invalid_actor); manager(invalid_actor);
cleanup(rsn, ctx); cleanup(std::move(rsn), ctx);
} }
} // namespace caf } // namespace caf
...@@ -66,7 +66,6 @@ result<message> drop(local_actor*, const type_erased_tuple*) { ...@@ -66,7 +66,6 @@ result<message> drop(local_actor*, const type_erased_tuple*) {
local_actor::local_actor(actor_config& cfg) local_actor::local_actor(actor_config& cfg)
: monitorable_actor(cfg), : monitorable_actor(cfg),
context_(cfg.host), context_(cfg.host),
planned_exit_reason_(exit_reason::not_exited),
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) { unexpected_handler_(print_and_drop) {
...@@ -80,12 +79,12 @@ local_actor::~local_actor() { ...@@ -80,12 +79,12 @@ local_actor::~local_actor() {
} }
void local_actor::destroy() { void local_actor::destroy() {
CAF_LOG_TRACE(CAF_ARG(planned_exit_reason())); CAF_LOG_TRACE(CAF_ARG(is_terminated()));
if (planned_exit_reason() == exit_reason::not_exited) { if (! is_cleaned_up()) {
on_exit(); on_exit();
cleanup(exit_reason::unreachable, nullptr); cleanup(exit_reason::unreachable, nullptr);
monitorable_actor::destroy();
} }
monitorable_actor::destroy();
} }
void local_actor::intrusive_ptr_add_ref_impl() { void local_actor::intrusive_ptr_add_ref_impl() {
...@@ -290,7 +289,7 @@ msg_type filter_msg(local_actor* self, mailbox_element& node) { ...@@ -290,7 +289,7 @@ msg_type filter_msg(local_actor* self, mailbox_element& node) {
node.sender->enqueue( node.sender->enqueue(
mailbox_element::make_joint(self->ctrl(), mailbox_element::make_joint(self->ctrl(),
node.mid.response_id(), node.mid.response_id(),
{}, sec::invalid_sys_key), {}, sec::unsupported_sys_key),
self->context()); self->context());
} }
}); });
...@@ -663,8 +662,8 @@ void local_actor::launch(execution_unit* eu, bool lazy, bool hide) { ...@@ -663,8 +662,8 @@ 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_ASSERT(ptr != nullptr);
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()) { if (is_detached()) {
// actor lives in its own thread // actor lives in its own thread
...@@ -719,7 +718,7 @@ resumable::resume_result local_actor::resume(execution_unit* eu, ...@@ -719,7 +718,7 @@ resumable::resume_result local_actor::resume(execution_unit* eu,
// actor lives in its own thread // actor lives in its own thread
CAF_ASSERT(dynamic_cast<blocking_actor*>(this) != 0); CAF_ASSERT(dynamic_cast<blocking_actor*>(this) != 0);
auto self = static_cast<blocking_actor*>(this); auto self = static_cast<blocking_actor*>(this);
auto rsn = exit_reason::normal; error rsn;
std::exception_ptr eptr = nullptr; std::exception_ptr eptr = nullptr;
try { try {
self->act(); self->act();
...@@ -736,7 +735,6 @@ resumable::resume_result local_actor::resume(execution_unit* eu, ...@@ -736,7 +735,6 @@ resumable::resume_result local_actor::resume(execution_unit* eu,
rsn = opt_reason ? *opt_reason rsn = opt_reason ? *opt_reason
: exit_reason::unhandled_exception; : exit_reason::unhandled_exception;
} }
self->planned_exit_reason(rsn);
try { try {
self->on_exit(); self->on_exit();
} }
...@@ -744,16 +742,14 @@ resumable::resume_result local_actor::resume(execution_unit* eu, ...@@ -744,16 +742,14 @@ resumable::resume_result local_actor::resume(execution_unit* eu,
// simply ignore exception // simply ignore exception
} }
// exit reason might have been changed by on_exit() // exit reason might have been changed by on_exit()
self->cleanup(self->planned_exit_reason(), context()); self->cleanup(std::move(rsn), context());
return resumable::done; return resumable::done;
} }
if (is_initialized() if (is_initialized() && (! has_behavior() || is_terminated())) {
&& (! has_behavior()
|| planned_exit_reason() != exit_reason::not_exited)) {
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");
CAF_LOG_DEBUG_IF(planned_exit_reason() != exit_reason::not_exited, CAF_LOG_DEBUG_IF(is_terminated(),
"resume called on an actor with exit reason"); "resume called on a terminated actor");
return resumable::done; return resumable::done;
} }
std::exception_ptr eptr = nullptr; std::exception_ptr eptr = nullptr;
...@@ -799,30 +795,27 @@ resumable::resume_result local_actor::resume(execution_unit* eu, ...@@ -799,30 +795,27 @@ resumable::resume_result local_actor::resume(execution_unit* eu,
} }
catch (actor_exited& what) { catch (actor_exited& what) {
CAF_LOG_INFO("actor died because of exception:" << CAF_ARG(what.reason())); CAF_LOG_INFO("actor died because of exception:" << CAF_ARG(what.reason()));
if (exit_reason() == exit_reason::not_exited) { if (! is_terminated())
quit(what.reason()); quit(what.reason());
}
} }
catch (std::exception& e) { catch (std::exception& e) {
CAF_LOG_INFO("actor died because of an exception, what: " << e.what()); CAF_LOG_INFO("actor died because of an exception, what: " << e.what());
static_cast<void>(e); // keep compiler happy when not logging static_cast<void>(e); // keep compiler happy when not logging
if (exit_reason() == exit_reason::not_exited) { if (! is_terminated())
quit(exit_reason::unhandled_exception); quit(exit_reason::unhandled_exception);
}
eptr = std::current_exception(); eptr = std::current_exception();
} }
catch (...) { catch (...) {
CAF_LOG_INFO("actor died because of an unknown exception"); CAF_LOG_INFO("actor died because of an unknown exception");
if (exit_reason() == exit_reason::not_exited) { if (! is_terminated())
quit(exit_reason::unhandled_exception); quit(exit_reason::unhandled_exception);
}
eptr = std::current_exception(); eptr = std::current_exception();
} }
if (eptr) { if (eptr) {
auto opt_reason = handle(eptr); auto opt_reason = handle(eptr);
if (opt_reason) { if (opt_reason) {
// use exit reason defined by custom handler // use exit reason defined by custom handler
planned_exit_reason(*opt_reason); quit(*opt_reason);
} }
} }
if (! finished()) { if (! finished()) {
...@@ -882,11 +875,11 @@ void local_actor::exec_single_event(execution_unit* ctx, ...@@ -882,11 +875,11 @@ void local_actor::exec_single_event(execution_unit* ctx,
return; return;
} }
} }
if (! has_behavior() || planned_exit_reason() != exit_reason::not_exited) { if (! 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");
CAF_LOG_DEBUG_IF(planned_exit_reason() != exit_reason::not_exited, CAF_LOG_DEBUG_IF(is_terminated(),
"resume called on an actor with exit reason"); "resume called on a terminated actor");
return; return;
} }
try { try {
...@@ -897,8 +890,7 @@ void local_actor::exec_single_event(execution_unit* ctx, ...@@ -897,8 +890,7 @@ void local_actor::exec_single_event(execution_unit* ctx,
auto eptr = std::current_exception(); auto eptr = std::current_exception();
auto opt_reason = this->handle(eptr); auto opt_reason = this->handle(eptr);
if (opt_reason) if (opt_reason)
planned_exit_reason(*opt_reason); quit(*opt_reason);
//finalize();
} }
} }
...@@ -1022,26 +1014,21 @@ void local_actor::load_state(deserializer&, const unsigned int) { ...@@ -1022,26 +1014,21 @@ void local_actor::load_state(deserializer&, const unsigned int) {
} }
bool local_actor::finished() { bool local_actor::finished() {
if (has_behavior() && planned_exit_reason() == exit_reason::not_exited) if (has_behavior() && ! is_terminated())
return false; return false;
CAF_LOG_DEBUG("actor either has no behavior or has set an exit reason"); CAF_LOG_DEBUG("actor either has no behavior or has set an exit reason");
on_exit(); on_exit();
bhvr_stack().clear(); bhvr_stack().clear();
bhvr_stack().cleanup(); bhvr_stack().cleanup();
auto rsn = planned_exit_reason(); cleanup(std::move(fail_state_), context());
if (rsn == exit_reason::not_exited) {
rsn = exit_reason::normal;
planned_exit_reason(rsn);
}
cleanup(rsn, context());
return true; return true;
} }
void local_actor::cleanup(exit_reason reason, execution_unit* host) { bool local_actor::cleanup(error&& fail_state, execution_unit* host) {
CAF_LOG_TRACE(CAF_ARG(reason)); CAF_LOG_TRACE(CAF_ARG(fail_state));
current_mailbox_element().reset(); current_mailbox_element().reset();
if (! mailbox_.closed()) { if (! mailbox_.closed()) {
detail::sync_request_bouncer f{reason}; detail::sync_request_bouncer f{fail_state};
mailbox_.close(f); mailbox_.close(f);
} }
awaited_responses_.clear(); awaited_responses_.clear();
...@@ -1050,16 +1037,18 @@ void local_actor::cleanup(exit_reason reason, execution_unit* host) { ...@@ -1050,16 +1037,18 @@ void local_actor::cleanup(exit_reason reason, execution_unit* host) {
for (auto& subscription : subscriptions_) for (auto& subscription : subscriptions_)
subscription->unsubscribe(me); subscription->unsubscribe(me);
subscriptions_.clear(); subscriptions_.clear();
monitorable_actor::cleanup(reason, host); monitorable_actor::cleanup(std::move(fail_state), host);
// tell registry we're done // tell registry we're done
is_registered(false); is_registered(false);
return true;
} }
void local_actor::quit(exit_reason reason) { void local_actor::quit(error x) {
CAF_LOG_TRACE(CAF_ARG(reason)); CAF_LOG_TRACE(CAF_ARG(x));
planned_exit_reason(reason); fail_state_ = std::move(x);
is_terminated(true);
if (is_blocking()) { if (is_blocking()) {
throw actor_exited(reason); throw actor_exited(fail_state_);
} }
} }
......
...@@ -27,6 +27,8 @@ ...@@ -27,6 +27,8 @@
#include "caf/system_messages.hpp" #include "caf/system_messages.hpp"
#include "caf/default_attachable.hpp" #include "caf/default_attachable.hpp"
#include "caf/detail/sync_request_bouncer.hpp"
#include "caf/scheduler/abstract_coordinator.hpp" #include "caf/scheduler/abstract_coordinator.hpp"
namespace caf { namespace caf {
...@@ -37,19 +39,19 @@ const char* monitorable_actor::name() const { ...@@ -37,19 +39,19 @@ const char* monitorable_actor::name() const {
void monitorable_actor::attach(attachable_ptr ptr) { void monitorable_actor::attach(attachable_ptr ptr) {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
if (ptr == nullptr) CAF_ASSERT(ptr);
return; error fail_state;
caf::exit_reason reason; auto attached = exclusive_critical_section([&] {
{ // lifetime scope of guard if (is_terminated()) {
std::unique_lock<std::mutex> guard{mtx_}; fail_state = fail_state_;
reason = get_exit_reason(); return false;
if (reason == exit_reason::not_exited) {
attach_impl(ptr);
return;
} }
} attach_impl(ptr);
return true;
});
CAF_LOG_DEBUG("cannot attach functor to terminated actor: call immediately"); CAF_LOG_DEBUG("cannot attach functor to terminated actor: call immediately");
ptr->actor_exited(reason, nullptr); if (! attached)
ptr->actor_exited(fail_state, nullptr);
} }
size_t monitorable_actor::detach(const attachable::token& what) { size_t monitorable_actor::detach(const attachable::token& what) {
...@@ -58,18 +60,23 @@ size_t monitorable_actor::detach(const attachable::token& what) { ...@@ -58,18 +60,23 @@ size_t monitorable_actor::detach(const attachable::token& what) {
return detach_impl(what, attachables_head_); return detach_impl(what, attachables_head_);
} }
void monitorable_actor::cleanup(exit_reason reason, execution_unit* host) { bool monitorable_actor::cleanup(error&& reason, execution_unit* host) {
CAF_LOG_TRACE(CAF_ARG(reason)); CAF_LOG_TRACE(CAF_ARG(reason));
CAF_ASSERT(reason != exit_reason::not_exited);
// move everything out of the critical section before processing it
attachable_ptr head; attachable_ptr head;
{ // lifetime scope of guard bool set_fail_state = exclusive_critical_section([&]() -> bool {
std::unique_lock<std::mutex> guard{mtx_}; if (! is_cleaned_up()) {
if (get_exit_reason() != exit_reason::not_exited) // local actors pass fail_state_ as first argument
return; if (&fail_state_ != &reason)
exit_reason_.store(reason, std::memory_order_relaxed); fail_state_ = std::move(reason);
attachables_head_.swap(head); attachables_head_.swap(head);
} flags(flags() | is_terminated_flag | is_cleaned_up_flag);
on_cleanup();
return true;
}
return false;
});
if (! set_fail_state)
return false;
CAF_LOG_INFO_IF(host && host->system().node() == node(), CAF_LOG_INFO_IF(host && host->system().node() == node(),
"cleanup" << CAF_ARG(id()) << CAF_ARG(reason)); "cleanup" << CAF_ARG(id()) << CAF_ARG(reason));
// send exit messages // send exit messages
...@@ -82,11 +89,30 @@ void monitorable_actor::cleanup(exit_reason reason, execution_unit* host) { ...@@ -82,11 +89,30 @@ void monitorable_actor::cleanup(exit_reason reason, execution_unit* host) {
delete_atom::value, id()), delete_atom::value, id()),
nullptr); nullptr);
} }
return true;
} }
monitorable_actor::monitorable_actor(actor_config& cfg) void monitorable_actor::on_cleanup() {
: abstract_actor(cfg), // nop
exit_reason_(exit_reason::not_exited) { }
void monitorable_actor::bounce(mailbox_element_ptr& what) {
error err;
shared_critical_section([&] {
err = fail_state_;
});
bounce(what, err);
}
void monitorable_actor::bounce(mailbox_element_ptr& what, const error& err) {
// make sure that a request always gets a response;
// the exit reason reflects the first actor on the
// forwarding chain that is out of service
detail::sync_request_bouncer rb{err};
rb(*what);
}
monitorable_actor::monitorable_actor(actor_config& cfg) : abstract_actor(cfg) {
// nop // nop
} }
...@@ -105,7 +131,7 @@ optional<exit_reason> monitorable_actor::handle(const std::exception_ptr& eptr) ...@@ -105,7 +131,7 @@ optional<exit_reason> monitorable_actor::handle(const std::exception_ptr& eptr)
return none; return none;
} }
bool monitorable_actor::link_impl(linking_operation op, const actor_addr& other) { bool monitorable_actor::link_impl(linking_operation op, abstract_actor* other) {
CAF_LOG_TRACE(CAF_ARG(op) << CAF_ARG(other)); CAF_LOG_TRACE(CAF_ARG(op) << CAF_ARG(other));
switch (op) { switch (op) {
case establish_link_op: case establish_link_op:
...@@ -120,82 +146,75 @@ bool monitorable_actor::link_impl(linking_operation op, const actor_addr& other) ...@@ -120,82 +146,75 @@ bool monitorable_actor::link_impl(linking_operation op, const actor_addr& other)
} }
} }
bool monitorable_actor::establish_link_impl(const actor_addr& other) { bool monitorable_actor::establish_link_impl(abstract_actor* x) {
CAF_LOG_TRACE(CAF_ARG(other)); CAF_LOG_TRACE(CAF_ARG(x));
if (other && other != this) { CAF_ASSERT(x);
std::unique_lock<std::mutex> guard{mtx_}; error fail_state;
auto ptr = actor_cast<actor>(other); bool send_exit_immediately = false;
if (! ptr) auto tmp = default_attachable::make_link(address(), x->address());
auto success = exclusive_critical_section([&]() -> bool {
if (is_terminated()) {
fail_state = fail_state_;
send_exit_immediately = true;
return false; return false;
auto reason = get_exit_reason(); }
// send exit message if already exited // add link if not already linked to other (checked by establish_backlink)
if (reason != exit_reason::not_exited) { if (x->establish_backlink(this)) {
guard.unlock();
ptr->enqueue(ctrl(), invalid_message_id,
make_message(exit_msg{address(), reason}), nullptr);
} else if (ptr->establish_backlink(address())) {
// add link if not already linked to other
// (checked by establish_backlink)
auto tmp = default_attachable::make_link(address(), other);
attach_impl(tmp); attach_impl(tmp);
return true; return true;
} }
} return false;
return false; });
if (send_exit_immediately)
x->enqueue(nullptr, invalid_message_id,
make_message(exit_msg{address(), fail_state}), nullptr);
return success;
} }
bool monitorable_actor::establish_backlink_impl(const actor_addr& other) { bool monitorable_actor::establish_backlink_impl(abstract_actor* x) {
CAF_LOG_TRACE(CAF_ARG(other)); CAF_LOG_TRACE(CAF_ARG(x));
auto reason = exit_reason::not_exited; CAF_ASSERT(x);
default_attachable::observe_token tk{other, default_attachable::link}; error fail_state;
if (other && other != this) { bool send_exit_immediately = false;
std::unique_lock<std::mutex> guard{mtx_}; default_attachable::observe_token tk{x->address(),
reason = get_exit_reason(); default_attachable::link};
if (reason == exit_reason::not_exited) { auto tmp = default_attachable::make_link(address(), x->address());
if (detach_impl(tk, attachables_head_, true, true) == 0) { auto success = exclusive_critical_section([&]() -> bool {
auto tmp = default_attachable::make_link(address(), other); if (is_terminated()) {
attach_impl(tmp); fail_state = fail_state_;
return true; send_exit_immediately = true;
} return false;
} }
} if (detach_impl(tk, attachables_head_, true, true) == 0) {
// send exit message without lock attach_impl(tmp);
if (reason != exit_reason::not_exited) { return true;
auto hdl = actor_cast<actor>(other); }
if (hdl) return false;
hdl->enqueue(ctrl(), invalid_message_id, });
make_message(exit_msg{address(), reason}), nullptr); if (send_exit_immediately)
} x->enqueue(nullptr, invalid_message_id,
return false; make_message(exit_msg{address(), fail_state}), nullptr);
return success;
} }
bool monitorable_actor::remove_link_impl(const actor_addr& other) { bool monitorable_actor::remove_link_impl(abstract_actor* x) {
CAF_LOG_TRACE(CAF_ARG(other)); CAF_LOG_TRACE(CAF_ARG(x));
if (other == invalid_actor_addr || other == this) default_attachable::observe_token tk{x->address(), default_attachable::link};
return false; auto success = exclusive_critical_section([&]() -> bool {
default_attachable::observe_token tk{other, default_attachable::link}; return detach_impl(tk, attachables_head_, true) > 0;
std::unique_lock<std::mutex> guard{mtx_}; });
// remove_backlink returns true if this actor is linked to other if (success)
auto hdl = actor_cast<actor>(other); x->remove_backlink(this);
if (! hdl) return success;
return false;
if (detach_impl(tk, attachables_head_, true) > 0) {
guard.unlock();
// tell remote side to remove link as well
hdl->remove_backlink(address());
return true;
}
return false;
} }
bool monitorable_actor::remove_backlink_impl(const actor_addr& other) { bool monitorable_actor::remove_backlink_impl(abstract_actor* x) {
CAF_LOG_TRACE(CAF_ARG(other)); CAF_LOG_TRACE(CAF_ARG(x));
default_attachable::observe_token tk{other, default_attachable::link}; default_attachable::observe_token tk{x->address(), default_attachable::link};
if (other && other != this) { auto success = exclusive_critical_section([&]() -> bool {
std::unique_lock<std::mutex> guard{mtx_};
return detach_impl(tk, attachables_head_, true) > 0; return detach_impl(tk, attachables_head_, true) > 0;
} });
return false; return success;
} }
size_t monitorable_actor::detach_impl(const attachable::token& what, size_t monitorable_actor::detach_impl(const attachable::token& what,
...@@ -224,10 +243,9 @@ bool monitorable_actor::handle_system_message(mailbox_element& node, ...@@ -224,10 +243,9 @@ bool monitorable_actor::handle_system_message(mailbox_element& node,
auto& msg = node.msg; auto& msg = node.msg;
if (! trap_exit && msg.size() == 1 && msg.match_element<exit_msg>(0)) { if (! trap_exit && msg.size() == 1 && msg.match_element<exit_msg>(0)) {
// exits for non-normal exit reasons // exits for non-normal exit reasons
auto& em = msg.get_as<exit_msg>(0); auto& em = msg.get_as_mutable<exit_msg>(0);
CAF_ASSERT(em.reason != exit_reason::not_exited); if (em.reason)
if (em.reason != exit_reason::normal) cleanup(std::move(em.reason), context);
cleanup(em.reason, context);
return true; return true;
} else if (msg.size() > 1 && msg.match_element<sys_atom>(0)) { } else if (msg.size() > 1 && msg.match_element<sys_atom>(0)) {
if (! node.sender) if (! node.sender)
...@@ -238,7 +256,7 @@ bool monitorable_actor::handle_system_message(mailbox_element& node, ...@@ -238,7 +256,7 @@ bool monitorable_actor::handle_system_message(mailbox_element& node,
[&](sys_atom, get_atom, std::string& what) { [&](sys_atom, get_atom, std::string& what) {
CAF_LOG_TRACE(CAF_ARG(what)); CAF_LOG_TRACE(CAF_ARG(what));
if (what != "info") { if (what != "info") {
err = sec::invalid_sys_key; err = sec::unsupported_sys_key;
return; return;
} }
res = mailbox_element::make_joint(ctrl(), res = mailbox_element::make_joint(ctrl(),
......
...@@ -90,7 +90,7 @@ void proxy_registry::erase(const key_type& nid) { ...@@ -90,7 +90,7 @@ void proxy_registry::erase(const key_type& nid) {
proxies_.erase(i); proxies_.erase(i);
} }
void proxy_registry::erase(const key_type& inf, actor_id aid, exit_reason rsn) { void proxy_registry::erase(const key_type& inf, actor_id aid, error rsn) {
CAF_LOG_TRACE(CAF_ARG(inf) << CAF_ARG(aid)); CAF_LOG_TRACE(CAF_ARG(inf) << CAF_ARG(aid));
auto i = proxies_.find(inf); auto i = proxies_.find(inf);
if (i != proxies_.end()) { if (i != proxies_.end()) {
...@@ -112,7 +112,7 @@ void proxy_registry::clear() { ...@@ -112,7 +112,7 @@ void proxy_registry::clear() {
proxies_.clear(); proxies_.clear();
} }
void proxy_registry::kill_proxy(strong_actor_ptr& ptr, exit_reason rsn) { void proxy_registry::kill_proxy(strong_actor_ptr& ptr, error rsn) {
if (! ptr) if (! ptr)
return; return;
auto pptr = static_cast<actor_proxy*>(actor_cast<abstract_actor*>(ptr)); auto pptr = static_cast<actor_proxy*>(actor_cast<abstract_actor*>(ptr));
......
...@@ -58,12 +58,11 @@ scoped_actor::~scoped_actor() { ...@@ -58,12 +58,11 @@ scoped_actor::~scoped_actor() {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
if (! self_) if (! self_)
return; return;
if (ptr()->is_registered()) { auto x = ptr();
if (x->is_registered())
CAF_SET_AID(prev_); CAF_SET_AID(prev_);
} if (! x->is_terminated())
auto r = ptr()->planned_exit_reason(); x->cleanup(exit_reason::normal, &context_);
ptr()->cleanup(r == exit_reason::not_exited ? exit_reason::normal : r,
&context_);
} }
blocking_actor* scoped_actor::ptr() const { blocking_actor* scoped_actor::ptr() const {
......
...@@ -33,7 +33,7 @@ const char* sec_strings[] = { ...@@ -33,7 +33,7 @@ const char* sec_strings[] = {
"no_actor_to_unpublish", "no_actor_to_unpublish",
"no_actor_published_at_port", "no_actor_published_at_port",
"state_not_serializable", "state_not_serializable",
"invalid_sys_key", "unsupported_sys_key",
"unsupported_sys_message", "unsupported_sys_message",
"disconnect_during_handshake", "disconnect_during_handshake",
"cannot_forward_to_invalid_actor", "cannot_forward_to_invalid_actor",
......
...@@ -34,6 +34,8 @@ sequencer::sequencer(strong_actor_ptr f, strong_actor_ptr g, ...@@ -34,6 +34,8 @@ sequencer::sequencer(strong_actor_ptr f, strong_actor_ptr g,
f_(std::move(f)), f_(std::move(f)),
g_(std::move(g)), g_(std::move(g)),
msg_types_(std::move(msg_types)) { msg_types_(std::move(msg_types)) {
CAF_ASSERT(f_);
CAF_ASSERT(g_);
// composed actor has dependency on constituent actors by default; // composed actor has dependency on constituent actors by default;
// if either constituent actor is already dead upon establishing // if either constituent actor is already dead upon establishing
// the dependency, the actor is spawned dead // the dependency, the actor is spawned dead
...@@ -45,43 +47,40 @@ sequencer::sequencer(strong_actor_ptr f, strong_actor_ptr g, ...@@ -45,43 +47,40 @@ sequencer::sequencer(strong_actor_ptr f, strong_actor_ptr g,
} }
void sequencer::enqueue(mailbox_element_ptr what, execution_unit* context) { void sequencer::enqueue(mailbox_element_ptr what, execution_unit* context) {
if (! what) auto down_msg_handler = [&](down_msg& dm) {
return; // not even an empty message
auto reason = get_exit_reason();
if (reason != exit_reason::not_exited) {
// actor has exited
auto& mid = what->mid;
if (mid.is_request()) {
// make sure that a request always gets a response;
// the exit reason reflects the first actor on the
// forwarding chain that is out of service
detail::sync_request_bouncer rb{reason};
rb(what->sender, mid);
}
return;
}
auto down_msg_handler = [&](const down_msg& dm) {
// quit if either `f` or `g` are no longer available // quit if either `f` or `g` are no longer available
if (dm.source == f_ || dm.source == g_) cleanup(std::move(dm.reason), context);
monitorable_actor::cleanup(dm.reason, context);
}; };
// handle and consume the system message;
// the only effect that MAY result from handling a system message
// is to exit the actor if it hasn't exited already;
// `handle_system_message()` is thread-safe, and if the actor
// has already exited upon the invocation, nothing is done
if (handle_system_message(*what, context, false, down_msg_handler)) if (handle_system_message(*what, context, false, down_msg_handler))
return; return;
strong_actor_ptr f;
strong_actor_ptr g;
error err;
shared_critical_section([&] {
f = f_;
g = g_;
err = fail_state_;
});
if (! f) {
// f and g are invalid only after the sequencer terminated
bounce(what, err);
return;
}
// process and forward the non-system message; // process and forward the non-system message;
// store `f` as the next stage in the forwarding chain // store `f` as the next stage in the forwarding chain
what->stages.push_back(f_); what->stages.push_back(std::move(f));
// forward modified message to `g` // forward modified message to `g`
g_->enqueue(std::move(what), context); g->enqueue(std::move(what), context);
} }
sequencer::message_types_set sequencer::message_types() const { sequencer::message_types_set sequencer::message_types() const {
return msg_types_; return msg_types_;
} }
void sequencer::on_cleanup() {
f_.reset();
g_.reset();
}
} // namespace decorator } // namespace decorator
} // namespace caf } // namespace caf
...@@ -116,6 +116,7 @@ behavior fan_out_fan_in(stateful_actor<splitter_state>* self, ...@@ -116,6 +116,7 @@ behavior fan_out_fan_in(stateful_actor<splitter_state>* self,
splitter::splitter(std::vector<strong_actor_ptr> workers, splitter::splitter(std::vector<strong_actor_ptr> workers,
message_types_set msg_types) message_types_set msg_types)
: monitorable_actor(actor_config{}.add_flag(is_actor_dot_decorator_flag)), : monitorable_actor(actor_config{}.add_flag(is_actor_dot_decorator_flag)),
num_workers(workers.size()),
workers_(std::move(workers)), workers_(std::move(workers)),
msg_types_(std::move(msg_types)) { msg_types_(std::move(msg_types)) {
// composed actor has dependency on constituent actors by default; // composed actor has dependency on constituent actors by default;
...@@ -128,38 +129,24 @@ splitter::splitter(std::vector<strong_actor_ptr> workers, ...@@ -128,38 +129,24 @@ splitter::splitter(std::vector<strong_actor_ptr> workers,
} }
void splitter::enqueue(mailbox_element_ptr what, execution_unit* context) { void splitter::enqueue(mailbox_element_ptr what, execution_unit* context) {
if (! what) auto down_msg_handler = [&](down_msg& dm) {
return; // not even an empty message // quit if any worker fails
auto reason = get_exit_reason(); cleanup(std::move(dm.reason), context);
if (reason != exit_reason::not_exited) {
// actor has exited
auto& mid = what->mid;
if (mid.is_request()) {
// make sure that a request always gets a response;
// the exit reason reflects the first actor on the
// forwarding chain that is out of service
detail::sync_request_bouncer rb{reason};
rb(what->sender, mid);
}
return;
}
auto down_msg_handler = [&](const down_msg& dm) {
// quit if either `f` or `g` are no longer available
auto pred = [&](const strong_actor_ptr& x) {
return actor_addr::compare(
x.get(), actor_cast<actor_control_block*>(dm.source)) == 0;
};
if (std::any_of(workers_.begin(), workers_.end(), pred))
monitorable_actor::cleanup(dm.reason, context);
}; };
// handle and consume the system message;
// the only effect that MAY result from handling a system message
// is to exit the actor if it hasn't exited already;
// `handle_system_message()` is thread-safe, and if the actor
// has already exited upon the invocation, nothing is done
if (handle_system_message(*what, context, false, down_msg_handler)) if (handle_system_message(*what, context, false, down_msg_handler))
return; return;
auto helper = context->system().spawn(fan_out_fan_in, workers_); std::vector<strong_actor_ptr> workers;
workers.reserve(num_workers);
error fail_state;
shared_critical_section([&] {
workers = workers_;
fail_state = fail_state_;
});
if (workers.empty()) {
bounce(what, fail_state);
return;
}
auto helper = context->system().spawn(fan_out_fan_in, std::move(workers));
helper->enqueue(std::move(what), context); helper->enqueue(std::move(what), context);
} }
......
...@@ -32,20 +32,17 @@ ...@@ -32,20 +32,17 @@
namespace caf { namespace caf {
namespace detail { namespace detail {
sync_request_bouncer::sync_request_bouncer(exit_reason r) sync_request_bouncer::sync_request_bouncer(error r) : rsn(std::move(r)) {
: rsn(r == exit_reason::not_exited ? exit_reason::normal : r) {
// nop // nop
} }
void sync_request_bouncer::operator()(const strong_actor_ptr& sender, void sync_request_bouncer::operator()(const strong_actor_ptr& sender,
const message_id& mid) const { const message_id& mid) const {
CAF_ASSERT(rsn != exit_reason::not_exited); if (sender && mid.is_request())
if (sender && mid.is_request()) {
sender->enqueue(nullptr, mid.response_id(), sender->enqueue(nullptr, mid.response_id(),
make_message(make_error(sec::request_receiver_down)), make_message(make_error(sec::request_receiver_down)),
// TODO: this breaks out of the execution unit // TODO: this breaks out of the execution unit
nullptr); nullptr);
}
} }
void sync_request_bouncer::operator()(const mailbox_element& e) const { void sync_request_bouncer::operator()(const mailbox_element& e) const {
......
...@@ -43,7 +43,6 @@ public: ...@@ -43,7 +43,6 @@ public:
} }
~testee() { ~testee() {
printf("%s %d\n", __FILE__, __LINE__);
--s_testees; --s_testees;
} }
......
...@@ -40,7 +40,7 @@ behavior testee(event_based_actor* self) { ...@@ -40,7 +40,7 @@ behavior testee(event_based_actor* self) {
} }
struct fixture { struct fixture {
void wait_until_exited() { void wait_until_is_terminated() {
self->receive( self->receive(
[](const down_msg&) { [](const down_msg&) {
// nop // nop
...@@ -53,7 +53,7 @@ struct fixture { ...@@ -53,7 +53,7 @@ struct fixture {
auto ptr = actor_cast<abstract_actor*>(handle); auto ptr = actor_cast<abstract_actor*>(handle);
auto dptr = dynamic_cast<monitorable_actor*>(ptr); auto dptr = dynamic_cast<monitorable_actor*>(ptr);
CAF_REQUIRE(dptr != nullptr); CAF_REQUIRE(dptr != nullptr);
return dptr->exited(); return dptr->is_terminated();
} }
actor_system system; actor_system system;
...@@ -85,7 +85,7 @@ CAF_TEST(lifetime_1) { ...@@ -85,7 +85,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_exited(); wait_until_is_terminated();
auto bound = dbl.bind(1); auto bound = dbl.bind(1);
CAF_CHECK(exited(bound)); CAF_CHECK(exited(bound));
} }
...@@ -96,29 +96,7 @@ CAF_TEST(lifetime_2) { ...@@ -96,29 +96,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_exited(); wait_until_is_terminated();
}
// 1) ignores down message not from the decorated actor
// 2) exits by receiving an exit message
// 3) exit has no effect on decorated actor
CAF_TEST(lifetime_3) {
auto dbl = system.spawn(testee);
auto bound = dbl.bind(1);
anon_send(bound, down_msg{self->address(),
exit_reason::kill});
CAF_CHECK(! exited(bound));
self->monitor(bound);
auto em_sender = system.spawn(testee);
em_sender->link_to(bound->address());
anon_send_exit(em_sender, exit_reason::kill);
wait_until_exited();
self->request(dbl, infinite, 1).receive(
[](int v) {
CAF_CHECK_EQUAL(v, 2);
}
);
anon_send_exit(dbl, exit_reason::kill);
} }
CAF_TEST(request_response_promise) { CAF_TEST(request_response_promise) {
......
...@@ -37,7 +37,7 @@ CAF_TEST(constructor_attach) { ...@@ -37,7 +37,7 @@ CAF_TEST(constructor_attach) {
class testee : public event_based_actor { class testee : public event_based_actor {
public: public:
testee(actor_config& cfg, actor buddy) : event_based_actor(cfg) { testee(actor_config& cfg, actor buddy) : event_based_actor(cfg) {
attach_functor([=](exit_reason reason) { attach_functor([=](const error& reason) {
send(buddy, done_atom::value, reason); send(buddy, done_atom::value, reason);
}); });
} }
...@@ -66,7 +66,7 @@ CAF_TEST(constructor_attach) { ...@@ -66,7 +66,7 @@ CAF_TEST(constructor_attach) {
quit(msg.reason); quit(msg.reason);
} }
}, },
[=](done_atom, exit_reason 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) {
quit(reason); quit(reason);
......
...@@ -527,7 +527,7 @@ CAF_TEST(constructor_attach) { ...@@ -527,7 +527,7 @@ CAF_TEST(constructor_attach) {
testee(actor_config& cfg, actor buddy) testee(actor_config& cfg, actor buddy)
: event_based_actor(cfg), : event_based_actor(cfg),
buddy_(buddy) { buddy_(buddy) {
attach_functor([=](exit_reason reason) { attach_functor([=](const error& reason) {
send(buddy, ok_atom::value, reason); send(buddy, ok_atom::value, reason);
}); });
} }
...@@ -562,7 +562,7 @@ CAF_TEST(constructor_attach) { ...@@ -562,7 +562,7 @@ CAF_TEST(constructor_attach) {
quit(msg.reason); quit(msg.reason);
} }
}, },
[=](ok_atom, exit_reason 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);
...@@ -670,18 +670,6 @@ CAF_TEST(kill_the_immortal) { ...@@ -670,18 +670,6 @@ CAF_TEST(kill_the_immortal) {
); );
} }
CAF_TEST(exit_reason_in_scoped_actor) {
scoped_actor self{system};
self->spawn<linked>(
[]() -> behavior {
return [] {
// nop
};
}
);
self->planned_exit_reason(exit_reason::unhandled_exception);
}
CAF_TEST(move_only_argument) { CAF_TEST(move_only_argument) {
using unique_int = std::unique_ptr<int>; using unique_int = std::unique_ptr<int>;
unique_int uptr{new int(42)}; unique_int uptr{new int(42)};
......
...@@ -126,8 +126,8 @@ CAF_TEST(single_res_function_view) { ...@@ -126,8 +126,8 @@ CAF_TEST(single_res_function_view) {
g(1, 0); g(1, 0);
CAF_ERROR("expected exception"); CAF_ERROR("expected exception");
} }
catch (actor_exited&) { catch (actor_exited& e) {
// nop CAF_MESSAGE("e: " << system.render(e.reason()));
} }
CAF_CHECK(g == nullptr); CAF_CHECK(g == nullptr);
g.assign(system.spawn(divider)); g.assign(system.spawn(divider));
......
...@@ -213,7 +213,6 @@ public: ...@@ -213,7 +213,6 @@ public:
\******************************************************************************/ \******************************************************************************/
behavior server(event_based_actor* self) { behavior server(event_based_actor* self) {
printf("server id: %d\n", (int) self->id());
return { return {
[=](idle_atom, actor worker) { [=](idle_atom, actor worker) {
self->become( self->become(
......
...@@ -55,7 +55,7 @@ second_stage::behavior_type typed_second_stage() { ...@@ -55,7 +55,7 @@ second_stage::behavior_type typed_second_stage() {
} }
struct fixture { struct fixture {
void wait_until_exited() { void wait_until_is_terminated() {
self->receive( self->receive(
[](const down_msg&) { [](const down_msg&) {
CAF_CHECK(true); CAF_CHECK(true);
...@@ -68,7 +68,7 @@ struct fixture { ...@@ -68,7 +68,7 @@ struct fixture {
auto ptr = actor_cast<abstract_actor*>(handle); auto ptr = actor_cast<abstract_actor*>(handle);
auto dptr = dynamic_cast<monitorable_actor*>(ptr); auto dptr = dynamic_cast<monitorable_actor*>(ptr);
CAF_REQUIRE(dptr != nullptr); CAF_REQUIRE(dptr != nullptr);
return dptr->exited(); return dptr->is_terminated();
} }
actor_system system; actor_system system;
...@@ -104,7 +104,7 @@ CAF_TEST(lifetime_1a) { ...@@ -104,7 +104,7 @@ CAF_TEST(lifetime_1a) {
auto f = system.spawn(testee); auto f = system.spawn(testee);
self->monitor(g); self->monitor(g);
anon_send_exit(g, exit_reason::kill); anon_send_exit(g, exit_reason::kill);
wait_until_exited(); wait_until_is_terminated();
auto h = f * g; auto h = f * g;
CAF_CHECK(exited(h)); CAF_CHECK(exited(h));
anon_send_exit(f, exit_reason::kill); anon_send_exit(f, exit_reason::kill);
...@@ -116,7 +116,7 @@ CAF_TEST(lifetime_1b) { ...@@ -116,7 +116,7 @@ CAF_TEST(lifetime_1b) {
auto f = system.spawn(testee); auto f = system.spawn(testee);
self->monitor(f); self->monitor(f);
anon_send_exit(f, exit_reason::kill); anon_send_exit(f, exit_reason::kill);
wait_until_exited(); wait_until_is_terminated();
auto h = f * g; auto h = f * g;
CAF_CHECK(exited(h)); CAF_CHECK(exited(h));
anon_send_exit(g, exit_reason::kill); anon_send_exit(g, exit_reason::kill);
...@@ -129,7 +129,7 @@ CAF_TEST(lifetime_2a) { ...@@ -129,7 +129,7 @@ CAF_TEST(lifetime_2a) {
auto h = f * g; auto h = f * g;
self->monitor(h); self->monitor(h);
anon_send(g, message{}); anon_send(g, message{});
wait_until_exited(); wait_until_is_terminated();
anon_send_exit(f, exit_reason::kill); anon_send_exit(f, exit_reason::kill);
} }
...@@ -140,41 +140,7 @@ CAF_TEST(lifetime_2b) { ...@@ -140,41 +140,7 @@ CAF_TEST(lifetime_2b) {
auto h = f * g; auto h = f * g;
self->monitor(h); self->monitor(h);
anon_send(f, message{}); anon_send(f, message{});
wait_until_exited(); wait_until_is_terminated();
anon_send_exit(g, exit_reason::kill);
}
// 1) ignores down message not from constituent actors
// 2) exits by receiving an exit message
// 3) exit has no effect on constituent actors
CAF_TEST(lifetime_3) {
auto g = system.spawn(testee);
auto f = system.spawn(testee);
auto h = f * g;
self->monitor(h);
anon_send(h, down_msg{self->address(), exit_reason::kill});
CAF_CHECK(! exited(h));
auto em_sender = system.spawn(testee);
em_sender->link_to(h.address());
anon_send_exit(em_sender, exit_reason::kill);
wait_until_exited();
self->request(f, infinite, 1).receive(
[](int v) {
CAF_CHECK_EQUAL(v, 2);
},
[](error) {
CAF_CHECK(false);
}
);
self->request(g, infinite, 1).receive(
[](int v) {
CAF_CHECK_EQUAL(v, 2);
},
[](error) {
CAF_CHECK(false);
}
);
anon_send_exit(f, exit_reason::kill);
anon_send_exit(g, exit_reason::kill); anon_send_exit(g, exit_reason::kill);
} }
......
...@@ -88,7 +88,7 @@ public: ...@@ -88,7 +88,7 @@ public:
void enqueue(mailbox_element_ptr, execution_unit*) override; void enqueue(mailbox_element_ptr, execution_unit*) override;
/// Called after this broker has finished execution. /// Called after this broker has finished execution.
void cleanup(exit_reason reason, execution_unit* host) override; bool cleanup(error&& reason, execution_unit* host) override;
/// Starts running this broker in the `middleman`. /// Starts running this broker in the `middleman`.
void launch(execution_unit* eu, bool lazy, bool hide); void launch(execution_unit* eu, bool lazy, bool hide);
......
...@@ -23,7 +23,6 @@ ...@@ -23,7 +23,6 @@
#include "caf/io/basp/header.hpp" #include "caf/io/basp/header.hpp"
#include "caf/io/basp/version.hpp" #include "caf/io/basp/version.hpp"
#include "caf/io/basp/instance.hpp" #include "caf/io/basp/instance.hpp"
#include "caf/io/basp/error_code.hpp"
#include "caf/io/basp/buffer_type.hpp" #include "caf/io/basp/buffer_type.hpp"
#include "caf/io/basp/message_type.hpp" #include "caf/io/basp/message_type.hpp"
#include "caf/io/basp/routing_table.hpp" #include "caf/io/basp/routing_table.hpp"
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2015 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#ifndef CAF_IO__HPP
#define CAF_IO__HPP
#include <cstdint>
namespace caf {
namespace io {
namespace basp {
/// @addtogroup BASP
/// Describes an error during forwarding of BASP messages.
enum class error_code : uint64_t {
/// Indicates that a forwarding node had no route
/// to the destination.
no_route_to_destination = 0x01,
/// Indicates that a forwarding node detected
/// a loop in the forwarding path.
loop_detected = 0x02
};
/// @relates error_code
constexpr const char* to_string(error_code x) {
return x == error_code::no_route_to_destination ? "no_route_to_destination"
: "loop_detected";
}
/// @}
} // namespace basp
} // namespace io
} // namespace caf
#endif // CAF_IO__HPP
...@@ -20,11 +20,12 @@ ...@@ -20,11 +20,12 @@
#ifndef CAF_IO_BASP_INSTANCE_HPP #ifndef CAF_IO_BASP_INSTANCE_HPP
#define CAF_IO_BASP_INSTANCE_HPP #define CAF_IO_BASP_INSTANCE_HPP
#include "caf/error.hpp"
#include "caf/io/hook.hpp" #include "caf/io/hook.hpp"
#include "caf/io/middleman.hpp" #include "caf/io/middleman.hpp"
#include "caf/io/basp/header.hpp" #include "caf/io/basp/header.hpp"
#include "caf/io/basp/error_code.hpp"
#include "caf/io/basp/buffer_type.hpp" #include "caf/io/basp/buffer_type.hpp"
#include "caf/io/basp/message_type.hpp" #include "caf/io/basp/message_type.hpp"
#include "caf/io/basp/routing_table.hpp" #include "caf/io/basp/routing_table.hpp"
...@@ -65,7 +66,7 @@ public: ...@@ -65,7 +66,7 @@ public:
/// Called whenever a remote actor died to destroy /// Called whenever a remote actor died to destroy
/// the proxy instance on our end. /// the proxy instance on our end.
virtual void kill_proxy(const node_id& nid, actor_id aid, virtual void kill_proxy(const node_id& nid, actor_id aid,
exit_reason rsn) = 0; const error& rsn) = 0;
/// Called whenever a `dispatch_message` arrived for a local actor. /// Called whenever a `dispatch_message` arrived for a local actor.
virtual void deliver(const node_id& source_node, actor_id source_actor, virtual void deliver(const node_id& source_node, actor_id source_actor,
...@@ -202,21 +203,12 @@ public: ...@@ -202,21 +203,12 @@ public:
void write_client_handshake(execution_unit* ctx, void write_client_handshake(execution_unit* ctx,
buffer_type& buf, const node_id& remote_side); buffer_type& buf, const node_id& remote_side);
/// Writes a `dispatch_error` to `buf`.
void write_dispatch_error(execution_unit* ctx,
buffer_type& buf,
const node_id& source_node,
const node_id& dest_node,
error_code ec,
const header& original_hdr,
buffer_type* payload);
/// Writes a `kill_proxy_instance` to `buf`. /// Writes a `kill_proxy_instance` to `buf`.
void write_kill_proxy_instance(execution_unit* ctx, void write_kill_proxy_instance(execution_unit* ctx,
buffer_type& buf, buffer_type& buf,
const node_id& dest_node, const node_id& dest_node,
actor_id aid, actor_id aid,
exit_reason rsn); const error& fail_state);
/// Writes a `heartbeat` to `buf`. /// Writes a `heartbeat` to `buf`.
void write_heartbeat(execution_unit* ctx, void write_heartbeat(execution_unit* ctx,
......
...@@ -63,7 +63,7 @@ struct basp_broker_state : proxy_registry::backend, basp::instance::callee { ...@@ -63,7 +63,7 @@ struct basp_broker_state : proxy_registry::backend, basp::instance::callee {
void proxy_announced(const node_id& nid, actor_id aid) override; void proxy_announced(const node_id& nid, actor_id aid) override;
// inherited from basp::instance::listener // inherited from basp::instance::listener
void kill_proxy(const node_id& nid, actor_id aid, exit_reason rsn) override; void kill_proxy(const node_id& nid, actor_id aid, const error& rsn) override;
// inherited from basp::instance::listener // inherited from basp::instance::listener
void deliver(const node_id& source_node, actor_id source_actor, void deliver(const node_id& source_node, actor_id source_actor,
......
...@@ -58,13 +58,13 @@ void abstract_broker::launch(execution_unit* eu, bool is_lazy, bool is_hidden) { ...@@ -58,13 +58,13 @@ void abstract_broker::launch(execution_unit* eu, bool is_lazy, bool is_hidden) {
eu->exec_later(this); eu->exec_later(this);
} }
void abstract_broker::cleanup(exit_reason reason, execution_unit* host) { bool abstract_broker::cleanup(error&& reason, execution_unit* host) {
CAF_LOG_TRACE(CAF_ARG(reason)); CAF_LOG_TRACE(CAF_ARG(reason));
close_all(); close_all();
CAF_ASSERT(doormen_.empty()); CAF_ASSERT(doormen_.empty());
CAF_ASSERT(scribes_.empty()); CAF_ASSERT(scribes_.empty());
cache_.clear(); cache_.clear();
local_actor::cleanup(reason, host); return local_actor::cleanup(std::move(reason), host);
} }
abstract_broker::~abstract_broker() { abstract_broker::~abstract_broker() {
......
...@@ -90,13 +90,13 @@ strong_actor_ptr basp_broker_state::make_proxy(node_id nid, actor_id aid) { ...@@ -90,13 +90,13 @@ strong_actor_ptr basp_broker_state::make_proxy(node_id nid, actor_id aid) {
auto res = make_actor<forwarding_actor_proxy, strong_actor_ptr>( auto res = make_actor<forwarding_actor_proxy, strong_actor_ptr>(
aid, nid, &(self->home_system()), cfg, self); aid, nid, &(self->home_system()), cfg, self);
auto selfptr = actor_cast<strong_actor_ptr>(self); auto selfptr = actor_cast<strong_actor_ptr>(self);
res->get()->attach_functor([=](exit_reason rsn) { res->get()->attach_functor([=](const error& rsn) {
mm->backend().post([=] { mm->backend().post([=] {
// using res->id() instead of aid keeps this actor instance alive // using res->id() instead of aid keeps this actor instance alive
// until the original instance terminates, thus preventing subtle // until the original instance terminates, thus preventing subtle
// bugs with attachables // bugs with attachables
auto bptr = static_cast<basp_broker*>(selfptr->get()); auto bptr = static_cast<basp_broker*>(selfptr->get());
if (! bptr->exited()) if (! bptr->is_terminated())
bptr->state.proxies().erase(nid, res->id(), rsn); bptr->state.proxies().erase(nid, res->id(), rsn);
}); });
}); });
...@@ -140,7 +140,7 @@ void basp_broker_state::finalize_handshake(const node_id& nid, actor_id aid, ...@@ -140,7 +140,7 @@ void basp_broker_state::finalize_handshake(const node_id& nid, actor_id aid,
strong_actor_ptr ptr; strong_actor_ptr ptr;
if (nid == this_node()) { if (nid == this_node()) {
// connected to self // connected to self
ptr = actor_cast<strong_actor_ptr>(system().registry().get(aid).first); ptr = actor_cast<strong_actor_ptr>(system().registry().get(aid));
CAF_LOG_INFO_IF(! ptr, "actor not found:" << CAF_ARG(aid)); CAF_LOG_INFO_IF(! ptr, "actor not found:" << CAF_ARG(aid));
} else { } else {
ptr = namespace_.get_or_put(nid, aid); ptr = namespace_.get_or_put(nid, aid);
...@@ -171,8 +171,8 @@ void basp_broker_state::proxy_announced(const node_id& nid, actor_id aid) { ...@@ -171,8 +171,8 @@ void basp_broker_state::proxy_announced(const node_id& nid, actor_id aid) {
CAF_LOG_TRACE(CAF_ARG(nid) << CAF_ARG(aid)); CAF_LOG_TRACE(CAF_ARG(nid) << CAF_ARG(aid));
// source node has created a proxy for one of our actors // source node has created a proxy for one of our actors
auto entry = system().registry().get(aid); auto entry = system().registry().get(aid);
auto send_kill_proxy_instance = [=](exit_reason rsn) { auto send_kill_proxy_instance = [=](error rsn) {
if (rsn == exit_reason::not_exited) if (! rsn)
rsn = exit_reason::unknown; rsn = exit_reason::unknown;
auto path = instance.tbl().lookup(nid); auto path = instance.tbl().lookup(nid);
if (! path) { if (! path) {
...@@ -184,29 +184,29 @@ void basp_broker_state::proxy_announced(const node_id& nid, actor_id aid) { ...@@ -184,29 +184,29 @@ void basp_broker_state::proxy_announced(const node_id& nid, actor_id aid) {
nid, aid, rsn); nid, aid, rsn);
instance.tbl().flush(*path); instance.tbl().flush(*path);
}; };
auto ptr = actor_cast<strong_actor_ptr>(entry.first); auto ptr = actor_cast<strong_actor_ptr>(entry);
if (! ptr || entry.second != exit_reason::not_exited) { if (! ptr) {
CAF_LOG_DEBUG("kill proxy immediately"); CAF_LOG_DEBUG("kill proxy immediately");
// kill immediately if actor has already terminated // kill immediately if actor has already terminated
send_kill_proxy_instance(entry.second); send_kill_proxy_instance(exit_reason::unknown);
} else { } else {
auto tmp = actor_cast<strong_actor_ptr>(self); // keep broker alive ... auto tmp = actor_cast<strong_actor_ptr>(self); // keep broker alive ...
auto mm = &system().middleman(); auto mm = &system().middleman();
ptr->get()->attach_functor([=](exit_reason reason) { ptr->get()->attach_functor([=](const error& fail_state) {
mm->backend().dispatch([=] { mm->backend().dispatch([=] {
CAF_LOG_TRACE(CAF_ARG(reason)); CAF_LOG_TRACE(CAF_ARG(fail_state));
auto bptr = static_cast<basp_broker*>(tmp->get()); auto bptr = static_cast<basp_broker*>(tmp->get());
// ... to make sure this is safe // ... to make sure this is safe
if (bptr == mm->named_broker<basp_broker>(atom("BASP")) if (bptr == mm->named_broker<basp_broker>(atom("BASP"))
&& ! bptr->exited()) && ! bptr->is_terminated())
send_kill_proxy_instance(reason); send_kill_proxy_instance(fail_state);
}); });
}); });
} }
} }
void basp_broker_state::kill_proxy(const node_id& nid, actor_id aid, void basp_broker_state::kill_proxy(const node_id& nid, actor_id aid,
exit_reason rsn) { const error& rsn) {
CAF_LOG_TRACE(CAF_ARG(nid) << CAF_ARG(aid) << CAF_ARG(rsn)); CAF_LOG_TRACE(CAF_ARG(nid) << CAF_ARG(aid) << CAF_ARG(rsn));
proxies().erase(nid, aid, rsn); proxies().erase(nid, aid, rsn);
} }
...@@ -223,7 +223,7 @@ void basp_broker_state::deliver(const node_id& src_nid, ...@@ -223,7 +223,7 @@ void basp_broker_state::deliver(const node_id& src_nid,
<< CAF_ARG(dest_aid) << CAF_ARG(msg) << CAF_ARG(mid)); << CAF_ARG(dest_aid) << CAF_ARG(msg) << CAF_ARG(mid));
strong_actor_ptr src; strong_actor_ptr src;
if (src_nid == this_node()) if (src_nid == this_node())
src = actor_cast<strong_actor_ptr>(system().registry().get(src_aid).first); src = actor_cast<strong_actor_ptr>(system().registry().get(src_aid));
else else
src = proxies().get_or_put(src_nid, src_aid); src = proxies().get_or_put(src_nid, src_aid);
strong_actor_ptr dest; strong_actor_ptr dest;
...@@ -240,7 +240,7 @@ void basp_broker_state::deliver(const node_id& src_nid, ...@@ -240,7 +240,7 @@ void basp_broker_state::deliver(const node_id& src_nid,
mid = message_id::make(); // override this since the message is async mid = message_id::make(); // override this since the message is async
dest = actor_cast<strong_actor_ptr>(system().registry().get(dest_name)); dest = actor_cast<strong_actor_ptr>(system().registry().get(dest_name));
} else { } else {
std::tie(dest, rsn) = system().registry().get(dest_aid); dest = system().registry().get(dest_aid);
} }
} }
if (! dest) { if (! dest) {
......
...@@ -99,8 +99,8 @@ bool kill_proxy_instance_valid(const header& hdr) { ...@@ -99,8 +99,8 @@ bool kill_proxy_instance_valid(const header& hdr) {
&& hdr.source_node != hdr.dest_node && hdr.source_node != hdr.dest_node
&& ! zero(hdr.source_actor) && ! zero(hdr.source_actor)
&& zero(hdr.dest_actor) && zero(hdr.dest_actor)
&& zero(hdr.payload_len) && ! zero(hdr.payload_len)
&& ! zero(hdr.operation_data); && zero(hdr.operation_data);
} }
bool heartbeat_valid(const header& hdr) { bool heartbeat_valid(const header& hdr) {
......
...@@ -24,7 +24,6 @@ ...@@ -24,7 +24,6 @@
#include "caf/binary_deserializer.hpp" #include "caf/binary_deserializer.hpp"
#include "caf/io/basp/version.hpp" #include "caf/io/basp/version.hpp"
#include "caf/io/basp/error_code.hpp"
namespace caf { namespace caf {
namespace io { namespace io {
...@@ -91,17 +90,12 @@ connection_state instance::handle(execution_unit* ctx, ...@@ -91,17 +90,12 @@ connection_state instance::handle(execution_unit* ctx,
} else { } else {
CAF_LOG_INFO("cannot forward message, no route to destination"); CAF_LOG_INFO("cannot forward message, no route to destination");
if (hdr.source_node != this_node_) { if (hdr.source_node != this_node_) {
// TODO: signalize error back to sending node
auto reverse_path = lookup(hdr.source_node); auto reverse_path = lookup(hdr.source_node);
if (! reverse_path) { if (! reverse_path) {
CAF_LOG_WARNING("cannot send error message: no route to source"); CAF_LOG_WARNING("cannot send error message: no route to source");
} else { } else {
write_dispatch_error(ctx, CAF_LOG_WARNING("not implemented yet: signalize forward failure");
reverse_path->wr_buf,
this_node_,
hdr.source_node,
error_code::no_route_to_destination,
hdr,
payload);
} }
} else { } else {
CAF_LOG_WARNING("lost packet with probably spoofed source"); CAF_LOG_WARNING("lost packet with probably spoofed source");
...@@ -190,10 +184,15 @@ connection_state instance::handle(execution_unit* ctx, ...@@ -190,10 +184,15 @@ connection_state instance::handle(execution_unit* ctx,
case message_type::announce_proxy_instance: case message_type::announce_proxy_instance:
callee_.proxy_announced(hdr.source_node, hdr.dest_actor); callee_.proxy_announced(hdr.source_node, hdr.dest_actor);
break; break;
case message_type::kill_proxy_instance: case message_type::kill_proxy_instance: {
callee_.kill_proxy(hdr.source_node, hdr.source_actor, if (! payload_valid())
static_cast<exit_reason>(hdr.operation_data)); return err();
binary_deserializer bd{ctx, *payload};
error fail_state;
bd >> fail_state;
callee_.kill_proxy(hdr.source_node, hdr.source_actor, fail_state);
break; break;
}
case message_type::heartbeat: { case message_type::heartbeat: {
CAF_LOG_TRACE("received heartbeat: " << CAF_ARG(hdr.source_node)); CAF_LOG_TRACE("received heartbeat: " << CAF_ARG(hdr.source_node));
callee_.handle_heartbeat(hdr.source_node); callee_.handle_heartbeat(hdr.source_node);
...@@ -408,36 +407,18 @@ void instance::write_client_handshake(execution_unit* ctx, ...@@ -408,36 +407,18 @@ void instance::write_client_handshake(execution_unit* ctx,
this_node_, remote_side, invalid_actor_id, invalid_actor_id); this_node_, remote_side, invalid_actor_id, invalid_actor_id);
} }
void instance::write_dispatch_error(execution_unit* ctx,
buffer_type& buf,
const node_id& source_node,
const node_id& dest_node,
error_code ec,
const header& original_hdr,
buffer_type* payload) {
CAF_LOG_TRACE(CAF_ARG(source_node) << CAF_ARG(dest_node)
<< CAF_ARG(ec) << CAF_ARG(original_hdr));
auto writer = make_callback([&](serializer& sink) {
sink << original_hdr;
if (payload)
sink.apply_raw(payload->size(), payload->data());
});
header hdr{message_type::kill_proxy_instance, 0,
static_cast<uint64_t>(ec),
source_node, dest_node,
invalid_actor_id, invalid_actor_id};
write(ctx, buf, hdr, &writer);
}
void instance::write_kill_proxy_instance(execution_unit* ctx, void instance::write_kill_proxy_instance(execution_unit* ctx,
buffer_type& buf, buffer_type& buf,
const node_id& dest_node, const node_id& dest_node,
actor_id aid, actor_id aid,
exit_reason rsn) { const error& rsn) {
CAF_LOG_TRACE(CAF_ARG(dest_node) << CAF_ARG(aid) << CAF_ARG(rsn)); CAF_LOG_TRACE(CAF_ARG(dest_node) << CAF_ARG(aid) << CAF_ARG(rsn));
header hdr{message_type::kill_proxy_instance, 0, static_cast<uint32_t>(rsn), auto writer = make_callback([&](serializer& sink) {
sink << rsn;
});
header hdr{message_type::kill_proxy_instance, 0, 0,
this_node_, dest_node, aid, invalid_actor_id}; this_node_, dest_node, aid, invalid_actor_id};
write(ctx, buf, hdr); write(ctx, buf, hdr, &writer);
} }
void instance::write_heartbeat(execution_unit* ctx, void instance::write_heartbeat(execution_unit* ctx,
......
...@@ -297,9 +297,9 @@ void middleman::stop() { ...@@ -297,9 +297,9 @@ void middleman::stop() {
for (auto& kvp : named_brokers_) { for (auto& kvp : named_brokers_) {
auto& hdl = kvp.second; auto& hdl = kvp.second;
auto ptr = static_cast<broker*>(actor_cast<abstract_actor*>(hdl)); auto ptr = static_cast<broker*>(actor_cast<abstract_actor*>(hdl));
if (! ptr->exited()) { if (! ptr->is_terminated()) {
ptr->context(&backend()); ptr->context(&backend());
ptr->planned_exit_reason(exit_reason::normal); ptr->is_terminated(true);
ptr->finished(); ptr->finished();
} }
} }
......
...@@ -331,8 +331,8 @@ public: ...@@ -331,8 +331,8 @@ public:
std::vector<strong_actor_ptr> stages; std::vector<strong_actor_ptr> stages;
message msg; message msg;
source >> stages >> msg; source >> stages >> msg;
auto src = actor_cast<strong_actor_ptr>(registry_->get(hdr.source_actor).first); auto src = actor_cast<strong_actor_ptr>(registry_->get(hdr.source_actor));
auto dest = actor_cast<actor>(registry_->get(hdr.dest_actor).first); auto dest = actor_cast<actor>(registry_->get(hdr.dest_actor));
CAF_REQUIRE(dest); CAF_REQUIRE(dest);
dest->enqueue(mailbox_element::make(src, message_id::make(), dest->enqueue(mailbox_element::make(src, message_id::make(),
std::move(stages), std::move(msg)), std::move(stages), std::move(msg)),
......
...@@ -77,9 +77,7 @@ struct http_state { ...@@ -77,9 +77,7 @@ struct http_state {
} }
~http_state() { ~http_state() {
aout(self_) << "http worker finished with exit reason: " aout(self_) << "http worker is destroyed";
<< self_->planned_exit_reason()
<< endl;
} }
std::vector<std::string> lines; std::vector<std::string> lines;
......
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