scheduler.hpp
// ---------------------------------------------------------------------
// pion: a Boost C++ framework for building lightweight HTTP interfaces
// ---------------------------------------------------------------------
// Copyright (C) 2007-2014 Splunk Inc. (https://github.com/splunk/pion)
//
// Distributed under the Boost Software License, Version 1.0.
// See http://www.boost.org/LICENSE_1_0.txt
//
#ifndef __PION_SCHEDULER_HEADER__
#define __PION_SCHEDULER_HEADER__
#include <vector>
#include <boost/asio.hpp>
#include <boost/assert.hpp>
#include <boost/bind.hpp>
#include <boost/function/function0.hpp>
#include <boost/cstdint.hpp>
#include <boost/shared_ptr.hpp>
#include <boost/noncopyable.hpp>
#include <boost/thread/thread.hpp>
#include <boost/thread/mutex.hpp>
#include <boost/thread/xtime.hpp>
#include <boost/thread/condition.hpp>
#include <pion/config.hpp>
#include <pion/logger.hpp>
namespace pion { // begin namespace pion
class PION_API scheduler :
private boost::noncopyable
{
public:
scheduler(void)
: m_logger(PION_GET_LOGGER("pion.scheduler")),
m_num_threads(DEFAULT_NUM_THREADS), m_active_users(0), m_is_running(false)
{}
virtual ~scheduler() {}
virtual void startup(void) {}
virtual void shutdown(void);
void join(void);
void add_active_user(void);
void remove_active_user(void);
inline bool is_running(void) const { return m_is_running; }
inline void set_num_threads(const boost::uint32_t n) { m_num_threads = n; }
inline boost::uint32_t get_num_threads(void) const { return m_num_threads; }
inline void set_logger(logger log_ptr) { m_logger = log_ptr; }
inline logger get_logger(void) { return m_logger; }
virtual boost::asio::io_service& get_io_service(void) = 0;
virtual void post(boost::function0<void> work_func) {
get_io_service().post(work_func);
}
void keep_running(boost::asio::io_service& my_service,
boost::asio::deadline_timer& my_timer);
inline static void sleep(boost::uint32_t sleep_sec, boost::uint32_t sleep_nsec) {
boost::system_time wakeup_time(get_wakeup_time(sleep_sec, sleep_nsec));
boost::thread::sleep(wakeup_time);
}
template <typename ConditionType, typename LockType>
inline static void sleep(ConditionType& wakeup_condition, LockType& wakeup_lock,
boost::uint32_t sleep_sec, boost::uint32_t sleep_nsec)
{
boost::system_time wakeup_time(get_wakeup_time(sleep_sec, sleep_nsec));
wakeup_condition.timed_wait(wakeup_lock, wakeup_time);
}
void process_service_work(boost::asio::io_service& service);
protected:
static boost::system_time get_wakeup_time(boost::uint32_t sleep_sec,
boost::uint32_t sleep_nsec);
virtual void stop_services(void) {}
virtual void stop_threads(void) {}
virtual void finish_services(void) {}
virtual void finish_threads(void) {}
static const boost::uint32_t DEFAULT_NUM_THREADS;
static const boost::uint32_t NSEC_IN_SECOND;
static const boost::uint32_t MICROSEC_IN_SECOND;
static const boost::uint32_t KEEP_RUNNING_TIMER_SECONDS;
boost::mutex m_mutex;
logger m_logger;
boost::condition m_no_more_active_users;
boost::condition m_scheduler_has_stopped;
boost::uint32_t m_num_threads;
boost::uint32_t m_active_users;
bool m_is_running;
};
class PION_API multi_thread_scheduler :
public scheduler
{
public:
multi_thread_scheduler(void) {}
virtual ~multi_thread_scheduler() {}
protected:
virtual void stop_threads(void) {
if (! m_thread_pool.empty()) {
PION_LOG_DEBUG(m_logger, "Waiting for threads to shutdown");
// wait until all threads in the pool have stopped
boost::thread current_thread;
for (ThreadPool::iterator i = m_thread_pool.begin();
i != m_thread_pool.end(); ++i)
{
// make sure we do not call join() for the current thread,
// since this may yield "undefined behavior"
if (**i != current_thread) (*i)->join();
}
}
}
virtual void finish_threads(void) { m_thread_pool.clear(); }
typedef std::vector<boost::shared_ptr<boost::thread> > ThreadPool;
ThreadPool m_thread_pool;
};
class PION_API single_service_scheduler :
public multi_thread_scheduler
{
public:
single_service_scheduler(void)
: m_service(), m_timer(m_service)
{}
virtual ~single_service_scheduler() { shutdown(); }
virtual boost::asio::io_service& get_io_service(void) { return m_service; }
virtual void startup(void);
protected:
virtual void stop_services(void) { m_service.stop(); }
virtual void finish_services(void) { m_service.reset(); }
boost::asio::io_service m_service;
boost::asio::deadline_timer m_timer;
};
class PION_API one_to_one_scheduler :
public multi_thread_scheduler
{
public:
one_to_one_scheduler(void)
: m_service_pool(), m_next_service(0)
{}
virtual ~one_to_one_scheduler() { shutdown(); }
virtual boost::asio::io_service& get_io_service(void) {
boost::mutex::scoped_lock scheduler_lock(m_mutex);
while (m_service_pool.size() < m_num_threads) {
boost::shared_ptr<service_pair_type> service_ptr(new service_pair_type());
m_service_pool.push_back(service_ptr);
}
if (++m_next_service >= m_num_threads)
m_next_service = 0;
BOOST_ASSERT(m_next_service < m_num_threads);
return m_service_pool[m_next_service]->first;
}
virtual boost::asio::io_service& get_io_service(boost::uint32_t n) {
BOOST_ASSERT(n < m_num_threads);
BOOST_ASSERT(n < m_service_pool.size());
return m_service_pool[n]->first;
}
virtual void startup(void);
protected:
virtual void stop_services(void) {
for (service_pool_type::iterator i = m_service_pool.begin(); i != m_service_pool.end(); ++i) {
(*i)->first.stop();
}
}
virtual void finish_services(void) { m_service_pool.clear(); }
struct service_pair_type {
service_pair_type(void) : first(), second(first) {}
boost::asio::io_service first;
boost::asio::deadline_timer second;
};
typedef std::vector<boost::shared_ptr<service_pair_type> > service_pool_type;
service_pool_type m_service_pool;
boost::uint32_t m_next_service;
};
} // end namespace pion
#endif
pion::scheduler::post
virtual void post(boost::function0< void > work_func)
Definition: scheduler.hpp:88
pion::multi_thread_scheduler::~multi_thread_scheduler
virtual ~multi_thread_scheduler()
virtual destructor
Definition: scheduler.hpp:208
pion::scheduler::m_scheduler_has_stopped
boost::condition m_scheduler_has_stopped
condition triggered when the scheduler has stopped
Definition: scheduler.hpp:183
pion::multi_thread_scheduler::finish_threads
virtual void finish_threads(void)
finishes all threads used to perform work
Definition: scheduler.hpp:231
pion::scheduler::stop_threads
virtual void stop_threads(void)
stops all threads used to perform work
Definition: scheduler.hpp:151
pion::scheduler::stop_services
virtual void stop_services(void)
stops all services used to schedule work
Definition: scheduler.hpp:148
pion::single_service_scheduler
Definition: scheduler.hpp:246
pion::one_to_one_scheduler::~one_to_one_scheduler
virtual ~one_to_one_scheduler()
virtual destructor
Definition: scheduler.hpp:297
pion::scheduler::set_logger
void set_logger(logger log_ptr)
sets the logger to be used
Definition: scheduler.hpp:75
pion::scheduler::m_is_running
bool m_is_running
true if the thread scheduler is running
Definition: scheduler.hpp:192
pion::scheduler::DEFAULT_NUM_THREADS
static const boost::uint32_t DEFAULT_NUM_THREADS
default number of worker threads in the thread pool
Definition: scheduler.hpp:161
pion::single_service_scheduler::m_timer
boost::asio::deadline_timer m_timer
timer used to periodically check for shutdown
Definition: scheduler.hpp:279
pion::single_service_scheduler::single_service_scheduler
single_service_scheduler(void)
constructs a new single_service_scheduler
Definition: scheduler.hpp:252
pion::multi_thread_scheduler
Definition: scheduler.hpp:199
pion::multi_thread_scheduler::multi_thread_scheduler
multi_thread_scheduler(void)
constructs a new single_service_scheduler
Definition: scheduler.hpp:205
pion::scheduler::finish_services
virtual void finish_services(void)
finishes all services used to schedule work
Definition: scheduler.hpp:154
pion::single_service_scheduler::stop_services
virtual void stop_services(void)
stops all services used to schedule work
Definition: scheduler.hpp:269
pion::scheduler::sleep
static void sleep(ConditionType &wakeup_condition, LockType &wakeup_lock, boost::uint32_t sleep_sec, boost::uint32_t sleep_nsec)
Definition: scheduler.hpp:122
pion::scheduler::m_no_more_active_users
boost::condition m_no_more_active_users
condition triggered when there are no more active users
Definition: scheduler.hpp:180
pion::single_service_scheduler::finish_services
virtual void finish_services(void)
finishes all services used to schedule work
Definition: scheduler.hpp:272
pion::scheduler::is_running
bool is_running(void) const
returns true if the scheduler is running
Definition: scheduler.hpp:66
pion::single_service_scheduler::m_service
boost::asio::io_service m_service
service used to manage async I/O events
Definition: scheduler.hpp:276
pion::scheduler::MICROSEC_IN_SECOND
static const boost::uint32_t MICROSEC_IN_SECOND
number of microseconds in one full second (10 ^ 6)
Definition: scheduler.hpp:167
pion::scheduler::m_num_threads
boost::uint32_t m_num_threads
total number of worker threads in the pool
Definition: scheduler.hpp:186
pion::one_to_one_scheduler::get_io_service
virtual boost::asio::io_service & get_io_service(void)
returns an async I/O service used to schedule work
Definition: scheduler.hpp:300
pion::multi_thread_scheduler::m_thread_pool
ThreadPool m_thread_pool
pool of threads used to perform work
Definition: scheduler.hpp:239
pion::scheduler::~scheduler
virtual ~scheduler()
virtual destructor
Definition: scheduler.hpp:46
pion::single_service_scheduler::get_io_service
virtual boost::asio::io_service & get_io_service(void)
returns an async I/O service used to schedule work
Definition: scheduler.hpp:260
pion::scheduler::get_num_threads
boost::uint32_t get_num_threads(void) const
returns the number of threads currently in use
Definition: scheduler.hpp:72
pion::scheduler::scheduler
scheduler(void)
constructs a new scheduler
Definition: scheduler.hpp:40
pion::scheduler::m_mutex
boost::mutex m_mutex
mutex to make class thread-safe
Definition: scheduler.hpp:174
pion::scheduler::sleep
static void sleep(boost::uint32_t sleep_sec, boost::uint32_t sleep_nsec)
Definition: scheduler.hpp:107
pion::scheduler::NSEC_IN_SECOND
static const boost::uint32_t NSEC_IN_SECOND
number of nanoseconds in one full second (10 ^ 9)
Definition: scheduler.hpp:164
pion::one_to_one_scheduler
Definition: scheduler.hpp:286
pion::scheduler::m_logger
logger m_logger
primary logging interface used by this class
Definition: scheduler.hpp:177
pion::scheduler::m_active_users
boost::uint32_t m_active_users
the scheduler will not shutdown until there are no more active users
Definition: scheduler.hpp:189
pion::scheduler::set_num_threads
void set_num_threads(const boost::uint32_t n)
sets the number of threads to be used (these are shared by all servers)
Definition: scheduler.hpp:69
pion::scheduler::KEEP_RUNNING_TIMER_SECONDS
static const boost::uint32_t KEEP_RUNNING_TIMER_SECONDS
number of seconds a timer should wait for to keep the IO services running
Definition: scheduler.hpp:170
pion::one_to_one_scheduler::service_pool_type
std::vector< boost::shared_ptr< service_pair_type > > service_pool_type
typedef for a pool of IO services
Definition: scheduler.hpp:349
pion::logger
Definition: logger.hpp:238
pion::one_to_one_scheduler::m_next_service
boost::uint32_t m_next_service
the next service to use for scheduling work
Definition: scheduler.hpp:356
pion::multi_thread_scheduler::ThreadPool
std::vector< boost::shared_ptr< boost::thread > > ThreadPool
typedef for a pool of worker threads
Definition: scheduler.hpp:235
pion::one_to_one_scheduler::service_pair_type
typedef for a pair object where first is an IO service and second is a deadline timer ...
Definition: scheduler.hpp:342
pion::one_to_one_scheduler::finish_services
virtual void finish_services(void)
finishes all services used to schedule work
Definition: scheduler.hpp:338
pion::one_to_one_scheduler::get_io_service
virtual boost::asio::io_service & get_io_service(boost::uint32_t n)
Definition: scheduler.hpp:318
pion::one_to_one_scheduler::stop_services
virtual void stop_services(void)
stops all services used to schedule work
Definition: scheduler.hpp:331
pion::single_service_scheduler::~single_service_scheduler
virtual ~single_service_scheduler()
virtual destructor
Definition: scheduler.hpp:257
pion
Definition: admin_rights.hpp:19
pion::scheduler::get_logger
logger get_logger(void)
returns the logger currently in use
Definition: scheduler.hpp:78
pion::scheduler::startup
virtual void startup(void)
Starts the thread scheduler (this is called automatically when necessary)
Definition: scheduler.hpp:49
pion::multi_thread_scheduler::stop_threads
virtual void stop_threads(void)
stops all threads used to perform work
Definition: scheduler.hpp:214
pion::one_to_one_scheduler::m_service_pool
service_pool_type m_service_pool
pool of IO services used to schedule work
Definition: scheduler.hpp:353
pion::scheduler::finish_threads
virtual void finish_threads(void)
finishes all threads used to perform work
Definition: scheduler.hpp:157
pion::one_to_one_scheduler::one_to_one_scheduler
one_to_one_scheduler(void)
constructs a new one_to_one_scheduler
Definition: scheduler.hpp:292
pion::scheduler
Definition: scheduler.hpp:34