text:
/** Source file for job_stream python extension.  Usage like:

from job_stream import Job, run

class AddOne(Job):
    def handleWork(self, work):
        self.emit(work + 1)

if __name__ == '__main__':
    run()

DEBUG EXECUTION::

LD_LIBRARY_PATH=/u/wwoods/dev/boost_1_55_0/stage/lib/:~/dev/yaml-cpp-0.5.1/build/:/usr/lib/openmpi/lib:/usr/lib YAML_CPP=~/dev/yaml-cpp-0.5.1/ bash -c "cmake .. && make -j8 test-python"
*/

#include <job_stream/debug_internals.h>
#include <job_stream/job_stream.h>
#include <job_stream/pythonType.h>

#include <boost/python.hpp>
#include <boost/python/raw_function.hpp>
#include <dlfcn.h>

//Macro to help with timing
#define COMBINE1(X, Y) X##Y
#define COMBINE(X, Y) COMBINE1(X, Y)
#define TIMER(typ) job_stream::processor::Processor::WorkTimer COMBINE(workTimer, __LINE__)(this->_shell->processor, job_stream::processor::Processor::typ)

namespace bp = boost::python;
using job_stream::python::SerializedPython;

namespace job_stream {
namespace python {

bp::object encodeObj, decodeStr;
bp::object object, repr;
bp::object stackAlreadyPrintedError;

} //python
} //job_stream

/** So, as per the OpenMPI FAQ, MPI plugins will fail to load if libmpi.so is
    loaded in a local namespace.  From what I can tell, that is what python is
    doing, since those are failing to load.  So, we load the dll manually, and
    publicly. */
class _DlOpener {
public:
    _DlOpener(const char* soName) {
        this->lib = dlopen(soName, RTLD_NOW | RTLD_GLOBAL);
    }
    ~_DlOpener() {
        dlclose(this->lib);
    }

private:
    void* lib;
} holdItOpenGlobally("libmpi.so");

/** Used to execute python code within job_stream's operations. */
class _PyGilAcquire {
public:
    _PyGilAcquire() {
        this->gilState = PyGILState_Ensure();
    }
    ~_PyGilAcquire() {
        PyGILState_Release(this->gilState);
    }
private:
    PyGILState_STATE gilState;
};

/** Used outside of our main thread to ensure the GIL is by default released for
    job_stream's operations. */
class _PyGilRelease {
public:
    _PyGilRelease() {
        this->_save = PyEval_SaveThread();
    }
    ~_PyGilRelease() {
        PyEval_RestoreThread(this->_save);
    }
private:
    PyThreadState* _save;
};

/** If an error occurred in python, print its stack trace and information, then
    raise a C++ exception.  To be used in catch blocks around python code. */
#define CHECK_PYTHON_ERROR(m) \
    if (PyErr_Occurred()) { \
        if (!PyErr_ExceptionMatches(job_stream::python::stackAlreadyPrintedError.ptr())) { \
            /* Print stack and move error into sys.last_type, sys.last_value, \
               sys.last_traceback.  Also clears the error. */ \
            PyErr_Print(); \
        } \
        ERROR("Python exception caught, stack trace printed: " << m); \
    }

/** Must be called with GIL locked
 * */
std::string getStrFromPy(bp::object o) {
    PyObject* ptr = o.ptr();
    if (PyBytes_Check(ptr)) {
        return std::string(PyBytes_AsString(ptr),
                PyBytes_Size(ptr));
    }

    bp::object bytes = o.attr("encode")("utf-8");
    ptr = bytes.ptr();
    return std::string(PyBytes_AsString(ptr),
            PyBytes_Size(ptr));
}

/** Must be called with GIL locked */
SerializedPython pythonToSerialized(const bp::object& o) {
    bp::object encoded(job_stream::python::encodeObj(o));
    return SerializedPython(getStrFromPy(encoded));
}

/** Must be called with GIL locked */
bp::object serializedToPython(const SerializedPython& sp) {
    try {
#if PYTHON_MAJOR < 3
        return job_stream::python::decodeStr(sp.data);
#else
        bp::object buffer(bp::handle<>(PyMemoryView_FromMemory(
                    const_cast<char*>(sp.data.c_str()), sp.data.size(),
                    PyBUF_READ)));
        return job_stream::python::decodeStr(buffer);
#endif //PYTHON_MAJOR <= 2
    }
    catch (...) {
        CHECK_PYTHON_ERROR("Error deserializing");
        throw;
    }
}

/** Must be called with GIL locked; translate a YAML node into a python object that can
    be read. */
bp::object configToPython(const YAML::LockedNode& n) {
    if (n.IsNull()) {
        return bp::object();
    }
    else if (n.IsSequence()) {
        bp::list l;
        for (int i = 0, m = n.size(); i < m; i++) {
            l.append(configToPython(n[i]));
        }
        return l;
    }
    else if (n.IsMap()) {
        bp::dict d;
        for (auto it = n.begin(); it != n.end(); it++) {
            YAML::GuardedNode gn;
            gn.set(it->second);
            d[it->first.as<std::string>()] = configToPython(gn.acquire());
        }
        return d;
    }
    else if (n.IsScalar()) {
        //We need to ascertain the type, somehow.  Assume it's a double, otherwise a str?
        #define TRY_TYPE(t) try { return bp::object(n.as<t>()); } catch (...) {}
        TRY_TYPE(int)
        TRY_TYPE(double)
        return bp::object(n.as<std::string>());
        #undef TRY_TYPE
    }
    else {
        ERROR("Unknown node type: " << n.Type());
    }
}

namespace job_stream {
namespace python {
std::istream& operator>>(std::istream& is,
        SerializedPython& sp) {
    //Happens outside the GIL!
    ERROR("Cannot do this");
    return is;
}

std::ostream& operator<<(std::ostream& os,
        const SerializedPython& sp) {
    //Happens outside the GIL!
    _PyGilAcquire gilLock;
    bp::object o = serializedToPython(sp);
    std::string s = getStrFromPy(job_stream::python::repr(o));
    return os << s;
}

} //python
} //job_stream

class PyFrame;
class PyJob;
class PyReducer;
void _timerHandleResult(job_stream::processor::Processor* p, bp::object res);

/*********************************************************************/
//Job

/** Since our jobs are technically allocated from Python, but job_stream expects
    jobs to belong to it (and subsequently frees them), we use a shell around
    the python job for interacting with job_stream.
    */
class PyJobShell : public job_stream::Job<PyJobShell, SerializedPython> {
public:
    //TODO - Remove _AutoRegister for this case, so we don't need NAME() or
    //pyHandleWork != 0 in base class.
    static const char* NAME() { return "_pyJobBase"; }

    PyJobShell() : _job(0) {}
    PyJobShell(PyJob* job) : _job(job) {}
    virtual ~PyJobShell();

    void postSetup() override;
    void handleWork(std::unique_ptr<SerializedPython> work) override;

private:
    friend class PyJob;

    PyJob* _job;
};

class PyJob {
public:
    PyJob() {}
    virtual ~PyJob() {}

    void forceCheckpoint(bool forceQuit) {
        this->_shell->forceCheckpoint(forceQuit);
    }

    void pyEmit(bp::object o) {
        this->pyEmit(o, "");
    }

    void pyEmit(bp::object o, const std::string& target) {
        SerializedPython obj = pythonToSerialized(o);

        //Let other threads do stuff while we're emitting
        _PyGilRelease gilLock;
        this->_shell->emit(obj, target);
    }

    void handleWork(std::unique_ptr<SerializedPython> work) {
        //Entry point to python!  Reacquire the GIL to deserialize and run our
        //code
        TIMER(TIME_IDLE); //out of gil
        _PyGilAcquire gilLock;
        try {
            TIMER(TIME_SYSTEM); //For serializing
            bp::object workObj = serializedToPython(*work);
            TIMER(TIME_USER); //For user code
            bp::object res = this->pyHandleWork(workObj);
            _timerHandleResult(this->_shell->processor, res);
        }
        catch (...) {
            CHECK_PYTHON_ERROR("PyJob::handleWork()");
            throw;
        }
    }

    void postSetup() {
        _PyGilAcquire gilLock;
        try {
            this->_pythonObject.attr("config") = configToPython(
                    this->_shell->config.get());
            this->pyPostSetup();
        }
        catch (...) {
            CHECK_PYTHON_ERROR("PyJob::postSetup()");
            throw;
        }
    }

    virtual bp::object pyHandleWork(bp::object work) {
        throw std::runtime_error("Python handleWork() not implemented");
    }

    virtual void pyPostSetup() {
        throw std::runtime_error("Python postSetup() not implemented");
    }

    void setPythonObject(bp::object o) {
        this->_pythonObject = o;
    }

    void setShell(PyJobShell* shell) {
        this->_shell = shell;
    }

    void releaseShell() {
        this->_shell = 0;
        _PyGilAcquire runningPyCode;
        //We no longer need to exist, so let the python GC clean up
        this->_pythonObject = bp::object();
    }

private:
    //Prevent our derivative class from being cleaned up.
    bp::object _pythonObject;
    PyJobShell* _shell;
};

class PyJobExt : public PyJob {
public:
    PyJobExt(PyObject* p) : self(p) {}
    PyJobExt(PyObject* p, const PyJob& j) : PyJob(j), self(p) {}
    virtual ~PyJobExt() {}

    bp::object pyHandleWork(bp::object work) override {
        return bp::call_method<bp::object>(this->self, "handleWork", work);
    }

    static bp::object default_pyHandleWork(PyJob& self_, bp::object work) {
        return self_.PyJob::pyHandleWork(work);
    }

    void pyPostSetup() override {
        bp::call_method<void>(this->self, "postSetup");
    }

    static void default_pyPostSetup(PyJob& self_) {
        self_.PyJob::pyPostSetup();
    }

private:
    PyObject* self;
};

PyJobShell::~PyJobShell() {
    if (this->_job) {
        this->_job->releaseShell();
        this->_job = 0;
    }
}
void PyJobShell::postSetup() {
    this->_job->postSetup();
}
void PyJobShell::handleWork(std::unique_ptr<SerializedPython> work) {
    this->_job->handleWork(std::move(work));
}

/*********************************************************************/
//Reducer

/** Since our reducers are technically allocated from Python, but job_stream expects
    jobs to belong to it (and subsequently frees them), we use a shell around
    the python reducer for interacting with job_stream.
    */
class PyReducerShell : public job_stream::Reducer<PyReducerShell,
        SerializedPython, SerializedPython> {
public:
    //TODO - Remove _AutoRegister for this case, so we don't need NAME() or
    //pyHandleWork != 0 in base class.
    static const char* NAME() { return "_pyReducerBase"; }

    PyReducerShell() : _reducer(0) {}
    PyReducerShell(PyReducer* reducer) : _reducer(reducer) {}
    virtual ~PyReducerShell();

    void postSetup() override;
    void handleInit(SerializedPython& current) override;
    void handleAdd(SerializedPython& current,
            std::unique_ptr<SerializedPython> work) override;
    void handleJoin(SerializedPython& current,
            std::unique_ptr<SerializedPython> other) override;
    void handleDone(SerializedPython& current) override;

private:
    friend class PyReducer;

    PyReducer* _reducer;
};

class PyReducer {
public:
    PyReducer() {}
    virtual ~PyReducer() {}

    void forceCheckpoint(bool forceQuit) {
        this->_shell->forceCheckpoint(forceQuit);
    }

    void pyEmit(bp::object o) {
        SerializedPython obj = pythonToSerialized(o);

        //Let other threads do stuff while we're emitting
        _PyGilRelease gilLock;
        this->_shell->emit(obj);
    }

    void pyRecur(bp::object o) {
        this->pyRecur(o, "");
    }

    void pyRecur(bp::object o, const std::string& target) {
        SerializedPython obj = pythonToSerialized(o);

        //Let other threads do stuff while we're recurring
        _PyGilRelease gilLock;
        this->_shell->recur(obj, target);
    }

