scheduler.cpp
// ---------------------------------------------------------------------
// 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
//
#include <boost/exception/diagnostic_information.hpp>
#include <boost/date_time/posix_time/posix_time_duration.hpp>
#include <pion/scheduler.hpp>
namespace pion { // begin namespace pion
// static members of scheduler
const boost::uint32_t scheduler::DEFAULT_NUM_THREADS = 8;
const boost::uint32_t scheduler::NSEC_IN_SECOND = 1000000000; // (10^9)
const boost::uint32_t scheduler::MICROSEC_IN_SECOND = 1000000; // (10^6)
const boost::uint32_t scheduler::KEEP_RUNNING_TIMER_SECONDS = 5;
// scheduler member functions
void scheduler::shutdown(void)
{
// lock mutex for thread safety
boost::mutex::scoped_lock scheduler_lock(m_mutex);
if (m_is_running) {
PION_LOG_INFO(m_logger, "Shutting down the thread scheduler");
while (m_active_users > 0) {
// first, wait for any active users to exit
PION_LOG_INFO(m_logger, "Waiting for " << m_active_users << " scheduler users to finish");
m_no_more_active_users.wait(scheduler_lock);
}
// shut everything down
m_is_running = false;
stop_services();
stop_threads();
finish_services();
finish_threads();
PION_LOG_INFO(m_logger, "The thread scheduler has shutdown");
// Make sure anyone waiting on shutdown gets notified
m_scheduler_has_stopped.notify_all();
} else {
// stop and finish everything to be certain that no events are pending
stop_services();
stop_threads();
finish_services();
finish_threads();
// Make sure anyone waiting on shutdown gets notified
// even if the scheduler did not startup successfully
m_scheduler_has_stopped.notify_all();
}
}
void scheduler::join(void)
{
boost::mutex::scoped_lock scheduler_lock(m_mutex);
while (m_is_running) {
// sleep until scheduler_has_stopped condition is signaled
m_scheduler_has_stopped.wait(scheduler_lock);
}
}
void scheduler::keep_running(boost::asio::io_service& my_service,
boost::asio::deadline_timer& my_timer)
{
if (m_is_running) {
// schedule this again to make sure the service doesn't complete
my_timer.expires_from_now(boost::posix_time::seconds(KEEP_RUNNING_TIMER_SECONDS));
my_timer.async_wait(boost::bind(&scheduler::keep_running, this,
boost::ref(my_service), boost::ref(my_timer)));
}
}
void scheduler::add_active_user(void)
{
if (!m_is_running) startup();
boost::mutex::scoped_lock scheduler_lock(m_mutex);
++m_active_users;
}
void scheduler::remove_active_user(void)
{
boost::mutex::scoped_lock scheduler_lock(m_mutex);
if (--m_active_users == 0)
m_no_more_active_users.notify_all();
}
boost::system_time scheduler::get_wakeup_time(boost::uint32_t sleep_sec,
boost::uint32_t sleep_nsec)
{
return boost::get_system_time() + boost::posix_time::seconds(sleep_sec) + boost::posix_time::microseconds(sleep_nsec / 1000);
}
void scheduler::process_service_work(boost::asio::io_service& service) {
while (m_is_running) {
try {
service.run();
} catch (std::exception& e) {
PION_LOG_ERROR(m_logger, boost::diagnostic_information(e));
} catch (...) {
PION_LOG_ERROR(m_logger, "caught unrecognized exception");
}
}
}
// single_service_scheduler member functions
void single_service_scheduler::startup(void)
{
// lock mutex for thread safety
boost::mutex::scoped_lock scheduler_lock(m_mutex);
if (! m_is_running) {
PION_LOG_INFO(m_logger, "Starting thread scheduler");
m_is_running = true;
// schedule a work item to make sure that the service doesn't complete
m_service.reset();
keep_running(m_service, m_timer);
// start multiple threads to handle async tasks
for (boost::uint32_t n = 0; n < m_num_threads; ++n) {
boost::shared_ptr<boost::thread> new_thread(new boost::thread( boost::bind(&scheduler::process_service_work,
this, boost::ref(m_service)) ));
m_thread_pool.push_back(new_thread);
}
}
}
// one_to_one_scheduler member functions
void one_to_one_scheduler::startup(void)
{
// lock mutex for thread safety
boost::mutex::scoped_lock scheduler_lock(m_mutex);
if (! m_is_running) {
PION_LOG_INFO(m_logger, "Starting thread scheduler");
m_is_running = true;
// make sure there are enough services initialized
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);
}
// schedule a work item for each service to make sure that it doesn't complete
for (service_pool_type::iterator i = m_service_pool.begin(); i != m_service_pool.end(); ++i) {
keep_running((*i)->first, (*i)->second);
}
// start multiple threads to handle async tasks
for (boost::uint32_t n = 0; n < m_num_threads; ++n) {
boost::shared_ptr<boost::thread> new_thread(new boost::thread( boost::bind(&scheduler::process_service_work,
this, boost::ref(m_service_pool[n]->first)) ));
m_thread_pool.push_back(new_thread);
}
}
}
} // end namespace pion
pion::scheduler::m_scheduler_has_stopped
boost::condition m_scheduler_has_stopped
condition triggered when the scheduler has stopped
Definition: scheduler.hpp:183
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::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::one_to_one_scheduler::startup
virtual void startup(void)
Starts the thread scheduler (this is called automatically when necessary)
Definition: scheduler.cpp:148
pion::scheduler::finish_services
virtual void finish_services(void)
finishes all services used to schedule work
Definition: scheduler.hpp:154
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::scheduler::get_wakeup_time
static boost::system_time get_wakeup_time(boost::uint32_t sleep_sec, boost::uint32_t sleep_nsec)
Definition: scheduler.cpp:102
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::remove_active_user
void remove_active_user(void)
unregisters an active user with the thread scheduler
Definition: scheduler.cpp:95
pion::scheduler::m_num_threads
boost::uint32_t m_num_threads
total number of worker threads in the pool
Definition: scheduler.hpp:186
pion::scheduler::m_mutex
boost::mutex m_mutex
mutex to make class thread-safe
Definition: scheduler.hpp:174
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::scheduler::m_logger
logger m_logger
primary logging interface used by this class
Definition: scheduler.hpp:177
pion::scheduler::shutdown
virtual void shutdown(void)
Stops the thread scheduler (this is called automatically when the program exits)
Definition: scheduler.cpp:27
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::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_pair_type
typedef for a pair object where first is an IO service and second is a deadline timer ...
Definition: scheduler.hpp:342
pion::scheduler::keep_running
void keep_running(boost::asio::io_service &my_service, boost::asio::deadline_timer &my_timer)
Definition: scheduler.cpp:77
pion
Definition: admin_rights.hpp:19
pion::scheduler::startup
virtual void startup(void)
Starts the thread scheduler (this is called automatically when necessary)
Definition: scheduler.hpp:49
pion::scheduler::add_active_user
void add_active_user(void)
Definition: scheduler.cpp:88
pion::single_service_scheduler::startup
virtual void startup(void)
Starts the thread scheduler (this is called automatically when necessary)
Definition: scheduler.cpp:123
pion::scheduler::join
void join(void)
the calling thread will sleep until the scheduler has stopped
Definition: scheduler.cpp:68
pion::scheduler::process_service_work
void process_service_work(boost::asio::io_service &service)
processes work passed to the asio service & handles uncaught exceptions
Definition: scheduler.cpp:108
pion::scheduler::finish_threads
virtual void finish_threads(void)
finishes all threads used to perform work
Definition: scheduler.hpp:157