platform/include/pion/platform/ReactionScheduler.hpp

// ------------------------------------------------------------------------
// Pion is a development platform for building Reactors that process Events
// ------------------------------------------------------------------------
// Copyright (C) 2007-2008 Atomic Labs, Inc.  (http://www.atomiclabs.com)
//
// Pion is free software: you can redistribute it and/or modify it under the
// terms of the GNU Affero General Public License as published by the Free
// Software Foundation, either version 3 of the License, or (at your option)
// any later version.
//
// Pion is distributed in the hope that it will be useful, but WITHOUT ANY
// WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
// FOR A PARTICULAR PURPOSE.  See the GNU Affero General Public License for
// more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with Pion.  If not, see <http://www.gnu.org/licenses/>.
//

#ifndef __PION_REACTIONSCHEDULER_HEADER__
#define __PION_REACTIONSCHEDULER_HEADER__

#include <vector>
#include <boost/shared_ptr.hpp>
#include <boost/function/function0.hpp>
#include <boost/thread/mutex.hpp>
#include <pion/PionConfig.hpp>
#include <pion/PionScheduler.hpp>
#include <pion/PionException.hpp>
#include <pion/PionLockedQueue.hpp>


namespace pion {        // begin namespace pion
namespace platform {    // begin namespace platform (Pion Platform Library)

class ReactionScheduler :
    public PionSingleServiceScheduler
{
public:
    
    class NoServiceException : public std::exception {
    public:
        virtual const char* what() const throw() {
            return "ReactionScheduler has no io_service available";
        }
    };

    
    ReactionScheduler(void) {}
    
    virtual ~ReactionScheduler() { shutdown(); }
    
    virtual void 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();
            keepRunning(m_service, m_timer);

            // start a thread that will be used to handle io_service requests
            m_service_thread.reset(new boost::thread( boost::bind(&PionScheduler::processServiceWork,
                                                                  this, boost::ref(m_service)) ));
            
            // 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(&ReactionScheduler::processReactionQueue, this) ));
                m_thread_pool.push_back(new_thread);
            }
        }
    }       
    
    virtual void stopThreads(void) {
        stopThreadInfo();
        PionSingleServiceScheduler::stopThreads();
        if (m_service_thread)
            m_service_thread->join();
        boost::mutex::scoped_lock thread_info_lock(m_thread_info_mutex);
        m_thread_info.clear();
    }

    virtual void finishThreads(void) {
        PionSingleServiceScheduler::finishThreads();
        m_thread_pool.clear();
        m_service_thread.reset();
    }

    virtual void post(boost::function0<void> work_func) {
        m_reaction_queue.push(work_func);
    }
    
    inline std::size_t getQueueSize(void) const { return m_reaction_queue.size(); }

    
protected:
    
    typedef boost::function0<void>      Reaction;

    typedef PionLockedQueue<Reaction>   ReactionQueue;

    typedef boost::shared_ptr<ReactionQueue::ConsumerThread>    ThreadInfoPtr;

    typedef std::vector<ThreadInfoPtr>  ThreadInfoVector;
    
    
    ThreadInfoPtr getThreadInfo(void) {
        ThreadInfoPtr info_ptr(new ReactionQueue::ConsumerThread);
        boost::mutex::scoped_lock thread_info_lock(m_thread_info_mutex);
        m_thread_info.push_back(info_ptr);
        return info_ptr;
    }

    void stopThreadInfo(void) {
        boost::mutex::scoped_lock thread_info_lock(m_thread_info_mutex);
        for (ThreadInfoVector::iterator i = m_thread_info.begin();
            i != m_thread_info.end(); ++i)
        {
            (*i)->stop();
        }
    }

    void processReactionQueue(void) {
        Reaction r;
        ThreadInfoPtr info_ptr(getThreadInfo());
        while (m_is_running) {
            try {
                while (m_is_running) {
                    if (m_reaction_queue.pop(r, *info_ptr) && m_is_running)
                        r();
                }
            } catch (std::exception& e) {
                PION_LOG_ERROR(m_logger, e.what());
            } catch (...) {
                PION_LOG_ERROR(m_logger, "caught unrecognized exception");
            }
        }
    }       
    
    
    ReactionQueue                       m_reaction_queue;

    boost::shared_ptr<boost::thread>    m_service_thread;

    ThreadInfoVector                    m_thread_info;

    boost::mutex                        m_thread_info_mutex;
};
    
    
}   // end namespace platform
}   // end namespace pion

#endif

---