ZmqServer.cpp

Go to the documentation of this file.

 
#include "rogue/Directives.h"
 
#include "rogue/interfaces/ZmqServer.h"
 
#include <inttypes.h>
#include <zmq.h>
 
#include <memory>
#include <string>
 
#include "rogue/GeneralError.h"
#include "rogue/GilRelease.h"
#include "rogue/ScopedGil.h"
 
#ifndef NO_PYTHON
    #include <boost/python.hpp>
namespace bp = boost::python;
#endif
 
rogue::interfaces::ZmqServerPtr rogue::interfaces::ZmqServer::create(const std::string& addr, uint16_t port) {
    rogue::interfaces::ZmqServerPtr ret = std::make_shared<rogue::interfaces::ZmqServer>(addr, port);
    return (ret);
}
 
void rogue::interfaces::ZmqServer::setup_python() {
#ifndef NO_PYTHON
 
    bp::class_<rogue::interfaces::ZmqServerWrap, rogue::interfaces::ZmqServerWrapPtr, boost::noncopyable>(
        "ZmqServer",
        bp::init<std::string, uint16_t>())
        .def("_doRequest", &rogue::interfaces::ZmqServer::doRequest, &rogue::interfaces::ZmqServerWrap::defDoRequest)
        .def("_doString", &rogue::interfaces::ZmqServer::doString, &rogue::interfaces::ZmqServerWrap::defDoString)
        .def("_publish", &rogue::interfaces::ZmqServer::publish)
        .def("port", &rogue::interfaces::ZmqServer::port)
        .def("_stop", &rogue::interfaces::ZmqServer::stop)
        .def("_start", &rogue::interfaces::ZmqServer::start);
#endif
}
 
rogue::interfaces::ZmqServer::ZmqServer(const std::string& addr, uint16_t port) {
    log_ = rogue::Logging::create("ZmqServer");
 
    this->addr_     = addr;
    this->zmqCtx_   = zmq_ctx_new();
    this->threadEn_ = false;
    this->basePort_ = port;
}
 
rogue::interfaces::ZmqServer::~ZmqServer() {
    stop();
    // stop() only frees the context when start() ran; covers construct-without-start.
    if (zmqCtx_ != nullptr) {
        zmq_ctx_destroy(zmqCtx_);
        zmqCtx_ = nullptr;
    }
}
 
void rogue::interfaces::ZmqServer::start() {
    std::string dummy;
    bool res = false;
    uint16_t port;
 
    if (this->threadEn_) return;
    port = this->basePort_;
 
    // Auto port
    if (port == 0) {
        for (this->basePort_ = 9099; this->basePort_ < (9099 + 100); this->basePort_ += 4) {
            res = this->tryConnect();
            if (res) break;
        }
    } else {
        res = this->tryConnect();
    }
 
    if (!res) {
        if (port == 0)
            throw(rogue::GeneralError::create("ZmqServer::ZmqServer",
                                              "Failed to auto bind server on interface %s.",
                                              this->addr_.c_str()));
        else
            throw(rogue::GeneralError::create("ZmqServer::ZmqServer",
                                              "Failed to bind server to port %" PRIu16
                                              " on interface %s. Another process may be using this port.",
                                              port + 1,
                                              this->addr_.c_str()));
    }
 
    log_->debug("Started Rogue server on %s at ports %" PRIu16 ":%" PRIu16 ":%" PRIu16,
                this->addr_.c_str(),
                this->basePort_,
                this->basePort_ + 1,
                this->basePort_ + 2);
 
    this->threadEn_ = true;
    this->rThread_  = new std::thread(&rogue::interfaces::ZmqServer::runThread, this);
    this->sThread_  = new std::thread(&rogue::interfaces::ZmqServer::strThread, this);
 
    // Send empty frame
    dummy = "null\n";
    zmq_send(this->zmqPub_, dummy.c_str(), dummy.size(), 0);
}
 
