Commit 257b210b authored by Neverlord's avatar Neverlord

Properly shutdown group modules

parent 57d9e752
......@@ -108,6 +108,11 @@ class abstract_group : public abstract_channel {
virtual ~module();
/**
* Stops all groups from this module.
*/
virtual void stop() = 0;
/**
* Returns the name of this module implementation.
* @threadsafe
......@@ -149,6 +154,11 @@ class abstract_group : public abstract_channel {
*/
virtual attachable_ptr subscribe(const actor_addr& who) = 0;
/**
* Stops any background actors or threads and IO handles.
*/
virtual void stop() = 0;
protected:
abstract_group(module_ptr module, std::string group_id);
// called by subscription objects
......
......@@ -32,11 +32,22 @@
namespace caf {
namespace detail {
class group_manager : public singleton_mixin<group_manager> {
class group_manager {
public:
friend class singleton_mixin<group_manager>;
inline void dispose() {
delete this;
}
public:
static inline group_manager* create_singleton() {
return new group_manager;
}
void stop();
inline void initialize() {
// nop
}
~group_manager();
......@@ -50,14 +61,12 @@ class group_manager : public singleton_mixin<group_manager> {
abstract_group::module_ptr get_module(const std::string& module_name);
private:
using modules_map = std::map<std::string, abstract_group::unique_module_ptr>;
modules_map m_mmap;
std::mutex m_mmap_mtx;
group_manager();
modules_map m_mmap;
std::mutex m_mmap_mtx;
};
} // namespace detail
......
......@@ -51,6 +51,10 @@ abstract_group::module::module(std::string mname) : m_name(std::move(mname)) {
// nop
}
void abstract_group::module::stop() {
// nop
}
const std::string& abstract_group::module::name() {
return m_name;
}
......
......@@ -48,6 +48,36 @@ using upgrade_to_unique_guard = upgrade_to_unique_lock<detail::shared_spinlock>;
class local_broker;
class local_group_module;
// simple handshake for one central component and X clients
class latch {
public:
latch(int value) : m_value(value) {
// nop
}
void wait() {
std::unique_lock<std::mutex> guard{m_mtx};
while (m_value != 0) {
m_cv.wait(guard);
}
}
void count_down() {
std::unique_lock<std::mutex> guard;
--m_value;
m_cv.notify_one();
}
~latch() {
CAF_REQUIRE(m_value == 0);
}
private:
volatile int m_value;
std::mutex m_mtx;
std::condition_variable m_cv;
};
class local_group : public abstract_group {
public:
void send_all_subscribers(const actor_addr& sender, const message& msg,
......@@ -100,26 +130,42 @@ class local_group : public abstract_group {
void serialize(serializer* sink);
void stop() override {
anon_send_exit(m_broker, exit_reason::user_shutdown);
m_latch.wait();
}
const actor& broker() const {
return m_broker;
}
local_group(bool spawn_local_broker, local_group_module* mod, std::string id);
~local_group();
protected:
detail::shared_spinlock m_mtx;
std::set<actor_addr> m_subscribers;
actor m_broker;
latch m_latch;
};
using local_group_ptr = intrusive_ptr<local_group>;
class local_broker : public event_based_actor {
public:
local_broker(local_group_ptr g) : m_group(std::move(g)) {
local_broker(local_group_ptr g, latch* hv)
: m_group(std::move(g)),
m_hv(hv) {
// nop
}
void on_exit() {
m_acquaintances.clear();
m_hv->count_down();
m_group.reset();
}
behavior make_behavior() override {
return {
on(atom("JOIN"), arg_match) >> [=](const actor& other) {
......@@ -182,7 +228,7 @@ class local_broker : public event_based_actor {
local_group_ptr m_group;
std::set<actor> m_acquaintances;
latch* m_hv;
};
// Send a "JOIN" message to the original group if a proxy
......@@ -201,7 +247,7 @@ class local_group_proxy : public local_group {
CAF_REQUIRE(m_broker == invalid_actor);
CAF_REQUIRE(remote_broker != invalid_actor);
m_broker = std::move(remote_broker);
m_proxy_broker = spawn<proxy_broker, hidden>(this);
m_proxy_broker = spawn<proxy_broker, hidden>(&m_latch, this);
}
attachable_ptr subscribe(const actor_addr& who) override {
......@@ -235,6 +281,11 @@ class local_group_proxy : public local_group {
make_message(atom("_Forward"), std::move(msg)), eu);
}
void stop() override {
anon_send_exit(m_proxy_broker, exit_reason::user_shutdown);
m_latch.wait();
}
private:
actor m_proxy_broker;
};
......@@ -243,18 +294,28 @@ using local_group_proxy_ptr = intrusive_ptr<local_group_proxy>;
class proxy_broker : public event_based_actor {
public:
proxy_broker(local_group_proxy_ptr grp) : m_group(std::move(grp)) {
proxy_broker(latch* grp_latch, local_group_proxy_ptr grp)
: m_latch(grp_latch),
m_group(std::move(grp)) {
// nop
}
behavior make_behavior() {
return {others >> [=] {
m_group->send_all_subscribers(current_sender(), current_message(),
host());
}};
return {
others >> [=] {
m_group->send_all_subscribers(current_sender(), current_message(),
host());
}
};
}
void on_exit() {
m_group.reset();
m_latch->count_down();
}
private:
latch* m_latch;
local_group_proxy_ptr m_group;
};
......@@ -277,6 +338,9 @@ class local_group_module : public abstract_group::module {
upgrade_to_unique_guard uguard(guard);
auto p = m_instances.insert(make_pair(identifier, tmp));
// someone might preempt us
if (p.first->second != tmp) {
tmp->stop();
}
return {p.first->second};
}
}
......@@ -313,6 +377,23 @@ class local_group_module : public abstract_group::module {
m_actor_utype->serialize(&ptr->broker(), sink);
}
void stop() override {
std::map<std::string, local_group_ptr> imap;
std::map<actor, local_group_ptr> pmap;
{ // critical section
exclusive_guard guard1{m_instances_mtx};
exclusive_guard guard2{m_proxies_mtx};
imap.swap(m_instances);
pmap.swap(m_proxies);
}
for (auto& kvp : imap) {
kvp.second->stop();
}
for (auto& kvp : pmap) {
kvp.second->stop();
}
}
private:
const uniform_type_info* m_actor_utype;
detail::shared_spinlock m_instances_mtx;
......@@ -322,10 +403,17 @@ class local_group_module : public abstract_group::module {
};
local_group::local_group(bool do_spawn, local_group_module* mod, std::string id)
: abstract_group(mod, std::move(id)) {
: abstract_group(mod, std::move(id)),
m_latch(1) {
if (do_spawn) {
m_broker = spawn<local_broker, hidden>(this);
m_broker = spawn<local_broker, hidden>(this, &m_latch);
}
// else: derived class spawns an actor and
// uses the latch for overriding stop()
}
local_group::~local_group() {
// nop
}
void local_group::serialize(serializer* sink) {
......@@ -338,6 +426,17 @@ std::atomic<size_t> m_ad_hoc_id;
} // namespace <anonymous>
void group_manager::stop() {
modules_map mm;
{ // critical section
std::lock_guard<std::mutex> guard(m_mmap_mtx);
mm.swap(m_mmap);
}
for (auto& kvp : mm) {
kvp.second->stop();
}
}
group_manager::~group_manager() {
// nop
}
......
......@@ -69,6 +69,8 @@ std::mutex& singletons::get_plugin_mutex() {
void singletons::stop_singletons() {
// stop singletons, i.e., make sure no background threads/actors are running
CAF_LOGF_DEBUG("stop group manager");
stop(s_group_manager);
CAF_LOGF_DEBUG("stop scheduler");
stop(s_scheduling_coordinator);
CAF_LOGF_DEBUG("stop plugins");
......@@ -77,8 +79,6 @@ void singletons::stop_singletons() {
}
CAF_LOGF_DEBUG("stop actor registry");
stop(s_actor_registry);
CAF_LOGF_DEBUG("stop group manager");
stop(s_group_manager);
CAF_LOGF_DEBUG("stop type info map");
stop(s_uniform_type_info_map);
stop(s_node_id);
......
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