TcpCore.cpp

Go to the documentation of this file.

 
#include "rogue/Directives.h"
 
#include "rogue/interfaces/stream/TcpCore.h"
 
#include <inttypes.h>
#include <zmq.h>
 
#include <cstring>
#include <memory>
#include <string>
 
#include "rogue/GeneralError.h"
#include "rogue/GilRelease.h"
#include "rogue/Logging.h"
#include "rogue/interfaces/stream/Buffer.h"
#include "rogue/interfaces/stream/Frame.h"
#include "rogue/interfaces/stream/FrameIterator.h"
#include "rogue/interfaces/stream/FrameLock.h"
 
namespace ris = rogue::interfaces::stream;
 
#ifndef NO_PYTHON
    #include <boost/python.hpp>
namespace bp = boost::python;
#endif
 
ris::TcpCorePtr ris::TcpCore::create(const std::string& addr, uint16_t port, bool server) {
    ris::TcpCorePtr r = std::make_shared<ris::TcpCore>(addr, port, server);
    return (r);
}
 
ris::TcpCore::TcpCore(const std::string& addr, uint16_t port, bool server) {
    int32_t opt;
    std::string logstr;
 
    this->server_ = server;
 
    logstr = "stream.TcpCore.";
    logstr.append(addr);
    logstr.append(".");
    if (server)
        logstr.append("Server.");
    else
        logstr.append("Client.");
    logstr.append(std::to_string(port));
 
    this->bridgeLog_ = rogue::Logging::create(logstr);
 
    // Format address
    this->pullAddr_ = "tcp://";
    this->pullAddr_.append(addr);
    this->pullAddr_.append(":");
    this->pushAddr_ = this->pullAddr_;
 
    this->zmqCtx_  = zmq_ctx_new();
    this->zmqPull_ = zmq_socket(this->zmqCtx_, ZMQ_PULL);
    this->zmqPush_ = zmq_socket(this->zmqCtx_, ZMQ_PUSH);
 
    // Throws before threadEn_=true skip the dtor's stop(); free the context here.
    try {
        // Don't buffer when no connection
        opt = 1;
        if (zmq_setsockopt(this->zmqPush_, ZMQ_IMMEDIATE, &opt, sizeof(int32_t)) != 0)
            throw(rogue::GeneralError("stream::TcpCore::TcpCore", "Failed to set socket immediate"));
 
        opt = 0;
        if (zmq_setsockopt(this->zmqPush_, ZMQ_LINGER, &opt, sizeof(int32_t)) != 0)
            throw(rogue::GeneralError("stream::TcpCore::TcpCore", "Failed to set socket linger"));
 
        if (zmq_setsockopt(this->zmqPull_, ZMQ_LINGER, &opt, sizeof(int32_t)) != 0)
            throw(rogue::GeneralError("stream::TcpCore::TcpCore", "Failed to set socket linger"));
 
        opt = 100;
        if (zmq_setsockopt(this->zmqPull_, ZMQ_RCVTIMEO, &opt, sizeof(int32_t)) != 0)
            throw(rogue::GeneralError("stream::TcpCore::TcpCore", "Failed to set socket receive timeout"));
 
        // Server mode
        if (server) {
            this->pullAddr_.append(std::to_string(static_cast<int64_t>(port)));
            this->pushAddr_.append(std::to_string(static_cast<int64_t>(port + 1)));
 
            this->bridgeLog_->debug("Creating pull server port: %s", this->pullAddr_.c_str());
 
            if (zmq_bind(this->zmqPull_, this->pullAddr_.c_str()) < 0)
                throw(rogue::GeneralError::create("stream::TcpCore::TcpCore",
                                                  "Failed to bind server to port %" PRIu16
                                                  " at address %s, another process may be using this port",
                                                  port,
                                                  addr.c_str()));
 
            this->bridgeLog_->debug("Creating push server port: %s", this->pushAddr_.c_str());
 
            if (zmq_bind(this->zmqPush_, this->pushAddr_.c_str()) < 0)
                throw(rogue::GeneralError::create("stream::TcpCore::TcpCore",
                                                  "Failed to bind server to port %" PRIu16
                                                  " at address %s, another process may be using this port",
                                                  port + 1,
                                                  addr.c_str()));
 
            // Client mode
        } else {
            this->pullAddr_.append(std::to_string(static_cast<int64_t>(port + 1)));
            this->pushAddr_.append(std::to_string(static_cast<int64_t>(port)));
 
            this->bridgeLog_->debug("Creating pull client port: %s", this->pullAddr_.c_str());
 
            if (zmq_connect(this->zmqPull_, this->pullAddr_.c_str()) < 0)
                throw(rogue::GeneralError::create("stream::TcpCore::TcpCore",
                                                  "Failed to connect to remote port %" PRIu16 " at address %s",
                                                  port + 1,
                                                  addr.c_str()));
 
            this->bridgeLog_->debug("Creating push client port: %s", this->pushAddr_.c_str());
 
            if (zmq_connect(this->zmqPush_, this->pushAddr_.c_str()) < 0)
                throw(rogue::GeneralError::create("stream::TcpCore::TcpCore",
                                                  "Failed to connect to remote port %" PRIu16 " at address %s",
                                                  port,
                                                  addr.c_str()));
        }
 
        this->bridgeLog_->debug("TCP stream bridge ready. pull=%s push=%s",
                                this->pullAddr_.c_str(),
                                this->pushAddr_.c_str());
 
        threadEn_     = true;
        this->thread_ = std::make_unique<std::thread>(&ris::TcpCore::runThread, this);
 
        // Set a thread name
#ifndef __MACH__
        pthread_setname_np(thread_->native_handle(), "TcpCore");
#endif
    } catch (...) {
        // ~std::thread on a joinable thread calls std::terminate(); join first.
        // RCVTIMEO=100 was already set on zmqPull_ above so the worker exits
        // within ~100 ms of threadEn_=false. Release the GIL around join() to
        // match stop(): the worker may be blocked in sendFrame() acquiring the
        // GIL to deliver into a Python slave, and join while holding the GIL
        // would deadlock.
        threadEn_ = false;
        if (thread_) {
            {
                rogue::GilRelease noGil;
                thread_->join();
            }
            thread_.reset();
        }
        if (zmqPull_ != nullptr) {
            zmq_close(zmqPull_);
            zmqPull_ = nullptr;
        }
        if (zmqPush_ != nullptr) {
            zmq_close(zmqPush_);
            zmqPush_ = nullptr;
        }
        if (zmqCtx_ != nullptr) {
            zmq_ctx_destroy(zmqCtx_);
            zmqCtx_ = nullptr;
        }
        throw;
    }
}
 
