ZmqClient.cpp

Go to the documentation of this file.

 
#include "rogue/Directives.h"
 
#include "rogue/interfaces/ZmqClient.h"
 
#include <inttypes.h>
#include <zmq.h>
 
#include <cstdio>
#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
 
namespace {
 
constexpr uint32_t kRetryLogIntervalSeconds = 30;
 
void logWaitRetry(rogue::LoggingPtr log, double seconds, uint32_t& lastLoggedSeconds) {
    uint32_t elapsedSeconds = static_cast<uint32_t>(seconds);
 
    if (elapsedSeconds == 0) return;
 
    if (lastLoggedSeconds == 0) {
        log->warning(
            "Timeout waiting for response after %" PRIu32 " seconds, server may be busy. Continuing to wait...",
            elapsedSeconds);
        lastLoggedSeconds = elapsedSeconds;
        return;
    }
 
    if (elapsedSeconds >= (lastLoggedSeconds + kRetryLogIntervalSeconds)) {
        log->warning("Still waiting for response after %" PRIu32 " seconds, server may be busy...", elapsedSeconds);
        lastLoggedSeconds = elapsedSeconds;
    }
}
 
void logWaitRecovered(rogue::LoggingPtr log, double seconds) {
    uint32_t elapsedSeconds = static_cast<uint32_t>(seconds);
 
    if (elapsedSeconds != 0) {
        log->warning("Received response from server after %" PRIu32 " seconds.", elapsedSeconds);
    }
}
 
}  // namespace
 
rogue::interfaces::ZmqClientPtr rogue::interfaces::ZmqClient::create(const std::string& addr, uint16_t port, bool doString) {
    rogue::interfaces::ZmqClientPtr ret = std::make_shared<rogue::interfaces::ZmqClient>(addr, port, doString);
    return (ret);
}
 
void rogue::interfaces::ZmqClient::setup_python() {
#ifndef NO_PYTHON
 
    bp::class_<rogue::interfaces::ZmqClientWrap, rogue::interfaces::ZmqClientWrapPtr, boost::noncopyable>(
        "ZmqClient",
        bp::init<std::string, uint16_t, bool>())
        .def("_doUpdate", &rogue::interfaces::ZmqClient::doUpdate, &rogue::interfaces::ZmqClientWrap::defDoUpdate)
        .def("_send", &rogue::interfaces::ZmqClient::send)
        .def("setTimeout", &rogue::interfaces::ZmqClient::setTimeout)
        .def("getDisp", &rogue::interfaces::ZmqClient::getDisp)
        .def("setDisp", &rogue::interfaces::ZmqClient::setDisp)
        .def("exec", &rogue::interfaces::ZmqClient::exec)
        .def("valueDisp", &rogue::interfaces::ZmqClient::valueDisp)
        .def("_stop", &rogue::interfaces::ZmqClient::stop);
#endif
}
 