void rogue::interfaces::ZmqServer::stop() {
    if (threadEn_) {
        rogue::GilRelease noGil;
        threadEn_ = false;
        log_->info("Waiting for server thread to exit");
        // Null-guard: a bad_alloc on the second start()-allocated thread leaves
        // one pointer valid and the other null.
        if (rThread_ != nullptr) {
            rThread_->join();
            delete rThread_;
            rThread_ = nullptr;
        }
        if (sThread_ != nullptr) {
            sThread_->join();
            delete sThread_;
            sThread_ = nullptr;
        }
        log_->info("Closing pub socket");
        zmq_close(this->zmqPub_);
        zmqPub_ = nullptr;
        log_->info("Closing request socket");
        zmq_close(this->zmqRep_);
        zmqRep_ = nullptr;
        log_->info("Closing string socket");
        zmq_close(this->zmqStr_);
        zmqStr_ = nullptr;
        log_->info("Destroying Context");
        zmq_ctx_destroy(this->zmqCtx_);
        zmqCtx_ = nullptr;
        log_->info("Zmq server done. Exiting");
    }
}
 
bool rogue::interfaces::ZmqServer::tryConnect() {
    std::string temp;
    uint32_t opt;
 
    log_->debug("Trying to serve on ports %" PRIu16 ":%" PRIu16 ":%" PRIu16,
                this->basePort_,
                this->basePort_ + 1,
                this->basePort_ + 2);
 
    this->zmqPub_ = zmq_socket(this->zmqCtx_, ZMQ_PUB);
    this->zmqRep_ = zmq_socket(this->zmqCtx_, ZMQ_REP);
    this->zmqStr_ = zmq_socket(this->zmqCtx_, ZMQ_REP);
 
    // Throws before start() skip the dtor's stop(); free the context here.
    try {
        opt = 0;
        if (zmq_setsockopt(this->zmqPub_, ZMQ_LINGER, &opt, sizeof(int32_t)) != 0)
            throw(rogue::GeneralError("ZmqServer::tryConnect", "Failed to set socket linger"));
 
        if (zmq_setsockopt(this->zmqRep_, ZMQ_LINGER, &opt, sizeof(int32_t)) != 0)
            throw(rogue::GeneralError("ZmqServer::tryConnect", "Failed to set socket linger"));
 
        if (zmq_setsockopt(this->zmqStr_, ZMQ_LINGER, &opt, sizeof(int32_t)) != 0)
            throw(rogue::GeneralError("ZmqServer::tryConnect", "Failed to set socket linger"));
 
        opt = 100;
        if (zmq_setsockopt(this->zmqRep_, ZMQ_RCVTIMEO, &opt, sizeof(int32_t)) != 0)
            throw(rogue::GeneralError("ZmqServer::tryConnect", "Failed to set socket receive timeout"));
 
        if (zmq_setsockopt(this->zmqStr_, ZMQ_RCVTIMEO, &opt, sizeof(int32_t)) != 0)
            throw(rogue::GeneralError("ZmqServer::tryConnect", "Failed to set socket receive timeout"));
    } catch (...) {
        zmq_close(this->zmqPub_);
        zmqPub_ = nullptr;
        zmq_close(this->zmqRep_);
        zmqRep_ = nullptr;
        zmq_close(this->zmqStr_);
        zmqStr_ = nullptr;
        throw;
    }
 
    // Setup publish port
    temp = "tcp://";
    temp.append(this->addr_);
    temp.append(":");
    temp.append(std::to_string(static_cast<int64_t>(this->basePort_)));
 
    // Null after close so the auto-port retry loop does not double-close stale handles.
    if (zmq_bind(this->zmqPub_, temp.c_str()) < 0) {
        zmq_close(this->zmqPub_);
        zmqPub_ = nullptr;
        zmq_close(this->zmqRep_);
        zmqRep_ = nullptr;
        zmq_close(this->zmqStr_);
        zmqStr_ = nullptr;
        log_->debug("Failed to bind publish socket to %s: %s", temp.c_str(), zmq_strerror(zmq_errno()));
        return false;
    }
 
    // Setup response port
    temp = "tcp://";
    temp.append(this->addr_);
    temp.append(":");
    temp.append(std::to_string(static_cast<int64_t>(this->basePort_ + 1)));
 
    if (zmq_bind(this->zmqRep_, temp.c_str()) < 0) {
        zmq_close(this->zmqPub_);
        zmqPub_ = nullptr;
        zmq_close(this->zmqRep_);
        zmqRep_ = nullptr;
        zmq_close(this->zmqStr_);
        zmqStr_ = nullptr;
        log_->debug("Failed to bind request socket to %s: %s", temp.c_str(), zmq_strerror(zmq_errno()));
        return false;
    }
 
    // Setup string port
    temp = "tcp://";
    temp.append(this->addr_);
    temp.append(":");
    temp.append(std::to_string(static_cast<int64_t>(this->basePort_ + 2)));
 
    if (zmq_bind(this->zmqStr_, temp.c_str()) < 0) {
        zmq_close(this->zmqPub_);
        zmqPub_ = nullptr;
        zmq_close(this->zmqRep_);
        zmqRep_ = nullptr;
        zmq_close(this->zmqStr_);
        zmqStr_ = nullptr;
        log_->debug("Failed to bind string request socket to %s: %s", temp.c_str(), zmq_strerror(zmq_errno()));
        return false;
    }
 
    return true;
}
 