    void handleAdd(SerializedPython& current,
            std::unique_ptr<SerializedPython> work) {
        //Entry point to python!  Reacquire the GIL to deserialize and run our
        //code
        TIMER(TIME_IDLE); //out of gil
        _PyGilAcquire gilLock;
        try {
            TIMER(TIME_SYSTEM); //Serializing
            bp::object stash = serializedToPython(current);
            bp::object workObj = serializedToPython(*work);
            {
                TIMER(TIME_USER); //User code
                bp::object res = this->pyHandleAdd(stash, workObj);
                _timerHandleResult(this->_shell->processor, res);
            }
            current = pythonToSerialized(stash);
        }
        catch (...) {
            CHECK_PYTHON_ERROR("PyReducer::handleAdd");
            throw;
        }
    }

    void handleDone(SerializedPython& current) {
        //Entry point to python!  Reacquire the GIL to deserialize and run our
        //code
        TIMER(TIME_IDLE);
        _PyGilAcquire gilLock;
        try {
            TIMER(TIME_SYSTEM);
            bp::object stash = serializedToPython(current);
            {
                TIMER(TIME_USER);
                bp::object res = this->pyHandleDone(stash);
                _timerHandleResult(this->_shell->processor, res);
            }
            current = pythonToSerialized(stash);
        }
        catch (...) {
            CHECK_PYTHON_ERROR("PyReducer::handleDone");
            throw;
        }
    }

    void handleInit(SerializedPython& current) {
        //Entry point to python!  Reacquire the GIL to deserialize and run our
        //code
        TIMER(TIME_IDLE);
        _PyGilAcquire gilLock;
        try {
            TIMER(TIME_SYSTEM);
            //Current is uninitialized; make it a python object()
            bp::object stash = job_stream::python::object();
            {
                TIMER(TIME_USER);
                bp::object res = this->pyHandleInit(stash);
                _timerHandleResult(this->_shell->processor, res);
            }
            current = pythonToSerialized(stash);
        }
        catch (...) {
            CHECK_PYTHON_ERROR("PyReducer::handleInit");
            throw;
        }
    }

    void handleJoin(SerializedPython& current,
            std::unique_ptr<SerializedPython> other) {
        //Entry point to python!  Reacquire the GIL to deserialize and run our
        //code
        TIMER(TIME_IDLE);
        _PyGilAcquire gilLock;
        try {
            TIMER(TIME_SYSTEM);
            bp::object stash = serializedToPython(current);
            bp::object otter = serializedToPython(*other);
            {
                TIMER(TIME_USER);
                bp::object res = this->pyHandleJoin(stash, otter);
                _timerHandleResult(this->_shell->processor, res);
            }
            current = pythonToSerialized(stash);
        }
        catch (...) {
            CHECK_PYTHON_ERROR("PyReducer::handleJoin");
            throw;
        }
    }

    void postSetup() {
        _PyGilAcquire gilLock;
        try {
            this->_pythonObject.attr("config") = configToPython(
                    this->_shell->config.get());
            this->pyPostSetup();
        }
        catch (...) {
            CHECK_PYTHON_ERROR("PyReducer::postSetup");
            throw;
        }
    }

    virtual bp::object pyHandleAdd(bp::object stash, bp::object work) {
        throw std::runtime_error("Python handleAdd() not implemented");
    }

    virtual bp::object pyHandleDone(bp::object stash) {
        throw std::runtime_error("Python handleDone() not implemented");
    }

    virtual bp::object pyHandleInit(bp::object stash) {
        throw std::runtime_error("Python handleInit() not implemented");
    }

    virtual bp::object pyHandleJoin(bp::object stash, bp::object other) {
        throw std::runtime_error("Python handleJoin() not implemented");
    }

    virtual void pyPostSetup() {
        throw std::runtime_error("Python postSetup() not implemented");
    }

    void setPythonObject(bp::object o) {
        this->_pythonObject = o;
    }

    void setShell(PyReducerShell* shell) {
        this->_shell = shell;
    }

    void releaseShell() {
        this->_shell = 0;
        _PyGilAcquire runningPyCode;
        //We no longer need to exist, so let the python GC clean up
        this->_pythonObject = bp::object();
    }

private:
    //Prevent our derivative class from being cleaned up.
    bp::object _pythonObject;
    PyReducerShell* _shell;
};

class PyReducerExt : public PyReducer {
public:
    PyReducerExt(PyObject* p) : self(p) {}
    PyReducerExt(PyObject* p, const PyReducer& j) : PyReducer(j), self(p) {}
    virtual ~PyReducerExt() {}

    bp::object  pyHandleAdd(bp::object stash, bp::object work) override {
        return bp::call_method<bp::object>(this->self, "handleAdd", stash, work);
    }

    static bp::object default_pyHandleAdd(PyReducer& self_, bp::object stash,
            bp::object work) {
        return self_.PyReducer::pyHandleAdd(stash, work);
    }

    bp::object pyHandleDone(bp::object stash) override {
        return bp::call_method<bp::object>(this->self, "handleDone", stash);
    }

    static bp::object default_pyHandleDone(PyReducer& self_, bp::object stash) {
        return self_.PyReducer::pyHandleDone(stash);
    }

    bp::object pyHandleInit(bp::object stash) override {
        return bp::call_method<bp::object>(this->self, "handleInit", stash);
    }

    static bp::object default_pyHandleInit(PyReducer& self_, bp::object stash) {
        return self_.PyReducer::pyHandleInit(stash);
    }

    bp::object pyHandleJoin(bp::object stash, bp::object other) override {
        return bp::call_method<bp::object>(this->self, "handleJoin", stash, other);
    }

    static bp::object default_pyHandleJoin(PyReducer& self_, bp::object stash,
            bp::object other) {
        return self_.PyReducer::pyHandleJoin(stash, other);
    }

    void pyPostSetup() override {
        bp::call_method<void>(this->self, "postSetup");
    }

    static void default_pyPostSetup(PyReducer& self_) {
        self_.PyReducer::pyPostSetup();
    }

private:
    PyObject* self;
};

PyReducerShell::~PyReducerShell() {
    if (this->_reducer) {
        this->_reducer->releaseShell();
        this->_reducer = 0;
    }
}
void PyReducerShell::postSetup() {
    this->_reducer->postSetup();
}
void PyReducerShell::handleInit(SerializedPython& current) {
    this->_reducer->handleInit(current);
}
void PyReducerShell::handleAdd(SerializedPython& current,
        std::unique_ptr<SerializedPython> work) {
    this->_reducer->handleAdd(current, std::move(work));
}
void PyReducerShell::handleJoin(SerializedPython& current,
        std::unique_ptr<SerializedPython> other) {
    this->_reducer->handleJoin(current, std::move(other));
}
void PyReducerShell::handleDone(SerializedPython& current) {
    this->_reducer->handleDone(current);
}

/*********************************************************************/
//Frame

/** Since our frames are technically allocated from Python, but job_stream expects
    jobs to belong to it (and subsequently frees them), we use a shell around
    the python reducer for interacting with job_stream.
    */
class PyFrameShell : public job_stream::Frame<PyFrameShell,
        SerializedPython, SerializedPython> {
public:
    //TODO - Remove _AutoRegister for this case, so we don't need NAME() or
    //pyHandleWork != 0 in base class.
    static const char* NAME() { return "_pyFrameBase"; }

    PyFrameShell() : _frame(0) {}
    PyFrameShell(PyFrame* frame) : _frame(frame) {}
    virtual ~PyFrameShell();

    void postSetup() override;
    void handleDone(SerializedPython& current) override;
    void handleFirst(SerializedPython& current,
            std::unique_ptr<SerializedPython> work) override;
    void handleNext(SerializedPython& current,
            std::unique_ptr<SerializedPython> work) override;

private:
    friend class PyFrame;

    PyFrame* _frame;
};

class PyFrame {
public:
    PyFrame() {}
    virtual ~PyFrame() {}

    void forceCheckpoint(bool forceQuit) {
        this->_shell->forceCheckpoint(forceQuit);
    }

    void pyEmit(bp::object o) {
        SerializedPython obj = pythonToSerialized(o);

        //Let other threads do stuff while we're emitting
        _PyGilRelease gilLock;
        this->_shell->emit(obj);
    }

    void pyRecur(bp::object o) {
        this->pyRecur(o, "");
    }

    void pyRecur(bp::object o, const std::string& target) {
        SerializedPython obj = pythonToSerialized(o);

        //Let other threads do stuff while we're recurring
        _PyGilRelease gilLock;
        this->_shell->recur(obj, target);
    }

    void handleDone(SerializedPython& current) {
        //Entry point to python!  Reacquire the GIL to deserialize and run our
        //code
        TIMER(TIME_IDLE);
        _PyGilAcquire gilLock;
        try {
            TIMER(TIME_SYSTEM);
            bp::object stash = serializedToPython(current);
            {
                TIMER(TIME_USER);
                bp::object res = this->pyHandleDone(stash);
                _timerHandleResult(this->_shell->processor, res);
            }
            current = pythonToSerialized(stash);
        }
        catch (...) {
            CHECK_PYTHON_ERROR("PyFrame::handleDone");
            throw;
        }
    }

    void handleFirst(SerializedPython& current,
            std::unique_ptr<SerializedPython> work) {
        //Entry point to python!  Reacquire the GIL to deserialize and run our
        //code
        TIMER(TIME_IDLE);
        _PyGilAcquire gilLock;
        try {
            TIMER(TIME_SYSTEM);
            bp::object stash = job_stream::python::object();
            bp::object workObj = serializedToPython(*work);
            {
                TIMER(TIME_USER);
                bp::object res = this->pyHandleFirst(stash, workObj);
                _timerHandleResult(this->_shell->processor, res);
            }
            current = pythonToSerialized(stash);
        }
        catch (...) {
            CHECK_PYTHON_ERROR("PyFrame::handleFirst");
            throw;
        }
    }

    void handleNext(SerializedPython& current,
            std::unique_ptr<SerializedPython> work) {
        //Entry point to python!  Reacquire the GIL to deserialize and run our
        //code
        TIMER(TIME_IDLE);
        _PyGilAcquire gilLock;
        try {
            TIMER(TIME_SYSTEM);
            bp::object stash = serializedToPython(current);
            bp::object workObj = serializedToPython(*work);
            {
                TIMER(TIME_USER);
                bp::object res = this->pyHandleNext(stash, workObj);
                _timerHandleResult(this->_shell->processor, res);
            }
            current = pythonToSerialized(stash);
        }
        catch (...) {
            CHECK_PYTHON_ERROR("PyFrame::handleNext");
            throw;
        }
    }

    void postSetup() {
        _PyGilAcquire gilLock;
        try {
            this->_pythonObject.attr("config") = configToPython(
                    this->_shell->config.get());
            this->pyPostSetup();
        }
        catch (...) {
            CHECK_PYTHON_ERROR("PyFrame::postSetup");
            throw;
        }
    }

    virtual bp::object pyHandleDone(bp::object stash) {
        throw std::runtime_error("Python handleDone() not implemented");
    }

    virtual bp::object pyHandleFirst(bp::object stash, bp::object work) {
        throw std::runtime_error("Python handleFirst() not implemented");
    }

    virtual bp::object pyHandleNext(bp::object stash, bp::object work) {
        throw std::runtime_error("Python handleNext() not implemented");
    }

    virtual void pyPostSetup() {
        throw std::runtime_error("Python postSetup() not implemented");
    }

    void setPythonObject(bp::object o) {
        this->_pythonObject = o;
    }

    void setShell(PyFrameShell* shell) {
        this->_shell = shell;
    }

    void releaseShell() {
        this->_shell = 0;
        _PyGilAcquire runningPyCode;
        //We no longer need to exist, so let the python GC clean up
        this->_pythonObject = bp::object();
    }

private:
    //Prevent our derivative class from being cleaned up.
    bp::object _pythonObject;
    PyFrameShell* _shell;
};

