Commit 1fa6bb93 authored by Dominik Charousset's avatar Dominik Charousset

Fix potential deadlock in link impl., relates #536

parent c3d10228
...@@ -94,18 +94,6 @@ public: ...@@ -94,18 +94,6 @@ 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;
/// Establishes a link relation between this actor and `x`
/// and returns whether the operation succeeded.
inline bool establish_backlink(abstract_actor* x) {
return link_impl(establish_backlink_op, x);
}
/// Removes the link relation between this actor and `x`
/// and returns whether the operation succeeded.
inline bool remove_backlink(abstract_actor* x) {
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
/// an empty set if this actor is untyped. /// an empty set if this actor is untyped.
virtual std::set<std::string> message_types() const; virtual std::set<std::string> message_types() const;
...@@ -133,13 +121,6 @@ public: ...@@ -133,13 +121,6 @@ public:
ctx); ctx);
} }
enum linking_operation {
establish_link_op,
establish_backlink_op,
remove_link_op,
remove_backlink_op
};
// flags storing runtime information used by ... // flags storing runtime information used by ...
static constexpr int has_timeout_flag = 0x0004; // single_timeout static constexpr int has_timeout_flag = 0x0004; // single_timeout
static constexpr int is_registered_flag = 0x0008; // (several actors) static constexpr int is_registered_flag = 0x0008; // (several actors)
...@@ -173,14 +154,67 @@ public: ...@@ -173,14 +154,67 @@ public:
/// Unsets `is_registered_flag` and calls `system().registry().dec_running()`. /// Unsets `is_registered_flag` and calls `system().registry().dec_running()`.
void unregister_from_system(); void unregister_from_system();
virtual bool link_impl(linking_operation op, abstract_actor* other) = 0; /// Causes the actor to establish a link to `other`.
virtual void add_link(abstract_actor* other) = 0;
/// @endcond /// Causes the actor to remove any established link to `other`.
virtual void remove_link(abstract_actor* other) = 0;
/// Adds an entry to `other` to the link table of this actor.
/// @warning Must be called inside a critical section, i.e.,
/// while holding `mtx_`.
virtual bool add_backlink(abstract_actor* other) = 0;
/// Removes an entry to `other` from the link table of this actor.
/// @warning Must be called inside a critical section, i.e.,
/// while holding `mtx_`.
virtual bool remove_backlink(abstract_actor* other) = 0;
/// Calls `fun` with exclusive access to an actor's state.
template <class F>
auto exclusive_critical_section(F fun) -> decltype(fun()) {
std::unique_lock<std::mutex> guard{mtx_};
return fun();
}
/// Calls `fun` with readonly access to an actor's state.
template <class F>
auto shared_critical_section(F fun) -> decltype(fun()) {
std::unique_lock<std::mutex> guard{mtx_};
return fun();
}
/// Calls `fun` with exclusive access to the state of both `p1` and `p2`. This
/// function guarantees that the order of acquiring the locks is always
/// identical, independently from the order of `p1` and `p2`.
template <class F>
static auto joined_exclusive_critical_section(abstract_actor* p1,
abstract_actor* p2,
F fun)
-> decltype(fun()) {
// Make sure to allocate locks always in the same order by starting on the
// actor with the lowest address.
CAF_ASSERT(p1 != p2 && p1 != nullptr && p2 != nullptr);
if (p1 < p2) {
std::unique_lock<std::mutex> guard1{p1->mtx_};
std::unique_lock<std::mutex> guard2{p2->mtx_};
return fun();
}
std::unique_lock<std::mutex> guard1{p2->mtx_};
std::unique_lock<std::mutex> guard2{p1->mtx_};
return fun();
}
/// @endcond
protected: protected:
/// Creates a new actor instance. /// Creates a new actor instance.
explicit abstract_actor(actor_config& cfg); explicit abstract_actor(actor_config& cfg);
// Guards potentially concurrent access to the state. For example,
// `exit_state_`, `attachables_`, and `links_` in a `monitorable_actor`.
mutable std::mutex mtx_;
private: private:
// prohibit copies, assigments, and heap allocations // prohibit copies, assigments, and heap allocations
void* operator new(size_t); void* operator new(size_t);
...@@ -189,8 +223,6 @@ private: ...@@ -189,8 +223,6 @@ private:
abstract_actor& operator=(const abstract_actor&) = delete; abstract_actor& operator=(const abstract_actor&) = delete;
}; };
std::string to_string(abstract_actor::linking_operation op);
} // namespace caf } // namespace caf
#endif // CAF_ABSTRACT_ACTOR_HPP #endif // CAF_ABSTRACT_ACTOR_HPP
...@@ -38,13 +38,6 @@ public: ...@@ -38,13 +38,6 @@ public:
~actor_proxy() override; ~actor_proxy() override;
/// Establishes a local link state that's
/// not synchronized back to the remote instance.
virtual void local_link_to(abstract_actor* other) = 0;
/// Removes a local link state.
virtual void local_unlink_from(abstract_actor* other) = 0;
/// Invokes cleanup code. /// Invokes cleanup code.
virtual void kill_proxy(execution_unit* ctx, error reason) = 0; virtual void kill_proxy(execution_unit* ctx, error reason) = 0;
}; };
......
...@@ -38,11 +38,9 @@ public: ...@@ -38,11 +38,9 @@ public:
void enqueue(mailbox_element_ptr what, execution_unit* context) override; void enqueue(mailbox_element_ptr what, execution_unit* context) override;
bool link_impl(linking_operation op, abstract_actor* other) override; bool add_backlink(abstract_actor* x) override;
void local_link_to(abstract_actor* other) override; bool remove_backlink(abstract_actor* x) override;
void local_unlink_from(abstract_actor* other) override;
void kill_proxy(execution_unit* ctx, error rsn) override; void kill_proxy(execution_unit* ctx, error rsn) override;
......
...@@ -59,7 +59,7 @@ public: ...@@ -59,7 +59,7 @@ public:
void link_to(const actor_addr& x) { void link_to(const actor_addr& x) {
auto ptr = actor_cast<strong_actor_ptr>(x); auto ptr = actor_cast<strong_actor_ptr>(x);
if (ptr && ptr->get() != this) if (ptr && ptr->get() != this)
link_impl(establish_link_op, ptr->get()); add_link(ptr->get());
} }
/// Links this actor to `x`. /// Links this actor to `x`.
...@@ -67,14 +67,14 @@ public: ...@@ -67,14 +67,14 @@ public:
void link_to(const ActorHandle& x) { void link_to(const ActorHandle& x) {
auto ptr = actor_cast<abstract_actor*>(x); auto ptr = actor_cast<abstract_actor*>(x);
if (ptr && ptr != this) if (ptr && ptr != this)
link_impl(establish_link_op, ptr); add_link(ptr);
} }
/// Unlinks this actor from `x`. /// Unlinks this actor from `x`.
void unlink_from(const actor_addr& x) { void unlink_from(const actor_addr& x) {
auto ptr = actor_cast<strong_actor_ptr>(x); auto ptr = actor_cast<strong_actor_ptr>(x);
if (ptr && ptr->get() != this) if (ptr && ptr->get() != this)
link_impl(remove_link_op, ptr->get()); remove_link(ptr->get());
} }
/// Links this actor to `x`. /// Links this actor to `x`.
...@@ -82,7 +82,7 @@ public: ...@@ -82,7 +82,7 @@ public:
void unlink_from(const ActorHandle& x) { void unlink_from(const ActorHandle& x) {
auto ptr = actor_cast<abstract_actor*>(x); auto ptr = actor_cast<abstract_actor*>(x);
if (ptr && ptr != this) if (ptr && ptr != this)
link_impl(remove_link_op, ptr); remove_link(ptr);
} }
/// @cond PRIVATE /// @cond PRIVATE
...@@ -94,6 +94,14 @@ public: ...@@ -94,6 +94,14 @@ public:
/// function is ignored by scheduled actors. /// function is ignored by scheduled actors.
virtual bool cleanup(error&& reason, execution_unit* host); virtual bool cleanup(error&& reason, execution_unit* host);
void add_link(abstract_actor* x) override;
void remove_link(abstract_actor* x) override;
bool add_backlink(abstract_actor* x) override;
bool remove_backlink(abstract_actor* x) override;
/// @endcond /// @endcond
protected: protected:
...@@ -115,16 +123,6 @@ protected: ...@@ -115,16 +123,6 @@ protected:
* here be dragons: end of public interface * * here be dragons: end of public interface *
****************************************************************************/ ****************************************************************************/
bool link_impl(linking_operation op, abstract_actor* other) override;
bool establish_link_impl(abstract_actor* x);
bool remove_link_impl(abstract_actor* x);
bool establish_backlink_impl(abstract_actor* x);
bool remove_backlink_impl(abstract_actor* x);
// precondition: `mtx_` is acquired // precondition: `mtx_` is acquired
inline void attach_impl(attachable_ptr& ptr) { inline void attach_impl(attachable_ptr& ptr) {
ptr->next.swap(attachables_head_); ptr->next.swap(attachables_head_);
...@@ -160,27 +158,10 @@ protected: ...@@ -160,27 +158,10 @@ protected:
return handle_system_message(x, context, trap_exit); return handle_system_message(x, context, trap_exit);
} }
// Calls `fun` with exclusive access to an actor's state.
template <class F>
auto exclusive_critical_section(F fun) -> decltype(fun()) {
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 // is protected by `mtx_` in actors that are not scheduled, but
// can be accessed without lock for event-based and blocking actors // can be accessed without lock for event-based and blocking actors
error fail_state_; error fail_state_;
// guards access to exit_state_, attachables_, links_,
// and enqueue operations if actor is thread-mapped
mutable std::mutex mtx_;
// only used in blocking and thread-mapped actors // only used in blocking and thread-mapped actors
mutable std::condition_variable cv_; mutable std::condition_variable cv_;
...@@ -190,8 +171,6 @@ protected: ...@@ -190,8 +171,6 @@ protected:
/// @endcond /// @endcond
}; };
std::string to_string(abstract_actor::linking_operation op);
} // namespace caf } // namespace caf
#endif // CAF_MONITORABLE_ACTOR_HPP #endif // CAF_MONITORABLE_ACTOR_HPP
...@@ -105,19 +105,4 @@ void abstract_actor::unregister_from_system() { ...@@ -105,19 +105,4 @@ void abstract_actor::unregister_from_system() {
home_system().registry().dec_running(); home_system().registry().dec_running();
} }
namespace {
const char* linking_operation_strings[] = {
"establish_link",
"establish_backlink",
"remove_link",
"remove_backlink"
};
} // namespace <anonymous>
std::string to_string(abstract_actor::linking_operation op) {
return detail::enum_to_string(op, linking_operation_strings);
}
} // namespace caf } // namespace caf
...@@ -85,52 +85,22 @@ void forwarding_actor_proxy::enqueue(mailbox_element_ptr what, ...@@ -85,52 +85,22 @@ void forwarding_actor_proxy::enqueue(mailbox_element_ptr what,
} }
} }
bool forwarding_actor_proxy::link_impl(linking_operation op, bool forwarding_actor_proxy::add_backlink(abstract_actor* x) {
abstract_actor* other) { if (monitorable_actor::add_backlink(x)) {
switch (op) { forward_msg(ctrl(), invalid_message_id,
case establish_link_op: make_message(link_atom::value, x->ctrl()));
if (establish_link_impl(other)) { return true;
// causes remote actor to link to (proxy of) other
// receiving peer will call: this->local_link_to(other)
forward_msg(ctrl(), invalid_message_id,
make_message(link_atom::value, other->ctrl()));
return true;
}
break;
case remove_link_op:
if (remove_link_impl(other)) {
// causes remote actor to unlink from (proxy of) other
forward_msg(ctrl(), invalid_message_id,
make_message(unlink_atom::value, other->ctrl()));
return true;
}
break;
case establish_backlink_op:
if (establish_backlink_impl(other)) {
// causes remote actor to unlink from (proxy of) other
forward_msg(ctrl(), invalid_message_id,
make_message(link_atom::value, other->ctrl()));
return true;
}
break;
case remove_backlink_op:
if (remove_backlink_impl(other)) {
// causes remote actor to unlink from (proxy of) other
forward_msg(ctrl(), invalid_message_id,
make_message(unlink_atom::value, other->ctrl()));
return true;
}
break;
} }
return false; return false;
} }
void forwarding_actor_proxy::local_link_to(abstract_actor* other) { bool forwarding_actor_proxy::remove_backlink(abstract_actor* x) {
establish_link_impl(other); if (monitorable_actor::remove_backlink(x)) {
} forward_msg(ctrl(), invalid_message_id,
make_message(unlink_atom::value, x->ctrl()));
void forwarding_actor_proxy::local_unlink_from(abstract_actor* other) { return true;
remove_link_impl(other); }
return false;
} }
void forwarding_actor_proxy::kill_proxy(execution_unit* ctx, error rsn) { void forwarding_actor_proxy::kill_proxy(execution_unit* ctx, error rsn) {
......
...@@ -116,47 +116,37 @@ monitorable_actor::monitorable_actor(actor_config& cfg) : abstract_actor(cfg) { ...@@ -116,47 +116,37 @@ monitorable_actor::monitorable_actor(actor_config& cfg) : abstract_actor(cfg) {
// nop // nop
} }
bool monitorable_actor::link_impl(linking_operation op, abstract_actor* other) { void monitorable_actor::add_link(abstract_actor* x) {
CAF_LOG_TRACE(CAF_ARG(op) << CAF_ARG(other)); // Add backlink on `x` first and add the local attachable only on success.
switch (op) {
case establish_link_op:
return establish_link_impl(other);
case establish_backlink_op:
return establish_backlink_impl(other);
case remove_link_op:
return remove_link_impl(other);
default:
CAF_ASSERT(op == remove_backlink_op);
return remove_backlink_impl(other);
}
}
bool monitorable_actor::establish_link_impl(abstract_actor* x) {
CAF_LOG_TRACE(CAF_ARG(x)); CAF_LOG_TRACE(CAF_ARG(x));
CAF_ASSERT(x); CAF_ASSERT(x != nullptr);
error fail_state; error fail_state;
bool send_exit_immediately = false; bool send_exit_immediately = false;
auto tmp = default_attachable::make_link(address(), x->address()); auto tmp = default_attachable::make_link(address(), x->address());
auto success = exclusive_critical_section([&]() -> bool { joined_exclusive_critical_section(this, x, [&] {
if (getf(is_terminated_flag)) { if (getf(is_terminated_flag)) {
fail_state = fail_state_; fail_state = fail_state_;
send_exit_immediately = true; send_exit_immediately = true;
return false; } else if (x->add_backlink(this)) {
}
// add link if not already linked to other (checked by establish_backlink)
if (x->establish_backlink(this)) {
attach_impl(tmp); attach_impl(tmp);
return true;
} }
return false;
}); });
if (send_exit_immediately) if (send_exit_immediately)
x->enqueue(nullptr, invalid_message_id, x->enqueue(nullptr, invalid_message_id,
make_message(exit_msg{address(), fail_state}), nullptr); make_message(exit_msg{address(), fail_state}), nullptr);
return success;
} }
bool monitorable_actor::establish_backlink_impl(abstract_actor* x) { void monitorable_actor::remove_link(abstract_actor* x) {
CAF_LOG_TRACE(CAF_ARG(x));
default_attachable::observe_token tk{x->address(), default_attachable::link};
joined_exclusive_critical_section(this, x, [&] {
if (x->remove_backlink(this))
detach_impl(tk, attachables_head_, true);
});
}
bool monitorable_actor::add_backlink(abstract_actor* x) {
// Called in an exclusive critical section.
CAF_LOG_TRACE(CAF_ARG(x)); CAF_LOG_TRACE(CAF_ARG(x));
CAF_ASSERT(x); CAF_ASSERT(x);
error fail_state; error fail_state;
...@@ -164,42 +154,25 @@ bool monitorable_actor::establish_backlink_impl(abstract_actor* x) { ...@@ -164,42 +154,25 @@ bool monitorable_actor::establish_backlink_impl(abstract_actor* x) {
default_attachable::observe_token tk{x->address(), default_attachable::observe_token tk{x->address(),
default_attachable::link}; default_attachable::link};
auto tmp = default_attachable::make_link(address(), x->address()); auto tmp = default_attachable::make_link(address(), x->address());
auto success = exclusive_critical_section([&]() -> bool { auto success = false;
if (getf(is_terminated_flag)) { if (getf(is_terminated_flag)) {
fail_state = fail_state_; fail_state = fail_state_;
send_exit_immediately = true; send_exit_immediately = true;
return false; } else if (detach_impl(tk, attachables_head_, true, true) == 0) {
} attach_impl(tmp);
if (detach_impl(tk, attachables_head_, true, true) == 0) { success = true;
attach_impl(tmp); }
return true;
}
return false;
});
if (send_exit_immediately) if (send_exit_immediately)
x->enqueue(nullptr, invalid_message_id, x->enqueue(nullptr, invalid_message_id,
make_message(exit_msg{address(), fail_state}), nullptr); make_message(exit_msg{address(), fail_state}), nullptr);
return success; return success;
} }
bool monitorable_actor::remove_link_impl(abstract_actor* x) { bool monitorable_actor::remove_backlink(abstract_actor* x) {
// Called in an exclusive critical section.
CAF_LOG_TRACE(CAF_ARG(x)); CAF_LOG_TRACE(CAF_ARG(x));
default_attachable::observe_token tk{x->address(), default_attachable::link}; default_attachable::observe_token tk{x->address(), default_attachable::link};
auto success = exclusive_critical_section([&]() -> bool { return detach_impl(tk, attachables_head_, true) > 0;
return detach_impl(tk, attachables_head_, true) > 0;
});
if (success)
x->remove_backlink(this);
return success;
}
bool monitorable_actor::remove_backlink_impl(abstract_actor* x) {
CAF_LOG_TRACE(CAF_ARG(x));
default_attachable::observe_token tk{x->address(), default_attachable::link};
auto success = exclusive_critical_section([&]() -> bool {
return detach_impl(tk, attachables_head_, true) > 0;
});
return success;
} }
size_t monitorable_actor::detach_impl(const attachable::token& what, size_t monitorable_actor::detach_impl(const attachable::token& what,
......
...@@ -256,7 +256,7 @@ void basp_broker_state::deliver(const node_id& src_nid, actor_id src_aid, ...@@ -256,7 +256,7 @@ void basp_broker_state::deliver(const node_id& src_nid, actor_id src_aid,
make_error(sec::remote_linking_failed)); make_error(sec::remote_linking_failed));
return; return;
} }
static_cast<actor_proxy*>(ptr->get())->local_link_to(src->get()); static_cast<actor_proxy*>(ptr->get())->add_link(src->get());
return; return;
} }
case unlink_atom::value.uint_value(): { case unlink_atom::value.uint_value(): {
...@@ -273,7 +273,7 @@ void basp_broker_state::deliver(const node_id& src_nid, actor_id src_aid, ...@@ -273,7 +273,7 @@ void basp_broker_state::deliver(const node_id& src_nid, actor_id src_aid,
CAF_LOG_DEBUG("received unlink for invalid actor, report error"); CAF_LOG_DEBUG("received unlink for invalid actor, report error");
return; return;
} }
static_cast<actor_proxy*>(ptr->get())->local_unlink_from(src->get()); static_cast<actor_proxy*>(ptr->get())->remove_link(src->get());
return; return;
} }
} }
......
...@@ -167,8 +167,7 @@ expected<uint16_t> middleman::publish_local_groups(uint16_t port, ...@@ -167,8 +167,7 @@ expected<uint16_t> middleman::publish_local_groups(uint16_t port,
auto result = publish(gn, port, in); auto result = publish(gn, port, in);
// link gn to our manager // link gn to our manager
if (result) if (result)
manager_->link_impl(abstract_actor::establish_link_op, manager_->add_link(actor_cast<abstract_actor*>(gn));
actor_cast<abstract_actor*>(gn));
else else
anon_send_exit(gn, exit_reason::user_shutdown); anon_send_exit(gn, exit_reason::user_shutdown);
return result; return result;
......
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