Commit bdcee9fa authored by Ilya Leoshkevich's avatar Ilya Leoshkevich

actor_count singleton

parent 0c5df318
#ifndef ACTOR_COUNT_HPP
#define ACTOR_COUNT_HPP
#include <atomic>
#include <boost/thread.hpp>
#include "cppa/attachable.hpp"
namespace cppa { namespace detail {
void inc_actor_count();
void dec_actor_count();
class actor_count {
public:
actor_count();
void inc();
void dec();
// @pre expected <= 1
void actor_count_wait_until(size_t expected);
// @pre expected <= 1
void wait_until(size_t expected);
static actor_count& get();
private:
static actor_count* s_instance;
boost::mutex m_ra_mtx;
boost::condition_variable m_ra_cv;
std::atomic<size_t> m_running_actors;
};
struct exit_observer : cppa::attachable
{
......
#include <limits>
#include <atomic>
#include <stdexcept>
#include <boost/thread.hpp>
#include "cppa/detail/actor_count.hpp"
namespace {
boost::mutex s_ra_mtx;
boost::condition_variable s_ra_cv;
std::atomic<size_t> s_running_actors(0);
typedef boost::unique_lock<boost::mutex> guard_type;
} // namespace <anonymous>
namespace cppa { namespace detail {
void inc_actor_count()
// TODO: free
actor_count* actor_count::s_instance = new actor_count();
actor_count& actor_count::get()
{
return *s_instance;
}
actor_count::actor_count() : m_running_actors(0)
{
}
void actor_count::inc()
{
++s_running_actors;
++m_running_actors;
}
void dec_actor_count()
void actor_count::dec()
{
size_t new_val = --s_running_actors;
size_t new_val = --m_running_actors;
if (new_val == std::numeric_limits<size_t>::max())
{
throw std::underflow_error("dec_actor_count()");
throw std::underflow_error("actor_count::dec()");
}
else if (new_val <= 1)
{
guard_type guard(s_ra_mtx);
s_ra_cv.notify_all();
guard_type guard(m_ra_mtx);
m_ra_cv.notify_all();
}
}
void actor_count_wait_until(size_t expected)
void actor_count::wait_until(size_t expected)
{
guard_type lock(s_ra_mtx);
while (s_running_actors.load() != expected)
guard_type lock(m_ra_mtx);
while (m_running_actors.load() != expected)
{
s_ra_cv.wait(lock);
m_ra_cv.wait(lock);
}
}
exit_observer::~exit_observer()
{
cppa::detail::dec_actor_count();
actor_count::get().dec();
}
} } // namespace cppa::detail
......@@ -34,7 +34,7 @@ void run_actor(cppa::intrusive_ptr<cppa::context> m_self,
catch (...) { }
delete behavior;
}
cppa::detail::dec_actor_count();
cppa::detail::actor_count::get().dec();
}
} // namespace <anonymous>
......@@ -43,7 +43,7 @@ namespace cppa { namespace detail {
actor_ptr mock_scheduler::spawn(actor_behavior* behavior)
{
inc_actor_count();
actor_count::get().inc();
CPPA_MEMORY_BARRIER();
intrusive_ptr<context> ctx(new detail::converted_thread_context);
boost::thread(run_actor, ctx, behavior).detach();
......
......@@ -43,21 +43,21 @@ scheduler::~scheduler()
void scheduler::await_others_done()
{
detail::actor_count_wait_until((unchecked_self() == nullptr) ? 0 : 1);
detail::actor_count::get().wait_until((unchecked_self() == nullptr) ? 0 : 1);
}
void scheduler::register_converted_context(context* what)
{
if (what)
{
detail::inc_actor_count();
detail::actor_count::get().inc();
what->attach(new detail::exit_observer);
}
}
attachable* scheduler::register_hidden_context()
{
detail::inc_actor_count();
detail::actor_count::get().inc();
return new detail::exit_observer;
}
......
......@@ -40,7 +40,7 @@ void task_scheduler::worker_loop(job_queue* jq, scheduled_actor* dummy)
{
if (!job->deref()) delete job;
CPPA_MEMORY_BARRIER();
dec_actor_count();
actor_count::get().dec();
});
}
}
......@@ -75,7 +75,7 @@ void task_scheduler::schedule(scheduled_actor* what)
actor_ptr task_scheduler::spawn(actor_behavior* behavior, scheduling_hint)
{
inc_actor_count();
actor_count::get().inc();
intrusive_ptr<scheduled_actor> ctx(new scheduled_actor(behavior,
enqueue_fun,
this));
......
......@@ -78,7 +78,7 @@ struct thread_pool_scheduler::worker
{
if (!m_job->deref()) delete m_job;
CPPA_MEMORY_BARRIER();
dec_actor_count();
actor_count::get().dec();
});
if (reschedule)
{
......@@ -184,7 +184,7 @@ actor_ptr thread_pool_scheduler::spawn(actor_behavior* behavior,
}
else
{
inc_actor_count();
actor_count::get().inc();
CPPA_MEMORY_BARRIER();
intrusive_ptr<scheduled_actor> ctx(new scheduled_actor(behavior,
enqueue_fun,
......
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