class PyFrameExt : public PyFrame {
public:
    PyFrameExt(PyObject* p) : self(p) {}
    PyFrameExt(PyObject* p, const PyFrame& j) : PyFrame(j), self(p) {}
    virtual ~PyFrameExt() {}

    bp::object pyHandleDone(bp::object stash) override {
        return bp::call_method<bp::object>(this->self, "handleDone", stash);
    }

    static bp::object default_pyHandleDone(PyFrame& self_, bp::object stash) {
        return self_.PyFrame::pyHandleDone(stash);
    }

    bp::object pyHandleFirst(bp::object stash, bp::object work) override {
        return bp::call_method<bp::object>(this->self, "handleFirst", stash, work);
    }

    static bp::object default_pyHandleFirst(PyFrame& self_, bp::object stash,
            bp::object work) {
        return self_.PyFrame::pyHandleFirst(stash, work);
    }

    bp::object pyHandleNext(bp::object stash, bp::object work) override {
        return bp::call_method<bp::object>(this->self, "handleNext", stash, work);
    }

    static bp::object default_pyHandleNext(PyFrame& self_, bp::object stash,
            bp::object work) {
        return self_.PyFrame::pyHandleNext(stash, work);
    }

    void pyPostSetup() override {
        bp::call_method<void>(this->self, "postSetup");
    }

    static void default_pyPostSetup(PyFrame& self_) {
        self_.PyFrame::pyPostSetup();
    }

private:
    PyObject* self;
};

PyFrameShell::~PyFrameShell() {
    if (this->_frame) {
        this->_frame->releaseShell();
        this->_frame = 0;
    }
}
void PyFrameShell::postSetup() {
    this->_frame->postSetup();
}
void PyFrameShell::handleDone(SerializedPython& current) {
    this->_frame->handleDone(current);
}
void PyFrameShell::handleFirst(SerializedPython& current,
        std::unique_ptr<SerializedPython> work) {
    this->_frame->handleFirst(current, std::move(work));
}
void PyFrameShell::handleNext(SerializedPython& current,
        std::unique_ptr<SerializedPython> work) {
    this->_frame->handleNext(current, std::move(work));
}

/*********************************************************************/
uint64_t _cpuThreadTimeMs() {
    return job_stream::processor::Processor::_getThreadCpuTimeMs();
}

/** Starts clock- and cpu- timers */
//Why is this a dangling pointer rather than just a thread_local std::vector?
//Because gcc has a bug that prevents it from being a non-primitive type.  So,
//a pointer is used as a workaround.
typedef std::tuple<uint64_t, uint64_t> PyTimeInfo;
thread_local std::vector<PyTimeInfo>* _pyTimers = 0;

void _timerStart() {
    if (_pyTimers == 0) {
        _pyTimers = new std::vector<PyTimeInfo>();
    }
    uint64_t wt = job_stream::message::Location::getCurrentTimeMs();
    uint64_t ct = _cpuThreadTimeMs();
    _pyTimers->emplace_back(wt, ct);
}

/** Returns the value since the latest clock- and cpu- timers */
bp::tuple _timerPop() {
    uint64_t ct = _cpuThreadTimeMs();
    uint64_t wt = job_stream::message::Location::getCurrentTimeMs();
    auto& tup = _pyTimers->back();
    //Unfortunately, the resolution of getCurrentTimeMs() seems to be slightly
    //less than that of _cpuThreadTimeMs.  This leads to bad situations where
    //the reported cpu time is greater than reported wall clock time, which
    //is impossible.  So, take the larger for wall-clock time.
    ct -= std::get<1>(tup);
    wt -= std::get<0>(tup);
    wt = std::max(ct, wt);
    auto res = bp::make_tuple(wt, ct);
    _pyTimers->pop_back();
    return res;
}

PyTimeInfo _timeInfoFromBp(bp::object o) {
    if (bp::len(o) != 2) {
        throw std::runtime_error("Not a tuple of size 2");
    }
    return std::make_tuple((uint64_t)bp::extract<uint32_t>(o[0]),
            (uint64_t)bp::extract<uint32_t>(o[1]));
}

/** Handles integration of timing information from a python method. */
void _timerHandleResult(job_stream::processor::Processor* p,
        bp::object res) {
    if (!res.is_none()) {
        //Timing information from an internal library; time is USER only,
        //all else assumed to be SYSTEM.
        PyTimeInfo usrT = _timeInfoFromBp(res);
        p->_modifyWorkTimer(job_stream::processor::Processor::TIME_USER,
                std::get<0>(usrT), std::get<1>(usrT));
    }
}

std::string checkpointInfo(bp::object o) {
    std::string fname = getStrFromPy(o);
    return job_stream::checkpointInfo(fname);
}

int getCpuCount() {
    return job_stream::getCpuCount();
}

int getHostCpuCount() {
    return job_stream::getHostCpuCount();
}

int getRank() {
    return job_stream::getRank();
}

bp::tuple invoke(bp::object progAndArgs, bp::list transientErrors,
        int maxRetries) {
    std::vector<std::string> cProg, cErrors;
    for (int i = 0, m = bp::len(progAndArgs); i < m; i++) {
        cProg.emplace_back(getStrFromPy(progAndArgs[i]));
    }
    for (int i = 0, m = bp::len(transientErrors); i < m; i++) {
        cErrors.emplace_back(getStrFromPy(transientErrors[i]));
    }
    std::tuple<std::string, std::string> results;
    {
        _PyGilRelease releaseGilForExecution;
        results = job_stream::invoke::run(cProg, cErrors, maxRetries);
    }

    return bp::make_tuple(std::get<0>(results), std::get<1>(results));
}

void registerEncoding(bp::object object, bp::object stackAlreadyPrintedError,
        bp::object encode, bp::object decode) {
    job_stream::python::object = object;
    job_stream::python::stackAlreadyPrintedError = stackAlreadyPrintedError;
    job_stream::python::encodeObj = encode;
    job_stream::python::decodeStr = decode;

    bp::handle<> builtinH(bp::borrowed(PyEval_GetBuiltins()));
    bp::object builtins(builtinH);
    job_stream::python::repr = builtins["repr"];
}

void registerFrame(bp::object name, bp::object cls) {
    std::string cname = getStrFromPy(name);
    job_stream::job::addReducer(cname,
            [cls, cname]() -> PyFrameShell* {
                _PyGilAcquire allocateInPython;
                try {
                    bp::object holder = cls();
                    //Pointer belongs to python!
                    PyFrame* r = bp::extract<PyFrame*>(holder);
                    r->setPythonObject(holder);
                    PyFrameShell* p = new PyFrameShell(r);
                    r->setShell(p);
                    return p;
                }
                catch (...) {
                    CHECK_PYTHON_ERROR("Frame allocation: " << cname);
                    throw;
                }
            });
}

void registerJob(bp::object name, bp::object cls) {
    std::string cname = getStrFromPy(name);
    job_stream::job::addJob(cname,
            [cls, cname]() -> PyJobShell* {
                _PyGilAcquire allocateInPython;
                try {
                    bp::object holder = cls();
                    //Remember, this pointer belongs to python!
                    PyJob* r = bp::extract<PyJob*>(holder);
                    r->setPythonObject(holder);
                    PyJobShell* p = new PyJobShell(r);
                    r->setShell(p);
                    return p;
                }
                catch (...) {
                    CHECK_PYTHON_ERROR("Job allocation: " << cname);
                    throw;
                }
            });
}

void registerReducer(bp::object name, bp::object cls) {
    std::string cname = getStrFromPy(name);
    job_stream::job::addReducer(cname,
            [cls, cname]() -> PyReducerShell* {
                _PyGilAcquire allocateInPython;
                try {
                    bp::object holder = cls();
                    //Pointer belongs to python!
                    PyReducer* r = bp::extract<PyReducer*>(holder);
                    r->setPythonObject(holder);
                    PyReducerShell* p = new PyReducerShell(r);
                    r->setShell(p);
                    return p;
                }
                catch (...) {
                    CHECK_PYTHON_ERROR("Reducer allocation: " << cname);
                    throw;
                }
            });
}

bp::object runProcessor(bp::tuple args, bp::dict kwargs) {
    ASSERT(bp::len(args) == 3, "runProcessor() takes exactly 3 non-keyword "
            "arguments: config, workList, handleResult");

    job_stream::SystemArguments sa;
    sa.config = getStrFromPy(args[0]);

    sa.checkExternalSignals = []() -> bool {
        _PyGilAcquire errCheck;
        if (PyErr_CheckSignals()) {
            return true;
        }
        return false;
    };
    sa.handleOutputCallback = [&args](std::unique_ptr<job_stream::AnyType> result) -> void {
        _PyGilAcquire outSaver;
        try {
            args[2](serializedToPython(*result->as<SerializedPython>()));
        }
        catch (...) {
            CHECK_PYTHON_ERROR("handleResult");
            throw;
        }
    };

    bp::object workList = bp::object(args[1]);
    for (int i = 0, m = bp::len(workList); i < m; i++) {
        job_stream::queueInitialWork(pythonToSerialized(
                    bp::object(workList[i])));
    }

    //Enumerate kwargs
    bp::list keys = kwargs.keys();
    for (int i = 0, m = bp::len(keys); i < m; i++) {
        std::string key = getStrFromPy(keys[i]);
        bp::object val = kwargs[key];

        if (key == "checkpointFile") {
            sa.checkpointFile = getStrFromPy(val);
        }
        else if (key == "checkpointInterval") {
            sa.checkpointIntervalMs = (int)(bp::extract<double>(val) * 1000);
        }
        else if (key == "checkpointSyncInterval") {
            sa.checkpointSyncIntervalMs = (int)(bp::extract<double>(val)
                    * 1000);
        }
        else {
            ERROR("Unrecognized kwarg to runProcessor: " << key);
        }
    }

    PyEval_InitThreads();
    {
        _PyGilRelease releaser;
        job_stream::runProcessor(sa);
    }

    return bp::object();
}

