Commit fb005263 authored by Dominik Charousset's avatar Dominik Charousset

Move detached thread counting to actor system

parent 4046105b
......@@ -72,15 +72,6 @@ public:
/// (must be either 0 or 1).
void await_running_count_equal(size_t expected) const;
/// Increases running-detached-threads-count by one.
void inc_detached_threads();
/// Decreases running-detached-threads-count by one.
void dec_detached_threads();
/// Blocks the caller until all detached threads are done.
void await_detached_threads();
/// Returns the actor associated with `key` or `invalid_actor`.
strong_actor_ptr get(atom_value key) const;
......@@ -115,10 +106,6 @@ private:
name_map named_entries_;
mutable detail::shared_spinlock named_entries_mtx_;
std::atomic<size_t> detached;
mutable std::mutex detached_mtx;
mutable std::condition_variable detached_cv;
actor_system& system_;
};
......
......@@ -20,11 +20,13 @@
#ifndef CAF_ACTOR_SYSTEM_HPP
#define CAF_ACTOR_SYSTEM_HPP
#include <mutex>
#include <atomic>
#include <string>
#include <memory>
#include <cstddef>
#include <functional>
#include <condition_variable>
#include "caf/fwd.hpp"
#include "caf/logger.hpp"
......@@ -408,9 +410,20 @@ public:
}
/// @cond PRIVATE
/// Increases running-detached-threads-count by one.
void inc_detached_threads();
/// Decreases running-detached-threads-count by one.
void dec_detached_threads();
/// Blocks the caller until all detached threads are done.
void await_detached_threads();
inline atom_value backend_name() const {
return backend_name_;
}
/// @endcond
private:
......@@ -458,6 +471,10 @@ private:
bool await_actors_before_shutdown_;
strong_actor_ptr config_serv_;
strong_actor_ptr spawn_serv_;
std::atomic<size_t> detached;
mutable std::mutex detached_mtx;
mutable std::condition_variable detached_cv;
};
} // namespace caf
......
......@@ -84,7 +84,7 @@ void actor_registry::put(actor_id key, strong_actor_ptr val) {
}
void actor_registry::erase(actor_id key) {
exclusive_guard guard(instances_mtx_);
exclusive_guard guard{instances_mtx_};
entries_.erase(key);
}
......@@ -120,24 +120,6 @@ void actor_registry::await_running_count_equal(size_t expected) const {
}
}
void actor_registry::inc_detached_threads() {
++detached;
}
void actor_registry::dec_detached_threads() {
if (--detached == 0) {
std::unique_lock<std::mutex> guard{detached_mtx};
detached_cv.notify_all();
}
}
void actor_registry::await_detached_threads() {
std::unique_lock<std::mutex> guard{detached_mtx};
while (detached != 0) {
detached_cv.wait(guard);
}
}
strong_actor_ptr actor_registry::get(atom_value key) const {
shared_guard guard{named_entries_mtx_};
auto i = named_entries_.find(key);
......@@ -155,6 +137,11 @@ void actor_registry::put(atom_value key, strong_actor_ptr value) {
named_entries_.emplace(key, std::move(value));
}
void actor_registry::erase(atom_value key) {
exclusive_guard guard{named_entries_mtx_};
named_entries_.erase(key);
}
auto actor_registry::named_actors() const -> name_map {
shared_guard guard{named_entries_mtx_};
return named_entries_;
......
......@@ -43,7 +43,7 @@ struct kvstate {
using topic_set = std::unordered_set<std::string>;
std::unordered_map<key_type, std::pair<mapped_type, subscriber_set>> data;
std::unordered_map<strong_actor_ptr, topic_set> subscribers;
const char* name = "caf.config_server";
static const char* name;
template <class Processor>
friend void serialize(Processor& proc, kvstate& x, const unsigned int) {
proc & x.data;
......@@ -51,6 +51,8 @@ struct kvstate {
}
};
const char* kvstate::name = "caf.config_server";
behavior config_serv_impl(stateful_actor<kvstate>* self) {
CAF_LOG_TRACE("");
std::string wildcard = "*";
......@@ -188,7 +190,8 @@ actor_system::actor_system(actor_system_config&& cfg)
groups_(*this),
middleman_(nullptr),
dummy_execution_unit_(this),
await_actors_before_shutdown_(true) {
await_actors_before_shutdown_(true),
detached(0) {
CAF_SET_LOGGER_SYS(this);
backend_name_ = cfg.middleman_network_backend;
for (auto& f : cfg.module_factories_) {
......@@ -280,12 +283,15 @@ actor_system::~actor_system() {
// release memory as soon as possible
spawn_serv_ = nullptr;
config_serv_ = nullptr;
registry_.erase(atom("SpawnServ"));
registry_.erase(atom("ConfigServ"));
// group module is the first one, relies on MM
groups_.stop();
// stop modules in reverse order
for (auto i = modules_.rbegin(); i != modules_.rend(); ++i)
if (*i)
(*i)->stop();
await_detached_threads();
registry_.stop();
logger_.stop();
CAF_SET_LOGGER_SYS(nullptr);
......@@ -370,4 +376,21 @@ void actor_system::await_all_actors_done() const {
registry_.await_running_count_equal(0);
}
void actor_system::inc_detached_threads() {
++detached;
}
void actor_system::dec_detached_threads() {
if (--detached == 0) {
std::unique_lock<std::mutex> guard{detached_mtx};
detached_cv.notify_all();
}
}
void actor_system::await_detached_threads() {
std::unique_lock<std::mutex> guard{detached_mtx};
while (detached != 0)
detached_cv.wait(guard);
}
} // namespace caf
......@@ -52,9 +52,9 @@ public:
: self_destroyed_(false),
self_(self),
state_(active),
registry(self->system().registry()) {
system_(self->system()) {
intrusive_ptr_add_ref(self->ctrl());
registry.inc_detached_threads();
system_.inc_detached_threads();
}
void run() {
......@@ -114,7 +114,7 @@ public:
// detached actor is destroyed and this object is unreachable
this_ptr->await_self_destroyed();
// signalize destruction of detached thread to registry
this_ptr->registry.dec_detached_threads();
this_ptr->system_.dec_detached_threads();
// done
delete this_ptr;
}
......@@ -141,7 +141,7 @@ private:
volatile bool self_destroyed_;
volatile local_actor* self_;
volatile worker_state state_;
actor_registry& registry;
actor_system& system_;
};
result<message> reflect(local_actor*, const type_erased_tuple* x) {
......
......@@ -237,7 +237,8 @@ void basp_broker_state::deliver(const node_id& src_nid, actor_id src_aid,
auto rsn = exit_reason::remote_link_unreachable;
CAF_LOG_INFO("cannot deliver message, destination not found");
self->parent().notify<hook::invalid_message_received>(src_nid, src,
0, mid, msg);
invalid_actor_id,
mid, msg);
if (mid.valid() && src) {
detail::sync_request_bouncer srb{rsn};
srb(src, mid);
......@@ -296,10 +297,10 @@ void basp_broker_state::learned_new_node(const node_id& nid) {
using namespace detail;
system().registry().put(tmp.id(), actor_cast<strong_actor_ptr>(tmp));
auto writer = make_callback([](serializer& sink) {
auto name = atom("SpawnServ");
auto name_atm = atom("SpawnServ");
std::vector<actor_id> stages;
auto msg = make_message(sys_atom::value, get_atom::value, "info");
sink << name << stages << msg;
sink << name_atm << stages << msg;
});
auto path = instance.tbl().lookup(nid);
if (! path) {
......@@ -399,10 +400,10 @@ void basp_broker_state::learned_new_node_indirectly(const node_id& nid) {
auto tmp = system().spawn<detached + hidden>(connection_helper, self);
system().registry().put(tmp.id(), actor_cast<strong_actor_ptr>(tmp));
auto writer = make_callback([](serializer& sink) {
auto name = atom("ConfigServ");
auto name_atm = atom("ConfigServ");
std::vector<actor_id> stages;
auto msg = make_message(get_atom::value, "basp.default-connectivity");
sink << name << stages << msg;
sink << name_atm << stages << msg;
});
basp::header hdr{basp::message_type::dispatch_message,
basp::header::named_receiver_flag,
......
......@@ -313,6 +313,7 @@ void middleman::stop() {
scoped_actor self{system(), true};
self->send_exit(manager_, exit_reason::user_shutdown);
self->wait_for(manager_);
invalidate(manager_);
}
void middleman::init(actor_system_config& cfg) {
......
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