Commit 590fe019 authored by Dominik Charousset's avatar Dominik Charousset

Move global detached_threads to actor registry

parent 98faf8b9
......@@ -41,7 +41,6 @@ set (LIBCAF_CORE_SRCS
src/deep_to_string.cpp
src/default_attachable.cpp
src/deserializer.cpp
src/detached_threads.cpp
src/duration.cpp
src/dynamic_message_data.cpp
src/error.cpp
......
......@@ -72,6 +72,15 @@ 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;
......@@ -97,7 +106,6 @@ private:
actor_registry(actor_system& sys);
std::atomic<size_t> running_;
mutable std::mutex running_mtx_;
mutable std::condition_variable running_cv_;
......@@ -107,6 +115,10 @@ 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_;
};
......
......@@ -22,7 +22,6 @@
#include "caf/scheduler/worker.hpp"
#include "caf/scheduler/coordinator.hpp"
#include "caf/scheduler/detached_threads.hpp"
#include "caf/scheduler/abstract_coordinator.hpp"
#endif // CAF_SCHEDULER_HPP
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2015 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#ifndef CAF_SCHEDULER_DETACHED_THREADS_HPP
#define CAF_SCHEDULER_DETACHED_THREADS_HPP
#include <cstddef>
namespace caf {
namespace scheduler {
/// Increases count for detached threads by one.
void inc_detached_threads();
/// Decreases count for detached threads by one.
void dec_detached_threads();
/// Blocks the caller until all detached threads are done.
void await_detached_threads();
} // namespace scheduler
} // namespace caf
#endif // CAF_SCHEDULER_DETACHED_THREADS_HPP
......@@ -38,8 +38,6 @@
#include "caf/event_based_actor.hpp"
#include "caf/uniform_type_info_map.hpp"
#include "caf/scheduler/detached_threads.hpp"
#include "caf/detail/shared_spinlock.hpp"
namespace caf {
......@@ -122,6 +120,24 @@ 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);
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2015 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENSE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#include "caf/scheduler/detached_threads.hpp"
#include <mutex>
#include <atomic>
#include <condition_variable>
namespace caf {
namespace scheduler {
namespace {
std::atomic<size_t> s_detached;
std::mutex s_detached_mtx;
std::condition_variable s_detached_cv;
} // namespace <anonymous>
void inc_detached_threads() {
++s_detached;
}
void dec_detached_threads() {
size_t new_val = --s_detached;
if (new_val == 0) {
std::unique_lock<std::mutex> guard(s_detached_mtx);
s_detached_cv.notify_all();
}
}
void await_detached_threads() {
std::unique_lock<std::mutex> guard{s_detached_mtx};
while (s_detached != 0) {
s_detached_cv.wait(guard);
}
}
} // namespace scheduler
} // namespace caf
......@@ -51,9 +51,10 @@ public:
private_thread(local_actor* self)
: self_destroyed_(false),
self_(self),
state_(active) {
state_(active),
registry(self->system().registry()) {
intrusive_ptr_add_ref(self->ctrl());
scheduler::inc_detached_threads();
registry.inc_detached_threads();
}
void run() {
......@@ -109,10 +110,12 @@ public:
static void exec(private_thread* this_ptr) {
this_ptr->run();
scheduler::dec_detached_threads();
// make sure to not destroy the private thread object before the
// 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();
// done
delete this_ptr;
}
......@@ -138,6 +141,7 @@ private:
volatile bool self_destroyed_;
volatile local_actor* self_;
volatile worker_state state_;
actor_registry& registry;
};
result<message> reflect(local_actor*, const type_erased_tuple* x) {
......
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