BOOST_PYTHON_MODULE(_job_stream) {
    bp::scope().attr("__doc__") = "C internals for job_stream python library; "
            "see https://github.com/wwoods/job_stream for more info";

    bp::def("_cpuThreadTimeMs", _cpuThreadTimeMs, "Returns a uint64_t of the "
            "current thread's CPU time, in milliseconds.");
    bp::def("_timerStart", _timerStart, "Starts a new clock- and cpu- timer");
    bp::def("_timerPop", _timerPop, "Pops a previously started timer, "
            "returning (clock-time, cpu-time) in MS");
    bp::def("checkpointInfo", checkpointInfo, "Returns a human-readable "
            "string with details of a checkpoint's state");
    bp::def("getCpuCount", getCpuCount, "Returns the number of CPUs in the cluster");
    bp::def("getHostCpuCount", getHostCpuCount, "Returns the number of CPUs "
            "desired on this machine.");
    bp::def("getRank", getRank, "Returns the mpi rank of this host");
    bp::def("invoke", invoke, "Invokes the given application.  See "
            "job_stream.invoke in the python module for more information.");
    bp::def("registerEncoding", registerEncoding, "Registers the encoding and "
            "decoding functions used by C code.");
    bp::def("registerFrame", registerFrame, "Registers a frame");
    bp::def("registerJob", registerJob, "Registers a job");
    bp::def("registerReducer", registerReducer, "Registers a reducer");
    //Does not accept docstring!!!
    bp::def("runProcessor", bp::raw_function(runProcessor));

    { //PyJob
        void (PyJob::*emit1)(bp::object) = &PyJob::pyEmit;
        void (PyJob::*emit2)(bp::object, const std::string&) = &PyJob::pyEmit;
        bp::class_<PyJob, PyJobExt>("Job", "A basic job")
                .def(bp::init<>())
                .enable_pickling()
                .def("emit", emit1, "Emit to only target")
                .def("emit", emit2, "Emit to specific target out of list")
                .def("handleWork", PyJobExt::default_pyHandleWork)
                .def("postSetup", PyJobExt::default_pyPostSetup)
                .def("_forceCheckpoint", &PyJob::forceCheckpoint, "Force a "
                    "checkpoint after this work.  If True is passed, cause the "
                    "program to crash afterwards.")
                ;
    }

    { //PyReducer
        void (PyReducer::*emit1)(bp::object) = &PyReducer::pyEmit;
        void (PyReducer::*recur1)(bp::object) = &PyReducer::pyRecur;
        void (PyReducer::*recur2)(bp::object, const std::string&) = &PyReducer::pyRecur;
        bp::class_<PyReducer, PyReducerExt>("Reducer", "A basic reducer")
                .def(bp::init<>())
                .enable_pickling()
                .def("emit", emit1, "Emit to only target (outside of reducer)")
                .def("recur", recur1, "Recur to only target")
                .def("recur", recur2, "Recur to specific target")
                .def("handleAdd", PyReducerExt::default_pyHandleAdd)
                .def("handleDone", PyReducerExt::default_pyHandleDone)
                .def("handleInit", PyReducerExt::default_pyHandleInit)
                .def("handleJoin", PyReducerExt::default_pyHandleJoin)
                .def("postSetup", PyReducerExt::default_pyPostSetup)
                .def("_forceCheckpoint", &PyReducer::forceCheckpoint, "Force a "
                    "checkpoint after this work.  If True is passed, cause the "
                    "program to crash afterwards.")
                ;
    }

    { //PyFrame
        void (PyFrame::*emit1)(bp::object) = &PyFrame::pyEmit;
        void (PyFrame::*recur1)(bp::object) = &PyFrame::pyRecur;
        void (PyFrame::*recur2)(bp::object, const std::string&) = &PyFrame::pyRecur;
        bp::class_<PyFrame, PyFrameExt>("Frame", "A basic frame")
                .def(bp::init<>())
                .enable_pickling()
                .def("emit", emit1, "Emit to only target (outside of frame)")
                .def("recur", recur1, "Recur to only target")
                .def("recur", recur2, "Recur to specific target")
                .def("handleDone", PyFrameExt::default_pyHandleDone)
                .def("handleFirst", PyFrameExt::default_pyHandleFirst)
                .def("handleNext", PyFrameExt::default_pyHandleNext)
                .def("postSetup", PyFrameExt::default_pyPostSetup)
                .def("_forceCheckpoint", &PyFrame::forceCheckpoint, "Force a "
                    "checkpoint after this work.  If True is passed, cause the "
                    "program to crash afterwards.")
                ;
    }
}


text:
#ifndef EDBA_TYPES_HPP
#define EDBA_TYPES_HPP

#include <edba/string_ref.hpp>
#include <edba/detail/utils.hpp>

#include <boost/config.hpp>
#include <boost/mpl/vector.hpp>
#include <boost/mpl/contains.hpp>
#include <boost/mpl/or.hpp>
#include <boost/variant/variant.hpp>
#include <boost/variant/get.hpp>
#include <boost/cstdint.hpp>

#include <string>
#include <ctime>
#include <iosfwd>

namespace edba {

class conn_info;
class session;
class session_monitor;
class statement;
class row;

namespace backend {

struct result_iface;
struct statement_iface;
struct connection_iface;

EDBA_ADD_INTRUSIVE_PTR_SUPPORT_FOR_TYPE(result_iface);
EDBA_ADD_INTRUSIVE_PTR_SUPPORT_FOR_TYPE(statement_iface);
EDBA_ADD_INTRUSIVE_PTR_SUPPORT_FOR_TYPE(connection_iface);

typedef boost::intrusive_ptr<result_iface> result_ptr;
typedef boost::intrusive_ptr<statement_iface> statement_ptr;
typedef boost::intrusive_ptr<connection_iface> connection_ptr;

}

/// Null type
struct null_type {};

namespace {
///
/// Global instance of null_type, can be used in bind expressions
///
null_type null;
}

/// Types natively supported by statement::bind method
typedef boost::mpl::vector<
    null_type
  , short
  , unsigned short
  , int
  , unsigned int
  , long
  , unsigned long
#ifndef BOOST_NO_LONG_LONG
  , long long
  , unsigned long long
#endif
  , float
  , double
  , long double
  , string_ref
  , std::tm
  , std::istream*
  > bind_types;

/// Types natively supported by backend::result::fetch method
typedef boost::mpl::vector<
    short*
  , unsigned short*
  , int*
  , unsigned int*
  , long*
  , unsigned long*
#ifndef BOOST_NO_LONG_LONG
  , long long*
  , unsigned long long*
#endif
  , float*
  , double*
  , long double*
  , std::string*
  , std::tm*
  , std::ostream*
  > fetch_types;

typedef boost::make_variant_over<bind_types>::type bind_types_variant;
typedef boost::make_variant_over<fetch_types>::type fetch_types_variant;

///
/// Extendable mechanism for binding arbitrary user type into statement.
///
template<typename T, typename Enable = void>
struct bind_conversion
{
    template<typename ColOrName>
    static void bind(statement& st, ColOrName col_or_name, const T& v)
    {
        BOOST_MPL_ASSERT_MSG(false, ADD_SPECIALIZATION_OF_BIND_CONVERSION_FOR_TYPE, (T));
    }
};

///
/// Extendable mechanism for fetching arbitrary user type from row.
///
template<typename T, typename Enable = void>
struct fetch_conversion
{
    template<typename ColOrName>
    static bool fetch(const row& res, ColOrName col_or_name, T& v)
    {
        BOOST_MPL_ASSERT_MSG(false, ADD_SPECIALIZATION_OF_FETCH_CONVERSION_FOR_TYPE, (T));
    }
};

}

typedef edba::backend::connection_iface* (*connect_function_type)(const edba::conn_info& cs, edba::session_monitor* sm);

#endif // EDBA_TYPES_HPP

ext: hpp


text:
// The contents of this file are in the public domain. See LICENSE_FOR_EXAMPLE_PROGRAMS.txt
/*

    This is an example illustrating the use of the linear_manifold_regularizer
    and empirical_kernel_map from the dlib C++ Library.

    This example program assumes you are familiar with some general elements of
    the library.  In particular, you should have at least read the svm_ex.cpp
    and matrix_ex.cpp examples.  You should also have read the empirical_kernel_map_ex.cpp
    example program as the present example builds upon it.

    This program shows an example of what is called semi-supervised learning.
    That is, a small amount of labeled data is augmented with a large amount
    of unlabeled data.  A learning algorithm is then run on all the data
    and the hope is that by including the unlabeled data we will end up with
    a better result.

    In this particular example we will generate 200,000 sample points of
    unlabeled data along with 2 samples of labeled data.  The sample points
    will be drawn randomly from two concentric circles.  One labeled data
    point will be drawn from each circle.  The goal is to learn to
    correctly separate the two circles using only the 2 labeled points
    and the unlabeled data.

    To do this we will first run an approximate form of k nearest neighbors
    to determine which of the unlabeled samples are closest together.  We will
    then make the manifold assumption, that is, we will assume that points close
    to each other should share the same classification label.

    Once we have determined which points are near neighbors we will use the
    empirical_kernel_map and linear_manifold_regularizer to transform all the
    data points into a new vector space where any linear rule will have similar
    output for points which we have decided are near neighbors.

    Finally, we will classify all the unlabeled data according to which of
    the two labeled points are nearest.  Normally this would not work but by
    using the manifold assumption we will be able to successfully classify
    all the unlabeled data.

    For further information on this subject you should begin with the following
    paper as it discusses a very similar application of manifold regularization.

        Beyond the Point Cloud: from Transductive to Semi-supervised Learning
        by Vikas Sindhwani, Partha Niyogi, and Mikhail Belkin

                    ******** SAMPLE PROGRAM OUTPUT ********

    Testing manifold regularization with an intrinsic_regularization_strength of 0.
    number of edges generated: 49998
    Running simple test...
    error: 0.37022
    error: 0.44036
    error: 0.376715
    error: 0.307545
    error: 0.463455
    error: 0.426065
    error: 0.416155
    error: 0.288295
    error: 0.400115
    error: 0.46347

    Testing manifold regularization with an intrinsic_regularization_strength of 10000.
    number of edges generated: 49998
    Running simple test...
    error: 0
    error: 0
    error: 0
    error: 0
    error: 0
    error: 0
    error: 0
    error: 0
    error: 0
    error: 0

*/

#include <dlib/manifold_regularization.h>
#include <dlib/svm.h>
#include <dlib/rand.h>
#include <dlib/statistics.h>
#include <iostream>
#include <vector>
#include <ctime>

using namespace std;
using namespace dlib;

// ----------------------------------------------------------------------------------------

// First let's make a typedef for the kind of samples we will be using.
typedef matrix<double, 0, 1> sample_type;

// We will be using the radial_basis_kernel in this example program.
typedef radial_basis_kernel<sample_type> kernel_type;

// ----------------------------------------------------------------------------------------

void generate_circle (
    std::vector<sample_type>& samples,
    double radius,
    const long num
);
/*!
    requires
        - num > 0
        - radius > 0
    ensures
        - generates num points centered at (0,0) with the given radius.  Adds these
          points into the given samples vector.
!*/

// ----------------------------------------------------------------------------------------

void test_manifold_regularization (
    const double intrinsic_regularization_strength
);
/*!
    ensures
        - Runs an example test using the linear_manifold_regularizer with the given
          intrinsic_regularization_strength.
!*/

// ----------------------------------------------------------------------------------------

int main()
{
    // Run the test without any manifold regularization.
    test_manifold_regularization(0);

    // Run the test with manifold regularization.  You can think of this number as
    // a measure of how much we trust the manifold assumption.  So if you are really
    // confident that you can select neighboring points which should have the same
    // classification then make this number big.
    test_manifold_regularization(10000.0);
}

// ----------------------------------------------------------------------------------------

void test_manifold_regularization (
    const double intrinsic_regularization_strength
)
{
    cout << "Testing manifold regularization with an intrinsic_regularization_strength of "
         << intrinsic_regularization_strength << ".\n";

    std::vector<sample_type> samples;

    // Declare an instance of the kernel we will be using.
    const kernel_type kern(0.1);

    const unsigned long num_points = 100000;

    // create a large dataset with two concentric circles.  There will be 100000 points on each circle
    // for a total of 200000 samples.
    generate_circle(samples, 2, num_points);  // circle of radius 2
    generate_circle(samples, 4, num_points);  // circle of radius 4

    // Create a set of sample_pairs that tells us which samples are "close" and should thus
    // be classified similarly.  These edges will be used to define the manifold regularizer.
    // To find these edges we use a simple function that samples point pairs randomly and
    // returns the top 5% with the shortest edges.
    std::vector<sample_pair> edges;
    find_percent_shortest_edges_randomly(samples, squared_euclidean_distance(), 0.05, 1000000, time(0), edges);

    cout << "number of edges generated: " << edges.size() << endl;

    empirical_kernel_map<kernel_type> ekm;

    // Since the circles are not linearly separable we will use an empirical kernel map to
    // map them into a space where they are separable.  We create an empirical_kernel_map
    // using a random subset of our data samples as basis samples.  Note, however, that even
    // though the circles are linearly separable in this new space given by the empirical_kernel_map
    // we still won't be able to correctly classify all the points given just the 2 labeled examples.
    // We will need to make use of the nearest neighbor information stored in edges.  To do that
    // we will use the linear_manifold_regularizer.
    ekm.load(kern, randomly_subsample(samples, 50));

    // Project all the samples into the span of our 50 basis samples
    for (unsigned long i = 0; i < samples.size(); ++i)
        samples[i] = ekm.project(samples[i]);

    // Now create the manifold regularizer.  The result is a transformation matrix that
    // embodies the manifold assumption discussed above.
    linear_manifold_regularizer<sample_type> lmr;
    // use_gaussian_weights is a function object that tells lmr how to weight each edge.  In this
    // case we let the weight decay as edges get longer.  So shorter edges are more important than
    // longer edges.
    lmr.build(samples, edges, use_gaussian_weights(0.1));
    const matrix<double> T = lmr.get_transformation_matrix(intrinsic_regularization_strength);

    // Apply the transformation generated by the linear_manifold_regularizer to
    // all our samples.
    for (unsigned long i = 0; i < samples.size(); ++i)
        samples[i] = T*samples[i];

    // For convenience, generate a projection_function and merge the transformation
    // matrix T into it.  That is, we will have: proj(x) == T*ekm.project(x).
    projection_function<kernel_type> proj = ekm.get_projection_function();
    proj.weights = T*proj.weights;

    cout << "Running simple test..." << endl;

    // Pick 2 different labeled points.  One on the inner circle and another on the outer.
    // For each of these test points we will see if using the single plane that separates
    // them is a good way to separate the concentric circles.  We also do this a bunch
    // of times with different randomly chosen points so we can see how robust the result is.
    for (int itr = 0; itr < 10; ++itr)
    {
        std::vector<sample_type> test_points;
        // generate a random point from the radius 2 circle
        generate_circle(test_points, 2, 1);
        // generate a random point from the radius 4 circle
        generate_circle(test_points, 4, 1);

        // project the two test points into kernel space.  Recall that this projection_function
        // has the manifold regularizer incorporated into it.
        const sample_type class1_point = proj(test_points[0]);
        const sample_type class2_point = proj(test_points[1]);

        double num_wrong = 0;

        // Now attempt to classify all the data samples according to which point
        // they are closest to.  The output of this program shows that without manifold
        // regularization this test will fail but with it it will perfectly classify
        // all the points.
        for (unsigned long i = 0; i < samples.size(); ++i)
        {
            double distance_to_class1 = length(samples[i] - class1_point);
            double distance_to_class2 = length(samples[i] - class2_point);

            bool predicted_as_class_1 = (distance_to_class1 < distance_to_class2);

            bool really_is_class_1 = (i < num_points);

            // now count how many times we make a mistake
            if (predicted_as_class_1 != really_is_class_1)
                ++num_wrong;
        }

        cout << "error: "<< num_wrong/samples.size() << endl;
    }

    cout << endl;
}