ris::TcpCore::~TcpCore() {
    this->stop();
}
 
// deprecated
void ris::TcpCore::close() {
    this->stop();
}
 
void ris::TcpCore::stop() {
    if (threadEn_) {
        rogue::GilRelease noGil;
        threadEn_ = false;
        thread_->join();
        thread_.reset();
        this->bridgeLog_->debug("Stopping TCP stream bridge. pull=%s push=%s",
                                this->pullAddr_.c_str(),
                                this->pushAddr_.c_str());
        // Serialize with in-flight acceptFrame() callers before destroying sockets.
        std::lock_guard<std::mutex> lock(bridgeMtx_);
        if (zmqPull_ != nullptr) {
            zmq_close(zmqPull_);
            zmqPull_ = nullptr;
        }
        if (zmqPush_ != nullptr) {
            zmq_close(zmqPush_);
            zmqPush_ = nullptr;
        }
        if (zmqCtx_ != nullptr) {
            zmq_ctx_destroy(zmqCtx_);
            zmqCtx_ = nullptr;
        }
    }
}
 
bool ris::TcpCore::rebuildPushSocket() {
    if (this->zmqPush_ != nullptr) {
        zmq_close(this->zmqPush_);
        this->zmqPush_ = nullptr;
    }
 
    this->zmqPush_ = zmq_socket(this->zmqCtx_, ZMQ_PUSH);
    if (this->zmqPush_ == nullptr) {
        bridgeLog_->error("Failed to create replacement push socket on %s: %s",
                          this->pushAddr_.c_str(),
                          zmq_strerror(zmq_errno()));
        return false;
    }
 
    int32_t opt = 1;
    if (zmq_setsockopt(this->zmqPush_, ZMQ_IMMEDIATE, &opt, sizeof(opt)) != 0) {
        bridgeLog_->error("Failed to set ZMQ_IMMEDIATE on rebuilt push socket: %s",
                          zmq_strerror(zmq_errno()));
        zmq_close(this->zmqPush_);
        this->zmqPush_ = nullptr;
        return false;
    }
 
    opt = 0;
    if (zmq_setsockopt(this->zmqPush_, ZMQ_LINGER, &opt, sizeof(opt)) != 0) {
        bridgeLog_->error("Failed to set ZMQ_LINGER on rebuilt push socket: %s",
                          zmq_strerror(zmq_errno()));
        zmq_close(this->zmqPush_);
        this->zmqPush_ = nullptr;
        return false;
    }
 
    if (this->server_) {
        if (zmq_bind(this->zmqPush_, this->pushAddr_.c_str()) < 0) {
            bridgeLog_->error("Failed to rebind push socket to %s: %s",
                              this->pushAddr_.c_str(),
                              zmq_strerror(zmq_errno()));
            zmq_close(this->zmqPush_);
            this->zmqPush_ = nullptr;
            return false;
        }
    } else {
        if (zmq_connect(this->zmqPush_, this->pushAddr_.c_str()) < 0) {
            bridgeLog_->error("Failed to reconnect push socket to %s: %s",
                              this->pushAddr_.c_str(),
                              zmq_strerror(zmq_errno()));
            zmq_close(this->zmqPush_);
            this->zmqPush_ = nullptr;
            return false;
        }
    }
 
    return true;
}
 