uint16_t rogue::interfaces::ZmqServer::port() {
    return this->basePort_;
}
 
std::string rogue::interfaces::ZmqServer::doString(const std::string& data) {
    return "";
}
 
#ifndef NO_PYTHON
 
void rogue::interfaces::ZmqServer::publish(bp::object value) {
    zmq_msg_t msg;
    Py_buffer valueBuf;
 
    if (!this->threadEn_) return;
 
    // PyErr_Print surfaces the Python-level error (e.g. TypeError from a
    // non-buffer argument) AND clears the thread's error indicator; without
    // it boost::python may observe "exception already set" when translating
    // our throw, masking the original cause.
    if (PyObject_GetBuffer(value.ptr(), &(valueBuf), PyBUF_SIMPLE) < 0) {
        PyErr_Print();
        throw(rogue::GeneralError::create("ZmqServer::publish", "Failed to extract object data"));
    }
 
    // zmq_msg_init_size returns -1 on allocation failure; the message is left
    // uninitialized so memcpy/zmq_msg_data must not run, and we still own
    // valueBuf and have to release it before propagating the error.
    if (zmq_msg_init_size(&msg, valueBuf.len) < 0) {
        PyBuffer_Release(&valueBuf);
        throw(rogue::GeneralError::create("ZmqServer::publish", "zmq_msg_init_size failed"));
    }
    memcpy(zmq_msg_data(&msg), valueBuf.buf, valueBuf.len);
    PyBuffer_Release(&valueBuf);
 
    rogue::GilRelease noGil;
    // On success, zmq_sendmsg transfers ownership and zeroes the message;
    // on failure (-1, e.g. ETERM during shutdown) we retain ownership and
    // must close to avoid leaking the libzmq message buffer.
    if (zmq_sendmsg(this->zmqPub_, &msg, 0) < 0) {
        zmq_msg_close(&msg);
        throw(rogue::GeneralError::create("ZmqServer::publish", "zmq_sendmsg failed"));
    }
}
 
bp::object rogue::interfaces::ZmqServer::doRequest(bp::object data) {
    bp::handle<> handle(bp::borrowed(Py_None));
    return bp::object(handle);
}
 
rogue::interfaces::ZmqServerWrap::ZmqServerWrap(std::string addr, uint16_t port)
    : rogue::interfaces::ZmqServer(addr, port) {}
 
bp::object rogue::interfaces::ZmqServerWrap::doRequest(bp::object data) {
    if (bp::override f = this->get_override("_doRequest")) {
        try {
            return (f(data));
        } catch (...) {
            PyErr_Print();
        }
    }
    return (rogue::interfaces::ZmqServer::doRequest(data));
}
 
bp::object rogue::interfaces::ZmqServerWrap::defDoRequest(bp::object data) {
    return (rogue::interfaces::ZmqServer::doRequest(data));
}
 
std::string rogue::interfaces::ZmqServerWrap::doString(const std::string& data) {
    {
        rogue::ScopedGil gil;
        if (bp::override f = this->get_override("_doString")) {
            try {
                return (f(data));
            } catch (...) {
                PyErr_Print();
            }
        }
    }
    return (rogue::interfaces::ZmqServer::doString(data));
}
 
std::string rogue::interfaces::ZmqServerWrap::defDoString(const std::string& data) {
    return (rogue::interfaces::ZmqServer::doString(data));
}
 
#endif
 