rogue::interfaces::ZmqClient::ZmqClient(const std::string& addr, uint16_t port, bool doString) {
    std::string temp;
    uint32_t val;
    uint32_t reqPort;
 
    this->doString_ = doString;
    this->zmqCtx_   = zmq_ctx_new();
    this->zmqSub_   = zmq_socket(this->zmqCtx_, ZMQ_SUB);
    this->zmqReq_   = zmq_socket(this->zmqCtx_, ZMQ_REQ);
 
    log_ = rogue::Logging::create("ZmqClient");
 
    if (!doString_) {
        // Setup sub port
        temp = "tcp://";
        temp.append(addr);
        temp.append(":");
        temp.append(std::to_string(static_cast<int64_t>(port)));
 
        if (zmq_setsockopt(this->zmqSub_, ZMQ_SUBSCRIBE, "", 0) != 0)
            throw(rogue::GeneralError("ZmqClient::ZmqClient", "Failed to set socket subscribe"));
 
        val = 0;
        if (zmq_setsockopt(this->zmqSub_, ZMQ_LINGER, &val, sizeof(int32_t)) != 0)
            throw(rogue::GeneralError("ZmqClient::ZmqClient", "Failed to set socket linger"));
 
        val = 100;
        if (zmq_setsockopt(this->zmqSub_, ZMQ_RCVTIMEO, &val, sizeof(int32_t)) != 0)
            throw(rogue::GeneralError("ZmqClient::ZmqClient", "Failed to set socket timeout"));
 
        if (zmq_connect(this->zmqSub_, temp.c_str()) < 0)
            throw(rogue::GeneralError::create("ZmqClient::ZmqClient",
                                              "Failed to connect to port %" PRIu16 " at address %s",
                                              port,
                                              addr.c_str()));
 
        reqPort = port + 1;
    } else {
        reqPort = port + 2;
    }
 
    // Setup request port
    temp = "tcp://";
    temp.append(addr);
    temp.append(":");
    temp.append(std::to_string(static_cast<int64_t>(reqPort)));
 
    waitRetry_ = false;  // Don't keep waiting after timeout
    timeout_   = 1000;   // 1 second
    if (zmq_setsockopt(this->zmqReq_, ZMQ_RCVTIMEO, &timeout_, sizeof(int32_t)) != 0)
        throw(rogue::GeneralError("ZmqClient::ZmqClient", "Failed to set socket timeout"));
 
    val = 1;
    if (zmq_setsockopt(this->zmqReq_, ZMQ_REQ_CORRELATE, &val, sizeof(int32_t)) != 0)
        throw(rogue::GeneralError("ZmqClient::ZmqClient", "Failed to set socket correlate"));
 
    if (zmq_setsockopt(this->zmqReq_, ZMQ_REQ_RELAXED, &val, sizeof(int32_t)) != 0)
        throw(rogue::GeneralError("ZmqClient::ZmqClient", "Failed to set socket relaxed"));
 
    val = 0;
    if (zmq_setsockopt(this->zmqReq_, ZMQ_LINGER, &val, sizeof(int32_t)) != 0)
        throw(rogue::GeneralError("ZmqClient::ZmqClient", "Failed to set socket linger"));
 
    if (zmq_connect(this->zmqReq_, temp.c_str()) < 0)
        throw(rogue::GeneralError::create("ZmqClient::ZmqClient",
                                          "Failed to connect to port %" PRIu32 " at address %s",
                                          reqPort,
                                          addr.c_str()));
 
    if (doString_) {
        threadEn_ = false;
        log_->info("Connected to Rogue server at port %" PRIu32, reqPort);
    } else {
        log_->info("Connected to Rogue server at ports %" PRIu16 ":%" PRIu32, port, reqPort);
 
        threadEn_ = true;
        thread_   = new std::thread(&rogue::interfaces::ZmqClient::runThread, this);
    }
    running_ = true;
}
 
rogue::interfaces::ZmqClient::~ZmqClient() {
    this->stop();
}
 
void rogue::interfaces::ZmqClient::stop() {
    if (running_) {
        running_ = false;
        if (threadEn_) {
            rogue::GilRelease noGil;
            waitRetry_ = false;
            threadEn_  = false;
            thread_->join();
        }
        if (!doString_) zmq_close(this->zmqSub_);
        zmq_close(this->zmqReq_);
        zmq_ctx_destroy(this->zmqCtx_);
    }
}
 
void rogue::interfaces::ZmqClient::setTimeout(uint32_t msecs, bool waitRetry) {
    waitRetry_ = waitRetry;
    timeout_   = msecs;
 
    log_->debug("Setting timeout to %" PRIu32 " msecs, waitRetry = %" PRIu8, timeout_, waitRetry_);
 
    if (zmq_setsockopt(this->zmqReq_, ZMQ_RCVTIMEO, &timeout_, sizeof(int32_t)) != 0)
        throw(rogue::GeneralError("ZmqClient::setTimeout", "Failed to set socket timeout"));
}
 
std::string rogue::interfaces::ZmqClient::sendString(const std::string& path, const std::string& attr, const std::string& arg) {
    std::string snd;
    std::string ret;
    zmq_msg_t msg;
    std::string data;
    double seconds = 0;
    uint32_t lastLoggedSeconds = 0;
 
    if (!doString_) throw rogue::GeneralError::create("ZmqClient::sendString", "Invalid send call in standard mode");
 
    snd = "{\"attr\": \"" + attr + "\",";
    snd += "\"path\": \"" + path + "\"";
 
    if (arg != "") snd += ",\"args\": [\"" + arg + "\"]";
 
    snd += "}";
 
    rogue::GilRelease noGil;
    zmq_send(this->zmqReq_, snd.c_str(), snd.size(), 0);
 
    while (1) {
        zmq_msg_init(&msg);
        if (zmq_recvmsg(this->zmqReq_, &msg, 0) <= 0) {
            seconds += static_cast<double>(timeout_) / 1000.0;
            if (waitRetry_) {
                logWaitRetry(log_, seconds, lastLoggedSeconds);
                zmq_msg_close(&msg);
            } else {
                throw rogue::GeneralError::create("ZmqClient::sendString",
                                                  "Timeout waiting for response after %d Seconds.",
                                                  static_cast<int>(seconds));
            }
        } else {
            break;
        }
    }
 
    logWaitRecovered(log_, seconds);
 
    data = std::string((const char*)zmq_msg_data(&msg), zmq_msg_size(&msg));
    zmq_msg_close(&msg);
    return data;
}
 