// ----------------------------------------------------------------------------------------

dlib::rand rnd;

void generate_circle (
    std::vector<sample_type>& samples,
    double radius,
    const long num
)
{
    sample_type m(2,1);

    for (long i = 0; i < num; ++i)
    {
        double sign = 1;
        if (rnd.get_random_double() < 0.5)
            sign = -1;
        m(0) = 2*radius*rnd.get_random_double()-radius;
        m(1) = sign*sqrt(radius*radius - m(0)*m(0));

        samples.push_back(m);
    }
}

// ----------------------------------------------------------------------------------------

ext: cpp

max_stars_repo_name: maxmert/nlp-mitie


text:
/* boost random/mersenne_twister.hpp header file
 *
 * Copyright Jens Maurer 2000-2001
 * Distributed under the Boost Software License, Version 1.0. (See
 * accompanying file LICENSE_1_0.txt or copy at
 * http://www.boost.org/LICENSE_1_0.txt)
 *
 * See http://www.boost.org for most recent version including documentation.
 *
 * $Id: mersenne_twister.hpp,v 1.20 2005/07/21 22:04:31 jmaurer Exp $
 *
 * Revision history
 *  2001-02-18  moved to individual header files
 */

#ifndef BOOST_RANDOM_MERSENNE_TWISTER_HPP
#define BOOST_RANDOM_MERSENNE_TWISTER_HPP

#include <iostream>
#include <algorithm>     // std::copy
#include <stdexcept>
#include <boost/config.hpp>
#include <boost/limits.hpp>
#include <boost/static_assert.hpp>
#include <boost/integer_traits.hpp>
#include <boost/cstdint.hpp>
#include <boost/random/linear_congruential.hpp>
#include <boost/detail/workaround.hpp>
#include <boost/random/detail/ptr_helper.hpp>

namespace boost {
namespace random {

// http://www.math.keio.ac.jp/matumoto/emt.html
template<class UIntType, int w, int n, int m, int r, UIntType a, int u,
  int s, UIntType b, int t, UIntType c, int l, UIntType val>
class mersenne_twister
{
public:
  typedef UIntType result_type;
  BOOST_STATIC_CONSTANT(int, word_size = w);
  BOOST_STATIC_CONSTANT(int, state_size = n);
  BOOST_STATIC_CONSTANT(int, shift_size = m);
  BOOST_STATIC_CONSTANT(int, mask_bits = r);
  BOOST_STATIC_CONSTANT(UIntType, parameter_a = a);
  BOOST_STATIC_CONSTANT(int, output_u = u);
  BOOST_STATIC_CONSTANT(int, output_s = s);
  BOOST_STATIC_CONSTANT(UIntType, output_b = b);
  BOOST_STATIC_CONSTANT(int, output_t = t);
  BOOST_STATIC_CONSTANT(UIntType, output_c = c);
  BOOST_STATIC_CONSTANT(int, output_l = l);

  BOOST_STATIC_CONSTANT(bool, has_fixed_range = false);

  mersenne_twister() { seed(); }

#if defined(__SUNPRO_CC) && (__SUNPRO_CC <= 0x520)
  // Work around overload resolution problem (Gennadiy E. Rozental)
  explicit mersenne_twister(const UIntType& value)
#else
  explicit mersenne_twister(UIntType value)
#endif
  { seed(value); }
  template<class It> mersenne_twister(It& first, It last) { seed(first,last); }

  template<class Generator>
  explicit mersenne_twister(Generator & gen) { seed(gen); }

  // compiler-generated copy ctor and assignment operator are fine

  void seed() { seed(UIntType(5489)); }

#if defined(__SUNPRO_CC) && (__SUNPRO_CC <= 0x520)
  // Work around overload resolution problem (Gennadiy E. Rozental)
  void seed(const UIntType& value)
#else
  void seed(UIntType value)
#endif
  {
    // New seeding algorithm from
    // http://www.math.sci.hiroshima-u.ac.jp/~m-mat/MT/MT2002/emt19937ar.html
    // In the previous versions, MSBs of the seed affected only MSBs of the
    // state x[].
    const UIntType mask = ~0u;
    x[0] = value & mask;
    for (i = 1; i < n; i++) {
      // See Knuth "The Art of Computer Programming" Vol. 2, 3rd ed., page 106
      x[i] = (1812433253UL * (x[i-1] ^ (x[i-1] >> (w-2))) + i) & mask;
    }
  }

  // For GCC, moving this function out-of-line prevents inlining, which may
  // reduce overall object code size.  However, MSVC does not grok
  // out-of-line definitions of member function templates.
  template<class Generator>
  void seed(Generator & gen)
  {
#ifndef BOOST_NO_LIMITS_COMPILE_TIME_CONSTANTS
    BOOST_STATIC_ASSERT(!std::numeric_limits<result_type>::is_signed);
#endif
    // I could have used std::generate_n, but it takes "gen" by value
    for(int j = 0; j < n; j++)
      x[j] = gen();
    i = n;
  }

  template<class It>
  void seed(It& first, It last)
  {
    int j;
    for(j = 0; j < n && first != last; ++j, ++first)
      x[j] = *first;
    i = n;
    if(first == last && j < n)
      throw std::invalid_argument("mersenne_twister::seed");
  }

  result_type min BOOST_PREVENT_MACRO_SUBSTITUTION () const { return 0; }
  result_type max BOOST_PREVENT_MACRO_SUBSTITUTION () const
  {
    // avoid "left shift count >= with of type" warning
    result_type res = 0;
    for(int i = 0; i < w; ++i)
      res |= (1u << i);
    return res;
  }

  result_type operator()();
  static bool validation(result_type v) { return val == v; }

#ifndef BOOST_NO_OPERATORS_IN_NAMESPACE

#ifndef BOOST_NO_MEMBER_TEMPLATE_FRIENDS
  template<class CharT, class Traits>
  friend std::basic_ostream<CharT,Traits>&
  operator<<(std::basic_ostream<CharT,Traits>& os, const mersenne_twister& mt)
  {
    for(int j = 0; j < mt.state_size; ++j)
      os << mt.compute(j) << " ";
    return os;
  }

  template<class CharT, class Traits>
  friend std::basic_istream<CharT,Traits>&
  operator>>(std::basic_istream<CharT,Traits>& is, mersenne_twister& mt)
  {
    for(int j = 0; j < mt.state_size; ++j)
      is >> mt.x[j] >> std::ws;
    // MSVC (up to 7.1) and Borland (up to 5.64) don't handle the template
    // value parameter "n" available from the class template scope, so use
    // the static constant with the same value
    mt.i = mt.state_size;
    return is;
  }
#endif

  friend bool operator==(const mersenne_twister& x, const mersenne_twister& y)
  {
    for(int j = 0; j < state_size; ++j)
      if(x.compute(j) != y.compute(j))
        return false;
    return true;
  }

  friend bool operator!=(const mersenne_twister& x, const mersenne_twister& y)
  { return !(x == y); }
#else
  // Use a member function; Streamable concept not supported.
  bool operator==(const mersenne_twister& rhs) const
  {
    for(int j = 0; j < state_size; ++j)
      if(compute(j) != rhs.compute(j))
        return false;
    return true;
  }

  bool operator!=(const mersenne_twister& rhs) const
  { return !(*this == rhs); }
#endif

private:
  // returns x(i-n+index), where index is in 0..n-1
  UIntType compute(unsigned int index) const
  {
    // equivalent to (i-n+index) % 2n, but doesn't produce negative numbers
    return x[ (i + n + index) % (2*n) ];
  }
  void twist(int block);

  // state representation: next output is o(x(i))
  //   x[0]  ... x[k] x[k+1] ... x[n-1]     x[n]     ... x[2*n-1]   represents
  //  x(i-k) ... x(i) x(i+1) ... x(i-k+n-1) x(i-k-n) ... x[i(i-k-1)]
  // The goal is to always have x(i-n) ... x(i-1) available for
  // operator== and save/restore.

