Commit b78cb1a3 authored by Dominik Charousset's avatar Dominik Charousset

Wait for BASP workers on MM shutdown

parent 6ac92c6b
......@@ -205,6 +205,10 @@ public:
return this_node_;
}
worker_hub& hub() {
return hub_;
}
actor_system& system() {
return callee_.proxies().system();
}
......
......@@ -19,6 +19,7 @@
#pragma once
#include <atomic>
#include <mutex>
#include "caf/fwd.hpp"
#include "caf/io/basp/fwd.hpp"
......@@ -47,7 +48,7 @@ public:
// -- properties -------------------------------------------------------------
/// Creates a new worker and adds it to the hub.
void push_new_worker(message_queue&, proxy_registry&);
void add_new_worker(message_queue&, proxy_registry&);
/// Add a worker to the hub.
void push(pointer ptr);
......@@ -62,10 +63,19 @@ public:
/// hub is currently empty.
pointer peek();
/// Waits until all workers are back at the hub.
void await_workers();
private:
// -- member variables -------------------------------------------------------
std::atomic<pointer> head_;
std::atomic<size_t> running_;
std::mutex mtx_;
std::condition_variable cv_;
};
} // namespace basp
......
......@@ -73,6 +73,11 @@ basp_broker::~basp_broker() {
// -- implementation of local_actor/broker -------------------------------------
void basp_broker::on_exit() {
// Wait until all pending messages of workers have been shipped.
// TODO: this blocks the calling thread. This is only safe because we know
// that the middleman calls this in its stop() function. However,
// ultimately we should find a nonblocking solution here.
instance.hub().await_workers();
// Release any obsolete state.
ctx.clear();
// Make sure all spawn servers are down before clearing the container.
......
......@@ -49,7 +49,7 @@ instance::instance(abstract_broker* parent, callee& lstnr)
auto workers = get_or(config(), "middleman.workers",
defaults::middleman::workers);
for (size_t i = 0; i < workers; ++i)
hub_.push_new_worker(queue_, proxies());
hub_.add_new_worker(queue_, proxies());
}
connection_state instance::handle(execution_unit* ctx,
......
......@@ -329,18 +329,12 @@ void middleman::stop() {
while (backend().try_run_once())
; // nop
}
named_brokers_.clear();
scoped_actor self{system(), true};
self->send_exit(manager_, exit_reason::kill);
if (!get_or(config(), "middleman.attach-utility-actors", false))
self->wait_for(manager_);
destroy(manager_);
// Note: we intentionally don't call `named_brokers_.clear()` here. The BASP
// broker must outlive the scheduler threads. The scheduler is stopped
// *after* the MM. However, the BASP workers still need to return to their
// hub safely, should some of them are still running at this point. By not
// clearing the container, we keep the BASP broker (and thus the BASP
// instance) alive until the MM module gets destroyed. At that point, all
// BASP workers are safe in their hub.
}
void middleman::init(actor_system_config& cfg) {
......
......@@ -26,11 +26,12 @@ namespace basp {
// -- constructors, destructors, and assignment operators ----------------------
worker_hub::worker_hub() : head_(nullptr) {
worker_hub::worker_hub() : head_(nullptr), running_(0) {
// nop
}
worker_hub::~worker_hub() {
await_workers();
auto head = head_.load();
while (head != nullptr) {
auto next = head->next_.load();
......@@ -41,18 +42,28 @@ worker_hub::~worker_hub() {
// -- properties ---------------------------------------------------------------
void worker_hub::push_new_worker(message_queue& queue,
proxy_registry& proxies) {
push(new worker(*this, queue, proxies));
void worker_hub::add_new_worker(message_queue& queue, proxy_registry& proxies) {
auto ptr = new worker(*this, queue, proxies);
auto next = head_.load();
for (;;) {
ptr->next_ = next;
if (head_.compare_exchange_strong(next, ptr))
return;
}
}
void worker_hub::push(pointer ptr) {
auto next = head_.load();
for (;;) {
ptr->next_ = next;
if (head_.compare_exchange_strong(next, ptr))
if (head_.compare_exchange_strong(next, ptr)) {
if (--running_ == 0) {
std::unique_lock<std::mutex> guard{mtx_};
cv_.notify_all();
}
return;
}
}
}
worker_hub::pointer worker_hub::pop() {
......@@ -61,15 +72,24 @@ worker_hub::pointer worker_hub::pop() {
return nullptr;
for (;;) {
auto next = result->next_.load();
if (head_.compare_exchange_strong(result, next))
if (head_.compare_exchange_strong(result, next)) {
if (result != nullptr)
++running_;
return result;
}
}
}
worker_hub::pointer worker_hub::peek() {
return head_.load();
}
void worker_hub::await_workers() {
std::unique_lock<std::mutex> guard{mtx_};
while (running_ != 0)
cv_.wait(guard);
}
} // namespace basp
} // namespace io
} // namespace caf
......@@ -105,7 +105,7 @@ CAF_TEST_FIXTURE_SCOPE(worker_tests, fixture)
CAF_TEST(deliver serialized message) {
CAF_MESSAGE("create the BASP worker");
CAF_REQUIRE_EQUAL(hub.peek(), nullptr);
hub.push_new_worker(queue, proxies);
hub.add_new_worker(queue, proxies);
CAF_REQUIRE_NOT_EQUAL(hub.peek(), nullptr);
auto w = hub.pop();
CAF_MESSAGE("create a fake message + BASP header");
......
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