platform/include/pion/platform/Reactor.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_REACTOR_HEADER__
#define __PION_REACTOR_HEADER__

#include <iosfwd>
#include <string>
#include <list>
#include <libxml/tree.h>
#include <boost/bind.hpp>
#include <boost/signal.hpp>
#include <boost/thread.hpp>
#include <boost/function.hpp>
#include <boost/function/function1.hpp>
#include <pion/PionConfig.hpp>
#include <pion/PionException.hpp>
#include <pion/platform/Event.hpp>
#include <pion/platform/Vocabulary.hpp>
#include <pion/platform/PlatformPlugin.hpp>
#include <pion/platform/ReactionScheduler.hpp>

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

class PION_PLATFORM_API Reactor
    : public PlatformPlugin
{
public:

    static const std::string        REACTOR_ELEMENT_NAME;
    
    static const std::string        RUNNING_ELEMENT_NAME;

    static const std::string        WORKSPACE_ELEMENT_NAME;

    static const std::string        X_COORDINATE_ELEMENT_NAME;

    static const std::string        Y_COORDINATE_ELEMENT_NAME;


    typedef boost::function1<void, EventPtr>    EventHandler;

    typedef std::vector<std::string>            QueryBranches;
    
    typedef StringDictionary                    QueryParams;
    
    enum ReactorType {
        TYPE_COLLECTION, TYPE_PROCESSING, TYPE_STORAGE
    };
    
    
    class AlreadyConnectedException : public PionException {
    public:
        AlreadyConnectedException(const std::string& reactor_id)
            : PionException("Reactor is already connected: ", reactor_id) {}
    };
    
    class ConnectionNotFoundException : public PionException {
    public:
        ConnectionNotFoundException(const std::string& reactor_id)
            : PionException("Tried removing an unknown connection: ", reactor_id) {}
    };
    
    class ConfigLockException : public PionException {
    public:
        ConfigLockException(const std::string& reactor_id)
            : PionException("Error obtaining Reactor configuration lock: ", reactor_id) {}
    };

    class UnknownSignalException : public PionException {
    public:
        UnknownSignalException(const std::string& reactor_id, const std::string& signal_id)
            : PionException("Unknown signal for Reactor " + reactor_id + ": ", signal_id) {}
    };

    class MissingWorkspaceException : public PionException {
    public:
        MissingWorkspaceException(const std::string& reactor_id)
            : PionException("Reactor configuration missing required Workspace parameter: ", reactor_id) {}
        MissingWorkspaceException(void)
            : PionException("Reactor configuration missing required Workspace parameter") {}
    };

    virtual ~Reactor() {}
    
    virtual void start(void) {
        ConfigWriteLock cfg_lock(*this);
        m_is_running = true;
    }
    
    virtual void stop(void) {
        ConfigWriteLock cfg_lock(*this);
        m_is_running = false;
    }
    
    virtual void reset(void) { clearStats(); }
    
    virtual void clearStats(void) {
        // atomic_count doesn't support assignment -- ugh!
        //m_events_in = m_events_out = boost::detail::atomic_count(0);
    }
    
    virtual void setConfig(const Vocabulary& v, const xmlNodePtr config_ptr);

    virtual void updateVocabulary(const Vocabulary& v);

    virtual void updateCodecs(void) {}
    
    virtual void updateDatabases(void) {}
    
    virtual void updateProtocols(void) {}
    
    virtual void query(std::ostream& out, const QueryBranches& branches,
        const QueryParams& qp);
    
    template <typename F>
    inline boost::signals::connection subscribe(const std::string& signal_id, F f) {
        ConfigWriteLock cfg_lock(*this);
        SignalMap::iterator it = m_signals.find(signal_id);
        if (it == m_signals.end())
            throw UnknownSignalException(getId(), signal_id);
        return it->second->connect(f);
    }

    inline void operator()(const EventPtr& e) {
        if ( isRunning() ) {
            ConfigReadLock cfg_lock(*this);
            // re-check after locking
            if ( isRunning() ) {
                ++m_events_in;
                process(e);
            }
        }
    }

    bool startOutRunning(const xmlNodePtr config_ptr, bool exec_start);

    void addConnection(Reactor& output_reactor);
    
    void addConnection(const std::string& connection_id, EventHandler connection_handler);
    
    void removeConnection(const std::string& connection_id);
    
    void clearConnections(void);

    inline void writeStatsXML(std::ostream& out) const {
        writeBeginReactorXML(out);
        writeStatsOnlyXML(out);
        writeEndReactorXML(out);
    }

    inline void setScheduler(ReactionScheduler& scheduler) { m_scheduler_ptr = & scheduler; }

    inline void setMultithreadBranches(bool b) { m_multithread_branches = b; }

    inline boost::uint32_t getEventsIn(void) const { return m_events_in; }

    inline boost::uint32_t getEventsOut(void) const { return m_events_out; }

    inline bool isRunning(void) const { return m_is_running; }
    
    inline ReactorType getType(void) const { return m_type; }

    inline std::string getWorkspace(void) const { return m_workspace_id; }

protected:

    typedef boost::signal3<void, const std::string&, const std::string&, void*> SignalType;

    typedef boost::shared_ptr<SignalType> SignalPtr;
    
    typedef PION_HASH_MAP<std::string, SignalPtr, PION_HASH_STRING> SignalMap;


    class ConfigReadLock {
    public:
        ConfigReadLock(const Reactor& r)
            : m_reactor_ref(r)
        {
            boost::uint16_t sleep_times = 0;
            do {
                while (m_reactor_ref.m_config_change_pending) {
                    if (++sleep_times > 50)
                        throw ConfigLockException(m_reactor_ref.getId());
                    boost::thread::sleep(boost::get_system_time()
                        + boost::posix_time::millisec(100));
                }
                ++m_reactor_ref.m_config_num_readers;
                if (m_reactor_ref.m_config_change_pending) {
                    --m_reactor_ref.m_config_num_readers;
                } else {
                    break;
                }
            } while (true);
        }
        ~ConfigReadLock() {
            --m_reactor_ref.m_config_num_readers;
        }
    private:
        const Reactor&  m_reactor_ref;
    };
    
    class ConfigWriteLock {
    public:
        ConfigWriteLock(Reactor& r)
            : m_reactor_ref(r), m_already_locked(r.m_config_change_pending)
        {
            if (! m_already_locked) {
                boost::uint16_t sleep_times = 0;
                m_reactor_ref.m_config_change_pending = true;
                while (m_reactor_ref.m_config_num_readers > 0) {
                    if (++sleep_times > 50)
                        throw ConfigLockException(m_reactor_ref.getId());
                    boost::thread::sleep(boost::get_system_time()
                        + boost::posix_time::millisec(100));
                }
            }
        }
        ~ConfigWriteLock() {
            if (! m_already_locked)
                m_reactor_ref.m_config_change_pending = false;
        }
    private:
        Reactor&    m_reactor_ref;
        bool        m_already_locked;
    };
        
        
    Reactor(const ReactorType type)
        : m_is_running(false), m_type(type), m_scheduler_ptr(NULL), 
        m_events_in(0), m_events_out(0),
        m_config_change_pending(false), m_config_num_readers(0)
    {}

    inline ReactionScheduler& getScheduler(void) {
        PION_ASSERT(m_scheduler_ptr != NULL);
        return *m_scheduler_ptr;
    }
        
    virtual void process(const EventPtr& e) {
        deliverEvent(e);
    }

    inline void deliverEvent(const EventPtr& e, bool return_immediately = false) {
        ++m_events_out;
        if (! m_connections.empty()) {
            if (m_multithread_branches) {
                // iterate through each Reactor after the first one and send the Event
                // using the scheduler.  This way, the entire thread pool will be used
                // for processing pipelines
                ConnectionMap::iterator i = m_connections.begin();
                
                // skip Reactors that are not running to avoid queueing work
                // to another thread when it is unnecessary.  Otherwise, if the
                // first Reactor is not running, all work will get queued rather
                // than the "longest path" being executed by the current thread
                while (i != m_connections.end() && ! i->second.isRunning())
                    ++i;

                if (i != m_connections.end()) {
                    // the first running reactor will be handled by the current thread
                    ConnectionMap::iterator cur_thread_reactor = i;
                    
                    // queue all other branches to be handled by other threads
                    while (++i != m_connections.end()) {
                        if (i->second.isRunning())
                            i->second.post(getScheduler(), e);
                    }
                
                    // send to the first Reactor using the same thread
                    // this helps to reduce context switching by ensuring
                    // that the longer processing chains remain unbroken
                    if (return_immediately)
                        cur_thread_reactor->second.post(getScheduler(), e);
                    else
                        cur_thread_reactor->second(e);
                }
            } else {
                // simple scheduling just iterates through connections and use the
                // same thread to carry the event through all the reaction chains
                for (ConnectionMap::iterator i = m_connections.begin();
                     i != m_connections.end(); ++i)
                {
                    if (i->second.isRunning()) {
                        if (return_immediately)
                            i->second.post(getScheduler(), e);
                        else
                            i->second(e);
                    }
                }
            }
        }
    }
    
    inline void deliverEvents(const EventContainer& events, bool return_immediately = false)
    {
        for (EventContainer::const_iterator event_it = events.begin();
                event_it != events.end(); ++event_it)
        {
            deliverEvent(*event_it, return_immediately);
        }
    }

    inline void incrementEventsIn(void) { ++m_events_in; }

    inline void publish(const std::string& signal_id) {
        SignalPtr signal_ptr(new SignalType);
        ConfigWriteLock cfg_lock(*this);
        m_signals.insert(std::make_pair(signal_id, signal_ptr));
    }

    inline void signalNoLock(const std::string& signal_id, void *ptr = NULL) {
        SignalMap::iterator it = m_signals.find(signal_id);
        if (it == m_signals.end())
            throw UnknownSignalException(getId(), signal_id);
        (*it->second)(getId(), signal_id, ptr);
    }

    inline void signal(const std::string& signal_id, void *ptr = NULL) {
        ConfigReadLock cfg_lock(*this);
        signalNoLock(signal_id, ptr);
    }

    void writeStatsOnlyXML(std::ostream& out) const;
    
    void writeBeginReactorXML(std::ostream& out) const;
    
    void writeEndReactorXML(std::ostream& out) const;       


    SignalMap                   m_signals;

    volatile bool               m_is_running;

    
private:

    class OutputConnection {
    public:
        ~OutputConnection() {}
        
        explicit OutputConnection(Reactor* reactor)
            : m_reactor_ptr(reactor), m_event_handler(boost::ref(*reactor))
        {}

        explicit OutputConnection(EventHandler handler)
            : m_reactor_ptr(NULL), m_event_handler(handler)
        {}

        OutputConnection(const OutputConnection& c)
            : m_reactor_ptr(c.m_reactor_ptr), m_event_handler(c.m_event_handler)
        {}

        inline bool isRunning(void) {
            return (m_reactor_ptr == NULL || m_reactor_ptr->isRunning());
        }
        
        inline void operator()(const EventPtr& event_ptr) {
            m_event_handler(event_ptr);
        }
        
        inline void post(ReactionScheduler& scheduler, const EventPtr& event_ptr) {
            scheduler.post(boost::bind<void>(m_event_handler, event_ptr));
        }

    private:
        
        Reactor *           m_reactor_ptr;
        
        EventHandler        m_event_handler;
    };
    
    typedef std::map<std::string, OutputConnection> ConnectionMap;
    
    
    static const std::string        EVENTS_IN_ELEMENT_NAME;
    
    static const std::string        EVENTS_OUT_ELEMENT_NAME;

    static const std::string        ID_ATTRIBUTE_NAME;


    const ReactorType               m_type;
    
    ReactionScheduler *             m_scheduler_ptr;
    
    ConnectionMap                   m_connections;

    std::string                     m_workspace_id;

    boost::detail::atomic_count     m_events_in;

    boost::detail::atomic_count     m_events_out;
    
    bool                            m_multithread_branches;
    
    mutable volatile bool                   m_config_change_pending;
    
    mutable boost::detail::atomic_count     m_config_num_readers;

    // allow configuration lock classes to modify private variables 
    friend class ConfigReadLock;
    friend class ConfigWriteLock;
};


//
// The following symbols must be defined for any reactor that you would
// like to be able to load dynamically using the ReactionEngine::load()
// function.  These are not required for any reactors that you only want to link
// directly into your programs.
//
// Make sure that you replace "REACTOR" with the name of your derived class.
// This name must also match the name of the object file (excluding the
// extension).  These symbols must be linked into your reactor's object file,
// not included in any headers that it may use (declarations are OK in headers
// but not the definitions).
//
// The "pion_create" function is used to create new instances of your reactor.
// The "pion_destroy" function is used to destroy instances of your reactor.
//
// extern "C" Reactor *pion_create_REACTOR(void) {
//      return new REACTOR;
// }
//
// extern "C" void pion_destroy_REACTOR(REACTOR *reactor_ptr) {
//      delete reactor_ptr;
// }
//


}   // end namespace platform
}   // end namespace pion

#endif

---