  UIntType x[2*n];
  int i;
};

#ifndef BOOST_NO_INCLASS_MEMBER_INITIALIZATION
//  A definition is required even for integral static constants
template<class UIntType, int w, int n, int m, int r, UIntType a, int u,
  int s, UIntType b, int t, UIntType c, int l, UIntType val>
const bool mersenne_twister<UIntType,w,n,m,r,a,u,s,b,t,c,l,val>::has_fixed_range;
template<class UIntType, int w, int n, int m, int r, UIntType a, int u,
  int s, UIntType b, int t, UIntType c, int l, UIntType val>
const int mersenne_twister<UIntType,w,n,m,r,a,u,s,b,t,c,l,val>::state_size;
template<class UIntType, int w, int n, int m, int r, UIntType a, int u,
  int s, UIntType b, int t, UIntType c, int l, UIntType val>
const int mersenne_twister<UIntType,w,n,m,r,a,u,s,b,t,c,l,val>::shift_size;
template<class UIntType, int w, int n, int m, int r, UIntType a, int u,
  int s, UIntType b, int t, UIntType c, int l, UIntType val>
const int mersenne_twister<UIntType,w,n,m,r,a,u,s,b,t,c,l,val>::mask_bits;
template<class UIntType, int w, int n, int m, int r, UIntType a, int u,
  int s, UIntType b, int t, UIntType c, int l, UIntType val>
const UIntType mersenne_twister<UIntType,w,n,m,r,a,u,s,b,t,c,l,val>::parameter_a;
template<class UIntType, int w, int n, int m, int r, UIntType a, int u,
  int s, UIntType b, int t, UIntType c, int l, UIntType val>
const int mersenne_twister<UIntType,w,n,m,r,a,u,s,b,t,c,l,val>::output_u;
template<class UIntType, int w, int n, int m, int r, UIntType a, int u,
  int s, UIntType b, int t, UIntType c, int l, UIntType val>
const int mersenne_twister<UIntType,w,n,m,r,a,u,s,b,t,c,l,val>::output_s;
template<class UIntType, int w, int n, int m, int r, UIntType a, int u,
  int s, UIntType b, int t, UIntType c, int l, UIntType val>
const UIntType mersenne_twister<UIntType,w,n,m,r,a,u,s,b,t,c,l,val>::output_b;
template<class UIntType, int w, int n, int m, int r, UIntType a, int u,
  int s, UIntType b, int t, UIntType c, int l, UIntType val>
const int mersenne_twister<UIntType,w,n,m,r,a,u,s,b,t,c,l,val>::output_t;
template<class UIntType, int w, int n, int m, int r, UIntType a, int u,
  int s, UIntType b, int t, UIntType c, int l, UIntType val>
const UIntType mersenne_twister<UIntType,w,n,m,r,a,u,s,b,t,c,l,val>::output_c;
template<class UIntType, int w, int n, int m, int r, UIntType a, int u,
  int s, UIntType b, int t, UIntType c, int l, UIntType val>
const int mersenne_twister<UIntType,w,n,m,r,a,u,s,b,t,c,l,val>::output_l;
#endif

template<class UIntType, int w, int n, int m, int r, UIntType a, int u,
  int s, UIntType b, int t, UIntType c, int l, UIntType val>
void mersenne_twister<UIntType,w,n,m,r,a,u,s,b,t,c,l,val>::twist(int block)
{
  const UIntType upper_mask = (~0u) << r;
  const UIntType lower_mask = ~upper_mask;

  if(block == 0) {
    for(int j = n; j < 2*n; j++) {
      UIntType y = (x[j-n] & upper_mask) | (x[j-(n-1)] & lower_mask);
      x[j] = x[j-(n-m)] ^ (y >> 1) ^ (y&1 ? a : 0);
    }
  } else if (block == 1) {
    // split loop to avoid costly modulo operations
    {  // extra scope for MSVC brokenness w.r.t. for scope
      for(int j = 0; j < n-m; j++) {
        UIntType y = (x[j+n] & upper_mask) | (x[j+n+1] & lower_mask);
        x[j] = x[j+n+m] ^ (y >> 1) ^ (y&1 ? a : 0);
      }
    }

    for(int j = n-m; j < n-1; j++) {
      UIntType y = (x[j+n] & upper_mask) | (x[j+n+1] & lower_mask);
      x[j] = x[j-(n-m)] ^ (y >> 1) ^ (y&1 ? a : 0);
    }
    // last iteration
    UIntType y = (x[2*n-1] & upper_mask) | (x[0] & lower_mask);
    x[n-1] = x[m-1] ^ (y >> 1) ^ (y&1 ? a : 0);
    i = 0;
  }
}

template<class UIntType, int w, int n, int m, int r, UIntType a, int u,
  int s, UIntType b, int t, UIntType c, int l, UIntType val>
inline typename mersenne_twister<UIntType,w,n,m,r,a,u,s,b,t,c,l,val>::result_type
mersenne_twister<UIntType,w,n,m,r,a,u,s,b,t,c,l,val>::operator()()
{
  if(i == n)
    twist(0);
  else if(i >= 2*n)
    twist(1);
  // Step 4
  UIntType z = x[i];
  ++i;
  z ^= (z >> u);
  z ^= ((z << s) & b);
  z ^= ((z << t) & c);
  z ^= (z >> l);
  return z;
}

} // namespace random

typedef random::mersenne_twister<uint32_t,32,351,175,19,0xccab8ee7,11,
  7,0x31b6ab00,15,0xffe50000,17, 0xa37d3c92> mt11213b;

// validation by experiment from mt19937.c
typedef random::mersenne_twister<uint32_t,32,624,397,31,0x9908b0df,11,
  7,0x9d2c5680,15,0xefc60000,18, 3346425566U> mt19937;

} // namespace boost

BOOST_RANDOM_PTR_HELPER_SPEC(boost::mt19937)

#endif // BOOST_RANDOM_MERSENNE_TWISTER_HPP

max_stars_repo_name: spxuw/RFIM

max_issues_repo_name: spxuw/RFIM

max_forks_repo_name: spxuw/RFIM


text:
/*
 * Copyright (c) 2013 Juniper Networks, Inc. All rights reserved.
 */

#include <boost/assign/list_of.hpp>
#include "rapidjson/document.h"
#include "rapidjson/stringbuffer.h"
#include "rapidjson/writer.h"
#include "query.h"

using boost::assign::map_list_of;

// compare flow records based on UUID
bool PostProcessingQuery::flow_record_comparator(
                            const QEOpServerProxy::ResultRowT& lhs,
                            const QEOpServerProxy::ResultRowT& rhs) {
    std::map<std::string, std::string>::const_iterator lhs_it, rhs_it;
    lhs_it = lhs.first.find(g_viz_constants.UUID_KEY);
    QE_ASSERT(lhs_it != lhs.first.end());
    rhs_it = rhs.first.find(g_viz_constants.UUID_KEY);
    QE_ASSERT(rhs_it != rhs.first.end());
    if (lhs_it->second < rhs_it->second) return true;
    if (lhs_it->second > rhs_it->second) return false;

    return false;
}

bool PostProcessingQuery::sort_field_comparator(
        const QEOpServerProxy::ResultRowT& lhs,
        const QEOpServerProxy::ResultRowT& rhs) {
    std::map<std::string, std::string>::const_iterator lhs_it, rhs_it;
    for (std::vector<sort_field_t>::iterator sort_it = sort_fields.begin();
         sort_it != sort_fields.end(); sort_it++) {
        lhs_it = lhs.first.find((*sort_it).name);
        QE_ASSERT(lhs_it != lhs.first.end());
        rhs_it = rhs.first.find((*sort_it).name);
        QE_ASSERT(rhs_it != rhs.first.end());
        if ((*sort_it).type == std::string("int") ||
            (*sort_it).type == std::string("long") ||
            (*sort_it).type == std::string("ipv4")) {
            uint64_t lhs_val = 0, rhs_val = 0;
            stringToInteger(lhs_it->second, lhs_val);
            stringToInteger(rhs_it->second, rhs_val);
            if (lhs_val < rhs_val) return true;
            if (lhs_val > rhs_val) return false;
        } else {
            if (lhs_it->second < rhs_it->second) return true;
            if (lhs_it->second > rhs_it->second) return false;
        }
    }

    return false;
}

bool PostProcessingQuery::flowseries_merge_processing(
        const QEOpServerProxy::BufferT *raw_result,
        QEOpServerProxy::BufferT* merged_result,
        fcid_rrow_map_t *fcid_rrow_map) {
    AnalyticsQuery *mquery = (AnalyticsQuery *)main_query;

    switch(mquery->selectquery_->flowseries_query_type()) {
    case SelectQuery::FS_SELECT_STATS:
        fs_stats_merge_processing(raw_result, merged_result);
        break;
    case SelectQuery::FS_SELECT_FLOW_TUPLE_STATS:
        fs_tuple_stats_merge_processing(raw_result, merged_result, fcid_rrow_map);
        break;
    default:
        return false;
    }

    return true;
}

void PostProcessingQuery::fs_merge_stats(
            const QEOpServerProxy::ResultRowT& input,
            QEOpServerProxy::ResultRowT& output) {
    QEOpServerProxy::OutRowT::iterator pit =
            output.first.find(SELECT_SUM_PACKETS);
    if (pit != output.first.end()) {
        uint64_t sum_pkts = 0;
        stringToInteger(pit->second, sum_pkts);
        uint64_t pkts = 0;
        stringToInteger(input.first.find(SELECT_SUM_PACKETS)->second, pkts);
        sum_pkts += pkts;
        pit->second = integerToString(sum_pkts);
    }
    QEOpServerProxy::OutRowT::iterator bit =
            output.first.find(SELECT_SUM_BYTES);
    if (bit != output.first.end()) {
        uint64_t sum_bytes = 0;
        stringToInteger(bit->second, sum_bytes);
        uint64_t bytes = 0;
        stringToInteger(input.first.find(SELECT_SUM_BYTES)->second, bytes);
        sum_bytes += bytes;
        bit->second = integerToString(sum_bytes);
    }
    if (input.second.get()) {
        fsMetaData *imetadata = static_cast<fsMetaData*>(input.second.get());
        fsMetaData *ometadata = static_cast<fsMetaData*>(output.second.get());
        ometadata->uuids.insert(imetadata->uuids.begin(),
                                imetadata->uuids.end());
    }
}

void PostProcessingQuery::fs_stats_merge_processing(
        const QEOpServerProxy::BufferT *raw_result,
        QEOpServerProxy::BufferT *merged_result) {
    if (!raw_result->size()) {
        return;
    }
    if (!merged_result->size()) {
        merged_result->reserve(raw_result->size());
        copy(raw_result->begin(), raw_result->end(),
             std::back_inserter(*merged_result));
        return;
    }
    assert(raw_result->size() == 1);
    assert(merged_result->size() == 1);
    QEOpServerProxy::ResultRowT& mresult_row = merged_result->at(0);
    const QEOpServerProxy::ResultRowT& rresult_row = raw_result->at(0);
    QE_TRACE(DEBUG, "fs_stats_merge_processing: merge_stats.");
    fs_merge_stats(rresult_row, mresult_row);
}

void PostProcessingQuery::fs_tuple_stats_merge_processing(
        const QEOpServerProxy::BufferT *raw_result,
        QEOpServerProxy::BufferT *merged_result,
        fcid_rrow_map_t *fcid_rrow_map) {

    for (size_t r = 0; r < merged_result->size(); ++r) {
        const QEOpServerProxy::ResultRowT& rresult_row = merged_result->at(r);
        const QEOpServerProxy::OutRowT& ro_row = rresult_row.first;
        QEOpServerProxy::OutRowT::const_iterator rfc_it =
            rresult_row.first.find(SELECT_FLOW_CLASS_ID);
        assert(rfc_it != ro_row.end());
        uint64_t rfc_id = 0;
        stringToInteger(rfc_it->second, rfc_id);
        if (fcid_rrow_map->find(rfc_id) == fcid_rrow_map->end()) {
            // flow_class_id rfc_id not found, add the result row to the map
            fcid_rrow_map->insert(std::pair<uint64_t,
                QEOpServerProxy::ResultRowT>(rfc_id, rresult_row));
        } else {
            // found an existing flow class id in the map
            QEOpServerProxy::ResultRowT& mresult_row =
                (fcid_rrow_map->find(rfc_id))->second;
            fs_merge_stats(rresult_row, mresult_row);
        }
    }
    for (size_t r = 0; r < raw_result->size(); ++r) {
        const QEOpServerProxy::ResultRowT& rresult_row = raw_result->at(r);
        const QEOpServerProxy::OutRowT& ro_row = rresult_row.first;
        QEOpServerProxy::OutRowT::const_iterator rfc_it =
            rresult_row.first.find(SELECT_FLOW_CLASS_ID);
        assert(rfc_it != ro_row.end());
        uint64_t rfc_id = 0;
        stringToInteger(rfc_it->second, rfc_id);
        if (fcid_rrow_map->find(rfc_id) == fcid_rrow_map->end()) {
            // flow_class_id rfc_id not found, add the result row to the map
            fcid_rrow_map->insert(std::pair<uint64_t,
                QEOpServerProxy::ResultRowT>(rfc_id, rresult_row));
        } else {
            // found an existing flow class id in the map
            QEOpServerProxy::ResultRowT& mresult_row =
                (fcid_rrow_map->find(rfc_id))->second;
            fs_merge_stats(rresult_row, mresult_row);
        }
    }
}

void PostProcessingQuery::fs_update_flow_count(
            QEOpServerProxy::ResultRowT& rrow) {
    QEOpServerProxy::OutRowT::iterator fit =
        rrow.first.find(SELECT_FLOW_COUNT);
    assert(fit != rrow.first.end());
    assert(rrow.second.get());
    fsMetaData *mdata =
        static_cast<fsMetaData*>(rrow.second.get());
    fit->second = integerToString(mdata->uuids.size());
}

