Commit a39a4c5a authored by Dominik Charousset's avatar Dominik Charousset

Fix shutdown issue when using blocking actors

parent b9f000ae
...@@ -12,6 +12,9 @@ is based on [Keep a Changelog](https://keepachangelog.com). ...@@ -12,6 +12,9 @@ is based on [Keep a Changelog](https://keepachangelog.com).
when setting a `timespan` parameter such as `caf.middleman.heartbeat-interval` when setting a `timespan` parameter such as `caf.middleman.heartbeat-interval`
via config file or CLI to `0s` and then printing the config parameter, e.g., via config file or CLI to `0s` and then printing the config parameter, e.g.,
via `--dump-config`. via `--dump-config`.
- Blocking actors now release their private thread before decrementing the
running-actors count to resolve a race condition during system shutdown that
could result in the system hanging (#1266).
### Removed ### Removed
......
...@@ -93,8 +93,8 @@ namespace { ...@@ -93,8 +93,8 @@ namespace {
class blocking_actor_runner : public resumable { class blocking_actor_runner : public resumable {
public: public:
explicit blocking_actor_runner(blocking_actor* self, explicit blocking_actor_runner(blocking_actor* self,
detail::private_thread* thread) detail::private_thread* thread, bool hidden)
: self_(self), thread_(thread) { : self_(self), thread_(thread), hidden_(hidden) {
intrusive_ptr_add_ref(self->ctrl()); intrusive_ptr_add_ref(self->ctrl());
} }
...@@ -127,7 +127,13 @@ public: ...@@ -127,7 +127,13 @@ public:
#endif #endif
self_->cleanup(std::move(rsn), ctx); self_->cleanup(std::move(rsn), ctx);
intrusive_ptr_release(self_->ctrl()); intrusive_ptr_release(self_->ctrl());
ctx->system().release_private_thread(thread_); auto& sys = ctx->system();
sys.release_private_thread(thread_);
if (!hidden_) {
[[maybe_unused]] auto count = sys.registry().dec_running();
CAF_LOG_DEBUG("actor" << self_->id() << "decreased running count to"
<< count);
}
return resumable::done; return resumable::done;
} }
...@@ -142,6 +148,7 @@ public: ...@@ -142,6 +148,7 @@ public:
private: private:
blocking_actor* self_; blocking_actor* self_;
detail::private_thread* thread_; detail::private_thread* thread_;
bool hidden_;
}; };
} // namespace } // namespace
...@@ -150,10 +157,17 @@ void blocking_actor::launch(execution_unit*, bool, bool hide) { ...@@ -150,10 +157,17 @@ void blocking_actor::launch(execution_unit*, bool, bool hide) {
CAF_PUSH_AID_FROM_PTR(this); CAF_PUSH_AID_FROM_PTR(this);
CAF_LOG_TRACE(CAF_ARG(hide)); CAF_LOG_TRACE(CAF_ARG(hide));
CAF_ASSERT(getf(is_blocking_flag)); CAF_ASSERT(getf(is_blocking_flag));
if (!hide) // Try to acquire a thread before incrementing the running count, since this
register_at_system(); // may throw.
auto thread = home_system().acquire_private_thread(); auto& sys = home_system();
thread->resume(new blocking_actor_runner(this, thread)); auto thread = sys.acquire_private_thread();
// Note: must *not* call register_at_system() to stop actor cleanup from
// decrementing the count before releasing the thread.
if (!hide) {
[[maybe_unused]] auto count = sys.registry().inc_running();
CAF_LOG_DEBUG("actor" << id() << "increased running count to" << count);
}
thread->resume(new blocking_actor_runner(this, thread, hide));
} }
blocking_actor::receive_while_helper blocking_actor::receive_while_helper
......
...@@ -67,7 +67,19 @@ private_thread* private_thread_pool::acquire() { ...@@ -67,7 +67,19 @@ private_thread* private_thread_pool::acquire() {
std::unique_lock guard{mtx_}; std::unique_lock guard{mtx_};
++running_; ++running_;
} }
#ifdef CAF_ENABLE_EXCEPTIONS
try {
return private_thread::launch(sys_);
} catch (...) {
{
std::unique_lock guard{mtx_};
--running_;
}
throw;
}
#else
return private_thread::launch(sys_); return private_thread::launch(sys_);
#endif
} }
void private_thread_pool::release(private_thread* ptr) { void private_thread_pool::release(private_thread* ptr) {
......
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