void rogue::interfaces::ZmqServer::runThread() {
    zmq_msg_t rxMsg;
    zmq_msg_t txMsg;
 
    log_->logThreadId();
    log_->info("Started Rogue server thread");
 
    while (threadEn_) {
        zmq_msg_init(&rxMsg);
 
        // Get the message
        if (zmq_recvmsg(this->zmqRep_, &rxMsg, 0) > 0) {
#ifndef NO_PYTHON
            // ZMQ_REP requires send-after-recv to keep the FSM healthy; the
            // catch blocks below send an empty reply so a single bad request
            // cannot wedge the socket into EFSM and drop all later requests.
            bool txInit = false;
            try {
                Py_buffer valueBuf;
                rogue::ScopedGil gil;
                PyObject* val = Py_BuildValue("y#", zmq_msg_data(&rxMsg), zmq_msg_size(&rxMsg));
 
                // PyErr_Print surfaces the Python-level error (e.g. MemoryError) AND clears
                // the thread's error indicator; without this, the pending exception leaks
                // into the next loop iteration's Py_BuildValue/bp::object calls.
                if (val == NULL) {
                    PyErr_Print();
                    throw(rogue::GeneralError::create("ZmqServer::runThread", "Failed to generate bytearray"));
                }
 
                bp::handle<> handle(val);
 
                bp::object ret = this->doRequest(bp::object(handle));
 
                if (PyObject_GetBuffer(ret.ptr(), &(valueBuf), PyBUF_SIMPLE) < 0) {
                    PyErr_Print();
                    throw(rogue::GeneralError::create("ZmqServer::runThread", "Failed to extract object data"));
                }
 
                // zmq_msg_init_size returns -1 on allocation failure; the message
                // stays uninitialized, so memcpy/zmq_msg_data must not run. Release
                // valueBuf before throwing so the catch block (which sends the
                // empty-REP keepalive) does not leak the Python buffer.
                if (zmq_msg_init_size(&txMsg, valueBuf.len) < 0) {
                    PyBuffer_Release(&valueBuf);
                    throw(rogue::GeneralError::create("ZmqServer::runThread", "zmq_msg_init_size failed"));
                }
                txInit = true;
                memcpy(zmq_msg_data(&txMsg), valueBuf.buf, valueBuf.len);
                PyBuffer_Release(&valueBuf);
 
                // On success, zmq_sendmsg transfers ownership and zeroes the
                // message; on failure (-1) we retain ownership and must close.
                // Throw so the catch block sends the empty REP keepalive that
                // clears the FSM -- otherwise the REP socket stays in the
                // "must send" state and every subsequent zmq_recvmsg returns
                // EFSM, wedging the worker on failed recvs until shutdown.
                if (zmq_sendmsg(this->zmqRep_, &txMsg, 0) < 0) {
                    zmq_msg_close(&txMsg);
                    txInit = false;
                    throw(rogue::GeneralError::create("ZmqServer::runThread", "zmq_sendmsg failed"));
                }
                txInit = false;
            } catch (const std::exception& e) {
                log_->warning("ZmqServer::runThread: dropping request after exception: %s", e.what());
                if (txInit) zmq_msg_close(&txMsg);
                zmq_send(this->zmqRep_, "", 0, 0);
            } catch (...) {
                log_->warning("ZmqServer::runThread: dropping request after unknown exception");
                if (txInit) zmq_msg_close(&txMsg);
                zmq_send(this->zmqRep_, "", 0, 0);
            }
#endif
        }
        // Close even on RCVTIMEO: zmq_recvmsg() initializes rxMsg regardless.
        zmq_msg_close(&rxMsg);
    }
    log_->info("Stopped Rogue server thread");
}
 
void rogue::interfaces::ZmqServer::strThread() {
    zmq_msg_t msg;
 
    log_->logThreadId();
    log_->info("Started Rogue string server thread");
 
    while (threadEn_) {
        zmq_msg_init(&msg);
 
        // Get the message
        if (zmq_recvmsg(this->zmqStr_, &msg, 0) > 0) {
            // ZMQ_REP requires send-after-recv to keep the FSM healthy; the
            // catch blocks below send an empty reply so a single bad request
            // cannot wedge the socket into EFSM and drop all later requests.
            try {
                std::string data((const char*)zmq_msg_data(&msg), zmq_msg_size(&msg));
                std::string ret = this->doString(data);
                zmq_send(this->zmqStr_, ret.c_str(), ret.size(), 0);
            } catch (const std::exception& e) {
                log_->warning("ZmqServer::strThread: dropping request after exception: %s", e.what());
                zmq_send(this->zmqStr_, "", 0, 0);
            } catch (...) {
                log_->warning("ZmqServer::strThread: dropping request after unknown exception");
                zmq_send(this->zmqStr_, "", 0, 0);
            }
        }
        // Close even on RCVTIMEO: zmq_recvmsg() initializes msg regardless.
        zmq_msg_close(&msg);
    }
    log_->info("Stopped Rogue string server thread");
}