zmq.cpp

Go to the documentation of this file.

/*
 * Zth (libzth), a cooperative userspace multitasking library.
 * Copyright (C) 2019-2022  Jochem Rutgers
 *
 * This Source Code Form is subject to the terms of the Mozilla Public
 * License, v. 2.0. If a copy of the MPL was not distributed with this
 * file, You can obtain one at https://mozilla.org/MPL/2.0/.
 */
 
#define ZTH_REDIRECT_ZMQ 0
 
#include <libzth/macros.h>
 
#ifdef ZTH_HAVE_LIBZMQ
 
#       include <libzth/poller.h>
#       include <libzth/waiter.h>
#       include <libzth/worker.h>
#       include <libzth/zmq.h>
 
namespace zth {
namespace zmq {
 
static void zmq_global_deinit()
{
        zth_dbg(zmq, "destroy context");
        zmq_ctx_term(zmq_context());
}
 
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());
 
        // Only do the deinit this when 0MQ was actually used.
        atexit(zmq_global_deinit);
        return zmq_ctx;
}
 
void* zmq_context()
{
        static void* 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);
 
        if((errno = poll(PollableFd(socket, Pollable::PollOut))))
                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);
 
        if((errno = poll(PollableFd(socket, Pollable::PollIn))))
                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);
 
        if((errno = poll(PollableFd(socket, Pollable::PollOut))))
                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);
 
        if((errno = poll(PollableFd(socket, Pollable::PollIn))))
                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);
 
        if((errno = poll(PollableFd(socket, Pollable::PollOut))))
                return -1;
 
        return ::zmq_send_const(socket, buf, len, flags | ZMQ_DONTWAIT);
}
 
} // namespace zmq
} // namespace zth
#else
static int no_zmq __attribute__((unused));
#endif // ZTH_HAVE_LIBZMQ