std::string rogue::interfaces::ZmqClient::getDisp(const std::string& path) {
    return sendString(path, "getDisp", "");
}
 
void rogue::interfaces::ZmqClient::setDisp(const std::string& path, const std::string& value) {
    sendString(path, "setDisp", value);
}
 
std::string rogue::interfaces::ZmqClient::exec(const std::string& path, const std::string& arg) {
    return sendString(path, "__call__", arg);
}
 
std::string rogue::interfaces::ZmqClient::valueDisp(const std::string& path) {
    return sendString(path, "valueDisp", "");
}
 
#ifndef NO_PYTHON
 
bp::object rogue::interfaces::ZmqClient::send(bp::object value) {
    zmq_msg_t txMsg;
    zmq_msg_t rxMsg;
    Py_buffer valueBuf;
    bp::object ret;
    double seconds = 0;
    uint32_t lastLoggedSeconds = 0;
 
    if (doString_) throw rogue::GeneralError::create("ZmqClient::send", "Invalid send call in string mode");
 
    if (PyObject_GetBuffer(value.ptr(), &(valueBuf), PyBUF_SIMPLE) < 0)
        throw(rogue::GeneralError::create("ZmqClient::send", "Failed to extract object data"));
 
    zmq_msg_init_size(&txMsg, valueBuf.len);
    memcpy(zmq_msg_data(&txMsg), valueBuf.buf, valueBuf.len);
    PyBuffer_Release(&valueBuf);
 
    {
        rogue::GilRelease noGil;
        zmq_sendmsg(this->zmqReq_, &txMsg, 0);
 
        while (1) {
            zmq_msg_init(&rxMsg);
            if (zmq_recvmsg(this->zmqReq_, &rxMsg, 0) <= 0) {
                seconds += static_cast<double>(timeout_) / 1000.0;
                if (waitRetry_) {
                    logWaitRetry(log_, seconds, lastLoggedSeconds);
                    zmq_msg_close(&rxMsg);
                } else {
                    throw rogue::GeneralError::create(
                        "ZmqClient::send",
                        "Timeout waiting for response after %d Seconds, server may be busy!",
                        static_cast<int>(seconds));
                }
            } else {
                break;
            }
        }
    }
 
    logWaitRecovered(log_, seconds);
 
    PyObject* val = Py_BuildValue("y#", zmq_msg_data(&rxMsg), zmq_msg_size(&rxMsg));
 
    if (val == NULL) throw(rogue::GeneralError::create("ZmqClient::send", "Failed to generate bytearray"));
 
    zmq_msg_close(&rxMsg);
 
    bp::handle<> handle(val);
    ret = bp::object(handle);
    return ret;
}
 
void rogue::interfaces::ZmqClient::doUpdate(bp::object data) {}
 
rogue::interfaces::ZmqClientWrap::ZmqClientWrap(const std::string& addr, uint16_t port, bool doString)
    : rogue::interfaces::ZmqClient(addr, port, doString) {}
 
void rogue::interfaces::ZmqClientWrap::doUpdate(bp::object data) {
    if (bp::override f = this->get_override("_doUpdate")) {
        try {
            f(data);
        } catch (...) {
            PyErr_Print();
        }
    }
    rogue::interfaces::ZmqClient::doUpdate(data);
}
 
void rogue::interfaces::ZmqClientWrap::defDoUpdate(bp::object data) {
    rogue::interfaces::ZmqClient::doUpdate(data);
}
 
#endif
 
void rogue::interfaces::ZmqClient::runThread() {
    zmq_msg_t msg;
 
    log_->logThreadId();
 
    while (threadEn_) {
        zmq_msg_init(&msg);
 
        // Get the message
        if (zmq_recvmsg(this->zmqSub_, &msg, 0) > 0) {
#ifndef NO_PYTHON
            rogue::ScopedGil gil;
            PyObject* val = Py_BuildValue("y#", zmq_msg_data(&msg), zmq_msg_size(&msg));
            bp::handle<> handle(val);
            bp::object dat = bp::object(handle);
            this->doUpdate(dat);
#endif
            zmq_msg_close(&msg);
        } else {
            zmq_msg_close(&msg);
        }
    }
}