platform/reactors/PythonReactor.hpp

// ------------------------------------------------------------------------
// Pion is a development platform for building Reactors that process Events
// ------------------------------------------------------------------------
// Copyright (C) 2007-2011 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_PYTHONREACTOR_HEADER__
#define __PION_PYTHONREACTOR_HEADER__

// NOTE: According to API docs, Python.h must be #include'd FIRST
#include <Python.h>
#include <string>
#include <boost/thread/tss.hpp>
#include <pion/PionConfig.hpp>
#include <pion/PionLogger.hpp>
#include <pion/PionHashMap.hpp>
#include <pion/PionException.hpp>
#include <pion/platform/Reactor.hpp>
#include <pion/platform/ReactionEngine.hpp>


namespace pion {        // begin namespace pion
namespace plugins {     // begin namespace plugins


class PythonReactor :
    public pion::platform::Reactor
{
public: 
    
    class UnknownTermException : public PionException {
    public:
        UnknownTermException(const std::string& term)
            : PionException("Unable to find required Vocabulary term: ", term) {}
    };

    class InternalPythonException : public PionException {
    public:
        InternalPythonException(const std::string& reactor_id)
            : PionException("PythonReactor internal API error: ", reactor_id) {}
    };

    class SourceFileNotFoundException : public PionException {
    public:
        SourceFileNotFoundException(const std::string& filename)
            : PionException("PythonReactor source code file not found: ", filename) {}
    };

    class ReadSourceFileException : public PionException {
    public:
        ReadSourceFileException(const std::string& filename)
            : PionException("PythonReactor unable to read source code from file: ", filename) {}
    };

    class FailedToCompileException : public PionException {
    public:
        FailedToCompileException(const std::string& filename)
            : PionException("PythonReactor compile failure: ", filename) {}
    };

    class NotCallableException : public PionException {
    public:
        NotCallableException(const std::string& name)
            : PionException("PythonReactor attribute defined is not callable: ", name) {}
    };

    class InitReactorObjectException : public PionException {
    public:
        InitReactorObjectException(const std::string& error_msg)
            : PionException("PythonReactor unable to initialize Reactor object: ", error_msg) {}
    };

    class EventConversionException : public PionException {
    public:
        EventConversionException(const std::string& error_msg)
            : PionException("PythonReactor event conversion error: ", error_msg) {}
    };


    PythonReactor(void);
    
    virtual ~PythonReactor();
    
    virtual void setConfig(const pion::platform::Vocabulary& v, const xmlNodePtr config_ptr);
    
    virtual void updateVocabulary(const pion::platform::Vocabulary& v);
    
    virtual void process(const pion::platform::EventPtr& e);
    
    virtual void query(std::ostream& out, const QueryBranches& branches,
        const QueryParams& qp);

    virtual void start(void);
    
    virtual void stop(void);
    

    bool deliverToConnections(PyObject *event_ptr);

    PyObject *getSession(PyObject *event_ptr);

    inline void setLogger(PionLogger log_ptr) { m_logger = log_ptr; }

    inline PionLogger getLogger(void) { return m_logger; }

    inline const pion::platform::Vocabulary& getVocabulary(void) const {
        PION_ASSERT(m_vocab_ptr);
        return *m_vocab_ptr;
    }
    
    static inline boost::uint64_t boost_msec_to_fsec(boost::uint64_t n);

    static inline boost::uint64_t boost_fsec_to_msec(boost::uint64_t n);

    
protected:

    void updateTerms(const pion::platform::Vocabulary& v);

    std::size_t getNumSessions(void) const;

    void flushSessions(void);

    static PyThreadState * initThreadState(void);

    static void releaseThreadState(PyThreadState *ptr);

    PyObject *findPythonFunction(PyObject *module_ptr, const std::string& func_name);

    void resetPythonSymbols(void);
    
    void compilePythonSource(void);
    
    void initPythonModule(void);
    
    void callPythonStart(void);
    
    void callPythonStop(void);
    
    std::string getSourceCodeFromFile(void);

    std::string getPythonError(void);

    
    class PythonLock {
    public:

        PythonLock(bool inverse = false)
            : m_inversed(inverse), m_thr_state_ptr(NULL)
        {
            m_thr_state_ptr = PythonReactor::initThreadState();
            if (m_inversed) {
                PyEval_ReleaseThread(m_thr_state_ptr);
            } else {
                PyEval_AcquireThread(m_thr_state_ptr);
            }
        }
        
        ~PythonLock() {
            if (m_inversed) {
                PyEval_AcquireThread(m_thr_state_ptr);
            } else {
                PyEval_ReleaseThread(m_thr_state_ptr);
            }
        }

    private:
    
        bool            m_inversed;

        PyThreadState * m_thr_state_ptr;
    };

    
private:
    
    typedef PION_HASH_MAP<pion::platform::Event::BlobType, PyObject*, HashPionIdBlob>   SessionMap;


    static const std::string        START_FUNCTION_NAME;

    static const std::string        STOP_FUNCTION_NAME;

    static const std::string        PROCESS_FUNCTION_NAME;

    static const std::string        FILENAME_ELEMENT_NAME;

    static const std::string        PYTHON_SOURCE_ELEMENT_NAME;

    static const std::string        OPEN_SESSIONS_ELEMENT_NAME;

    static boost::mutex             m_init_mutex;
    
    static boost::uint32_t          m_init_num;

    PionLogger                      m_logger;

    std::string                     m_source;

    std::string                     m_source_file;

    PyObject *                      m_byte_code;
    
    PyObject *                      m_module;
    
    PyObject *                      m_start_func;
    
    PyObject *                      m_stop_func;
    
    PyObject *                      m_process_func;
    
    PyObject *                      m_reactor_ptr;
    
    pion::platform::VocabularyPtr   m_vocab_ptr;
    
    mutable boost::mutex            m_sessions_mutex;

    SessionMap                      m_sessions;

    static PyInterpreterState *     m_interp_ptr;

    static boost::thread_specific_ptr<PyThreadState> *  m_state_ptr;

    static const std::string            VOCAB_CLICKSTREAM_SESSION_EVENT;
    pion::platform::Vocabulary::TermRef m_session_event_term_ref;

    static const std::string            VOCAB_CLICKSTREAM_SESSION_ID;
    pion::platform::Vocabulary::TermRef m_session_id_term_ref;
};


// inline functions for PythonReactor

inline boost::uint64_t PythonReactor::boost_msec_to_fsec(boost::uint64_t n) {
    switch (boost::posix_time::time_duration::resolution()) {
    case boost::date_time::sec:
        n /= 1000000;
        break;
    case boost::date_time::tenth:
        n /= 100000;
        break;
    case boost::date_time::hundreth:
        n /= 10000;
        break;
    case boost::date_time::milli:
        n /= 1000;
        break;
    case boost::date_time::ten_thousandth:
        n /= 100;
        break;
    case boost::date_time::micro:
        // good to go
        break;
    case boost::date_time::nano:
        n *= 1000;
        break;
    case boost::date_time::NumResolutions:
        // shouldn't happen
        break;
    }
    return n;
}

inline boost::uint64_t PythonReactor::boost_fsec_to_msec(boost::uint64_t n) {
    switch (boost::posix_time::time_duration::resolution()) {
    case boost::date_time::sec:
        n *= 1000000;
        break;
    case boost::date_time::tenth:
        n *= 100000;
        break;
    case boost::date_time::hundreth:
        n *= 10000;
        break;
    case boost::date_time::milli:
        n *= 1000;
        break;
    case boost::date_time::ten_thousandth:
        n *= 100;
        break;
    case boost::date_time::micro:
        // good to go
        break;
    case boost::date_time::nano:
        n /= 1000;
        break;
    case boost::date_time::NumResolutions:
        // shouldn't happen
        break;
    }
    return n;
}


}   // end namespace plugins
}   // end namespace pion

#endif

---