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,
             (bp::arg("msecs"), bp::arg("waitRetry"), bp::arg("maxRetries") = 0u))
        .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");
 
    // running_ is the dtor's cleanup sentinel; it is set last on success, so a
    // throw in here would otherwise leak the context and sockets.
    try {
        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
        maxRetries_ = 0;      // unbounded retry budget when waitRetry_ becomes true
        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;
    } catch (...) {
        if (zmqSub_ != nullptr) {
            zmq_close(zmqSub_);
            zmqSub_ = nullptr;
        }
        if (zmqReq_ != nullptr) {
            zmq_close(zmqReq_);
            zmqReq_ = nullptr;
        }
        if (zmqCtx_ != nullptr) {
            zmq_ctx_destroy(zmqCtx_);
            zmqCtx_ = nullptr;
        }
        throw;
    }
}
 
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();
            delete thread_;
            thread_ = nullptr;
        }
        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, uint32_t maxRetries) {
    // ZMQ sockets are not thread-safe; serialize zmqReq_ access.
    rogue::GilRelease noGil;
    std::lock_guard<std::mutex> lock(reqLock_);
 
    waitRetry_  = waitRetry;
    timeout_    = msecs;
    maxRetries_ = maxRetries;
 
    // %d (not PRIu8): bool promotes to int through varargs.
    log_->debug("Setting timeout to %" PRIu32 " msecs, waitRetry = %d, maxRetries = %" PRIu32,
                timeout_,
                waitRetry_,
                maxRetries_);
 
    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;
    std::lock_guard<std::mutex> lock(reqLock_);
    zmq_send(this->zmqReq_, snd.c_str(), snd.size(), 0);
 
    uint32_t retryCount = 0;
 
    while (1) {
        zmq_msg_init(&msg);
        if (zmq_recvmsg(this->zmqReq_, &msg, 0) <= 0) {
            seconds += static_cast<double>(timeout_) / 1000.0;
            // maxRetries_ == 0 preserves the historic unbounded-retry contract
            // (issue #1236). Caps the loop only when the caller explicitly
            // requested a finite retry budget via setTimeout(..., maxRetries).
            const bool retryBudgetExhausted = (maxRetries_ != 0 && ++retryCount >= maxRetries_);
            if (waitRetry_ && !retryBudgetExhausted) {
                logWaitRetry(log_, seconds, lastLoggedSeconds);
                zmq_msg_close(&msg);
            } else {
                zmq_msg_close(&msg);
                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");
 
    // 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("ZmqClient::send", "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(&txMsg, valueBuf.len) < 0) {
        PyBuffer_Release(&valueBuf);
        throw(rogue::GeneralError::create("ZmqClient::send", "zmq_msg_init_size failed"));
    }
    memcpy(zmq_msg_data(&txMsg), valueBuf.buf, valueBuf.len);
    PyBuffer_Release(&valueBuf);
 
    {
        rogue::GilRelease noGil;
        std::lock_guard<std::mutex> lock(reqLock_);
        // 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. Throwing here
        // also prevents the recv loop below from blocking on a request that
        // was never put on the wire.
        if (zmq_sendmsg(this->zmqReq_, &txMsg, 0) < 0) {
            // Capture errno before zmq_msg_close, which may overwrite it.
            // ETERM here is the issue #1234 signature: VirtualClient.__init__
            // tore down the context via _cleanupFailedInit() but returned
            // without raising, so the caller is now sending on a dead socket.
            int err = zmq_errno();
            zmq_msg_close(&txMsg);
            throw rogue::GeneralError::create("ZmqClient::send",
                                              "zmq_sendmsg failed: errno=%d %s",
                                              err,
                                              zmq_strerror(err));
        }
 
        uint32_t retryCount = 0;
 
        while (1) {
            zmq_msg_init(&rxMsg);
            if (zmq_recvmsg(this->zmqReq_, &rxMsg, 0) <= 0) {
                seconds += static_cast<double>(timeout_) / 1000.0;
                // maxRetries_ == 0 preserves the historic unbounded-retry contract
                // (issue #1236). Caps the loop only when the caller explicitly
                // requested a finite retry budget via setTimeout(..., maxRetries).
                const bool retryBudgetExhausted = (maxRetries_ != 0 && ++retryCount >= maxRetries_);
                if (waitRetry_ && !retryBudgetExhausted) {
                    logWaitRetry(log_, seconds, lastLoggedSeconds);
                    zmq_msg_close(&rxMsg);
                } else {
                    zmq_msg_close(&rxMsg);
                    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));
 
    // PyErr_Print surfaces the underlying Python error (e.g. MemoryError) AND
    // clears the thread's error indicator; without it boost::python may
    // observe "exception already set" when translating our throw.
    if (val == NULL) {
        PyErr_Print();
        zmq_msg_close(&rxMsg);
        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
            try {
                rogue::ScopedGil gil;
                PyObject* val = Py_BuildValue("y#", zmq_msg_data(&msg), zmq_msg_size(&msg));
 
                // 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("ZmqClient::runThread", "Failed to generate bytearray"));
                }
 
                bp::handle<> handle(val);
                bp::object dat = bp::object(handle);
                this->doUpdate(dat);
            } catch (const std::exception& e) {
                log_->warning("ZmqClient::runThread: dropping update after exception: %s", e.what());
            } catch (...) {
                log_->warning("ZmqClient::runThread: dropping update after unknown exception");
            }
#endif
        }
        // Close even on RCVTIMEO: zmq_recvmsg() initializes msg regardless.
        zmq_msg_close(&msg);
    }
}