void ris::TcpCore::acceptFrame(ris::FramePtr frame) {
    uint32_t x;
    uint8_t* data;
    uint16_t flags;
    uint8_t chan;
    uint8_t err;
    zmq_msg_t msg[4];
 
    rogue::GilRelease noGil;
    ris::FrameLockPtr frLock = frame->lock();
    std::lock_guard<std::mutex> lock(bridgeMtx_);
 
    if (!threadEn_) {
        bridgeLog_->debug("Dropping frame on stopped bridge %s (payload=%" PRIu32 ")",
                          this->pushAddr_.c_str(),
                          frame->getPayload());
        return;
    }
 
    // Retry rebuild if a previous failure left zmqPush_ null.
    if (this->zmqPush_ == nullptr) {
        if (!rebuildPushSocket()) {
            bridgeLog_->warning("Push socket unavailable on %s; dropping frame "
                                "(payload=%" PRIu32 ")",
                                this->pushAddr_.c_str(),
                                frame->getPayload());
            return;
        }
    }
 
    if (zmq_msg_init_size(&(msg[0]), 2) < 0) {
        bridgeLog_->warning("Failed to init message header");
        return;
    }
    if (zmq_msg_init_size(&(msg[1]), 1) < 0) {
        bridgeLog_->warning("Failed to init message header");
        zmq_msg_close(&(msg[0]));
        return;
    }
    if (zmq_msg_init_size(&(msg[2]), 1) < 0) {
        bridgeLog_->warning("Failed to init message header");
        zmq_msg_close(&(msg[0]));
        zmq_msg_close(&(msg[1]));
        return;
    }
 
    if (zmq_msg_init_size(&(msg[3]), frame->getPayload()) < 0) {
        bridgeLog_->warning("Failed to init message with size %" PRIu32, frame->getPayload());
        zmq_msg_close(&(msg[0]));
        zmq_msg_close(&(msg[1]));
        zmq_msg_close(&(msg[2]));
        return;
    }
 
    flags = frame->getFlags();
    std::memcpy(zmq_msg_data(&(msg[0])), &flags, 2);
 
    chan = frame->getChannel();
    std::memcpy(zmq_msg_data(&(msg[1])), &chan, 1);
 
    err = frame->getError();
    std::memcpy(zmq_msg_data(&(msg[2])), &err, 1);
 
    // Copy data
    ris::FrameIterator iter = frame->begin();
    data                    = reinterpret_cast<uint8_t*>(zmq_msg_data(&(msg[3])));
    ris::fromFrame(iter, frame->getPayload(), data);
 
    bool sendFailed = false;
    for (x = 0; x < 4; x++) {
        if (zmq_sendmsg(this->zmqPush_, &(msg[x]), (x == 3) ? 0 : ZMQ_SNDMORE) < 0) {
            bridgeLog_->warning("Failed to push message part %" PRIu32 " (frame size %" PRIu32 ") on %s: %s",
                                x,
                                frame->getPayload(),
                                this->pushAddr_.c_str(),
                                zmq_strerror(zmq_errno()));
            sendFailed = true;
            zmq_msg_close(&(msg[x]));
            for (uint32_t y = x + 1; y < 4; y++) zmq_msg_close(&(msg[y]));
            break;
        }
    }
 
    if (sendFailed) {
        bridgeLog_->error("Multi-part frame failed mid-stream on part %" PRIu32
                          "; peer may have received a torso-only frame. "
                          "Resetting push socket to clear PUSH multipart FSM.",
                          x);
 
        if (!rebuildPushSocket()) {
            bridgeLog_->error("Unable to recover TcpCore push socket on %s; "
                              "next acceptFrame() will retry the rebuild",
                              this->pushAddr_.c_str());
        }
        return;
    }
 
    bridgeLog_->debug("Pushed TCP frame with size %" PRIu32 " on %s", frame->getPayload(), this->pushAddr_.c_str());
}
 
