Commit 37b08840 authored by Dominik Charousset's avatar Dominik Charousset

Make sure actors empty their mailbox on shutdown

parent fa464bf5
...@@ -78,6 +78,8 @@ public: ...@@ -78,6 +78,8 @@ public:
void* subtype_ptr() override; void* subtype_ptr() override;
static void cleanup_and_release(resumable*);
protected: protected:
void stop_actors(); void stop_actors();
......
...@@ -123,9 +123,7 @@ protected: ...@@ -123,9 +123,7 @@ protected:
w->get_thread().join(); w->get_thread().join();
} }
// run cleanup code for each resumable // run cleanup code for each resumable
auto f = [](resumable* job) { auto f = &abstract_coordinator::cleanup_and_release;
intrusive_ptr_release(job);
};
for (auto& w : workers_) for (auto& w : workers_)
policy_.foreach_resumable(w.get(), f); policy_.foreach_resumable(w.get(), f);
policy_.foreach_central_resumable(this, f); policy_.foreach_central_resumable(this, f);
......
...@@ -348,5 +348,43 @@ abstract_coordinator::abstract_coordinator(actor_system& sys) ...@@ -348,5 +348,43 @@ abstract_coordinator::abstract_coordinator(actor_system& sys)
// nop // nop
} }
void abstract_coordinator::cleanup_and_release(resumable* ptr) {
class dummy_unit : public execution_unit {
public:
dummy_unit(local_actor* ptr) : execution_unit(&ptr->home_system()) {
// nop
}
void exec_later(resumable* ptr) override {
resumables.push_back(ptr);
}
std::vector<resumable*> resumables;
};
switch (ptr->subtype()) {
case resumable::scheduled_actor:
case resumable::io_actor: {
auto dptr = static_cast<local_actor*>(ptr);
dummy_unit dummy{dptr};
dptr->cleanup(make_error(exit_reason::user_shutdown), &dummy);
while (! dummy.resumables.empty()) {
auto sub = dummy.resumables.back();
dummy.resumables.pop_back();
switch (sub->subtype()) {
case resumable::scheduled_actor:
case resumable::io_actor: {
auto dsub = static_cast<local_actor*>(sub);
dsub->cleanup(make_error(exit_reason::user_shutdown), &dummy);
}
default:
break;
}
}
break;
}
default:
break;
}
intrusive_ptr_release(ptr);
}
} // namespace scheduler } // namespace scheduler
} // namespace caf } // namespace caf
...@@ -695,18 +695,17 @@ default_multiplexer::~default_multiplexer() { ...@@ -695,18 +695,17 @@ default_multiplexer::~default_multiplexer() {
nonblocking(pipe_.first, true); nonblocking(pipe_.first, true);
auto ptr = pipe_reader_.try_read_next(); auto ptr = pipe_reader_.try_read_next();
while (ptr) { while (ptr) {
intrusive_ptr_release(ptr); scheduler::abstract_coordinator::cleanup_and_release(ptr);
ptr = pipe_reader_.try_read_next(); ptr = pipe_reader_.try_read_next();
} }
// do cleanup for pipe reader manually, since WSACleanup needs to happen last // do cleanup for pipe reader manually, since WSACleanup needs to happen last
closesocket(pipe_reader_.fd()); closesocket(pipe_reader_.fd());
pipe_reader_.init(invalid_native_socket); pipe_reader_.init(invalid_native_socket);
# ifdef CAF_WINDOWS # ifdef CAF_WINDOWS
WSACleanup(); WSACleanup();
# endif # endif
} }
void default_multiplexer::exec_later(resumable* ptr) { void default_multiplexer::exec_later(resumable* ptr) {
CAF_ASSERT(ptr); CAF_ASSERT(ptr);
switch (ptr->subtype()) { switch (ptr->subtype()) {
......
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