bool PostProcessingQuery::merge_processing(
        const QEOpServerProxy::BufferT& input,
        QEOpServerProxy::BufferT& output)
{
    AnalyticsQuery *mquery = (AnalyticsQuery *)main_query;

    if (status_details != 0)
    {
        QE_TRACE(DEBUG,
             "No need to process query, as there were errors previously");
        return false;
    }

    if (mquery->table() == g_viz_constants.FLOW_SERIES_TABLE) {
        fcid_rrow_map_t fcid_rrow_map;
        if (flowseries_merge_processing(&input, &output, &fcid_rrow_map)) {
            if (fcid_rrow_map.size() != 0) {
                output.clear();
                for(fcid_rrow_map_t::iterator it = fcid_rrow_map.begin(); it != fcid_rrow_map.end(); ++it ) {
                    output.push_back(it->second);
                }
            }
            status_details = 0;
            return true;
        }
    }

    // Check if the result has to be sorted
    if (sorted) {
        QEOpServerProxy::BufferT *merged_result = &output;
        const QEOpServerProxy::BufferT *raw_result1 = &(input);

        if (result_.get() == NULL) {
            size_t merged_result_size = merged_result->size();
            merged_result->reserve(merged_result_size + raw_result1->size());
            copy(raw_result1->begin(), raw_result1->end(),
                 std::back_inserter(*merged_result));
            if (merged_result_size) {
                if (sorting_type == ASCENDING) {
                    std::inplace_merge(merged_result->begin(),
                        merged_result->begin() + merged_result_size,
                        merged_result->end(),
                        boost::bind(&PostProcessingQuery::sort_field_comparator,
                                    this, _1, _2));
                } else {
                    std::inplace_merge(merged_result->rbegin(),
                        merged_result->rbegin() + raw_result1->size(),
                        merged_result->rend(),
                        boost::bind(&PostProcessingQuery::sort_field_comparator,
                                    this, _1, _2));
                }
            }

            goto sort_done;
        }

        QEOpServerProxy::BufferT *raw_result2 = result_.get();
        size_t size1 = raw_result1->size();
        size_t size2 = raw_result2->size();
        QE_TRACE(DEBUG, "Merging results from vectors of size:" <<
                size1 << " and " << size2);
        merged_result->reserve(raw_result1->size() + raw_result2->size());
        if (sorting_type == ASCENDING) {
            std::merge(raw_result1->begin(), raw_result1->end(),
                raw_result2->begin(), raw_result2->end(),
                std::back_inserter(*merged_result),
                boost::bind(&PostProcessingQuery::sort_field_comparator,
                            this, _1, _2));
        } else {
            std::merge(raw_result1->rbegin(), raw_result1->rend(),
                raw_result2->rbegin(), raw_result2->rend(),
                std::back_inserter(*merged_result),
                boost::bind(&PostProcessingQuery::sort_field_comparator,
                            this, _1, _2));
        }
    }

sort_done:
    if (!sorted && (mquery->table() == g_viz_constants.FLOW_TABLE))
    {
        QE_TRACE(DEBUG, "Merge_Processing: Adding inputs to output");
        QEOpServerProxy::BufferT *merged_result = &output;
        const QEOpServerProxy::BufferT *raw_result1 = &(input);

        if (result_.get() == NULL)
        {
            merged_result->reserve(raw_result1->size());
            copy(raw_result1->begin(), raw_result1->end(),
                std::back_inserter(*merged_result));
        } else {

            QEOpServerProxy::BufferT *raw_result2 = result_.get();
            size_t size1 = raw_result1->size();
            size_t size2 = raw_result2->size();
            QE_TRACE(DEBUG, "Merging results from vectors of size:" <<
                    size1 << " and " << size2);
            merged_result->reserve(raw_result1->size() + raw_result2->size());
            copy(raw_result1->begin(), raw_result1->end(),
                std::back_inserter(*merged_result));
            copy(raw_result2->begin(), raw_result2->end(),
                std::back_inserter(*merged_result));
        }
        QE_TRACE(DEBUG, "Merge_Processing: Done adding inputs to output");
    }

    // Have the result ready and processing is done
    status_details = 0;
    return true;
}

bool PostProcessingQuery::final_merge_processing(
const std::vector<boost::shared_ptr<QEOpServerProxy::BufferT> >& inputs,
                        QEOpServerProxy::BufferT& output)
{
    AnalyticsQuery *mquery = (AnalyticsQuery *)main_query;

    if (status_details != 0)
    {
        QE_TRACE(DEBUG,
             "No need to process query, as there were errors previously");
        return false;
    }

    if (mquery->table() == g_viz_constants.FLOW_SERIES_TABLE) {
        fcid_rrow_map_t fcid_rrow_map;
        bool status = false;
        for (size_t i = 0; i < inputs.size(); i++) {
            status = flowseries_merge_processing(inputs[i].get(),
                                    &output, &fcid_rrow_map);
        }
        if (status) {
            bool is_select_flow_count =
                mquery->selectquery_->is_present_in_select_column_fields(
                                                        SELECT_FLOW_COUNT);
            if (fcid_rrow_map.size() != 0) {
                for(fcid_rrow_map_t::iterator it = fcid_rrow_map.begin();
                    it != fcid_rrow_map.end(); ++it ) {
                    if (is_select_flow_count) {
                        fs_update_flow_count(it->second);
                    }
                    output.push_back(it->second);
                }
            }

            if (mquery->selectquery_->flowseries_query_type() ==
                SelectQuery::FS_SELECT_STATS && is_select_flow_count) {
                if (output.size()) {
                    fs_update_flow_count(output[0]);
                }
            }

            if (sorted) {
                QEOpServerProxy::BufferT *merged_result = &output;
                if (sorting_type == ASCENDING) {
                    std::sort(merged_result->begin(), merged_result->end(),
                              boost::bind(
                                &PostProcessingQuery::sort_field_comparator,
                                          this, _1, _2));
                } else {
                    std::sort(merged_result->rbegin(), merged_result->rend(),
                              boost::bind(
                                &PostProcessingQuery::sort_field_comparator,
                                          this, _1, _2));
                }
            }
            goto limit;
        }
    }

    if (mquery->table() == g_viz_constants.FLOW_TABLE)
    {
        QE_TRACE(DEBUG, "Final_Merge_Processing: Uniquify flow records");
        // uniquify the records
        std::set<QEOpServerProxy::ResultRowT,
                 bool(*)(const QEOpServerProxy::ResultRowT&,
                         const QEOpServerProxy::ResultRowT&)>
                 result_row_set(&PostProcessingQuery::flow_record_comparator);

        for (size_t i = 0; i < inputs.size(); i++)
        {
            QEOpServerProxy::BufferT *raw_result = inputs[i].get();
            result_row_set.insert(raw_result->begin(), raw_result->end());
        }

        QEOpServerProxy::BufferT *merged_result = &output;
        merged_result->reserve(result_row_set.size());
        copy(result_row_set.begin(), result_row_set.end(),
            std::back_inserter(*merged_result));

        QE_TRACE(DEBUG, "Final_Merge_Processing: Done uniquify flow records");
        // Check if the result has to be sorted
        if (sorted) {
            if (sorting_type == ASCENDING) {
                std::sort(merged_result->begin(), merged_result->end(),
                          boost::bind(&PostProcessingQuery::sort_field_comparator,
                                      this, _1, _2));
            } else {
                std::sort(merged_result->rbegin(), merged_result->rend(),
                          boost::bind(&PostProcessingQuery::sort_field_comparator,
                                      this, _1, _2));
            }
        }
    } else {  // For non-flow-record queries
        // Check if the result has to be sorted
        if (sorted) {
            QEOpServerProxy::BufferT *merged_result = &output;
            size_t final_vector_size = 0;
            for (size_t i = 0; i < inputs.size(); i++)
                final_vector_size += inputs[i]->size();

            QE_TRACE(DEBUG, "Merging results between " << inputs.size()
                    << " vectors with final vector size:" << final_vector_size);

            merged_result->reserve(final_vector_size);

            for (size_t i = 0; i < inputs.size(); i++)
            {
                QEOpServerProxy::BufferT *raw_result = inputs[i].get();
                if (sorting_type == ASCENDING) {
                    QEOpServerProxy::BufferT::iterator
                        current_result_end = merged_result->end();
                    copy(raw_result->begin(), raw_result->end(),
                        std::back_inserter(*merged_result));
                    std::inplace_merge(merged_result->begin(),
                    current_result_end, merged_result->end(),
    boost::bind(&PostProcessingQuery::sort_field_comparator, this, _1, _2));
                } else {
                    QEOpServerProxy::BufferT::reverse_iterator
                        current_result_end = merged_result->rbegin();
                    copy(raw_result->begin(), raw_result->end(),
                        std::back_inserter(*merged_result));
                    std::inplace_merge(merged_result->rbegin(),
                    current_result_end, merged_result->rend(),
    boost::bind(&PostProcessingQuery::sort_field_comparator, this, _1, _2));
                }
            }
        }
    }

limit:
    if (limit) {
        QEOpServerProxy::BufferT *merged_result = &output;
        QE_TRACE(DEBUG, "Apply Limit [" << limit << "]");
        if (merged_result->size() > (size_t)limit) {
            merged_result->resize(limit);
        }
    }

    // Have the result ready and processing is done
    status_details = 0;
    return true;
}