void ris::TcpCore::runThread() {
    ris::FramePtr frame;
    uint64_t more;
    size_t moreSize;
    uint8_t* data;
    uint32_t size;
    uint32_t msgCnt;
    uint32_t x;
    zmq_msg_t msg[4];
    uint16_t flags;
    uint8_t chan;
    uint8_t err;
 
    bridgeLog_->logThreadId();
 
    while (threadEn_) {
        for (x = 0; x < 4; x++) zmq_msg_init(&(msg[x]));
        msgCnt = 0;
        x      = 0;
 
        // Get message
        do {
            // Get the message
            if (zmq_recvmsg(this->zmqPull_, &(msg[x]), 0) > 0) {
                if (x != 3) x++;
                msgCnt++;
 
                // Is there more data?
                more     = 0;
                moreSize = 8;
                zmq_getsockopt(this->zmqPull_, ZMQ_RCVMORE, &more, &moreSize);
            } else {
                more = 1;
            }
        } while (threadEn_ && more);
 
        // Proper message received
        if (threadEn_ && (msgCnt == 4)) {
            // Check sizes
            if ((zmq_msg_size(&(msg[0])) != 2) || (zmq_msg_size(&(msg[1])) != 1) || (zmq_msg_size(&(msg[2])) != 1)) {
                bridgeLog_->warning(
                    "Bad message sizes. flags=%zu channel=%zu error=%zu",
                    zmq_msg_size(&(msg[0])),
                    zmq_msg_size(&(msg[1])),
                    zmq_msg_size(&(msg[2])));
                for (x = 0; x < msgCnt; x++) zmq_msg_close(&(msg[x]));
                continue;  // while (1)
            }
 
            // Get fields
            std::memcpy(&flags, zmq_msg_data(&(msg[0])), 2);
            std::memcpy(&chan, zmq_msg_data(&(msg[1])), 1);
            std::memcpy(&err, zmq_msg_data(&(msg[2])), 1);
 
            // Get message info
            data = reinterpret_cast<uint8_t*>(zmq_msg_data(&(msg[3])));
            size = zmq_msg_size(&(msg[3]));
 
            // Generate frame
            frame = reqLocalFrame(size, false);
            frame->setPayload(size);
 
            // Copy data
            ris::FrameIterator iter = frame->begin();
            ris::toFrame(iter, size, data);
 
            // Set frame meta data and send
            frame->setFlags(flags);
            frame->setChannel(chan);
            frame->setError(err);
 
            bridgeLog_->debug("Pulled frame with size %" PRIu32, frame->getPayload());
            sendFrame(frame);
        }
 
        for (x = 0; x < msgCnt; x++) zmq_msg_close(&(msg[x]));
    }
}
 
void ris::TcpCore::setup_python() {
#ifndef NO_PYTHON
 
#pragma GCC diagnostic push
#pragma GCC diagnostic ignored "-Wdeprecated-declarations"
    bp::class_<ris::TcpCore, ris::TcpCorePtr, bp::bases<ris::Master, ris::Slave>, boost::noncopyable>("TcpCore",
                                                                                                      bp::no_init)
        .def("close", &ris::TcpCore::close);
#pragma GCC diagnostic pop
 
    bp::implicitly_convertible<ris::TcpCorePtr, ris::MasterPtr>();
    bp::implicitly_convertible<ris::TcpCorePtr, ris::SlavePtr>();
#endif
}