Commit d213894c authored by Dominik Charousset's avatar Dominik Charousset

Await detached threads in await_all_actors_done

parent a7468026
...@@ -41,6 +41,7 @@ set (LIBCAF_CORE_SRCS ...@@ -41,6 +41,7 @@ set (LIBCAF_CORE_SRCS
src/decorated_tuple.cpp src/decorated_tuple.cpp
src/default_attachable.cpp src/default_attachable.cpp
src/deserializer.cpp src/deserializer.cpp
src/detached_threads.cpp
src/duration.cpp src/duration.cpp
src/either.cpp src/either.cpp
src/event_based_actor.cpp src/event_based_actor.cpp
......
...@@ -23,6 +23,7 @@ ...@@ -23,6 +23,7 @@
#include "caf/set_scheduler.hpp" #include "caf/set_scheduler.hpp"
#include "caf/scheduler/worker.hpp" #include "caf/scheduler/worker.hpp"
#include "caf/scheduler/coordinator.hpp" #include "caf/scheduler/coordinator.hpp"
#include "caf/scheduler/detached_threads.hpp"
#include "caf/scheduler/abstract_coordinator.hpp" #include "caf/scheduler/abstract_coordinator.hpp"
#endif // CAF_SCHEDULER_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
...@@ -17,15 +17,18 @@ ...@@ -17,15 +17,18 @@
* http://www.boost.org/LICENSE_1_0.txt. * * http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/ ******************************************************************************/
#include "caf/detail/actor_registry.hpp"
#include <mutex> #include <mutex>
#include <limits> #include <limits>
#include <stdexcept> #include <stdexcept>
#include "caf/locks.hpp"
#include "caf/attachable.hpp" #include "caf/attachable.hpp"
#include "caf/exit_reason.hpp" #include "caf/exit_reason.hpp"
#include "caf/detail/actor_registry.hpp"
#include "caf/locks.hpp" #include "caf/scheduler/detached_threads.hpp"
#include "caf/detail/logging.hpp" #include "caf/detail/logging.hpp"
#include "caf/detail/shared_spinlock.hpp" #include "caf/detail/shared_spinlock.hpp"
...@@ -119,6 +122,9 @@ void actor_registry::await_running_count_equal(size_t expected) { ...@@ -119,6 +122,9 @@ void actor_registry::await_running_count_equal(size_t expected) {
CAF_LOG_DEBUG("count = " << m_running.load()); CAF_LOG_DEBUG("count = " << m_running.load());
m_running_cv.wait(guard); m_running_cv.wait(guard);
} }
// also wait for all detached threads to make sure
// destructore were called correctly
scheduler::await_detached_threads();
} }
} // namespace detail } // namespace detail
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| 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
...@@ -25,9 +25,12 @@ ...@@ -25,9 +25,12 @@
#include "caf/local_actor.hpp" #include "caf/local_actor.hpp"
#include "caf/default_attachable.hpp" #include "caf/default_attachable.hpp"
#include "caf/scheduler/detached_threads.hpp"
#include "caf/detail/logging.hpp" #include "caf/detail/logging.hpp"
#include "caf/detail/sync_request_bouncer.hpp" #include "caf/detail/sync_request_bouncer.hpp"
namespace caf { namespace caf {
// local actors are created with a reference count of one that is adjusted // local actors are created with a reference count of one that is adjusted
...@@ -464,19 +467,25 @@ void local_actor::launch(execution_unit* eu, bool lazy, bool hide) { ...@@ -464,19 +467,25 @@ void local_actor::launch(execution_unit* eu, bool lazy, bool hide) {
// actor lives in its own thread // actor lives in its own thread
CAF_PUSH_AID(id()); CAF_PUSH_AID(id());
CAF_LOG_TRACE(CAF_ARG(lazy) << ", " << CAF_ARG(hide)); CAF_LOG_TRACE(CAF_ARG(lazy) << ", " << CAF_ARG(hide));
intrusive_ptr<local_actor> mself{this}; if (!hide) {
attach_to_scheduler(); // hiding the actor also hides the thread
std::thread([=] { scheduler::inc_detached_threads();
CAF_PUSH_AID(id()); }
//intrusive_ptr<local_actor> mself{this};
std::thread([hide](intrusive_ptr<local_actor> mself) {
CAF_PUSH_AID(mself->id());
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
auto max_throughput = std::numeric_limits<size_t>::max(); auto max_throughput = std::numeric_limits<size_t>::max();
while (resume(nullptr, max_throughput) != resumable::done) { while (mself->resume(nullptr, max_throughput) != resumable::done) {
// await new data before resuming actor // await new data before resuming actor
await_data(); mself->await_data();
CAF_ASSERT(mailbox().blocked() == false); CAF_ASSERT(mself->mailbox().blocked() == false);
}
mself.reset();
if (!hide) {
scheduler::dec_detached_threads();
} }
detach_from_scheduler(); }, intrusive_ptr<local_actor>{this}).detach();
}).detach();
return; return;
} }
// actor is cooperatively scheduled // actor is cooperatively scheduled
...@@ -503,9 +512,7 @@ void local_actor::enqueue(mailbox_element_ptr ptr, execution_unit* eu) { ...@@ -503,9 +512,7 @@ void local_actor::enqueue(mailbox_element_ptr ptr, execution_unit* eu) {
auto mid = ptr->mid; auto mid = ptr->mid;
auto sender = ptr->sender; auto sender = ptr->sender;
// returns false if mailbox has been closed // returns false if mailbox has been closed
if (!mailbox().synchronized_enqueue(m_mtx, if (!mailbox().synchronized_enqueue(m_mtx, m_cv, ptr.release())) {
m_cv,
ptr.release())) {
if (mid.is_request()) { if (mid.is_request()) {
detail::sync_request_bouncer srb{exit_reason()}; detail::sync_request_bouncer srb{exit_reason()};
srb(sender, mid); srb(sender, mid);
......
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