query_status_t PostProcessingQuery::process_query() {
    if (status_details != 0)
    {
        QE_TRACE(DEBUG,
             "No need to process query, as there were errors previously");
        return QUERY_FAILURE;
    }

    AnalyticsQuery *mquery = (AnalyticsQuery *)main_query;
    result_ = mquery->selectquery_->result_;
    mresult_ = mquery->selectquery_->mresult_;
    QEOpServerProxy::BufferT *raw_result = result_.get();

    /* filter are ANDs over OR
     * [ [ e1 AND e2 ] OR [ e3 ] ]
     */
    /* below is filter processing for stats table queries
     */
    if (filter_list.size()) {
        size_t num_filtered=0;
        MapBufT::iterator kt = mresult_->end();
        for (MapBufT::iterator it = mresult_->begin();
                it!= mresult_->end(); it++) {

            if (kt!=mresult_->end()) {
                mresult_->erase(kt);
                kt = mresult_->end();
            }
            std::map<std::string, QEOpServerProxy::SubVal>& attrs = it->second.first;
            bool delete_row = true;
            std::string unknown_attr;
            for (size_t j = 0; j < filter_list.size(); j++) {
                std::vector<filter_match_t>& filter_and = filter_list[j];
                bool and_check = true;

                for (size_t k = 0; k < filter_and.size(); k++) {
                    std::map<std::string, QEOpServerProxy::SubVal>::const_iterator iter =
                        attrs.find(filter_and[k].name);
                    if (iter == attrs.end()) {
                        unknown_attr = filter_and[k].name;
                        break;
                    } else {
                        unknown_attr.clear();
                    }
                    std::ostringstream vstream;
                    vstream << iter->second;

                    switch(filter_and[k].op) {
                        case EQUAL:
                            if (filter_and[k].value != vstream.str())
                              {
                                and_check = false;
                              }
                            break;
                        case  NOT_EQUAL:
                            if (filter_and[k].value == vstream.str())
                              {
                                and_check = false;
                              }
                            break;
                        default:
                            // upsupported filter operation
                            QE_TRACE(DEBUG, "Unsupported filter operation " << filter_and[k].op);
                            break;
                    }
                    if (and_check == false)
                        break;
                }
                if (and_check == true) {
                    delete_row = false;
                    break;
                }
            }
            if (unknown_attr.size()) {
                QE_TRACE(DEBUG, "Unknown filter attr in row " << unknown_attr);
            }
            if (delete_row) {
                num_filtered++;
                kt = it;
            }
        }
        if (kt!=mresult_->end()) {
            mresult_->erase(kt);
            kt = mresult_->end();
        }
        QE_TRACE(DEBUG, "# of entries filtered is " << num_filtered);

    }

    /* below is filter processing for non stats table queries
     */
    if (filter_list.size() != 0) {
        QEOpServerProxy::BufferT filtered_table;
        // do filter operation
        QE_TRACE(DEBUG, "Doing filter operation");
        for (size_t i = 0; i < raw_result->size(); i++) {
            QEOpServerProxy::ResultRowT row = (*raw_result)[i];
            bool delete_row = true;

            for (size_t j = 0; j < filter_list.size(); j++) {
                std::vector<filter_match_t>& filter_and = filter_list[j];
                bool and_check = true;

                for (size_t k = 0; k < filter_and.size(); k++) {
                    std::map<std::string, std::string>::iterator iter;
                    iter = row.first.find(filter_and[k].name);
                    if (iter == row.first.end())
                      {
                        if (!(filter_and[k].ignore_col_absence)) {
                            and_check = false;
                            break;
                        }
                        continue;
                      }

                    switch(filter_and[k].op)
                      {
                        case EQUAL:
                            if (filter_and[k].value != iter->second)
                              {
                                and_check = false;
                              }
                            break;

                        case NOT_EQUAL:
                            if (filter_and[k].value == iter->second)
                              {
                                and_check = false;
                              }
                            break;

                        case LEQ:
                              {
                                int filter_value =
                                    atoi(filter_and[k].value.c_str());
                                int column_value= atoi(iter->second.c_str());
                                if (column_value > filter_value)
                                  {
                                    and_check = false;
                                  }
                                break;
                              }

                        case GEQ:
                              {
                                int filter_value =
                                    atoi(filter_and[k].value.c_str());
                                int column_value= atoi(iter->second.c_str());
                                if (column_value < filter_value)
                                  {
                                    and_check = false;
                                  }
                                break;
                              }

                        case REGEX_MATCH:
                              {
                                if (!boost::regex_match(iter->second,
                                            filter_and[k].match_e))
                                  {
                                    and_check = false;
                                  }
                                break;
                              }

                        default:
                            // upsupported filter operation
                            QE_LOG(ERROR, "Unsupported filter operation: " <<
                                    filter_and[k].op);
                            return QUERY_FAILURE;
                      }
                    if (and_check == false)
                        break;
                }

                if (and_check == true) {
                    QE_TRACE(DEBUG, "filter out entry #:" << i);
                    delete_row = false;
                    break;
                }
            }
            if (!delete_row) {
                filtered_table.push_back(row);
            }
        }
        *raw_result = filtered_table;
    }

    // Check if the result has to be sorted
    if (sorted) {
        if (sorting_type == ASCENDING) {
            std::sort(raw_result->begin(), raw_result->end(),
                      boost::bind(&PostProcessingQuery::sort_field_comparator,
                                  this, _1, _2));
        } else {
            std::sort(raw_result->rbegin(), raw_result->rend(),
                      boost::bind(&PostProcessingQuery::sort_field_comparator,
                                  this, _1, _2));
        }
    }

    // If the flow series query is parallelized, we should apply the limit
    // only after the result from all the tasks are merged
    // (@ final_merge_processing).
    if ((mquery->table() != g_viz_constants.FLOW_SERIES_TABLE ||
        (mquery->table() == g_viz_constants.FLOW_SERIES_TABLE &&
        !mquery->is_query_parallelized())) && limit) {
        QE_TRACE(DEBUG, "Apply Limit [" << limit << "]");
        if (raw_result->size() > (size_t)limit) {
            raw_result->resize(limit);
        }
    }

    if (IS_TRACE_ENABLED(POSTPROCESS_RESULT_TRACE))
    {
        std::vector<QEOpServerProxy::ResultRowT>::iterator res_it;
        QE_TRACE(DEBUG, "== Post Processing Result ==");
        for (res_it = raw_result->begin(); res_it != raw_result->end();
             ++res_it) {
            std::vector<final_result_col> row_entry;
            std::map<std::string, std::string>::iterator map_it;
            for (map_it = (*res_it).first.begin();
                 map_it != (*res_it).first.end(); ++map_it) {
                final_result_col col;
                col.set_col(map_it->first); col.set_value(map_it->second);
                row_entry.push_back(col);
                //QE_TRACE(DEBUG, map_it->first << " : " << map_it->second);
            }
            FINAL_RESULT_ROW_TRACE(QeTraceBuf, mquery->query_id, row_entry);
        }
    }

#if 0
    //if ((limit) && (!sorted))
    for (int i = 0 ; i < 200000; i++)
    raw_result->push_back(map_list_of(
            "destvn","abc-cor\"poration:front-end-network:001")(
            "sourceip","168430090")("destip","3232238090")(
            "sourcevn","abc-corporation:front-end-network:002")(
            "protocol","80")("dport","62000")("sport","1000")(
            "sum(packets)","4294967196")
        );
#endif

    // Have the result ready and processing is done
    status_details = 0;
    parent_query->subquery_processed(this);
    return QUERY_SUCCESS;
}

ext: cc


text:
#include <pars/service.hpp>

#define STB_IMAGE_WRITE_IMPLEMENTATION

#include <cstdint>
#include <string>

#include <boost/asio.hpp>
#include <google/protobuf/util/json_util.h>
#include <cppzmq/zmq.hpp>
#include <stb/stb_image_write.h>

#include <pars/pipeline.hpp>

#include <image.pb.h>
#include <settings.pb.h>

namespace pars
{
void run      (const std::string& address)
{
  pipeline pipeline;

  zmq::context_t context(1);
  zmq::socket_t  socket (context, ZMQ_PAIR);
  if (pipeline.communicator()->rank() == 0)
  {
    boost::asio::io_service        io_service;
    boost::asio::ip::tcp::resolver resolver(io_service);
    std::string                    host_name = boost::asio::ip::host_name();
    std::cout << "Host name is " << host_name << ".\n";
    std::cout << "IP addresses are: \n";
    std::for_each(resolver.resolve({host_name, ""}), {}, [ ] (const auto& re)
    {
      std::cout << re.endpoint().address() << '\n';
    });

    socket.bind(address);
  }

  while (true)
  {
    std::string request_string;

    if (pipeline.communicator()->rank() == 0)
    {
      zmq::message_t request;
      socket.recv  (&request);
      request_string = std::string(static_cast<char*>(request.data()), request.size());
    }

    boost::mpi::broadcast(*pipeline.communicator(), request_string, 0);

    settings settings;
    settings.ParseFromString(request_string);

    auto result = pipeline.execute(settings);

    if (pipeline.communicator()->rank() == 0)
    {
      std::string buffer;
      result.first.SerializeToString(&buffer);

      zmq::message_t response(buffer.size());
      memcpy(response.data(), buffer.data(), buffer.size());
      socket.send(response);
    }
  }
}
void benchmark(const std::size_t thread_count, const std::string& settings_filepath)
{
  pipeline pipeline(thread_count);

  settings      settings;
  std::ifstream settings_file  (settings_filepath);
  const auto    settings_string = std::string(std::istreambuf_iterator<char>(settings_file), std::istreambuf_iterator<char>());
  JsonStringToMessage(settings_string, &settings, google::protobuf::util::JsonParseOptions());

  auto ftle_mode = settings.mode().find("ftle") != std::string::npos;
  if (ftle_mode)
  {
    auto result = pipeline.execute_ftle(settings);

    result.gather();
    result.to_csv(settings_filepath + ".csv");

    if (pipeline.communicator()->rank() == 0)
      std::cout << "Saved benchmark.\n";
  }
  else
  {
    auto result = pipeline.execute     (settings);

    result.second.gather();
    result.second.to_csv(settings_filepath + ".csv");

    if (pipeline.communicator()->rank() == 0)
    {
      std::cout << "Saved benchmark.\n";
      const auto filepath = settings_filepath + ".png";
      stbi_write_png(filepath.c_str(), result.first.size(0), result.first.size(1), 4, result.first.data().c_str(), result.first.size(0) * sizeof(std::uint32_t));
      std::cout << "Saved image.\n";
    }
  }
}
}

max_stars_repo_name: acdemiralp/pars

max_issues_repo_name: acdemiralp/pars

max_forks_repo_name: acdemiralp/pars


text:
// Copyright © 2016 Venture Media Labs.
//
// This file is part of mxml. The full mxml copyright notice, including
// terms governing use, modification, and redistribution, is contained in the
// file LICENSE at the root of the source code distribution tree.

#include <lxml/lxml.h>
#include <mxml/parsing/ScoreHandler.h>
#include <sstream>
#include <boost/test/unit_test.hpp>

using namespace mxml;
using namespace mxml::parsing;

BOOST_AUTO_TEST_CASE(parseMusicXML) {
    const char musicXML[] =
        "<?xml version=\"1.0\" encoding=\"UTF-8\"?>\n"
        "<!DOCTYPE score-partwise PUBLIC \"-//Recordare//DTD MusicXML 2.0 Partwise//EN\" \"http://www.musicxml.org/dtds/partwise.dtd\">\n"
        "<score-partwise>\n"
        "  <part-list>\n"
        "    <score-part id=\"P1\">\n"
        "      <part-name>Piano</part-name>\n"
        "    </score-part>\n"
        "  </part-list>\n"
        "  <part id=\"P1\">\n"
        "    <measure number=\"22\">\n"
        "      <attributes>\n"
        "        <divisions>24</divisions>\n"
        "        <key>\n"
        "          <fifths>0</fifths>\n"
        "          <mode>major</mode>\n"
        "        </key>\n"
        "        <time symbol=\"common\">\n"
        "          <beats>4</beats>\n"
        "          <beat-type>4</beat-type>\n"
        "        </time>\n"
        "        <staves>2</staves>\n"
        "        <clef number=\"1\">\n"
        "          <sign>G</sign>\n"
        "          <line>2</line>\n"
        "        </clef>\n"
        "        <clef number=\"2\">\n"
        "          <sign>F</sign>\n"
        "          <line>4</line>\n"
        "        </clef>\n"
        "      </attributes>\n"
        "      <note>\n"
        "        <pitch>\n"
        "          <step>C</step>\n"
        "          <octave>6</octave>\n"
        "        </pitch>\n"
        "        <duration>36</duration>\n"
        "        <voice>1</voice>\n"
        "        <type>quarter</type>\n"
        "        <dot/>\n"
        "        <stem>down</stem>\n"
        "        <staff>1</staff>\n"
        "      </note>\n"
        "    </measure>\n"
        "  </part>\n"
        "</score-partwise>";

    ScoreHandler handler;
    std::stringstream ss(musicXML);
    lxml::parse(ss, "test", handler);

    const dom::Score& score = *handler.result();
    BOOST_CHECK_EQUAL(score.parts().size(), 1);

    const dom::Part& part = *score.parts()[0];
    BOOST_CHECK_EQUAL(part.parent(), &score);
    BOOST_CHECK_EQUAL(part.measures().size(), 1);

    const dom::Measure& measure = *part.measures()[0];
    BOOST_CHECK_EQUAL(measure.parent(), &part);
    BOOST_CHECK_EQUAL(measure.index(), 0);
    BOOST_CHECK_EQUAL(measure.number(), "22");
    BOOST_CHECK_EQUAL(measure.nodes().size(), 2);

    const dom::Attributes& attributes = dynamic_cast<dom::Attributes&>(*measure.nodes()[0]);
    BOOST_CHECK_EQUAL(attributes.divisions(), 24);
    BOOST_CHECK_EQUAL(attributes.staves(), 2);

    const dom::Chord& chord = dynamic_cast<dom::Chord&>(*measure.nodes()[1]);
    BOOST_CHECK_EQUAL(chord.notes().size(), 1);

    const dom::Note& note = *chord.notes()[0];
    BOOST_CHECK(note.pitch);
    BOOST_CHECK(note.pitch->step() == dom::Pitch::Step::C);
    BOOST_CHECK_EQUAL(note.pitch->octave(), 6);
    BOOST_CHECK_EQUAL(note.duration(), 36);
    BOOST_CHECK(note.type() == dom::Note::Type::Quarter);
    BOOST_CHECK(note.stem() == dom::Stem::Down);
    BOOST_CHECK_EQUAL(note.staff(), 1);
    BOOST_CHECK(note.dot);
}