zmq.cpp

Go to the documentation of this file.

/*
 * SPDX-FileCopyrightText: 2019-2026 Jochem Rutgers
 *
 * SPDX-License-Identifier: MPL-2.0
 */
 
#define ZTH_REDIRECT_ZMQ 0
 
#include <libzth/zmq.h>
 
#ifdef ZTH_HAVE_LIBZMQ
 
#  include <libzth/poller.h>
#  include <libzth/waiter.h>
#  include <libzth/worker.h>
 
namespace zth {
namespace zmq {
 
static void*& zmq_context_storage()
{
        // NOLINTNEXTLINE(cppcoreguidelines-avoid-non-const-global-variables)
        static void* zmq_ctx = nullptr;
        return zmq_ctx;
}
 
#  ifndef ZTH_OS_WINDOWS
static void zmq_global_deinit()
{
        void*& zmq_ctx = zmq_context_storage();
        if(!zmq_ctx)
                return;
 
        zth_dbg(zmq, "destroy context");
        (void)zmq_ctx_term(zmq_ctx);
        zmq_ctx = nullptr;
}
#  endif
 
static void* zmq_global_init()
{
        int major = 0;
        int minor = 0;
        int patch = 0;
        zmq_version(&major, &minor, &patch);
        zth_dbg(banner, "0MQ version is %d.%d.%d", major, minor, patch);
 
        zth_dbg(zmq, "new context");
        void* zmq_ctx = zmq_ctx_new();
 
        if(!zmq_ctx)
                zth_abort("0MQ context creation failed; %s", err(errno).c_str());
 
        // Windows shutdown order can tear down Winsock before process-exit hooks,
        // which makes libzmq's socket cleanup assert with WSANOTINITIALISED.
        // Keep explicit cleanup on non-Windows only.
#  ifndef ZTH_OS_WINDOWS
        // Only do the deinit this when 0MQ was actually used.
        (void)atexit(zmq_global_deinit);
#  endif
        zmq_context_storage() = zmq_ctx;
        return zmq_ctx;
}
 
void* zmq_context()
{
        void*& zmq_ctx = zmq_context_storage();
        if(!zmq_ctx)
                zmq_ctx = zmq_global_init();
        return zmq_ctx;
}
 
void* zmq_socket(int type)
{
        return ::zmq_socket(zmq_context(), type);
}
 
int zmq_msg_send(zmq_msg_t* msg, void* socket, int flags)
{
        perf_syscall("zmq_msg_send()");
 
        int res = ::zmq_msg_send(msg, socket, flags | ZMQ_DONTWAIT);
        int err = res == -1 ? zmq_errno() : 0;
        if(err != EAGAIN || (flags & ZMQ_DONTWAIT))
                return res;
 
        zth_dbg(zmq, "[%s] zmq_msg_send(%p) hand-off", zth::currentFiber().str().c_str(), socket);
 
        errno = poll(PollableFd(socket, Pollable::PollOut));
        if(errno)
                return -1;
 
        return ::zmq_msg_send(msg, socket, flags | ZMQ_DONTWAIT);
}
 
int zmq_msg_recv(zmq_msg_t* msg, void* socket, int flags)
{
        perf_syscall("zmq_msg_recv()");
 
        int res = ::zmq_msg_recv(msg, socket, flags | ZMQ_DONTWAIT);
        int err = res == -1 ? zmq_errno() : 0;
        if(err != EAGAIN || (flags & ZMQ_DONTWAIT))
                return res;
 
        zth_dbg(zmq, "[%s] zmq_msg_recv(%p) hand-off", zth::currentFiber().str().c_str(), socket);
 
        errno = poll(PollableFd(socket, Pollable::PollIn));
        if(errno)
                return -1;
 
        return ::zmq_msg_recv(msg, socket, flags | ZMQ_DONTWAIT);
}
 
int zmq_send(void* socket, void const* buf, size_t len, int flags)
{
        perf_syscall("zmq_send()");
 
        int res = ::zmq_send(socket, buf, len, flags | ZMQ_DONTWAIT);
        int err = res == -1 ? zmq_errno() : 0;
        if(err != EAGAIN || (flags & ZMQ_DONTWAIT))
                return res;
 
        zth_dbg(zmq, "[%s] zmq_send(%p) hand-off", zth::currentFiber().str().c_str(), socket);
 
        errno = poll(PollableFd(socket, Pollable::PollOut));
        if(errno)
                return -1;
 
        return ::zmq_send(socket, buf, len, flags | ZMQ_DONTWAIT);
}
 
int zmq_recv(void* socket, void* buf, size_t len, int flags)
{
        perf_syscall("zmq_recv()");
 
        int res = ::zmq_recv(socket, buf, len, flags | ZMQ_DONTWAIT);
        int err = res == -1 ? zmq_errno() : 0;
        if(err != EAGAIN || (flags & ZMQ_DONTWAIT))
                return res;
 
        zth_dbg(zmq, "[%s] zmq_recv(%p) hand-off", zth::currentFiber().str().c_str(), socket);
 
        errno = poll(PollableFd(socket, Pollable::PollIn));
        if(errno)
                return -1;
 
        return ::zmq_recv(socket, buf, len, flags | ZMQ_DONTWAIT);
}
 
int zmq_send_const(void* socket, void const* buf, size_t len, int flags)
{
        perf_syscall("zmq_send_const()");
 
        int res = ::zmq_send_const(socket, buf, len, flags | ZMQ_DONTWAIT);
        int err = res == -1 ? zmq_errno() : 0;
        if(err != EAGAIN || (flags & ZMQ_DONTWAIT))
                return res;
 
        zth_dbg(zmq, "[%s] zmq_send_const(%p) hand-off", zth::currentFiber().str().c_str(), socket);
 
        errno = poll(PollableFd(socket, Pollable::PollOut));
        if(errno)
                return -1;
 
        return ::zmq_send_const(socket, buf, len, flags | ZMQ_DONTWAIT);
}
 
} // namespace zmq
} // namespace zth
#else
static int const no_zmq __attribute__((unused)) = 0;
#endif // ZTH_HAVE_LIBZMQ