/* SPDX-License-Identifier: MPL-2.0 */
// "Tell them I was a writer.
// A maker of software.
// A humanist. A father.
// And many things.
// But above all, a writer.
// Thank You. :)"
// - Pieter Hintjens
#include "precompiled.hpp"
#define ZMQ_TYPE_UNSAFE
#include "macros.hpp"
#include "poller.hpp"
#include "peer.hpp"
#if !defined ZMQ_HAVE_POLLER
// On AIX platform, poll.h has to be included first to get consistent
// definition of pollfd structure (AIX uses 'reqevents' and 'retnevents'
// instead of 'events' and 'revents' and defines macros to map from POSIX-y
// names to AIX-specific names).
#if defined ZMQ_POLL_BASED_ON_POLL && !defined ZMQ_HAVE_WINDOWS
#include <poll.h>
#endif
#include "polling_util.hpp"
#endif
// TODO: determine if this is an issue, since zmq.h is being loaded from pch.
// zmq.h must be included *after* poll.h for AIX to build properly
//#include "../include/zmq.h"
#if !defined ZMQ_HAVE_WINDOWS
#include <unistd.h>
#ifdef ZMQ_HAVE_VXWORKS
#include <strings.h>
#endif
#endif
// XSI vector I/O
#if defined ZMQ_HAVE_UIO
#include <sys/uio.h>
#else
struct iovec
{
void *iov_base;
size_t iov_len;
};
#endif
#include <string.h>
#include <stdlib.h>
#include <new>
#include <climits>
#include "proxy.hpp"
#include "socket_base.hpp"
#include "stdint.hpp"
#include "config.hpp"
#include "likely.hpp"
#include "clock.hpp"
#include "ctx.hpp"
#include "err.hpp"
#include "msg.hpp"
#include "fd.hpp"
#include "metadata.hpp"
#include "socket_poller.hpp"
#include "timers.hpp"
#include "ip.hpp"
#include "address.hpp"
#ifdef ZMQ_HAVE_PPOLL
#include "polling_util.hpp"
#include <sys/select.h>
#endif
#if defined ZMQ_HAVE_OPENPGM
#define __PGM_WININT_H__
#include <pgm/pgm.h>
#endif
// Compile time check whether msg_t fits into zmq_msg_t.
typedef char
check_msg_t_size[sizeof (zmq::msg_t) == sizeof (zmq_msg_t) ? 1 : -1];
void zmq_version (int *major_, int *minor_, int *patch_)
{
*major_ = ZMQ_VERSION_MAJOR;
*minor_ = ZMQ_VERSION_MINOR;
*patch_ = ZMQ_VERSION_PATCH;
}
const char *zmq_strerror (int errnum_)
{
return zmq::errno_to_string (errnum_);
}
int zmq_errno (void)
{
return errno;
}
// New context API
void *zmq_ctx_new (void)
{
// We do this before the ctx constructor since its embedded mailbox_t
// object needs the network to be up and running (at least on Windows).
if (!zmq::initialize_network ()) {
return NULL;
}
// Create 0MQ context.
zmq::ctx_t *ctx = new (std::nothrow) zmq::ctx_t;
if (ctx) {
if (!ctx->valid ()) {
delete ctx;
return NULL;
}
}
return ctx;
}
int zmq_ctx_term (void *ctx_)
{
if (!ctx_ || !(static_cast<zmq::ctx_t *> (ctx_))->check_tag ()) {
errno = EFAULT;
return -1;
}
const int rc = (static_cast<zmq::ctx_t *> (ctx_))->terminate ();
const int en = errno;
// Shut down only if termination was not interrupted by a signal.
if (!rc || en != EINTR) {
zmq::shutdown_network ();
}
errno = en;
return rc;
}
int zmq_ctx_shutdown (void *ctx_)
{
if (!ctx_ || !(static_cast<zmq::ctx_t *> (ctx_))->check_tag ()) {
errno = EFAULT;
return -1;
}
return (static_cast<zmq::ctx_t *> (ctx_))->shutdown ();
}
int zmq_ctx_set (void *ctx_, int option_, int optval_)
{
return zmq_ctx_set_ext (ctx_, option_, &optval_, sizeof (int));
}
int zmq_ctx_set_ext (void *ctx_,
int option_,
const void *optval_,
size_t optvallen_)
{
if (!ctx_ || !(static_cast<zmq::ctx_t *> (ctx_))->check_tag ()) {
errno = EFAULT;
return -1;
}
return (static_cast<zmq::ctx_t *> (ctx_))
->set (option_, optval_, optvallen_);
}
int zmq_ctx_get (void *ctx_, int option_)
{
if (!ctx_ || !(static_cast<zmq::ctx_t *> (ctx_))->check_tag ()) {
errno = EFAULT;
return -1;
}
return (static_cast<zmq::ctx_t *> (ctx_))->get (option_);
}
int zmq_ctx_get_ext (void *ctx_, int option_, void *optval_, size_t *optvallen_)
{
if (!ctx_ || !(static_cast<zmq::ctx_t *> (ctx_))->check_tag ()) {
errno = EFAULT;
return -1;
}
return (static_cast<zmq::ctx_t *> (ctx_))
->get (option_, optval_, optvallen_);
}
// Stable/legacy context API
void *zmq_init (int io_threads_)
{
if (io_threads_ >= 0) {
void *ctx = zmq_ctx_new ();
zmq_ctx_set (ctx, ZMQ_IO_THREADS, io_threads_);
return ctx;
}
errno = EINVAL;
return NULL;
}
int zmq_term (void *ctx_)
{
return zmq_ctx_term (ctx_);
}
int zmq_ctx_destroy (void *ctx_)
{
return zmq_ctx_term (ctx_);
}
// Sockets
static zmq::socket_base_t *as_socket_base_t (void *s_)
{
zmq::socket_base_t *s = static_cast<zmq::socket_base_t *> (s_);
if (!s_ || !s->check_tag ()) {
errno = ENOTSOCK;
return NULL;
}
return s;
}
void *zmq_socket (void *ctx_, int type_)
{
if (!ctx_ || !(static_cast<zmq::ctx_t *> (ctx_))->check_tag ()) {
errno = EFAULT;
return NULL;
}
zmq::ctx_t *ctx = static_cast<zmq::ctx_t *> (ctx_);
zmq::socket_base_t *s = ctx->create_socket (type_);
return static_cast<void *> (s);
}
int zmq_close (void *s_)
{
zmq::socket_base_t *s = as_socket_base_t (s_);
if (!s)
return -1;
s->close ();
return 0;
}
int zmq_setsockopt (void *s_,
int option_,
const void *optval_,
size_t optvallen_)
{
zmq::socket_base_t *s = as_socket_base_t (s_);
if (!s)
return -1;
return s->setsockopt (option_, optval_, optvallen_);
}
int zmq_getsockopt (void *s_, int option_, void *optval_, size_t *optvallen_)
{
zmq::socket_base_t *s = as_socket_base_t (s_);
if (!s)
return -1;
return s->getsockopt (option_, optval_, optvallen_);
}
int zmq_socket_monitor_versioned (
void *s_, const char *addr_, uint64_t events_, int event_version_, int type_)
{
zmq::socket_base_t *s = as_socket_base_t (s_);
if (!s)
return -1;
return s->monitor (addr_, events_, event_version_, type_);
}
int zmq_socket_monitor (void *s_, const char *addr_, int events_)
{
return zmq_socket_monitor_versioned (s_, addr_, events_, 1, ZMQ_PAIR);
}
int zmq_join (void *s_, const char *group_)
{
zmq::socket_base_t *s = as_socket_base_t (s_);
if (!s)
return -1;
return s->join (group_);
}
int zmq_leave (void *s_, const char *group_)
{
zmq::socket_base_t *s = as_socket_base_t (s_);
if (!s)
return -1;
return s->leave (group_);
}
int zmq_bind (void *s_, const char *addr_)
{
zmq::socket_base_t *s = as_socket_base_t (s_);
if (!s)
return -1;
return s->bind (addr_);
}
int zmq_connect (void *s_, const char *addr_)
{
zmq::socket_base_t *s = as_socket_base_t (s_);
if (!s)
return -1;
return s->connect (addr_);
}
uint32_t zmq_connect_peer (void *s_, const char *addr_)
{
zmq::peer_t *s = static_cast<zmq::peer_t *> (s_);
if (!s_ || !s->check_tag ()) {
errno = ENOTSOCK;
return 0;
}
int socket_type;
size_t socket_type_size = sizeof (socket_type);
if (s->getsockopt (ZMQ_TYPE, &socket_type, &socket_type_size) != 0)
return 0;
if (socket_type != ZMQ_PEER) {
errno = ENOTSUP;
return 0;
}
return s->connect_peer (addr_);
}
int zmq_disconnect_peer (void *s_, uint32_t routing_id_)
{
zmq::socket_base_t *s = as_socket_base_t (s_);
if (!s)
return -1;
return s->disconnect_peer (routing_id_);
}
int zmq_unbind (void *s_, const char *addr_)
{
zmq::socket_base_t *s = as_socket_base_t (s_);
if (!s)
return -1;
return s->term_endpoint (addr_);
}
int zmq_disconnect (void *s_, const char *addr_)
{
zmq::socket_base_t *s = as_socket_base_t (s_);
if (!s)
return -1;
return s->term_endpoint (addr_);
}
// Sending functions.
static inline int
s_sendmsg (zmq::socket_base_t *s_, zmq_msg_t *msg_, int flags_)
{
size_t sz = zmq_msg_size (msg_);
const int rc = s_->send (reinterpret_cast<zmq::msg_t *> (msg_), flags_);
if (unlikely (rc < 0))
return -1;
// This is what I'd like to do, my C++ fu is too weak -- PH 2016/02/09
// int max_msgsz = s_->parent->get (ZMQ_MAX_MSGSZ);
size_t max_msgsz = INT_MAX;
// Truncate returned size to INT_MAX to avoid overflow to negative values
return static_cast<int> (sz < max_msgsz ? sz : max_msgsz);
}
/* To be deprecated once zmq_msg_send() is stable */
int zmq_sendmsg (void *s_, zmq_msg_t *msg_, int flags_)
{
return zmq_msg_send (msg_, s_, flags_);
}
int zmq_send (void *s_, const void *buf_, size_t len_, int flags_)
{
zmq::socket_base_t *s = as_socket_base_t (s_);
if (!s)
return -1;
zmq_msg_t msg;
int rc = zmq_msg_init_buffer (&msg, buf_, len_);
if (unlikely (rc < 0))
return -1;
rc = s_sendmsg (s, &msg, flags_);
if (unlikely (rc < 0)) {
const int err = errno;
const int rc2 = zmq_msg_close (&msg);
errno_assert (rc2 == 0);
errno = err;
return -1;
}
// Note the optimisation here. We don't close the msg object as it is
// empty anyway. This may change when implementation of zmq_msg_t changes.
return rc;
}
int zmq_send_const (void *s_, const void *buf_, size_t len_, int flags_)
{
zmq::socket_base_t *s = as_socket_base_t (s_);
if (!s)
return -1;
zmq_msg_t msg;
int rc =
zmq_msg_init_data (&msg, const_cast<void *> (buf_), len_, NULL, NULL);
if (rc != 0)
return -1;
rc = s_sendmsg (s, &msg, flags_);
if (unlikely (rc < 0)) {
const int err = errno;
const int rc2 = zmq_msg_close (&msg);
errno_assert (rc2 == 0);
errno = err;
return -1;
}
// Note the optimisation here. We don't close the msg object as it is
// empty anyway. This may change when implementation of zmq_msg_t changes.
return rc;
}
// Send multiple messages.
// TODO: this function has no man page
//
// If flag bit ZMQ_SNDMORE is set the vector is treated as
// a single multi-part message, i.e. the last message has
// ZMQ_SNDMORE bit switched off.
//
int zmq_sendiov (void *s_, iovec *a_, size_t count_, int flags_)
{
zmq::socket_base_t *s = as_socket_base_t (s_);
if (!s)
return -1;
if (unlikely (count_ <= 0 || !a_)) {
errno = EINVAL;
return -1;
}
int rc = 0;
zmq_msg_t msg;
for (size_t i = 0; i < count_; ++i) {
rc = zmq_msg_init_size (&msg, a_[i].iov_len);
if (rc != 0) {
rc = -1;
break;
}
memcpy (zmq_msg_data (&msg), a_[i].iov_base, a_[i].iov_len);
if (i == count_ - 1)
flags_ = flags_ & ~ZMQ_SNDMORE;
rc = s_sendmsg (s, &msg, flags_);
if (unlikely (rc < 0)) {
const int err = errno;
const int rc2 = zmq_msg_close (&msg);
errno_assert (rc2 == 0);
errno = err;
rc = -1;
break;
}
}
return rc;
}
// Receiving functions.
static int s_recvmsg (zmq::socket_base_t *s_, zmq_msg_t *msg_, int flags_)
{
const int rc = s_->recv (reinterpret_cast<zmq::msg_t *> (msg_), flags_);
if (unlikely (rc < 0))
return -1;
// Truncate returned size to INT_MAX to avoid overflow to negative values
const size_t sz = zmq_msg_size (msg_);
return static_cast<int> (sz < INT_MAX ? sz : INT_MAX);
}
/* To be deprecated once zmq_msg_recv() is stable */
int zmq_recvmsg (void *s_, zmq_msg_t *msg_, int flags_)
{
return zmq_msg_recv (msg_, s_, flags_);
}
int zmq_recv (void *s_, void *buf_, size_t len_, int flags_)
{
zmq::socket_base_t *s = as_socket_base_t (s_);
if (!s)
return -1;
zmq_msg_t msg;
int rc = zmq_msg_init (&msg);
errno_assert (rc == 0);
const int nbytes = s_recvmsg (s, &msg, flags_);
if (unlikely (nbytes < 0)) {
const int err = errno;
rc = zmq_msg_close (&msg);
errno_assert (rc == 0);
errno = err;
return -1;
}
// An oversized message is silently truncated.
const size_t to_copy = size_t (nbytes) < len_ ? size_t (nbytes) : len_;
// We explicitly allow a null buffer argument if len is zero
if (to_copy) {
assert (buf_);
memcpy (buf_, zmq_msg_data (&msg), to_copy);
}
rc = zmq_msg_close (&msg);
errno_assert (rc == 0);
return nbytes;
}
// Receive a multi-part message
//
// Receives up to *count_ parts of a multi-part message.
// Sets *count_ to the actual number of parts read.
// ZMQ_RCVMORE is set to indicate if a complete multi-part message was read.
// Returns number of message parts read, or -1 on error.
//
// Note: even if -1 is returned, some parts of the message
// may have been read. Therefore the client must consult
// *count_ to retrieve message parts successfully read,
// even if -1 is returned.
//
// The iov_base* buffers of each iovec *a_ filled in by this
// function may be freed using free().
// TODO: this function has no man page
//
int zmq_recviov (void *s_, iovec *a_, size_t *count_, int flags_)
{
zmq::socket_base_t *s = as_socket_base_t (s_);
if (!s)
return -1;
if (unlikely (!count_ || *count_ <= 0 || !a_)) {
errno = EINVAL;
return -1;
}
const size_t count = *count_;
int nread = 0;
bool recvmore = true;
*count_ = 0;
for (size_t i = 0; recvmore && i < count; ++i) {
zmq_msg_t msg;
int rc = zmq_msg_init (&msg);
errno_assert (rc == 0);
const int nbytes = s_recvmsg (s, &msg, flags_);
if (unlikely (nbytes < 0)) {
const int err = errno;
rc = zmq_msg_close (&msg);
errno_assert (rc == 0);
errno = err;
nread = -1;
break;
}
a_[i].iov_len = zmq_msg_size (&msg);
a_[i].iov_base = static_cast<char *> (malloc (a_[i].iov_len));
if (unlikely (!a_[i].iov_base)) {
errno = ENOMEM;
return -1;
}
memcpy (a_[i].iov_base, static_cast<char *> (zmq_msg_data (&msg)),
a_[i].iov_len);
// Assume zmq_socket ZMQ_RVCMORE is properly set.
const zmq::msg_t *p_msg = reinterpret_cast<const zmq::msg_t *> (&msg);
recvmore = p_msg->flags () & zmq::msg_t::more;
rc = zmq_msg_close (&msg);
errno_assert (rc == 0);
++*count_;
++nread;
}
return nread;
}
// Message manipulators.
int zmq_msg_init (zmq_msg_t *msg_)
{
return (reinterpret_cast<zmq::msg_t *> (msg_))->init ();
}
int zmq_msg_init_size (zmq_msg_t *msg_, size_t size_)
{
return (reinterpret_cast<zmq::msg_t *> (msg_))->init_size (size_);
}
int zmq_msg_init_buffer (zmq_msg_t *msg_, const void *buf_, size_t size_)
{
return (reinterpret_cast<zmq::msg_t *> (msg_))->init_buffer (buf_, size_);
}
int zmq_msg_init_data (
zmq_msg_t *msg_, void *data_, size_t size_, zmq_free_fn *ffn_, void *hint_)
{
return (reinterpret_cast<zmq::msg_t *> (msg_))
->init_data (data_, size_, ffn_, hint_);
}
int zmq_msg_send (zmq_msg_t *msg_, void *s_, int flags_)
{
zmq::socket_base_t *s = as_socket_base_t (s_);
if (!s)
return -1;
return s_sendmsg (s, msg_, flags_);
}
int zmq_msg_recv (zmq_msg_t *msg_, void *s_, int flags_)
{
zmq::socket_base_t *s = as_socket_base_t (s_);
if (!s)
return -1;
return s_recvmsg (s, msg_, flags_);
}
int zmq_msg_close (zmq_msg_t *msg_)
{
return (reinterpret_cast<zmq::msg_t *> (msg_))->close ();
}
int zmq_msg_move (zmq_msg_t *dest_, zmq_msg_t *src_)
{
return (reinterpret_cast<zmq::msg_t *> (dest_))
->move (*reinterpret_cast<zmq::msg_t *> (src_));
}
int zmq_msg_copy (zmq_msg_t *dest_, zmq_msg_t *src_)
{
return (reinterpret_cast<zmq::msg_t *> (dest_))
->copy (*reinterpret_cast<zmq::msg_t *> (src_));
}
void *zmq_msg_data (zmq_msg_t *msg_)
{
return (reinterpret_cast<zmq::msg_t *> (msg_))->data ();
}
size_t zmq_msg_size (const zmq_msg_t *msg_)
{
return ((zmq::msg_t *) msg_)->size ();
}
int zmq_msg_more (const zmq_msg_t *msg_)
{
return zmq_msg_get (msg_, ZMQ_MORE);
}
int zmq_msg_get (const zmq_msg_t *msg_, int property_)
{
const char *fd_string;
switch (property_) {
case ZMQ_MORE:
return (((zmq::msg_t *) msg_)->flags () & zmq::msg_t::more) ? 1 : 0;
case ZMQ_SRCFD:
fd_string = zmq_msg_gets (msg_, "__fd");
if (fd_string == NULL)
return -1;
return atoi (fd_string);
case ZMQ_SHARED:
return (((zmq::msg_t *) msg_)->is_cmsg ())
|| (((zmq::msg_t *) msg_)->flags () & zmq::msg_t::shared)
? 1
: 0;
default:
errno = EINVAL;
return -1;
}
}
int zmq_msg_set (zmq_msg_t *, int, int)
{
// No properties supported at present
errno = EINVAL;
return -1;
}
int zmq_msg_set_routing_id (zmq_msg_t *msg_, uint32_t routing_id_)
{
return (reinterpret_cast<zmq::msg_t *> (msg_))
->set_routing_id (routing_id_);
}
uint32_t zmq_msg_routing_id (zmq_msg_t *msg_)
{
return (reinterpret_cast<zmq::msg_t *> (msg_))->get_routing_id ();
}
int zmq_msg_set_group (zmq_msg_t *msg_, const char *group_)
{
return (reinterpret_cast<zmq::msg_t *> (msg_))->set_group (group_);
}
const char *zmq_msg_group (zmq_msg_t *msg_)
{
return (reinterpret_cast<zmq::msg_t *> (msg_))->group ();
}
// Get message metadata string
const char *zmq_msg_gets (const zmq_msg_t *msg_, const char *property_)
{
const zmq::metadata_t *metadata =
reinterpret_cast<const zmq::msg_t *> (msg_)->metadata ();
const char *value = NULL;
if (metadata)
value = metadata->get (std::string (property_));
if (value)
return value;
errno = EINVAL;
return NULL;
}
// Polling.
#if defined ZMQ_HAVE_POLLER
static int zmq_poller_poll (zmq_pollitem_t *items_, int nitems_, long timeout_)
{
// implement zmq_poll on top of zmq_poller
int rc;
zmq_poller_event_t *events;
zmq::socket_poller_t poller;
events = new (std::nothrow) zmq_poller_event_t[nitems_];
alloc_assert (events);
bool repeat_items = false;
// Register sockets with poller
for (int i = 0; i < nitems_; i++) {
items_[i].revents = 0;
bool modify = false;
short e = items_[i].events;
if (items_[i].socket) {
// Poll item is a 0MQ socket.
for (int j = 0; j < i; ++j) {
// Check for repeat entries
if (items_[j].socket == items_[i].socket) {
repeat_items = true;
modify = true;
e |= items_[j].events;
}
}
if (modify) {
rc = zmq_poller_modify (&poller, items_[i].socket, e);
} else {
rc = zmq_poller_add (&poller, items_[i].socket, NULL, e);
}
if (rc < 0) {
delete[] events;
return rc;
}
} else {
// Poll item is a raw file descriptor.
for (int j = 0; j < i; ++j) {
// Check for repeat entries
if (!items_[j].socket && items_[j].fd == items_[i].fd) {
repeat_items = true;
modify = true;
e |= items_[j].events;
}
}
if (modify) {
rc = zmq_poller_modify_fd (&poller, items_[i].fd, e);
} else {
rc = zmq_poller_add_fd (&poller, items_[i].fd, NULL, e);
}
if (rc < 0) {
delete[] events;
return rc;
}
}
}
// Wait for events
rc = zmq_poller_wait_all (&poller, events, nitems_, timeout_);
if (rc < 0) {
delete[] events;
if (zmq_errno () == EAGAIN) {
return 0;
}
return rc;
}
// Transform poller events into zmq_pollitem events.
// items_ contains all items, while events only contains fired events.
// If no sockets are repeated (likely), the two are still co-ordered, so step through the items
// checking for matches only on the first event.
// If there are repeat items, they cannot be assumed to be co-ordered,
// so each pollitem must check fired events from the beginning.
int j_start = 0, found_events = rc;
for (int i = 0; i < nitems_; i++) {
for (int j = j_start; j < found_events; ++j) {
if ((items_[i].socket && items_[i].socket == events[j].socket)
|| (!(items_[i].socket || events[j].socket)
&& items_[i].fd == events[j].fd)) {
items_[i].revents = events[j].events & items_[i].events;
if (!repeat_items) {
// no repeats, we can ignore events we've already seen
j_start++;
}
break;
}
if (!repeat_items) {
// no repeats, never have to look at j > j_start
break;
}
}
}
// Cleanup
delete[] events;
return rc;
}
#endif // ZMQ_HAVE_POLLER
int zmq_poll (zmq_pollitem_t *items_, int nitems_, long timeout_)
{
#if defined ZMQ_HAVE_POLLER
// if poller is present, use that if there is at least 1 thread-safe socket,
// otherwise fall back to the previous implementation as it's faster.
for (int i = 0; i != nitems_; i++) {
if (items_[i].socket) {
zmq::socket_base_t *s = as_socket_base_t (items_[i].socket);
if (s) {
if (s->is_thread_safe ())
return zmq_poller_poll (items_, nitems_, timeout_);
} else {
//as_socket_base_t returned NULL : socket is invalid
return -1;
}
}
}
#endif // ZMQ_HAVE_POLLER
#if defined ZMQ_POLL_BASED_ON_POLL || defined ZMQ_POLL_BASED_ON_SELECT
if (unlikely (nitems_ < 0)) {
errno = EINVAL;
return -1;
}
if (unlikely (nitems_ == 0)) {
if (timeout_ == 0)
return 0;
#if defined ZMQ_HAVE_WINDOWS
Sleep (timeout_ > 0 ? timeout_ : INFINITE);
return 0;
#elif defined ZMQ_HAVE_VXWORKS
struct timespec ns_;
ns_.tv_sec = timeout_ / 1000;
ns_.tv_nsec = timeout_ % 1000 * 1000000;
return nanosleep (&ns_, 0);
#else
return usleep (timeout_ * 1000);
#endif
}
if (!items_) {
errno = EFAULT;
return -1;
}
zmq::clock_t clock;
uint64_t now = 0;
uint64_t end = 0;
#if defined ZMQ_POLL_BASED_ON_POLL
zmq::fast_vector_t<pollfd, ZMQ_POLLITEMS_DFLT> pollfds (nitems_);
// Build pollset for poll () system call.
for (int i = 0; i != nitems_; i++) {
// If the poll item is a 0MQ socket, we poll on the file descriptor
// retrieved by the ZMQ_FD socket option.
if (items_[i].socket) {
size_t zmq_fd_size = sizeof (zmq::fd_t);
if (zmq_getsockopt (items_[i].socket, ZMQ_FD, &pollfds[i].fd,
&zmq_fd_size)
== -1) {
return -1;
}
pollfds[i].events = items_[i].events ? POLLIN : 0;
}
// Else, the poll item is a raw file descriptor. Just convert the
// events to normal POLLIN/POLLOUT for poll ().
else {
pollfds[i].fd = items_[i].fd;
pollfds[i].events =
(items_[i].events & ZMQ_POLLIN ? POLLIN : 0)
| (items_[i].events & ZMQ_POLLOUT ? POLLOUT : 0)
| (items_[i].events & ZMQ_POLLPRI ? POLLPRI : 0);
}
}
#else
// Ensure we do not attempt to select () on more than FD_SETSIZE
// file descriptors.
// TODO since this function is called by a client, we could return errno EINVAL/ENOMEM/... here
zmq_assert (nitems_ <= FD_SETSIZE);
zmq::optimized_fd_set_t pollset_in (nitems_);
FD_ZERO (pollset_in.get ());
zmq::optimized_fd_set_t pollset_out (nitems_);
FD_ZERO (pollset_out.get ());
zmq::optimized_fd_set_t pollset_err (nitems_);
FD_ZERO (pollset_err.get ());
zmq::fd_t maxfd = 0;
// Build the fd_sets for passing to select ().
for (int i = 0; i != nitems_; i++) {
// If the poll item is a 0MQ socket we are interested in input on the
// notification file descriptor retrieved by the ZMQ_FD socket option.
if (items_[i].socket) {
size_t zmq_fd_size = sizeof (zmq::fd_t);
zmq::fd_t notify_fd;
if (zmq_getsockopt (items_[i].socket, ZMQ_FD, ¬ify_fd,
&zmq_fd_size)
== -1)
return -1;
if (items_[i].events) {
FD_SET (notify_fd, pollset_in.get ());
if (maxfd < notify_fd)
maxfd = notify_fd;
}
}
// Else, the poll item is a raw file descriptor. Convert the poll item
// events to the appropriate fd_sets.
else {
if (items_[i].events & ZMQ_POLLIN)
FD_SET (items_[i].fd, pollset_in.get ());
if (items_[i].events & ZMQ_POLLOUT)
FD_SET (items_[i].fd, pollset_out.get ());
if (items_[i].events & ZMQ_POLLERR)
FD_SET (items_[i].fd, pollset_err.get ());
if (maxfd < items_[i].fd)
maxfd = items_[i].fd;
}
}
zmq::optimized_fd_set_t inset (nitems_);
zmq::optimized_fd_set_t outset (nitems_);
zmq::optimized_fd_set_t errset (nitems_);
#endif
bool first_pass = true;
int nevents = 0;
while (true) {
#if defined ZMQ_POLL_BASED_ON_POLL
// Compute the timeout for the subsequent poll.
const zmq::timeout_t timeout =
zmq::compute_timeout (first_pass, timeout_, now, end);
// Wait for events.
{
const int rc = poll (&pollfds[0], nitems_, timeout);
if (rc == -1 && errno == EINTR) {
return -1;
}
errno_assert (rc >= 0);
}
// Check for the events.
for (int i = 0; i != nitems_; i++) {
items_[i].revents = 0;
// The poll item is a 0MQ socket. Retrieve pending events
// using the ZMQ_EVENTS socket option.
if (items_[i].socket) {
size_t zmq_events_size = sizeof (uint32_t);
uint32_t zmq_events;
if (zmq_getsockopt (items_[i].socket, ZMQ_EVENTS, &zmq_events,
&zmq_events_size)
== -1) {
return -1;
}
if ((items_[i].events & ZMQ_POLLOUT)
&& (zmq_events & ZMQ_POLLOUT))
items_[i].revents |= ZMQ_POLLOUT;
if ((items_[i].events & ZMQ_POLLIN)
&& (zmq_events & ZMQ_POLLIN))
items_[i].revents |= ZMQ_POLLIN;
}
// Else, the poll item is a raw file descriptor, simply convert
// the events to zmq_pollitem_t-style format.
else {
if (pollfds[i].revents & POLLIN)
items_[i].revents |= ZMQ_POLLIN;
if (pollfds[i].revents & POLLOUT)
items_[i].revents |= ZMQ_POLLOUT;
if (pollfds[i].revents & POLLPRI)
items_[i].revents |= ZMQ_POLLPRI;
if (pollfds[i].revents & ~(POLLIN | POLLOUT | POLLPRI))
items_[i].revents |= ZMQ_POLLERR;
}
if (items_[i].revents)
nevents++;
}
#else
// Compute the timeout for the subsequent poll.
timeval timeout;
timeval *ptimeout;
if (first_pass) {
timeout.tv_sec = 0;
timeout.tv_usec = 0;
ptimeout = &timeout;
} else if (timeout_ < 0)
ptimeout = NULL;
else {
timeout.tv_sec = static_cast<long> ((end - now) / 1000);
timeout.tv_usec = static_cast<long> ((end - now) % 1000 * 1000);
ptimeout = &timeout;
}
// Wait for events. Ignore interrupts if there's infinite timeout.
while (true) {
memcpy (inset.get (), pollset_in.get (),
zmq::valid_pollset_bytes (*pollset_in.get ()));
memcpy (outset.get (), pollset_out.get (),
zmq::valid_pollset_bytes (*pollset_out.get ()));
memcpy (errset.get (), pollset_err.get (),
zmq::valid_pollset_bytes (*pollset_err.get ()));
#if defined ZMQ_HAVE_WINDOWS
int rc =
select (0, inset.get (), outset.get (), errset.get (), ptimeout);
if (unlikely (rc == SOCKET_ERROR)) {
errno = zmq::wsa_error_to_errno (WSAGetLastError ());
wsa_assert (errno == ENOTSOCK);
return -1;
}
#else
int rc = select (maxfd + 1, inset.get (), outset.get (),
errset.get (), ptimeout);
if (unlikely (rc == -1)) {
errno_assert (errno == EINTR || errno == EBADF);
return -1;
}
#endif
break;
}
// Check for the events.
for (int i = 0; i != nitems_; i++) {
items_[i].revents = 0;
// The poll item is a 0MQ socket. Retrieve pending events
// using the ZMQ_EVENTS socket option.
if (items_[i].socket) {
size_t zmq_events_size = sizeof (uint32_t);
uint32_t zmq_events;
if (zmq_getsockopt (items_[i].socket, ZMQ_EVENTS, &zmq_events,
&zmq_events_size)
== -1)
return -1;
if ((items_[i].events & ZMQ_POLLOUT)
&& (zmq_events & ZMQ_POLLOUT))
items_[i].revents |= ZMQ_POLLOUT;
if ((items_[i].events & ZMQ_POLLIN)
&& (zmq_events & ZMQ_POLLIN))
items_[i].revents |= ZMQ_POLLIN;
}
// Else, the poll item is a raw file descriptor, simply convert
// the events to zmq_pollitem_t-style format.
else {
if (FD_ISSET (items_[i].fd, inset.get ()))
items_[i].revents |= ZMQ_POLLIN;
if (FD_ISSET (items_[i].fd, outset.get ()))
items_[i].revents |= ZMQ_POLLOUT;
if (FD_ISSET (items_[i].fd, errset.get ()))
items_[i].revents |= ZMQ_POLLERR;
}
if (items_[i].revents)
nevents++;
}
#endif
// If timeout is zero, exit immediately whether there are events or not.
if (timeout_ == 0)
break;
// If there are events to return, we can exit immediately.
if (nevents)
break;
// At this point we are meant to wait for events but there are none.
// If timeout is infinite we can just loop until we get some events.
if (timeout_ < 0) {
if (first_pass)
first_pass = false;
continue;
}
// The timeout is finite and there are no events. In the first pass
// we get a timestamp of when the polling have begun. (We assume that
// first pass have taken negligible time). We also compute the time
// when the polling should time out.
if (first_pass) {
now = clock.now_ms ();
end = now + timeout_;
if (now == end)
break;
first_pass = false;
continue;
}
// Find out whether timeout have expired.
now = clock.now_ms ();
if (now >= end)
break;
}
return nevents;
#else
// Exotic platforms that support neither poll() nor select().
errno = ENOTSUP;
return -1;
#endif
}
#ifdef ZMQ_HAVE_PPOLL
// return values of 0 or -1 should be returned from zmq_poll; return value 1 means items passed checks
int zmq_poll_check_items_ (zmq_pollitem_t *items_, int nitems_, long timeout_)
{
if (unlikely (nitems_ < 0)) {
errno = EINVAL;
return -1;
}
if (unlikely (nitems_ == 0)) {
if (timeout_ == 0)
return 0;
#if defined ZMQ_HAVE_WINDOWS
Sleep (timeout_ > 0 ? timeout_ : INFINITE);
return 0;
#elif defined ZMQ_HAVE_VXWORKS
struct timespec ns_;
ns_.tv_sec = timeout_ / 1000;
ns_.tv_nsec = timeout_ % 1000 * 1000000;
return nanosleep (&ns_, 0);
#else
return usleep (timeout_ * 1000);
#endif
}
if (!items_) {
errno = EFAULT;
return -1;
}
return 1;
}
struct zmq_poll_select_fds_t_
{
explicit zmq_poll_select_fds_t_ (int nitems_) :
pollset_in (nitems_),
pollset_out (nitems_),
pollset_err (nitems_),
inset (nitems_),
outset (nitems_),
errset (nitems_),
maxfd (0)
{
FD_ZERO (pollset_in.get ());
FD_ZERO (pollset_out.get ());
FD_ZERO (pollset_err.get ());
}
zmq::optimized_fd_set_t pollset_in;
zmq::optimized_fd_set_t pollset_out;
zmq::optimized_fd_set_t pollset_err;
zmq::optimized_fd_set_t inset;
zmq::optimized_fd_set_t outset;
zmq::optimized_fd_set_t errset;
zmq::fd_t maxfd;
};
zmq_poll_select_fds_t_
zmq_poll_build_select_fds_ (zmq_pollitem_t *items_, int nitems_, int &rc)
{
// Ensure we do not attempt to select () on more than FD_SETSIZE
// file descriptors.
// TODO since this function is called by a client, we could return errno EINVAL/ENOMEM/... here
zmq_assert (nitems_ <= FD_SETSIZE);
zmq_poll_select_fds_t_ fds (nitems_);
// Build the fd_sets for passing to select ().
for (int i = 0; i != nitems_; i++) {
// If the poll item is a 0MQ socket we are interested in input on the
// notification file descriptor retrieved by the ZMQ_FD socket option.
if (items_[i].socket) {
size_t zmq_fd_size = sizeof (zmq::fd_t);
zmq::fd_t notify_fd;
if (zmq_getsockopt (items_[i].socket, ZMQ_FD, ¬ify_fd,
&zmq_fd_size)
== -1) {
rc = -1;
return fds;
}
if (items_[i].events) {
FD_SET (notify_fd, fds.pollset_in.get ());
if (fds.maxfd < notify_fd)
fds.maxfd = notify_fd;
}
}
// Else, the poll item is a raw file descriptor. Convert the poll item
// events to the appropriate fd_sets.
else {
if (items_[i].events & ZMQ_POLLIN)
FD_SET (items_[i].fd, fds.pollset_in.get ());
if (items_[i].events & ZMQ_POLLOUT)
FD_SET (items_[i].fd, fds.pollset_out.get ());
if (items_[i].events & ZMQ_POLLERR)
FD_SET (items_[i].fd, fds.pollset_err.get ());
if (fds.maxfd < items_[i].fd)
fds.maxfd = items_[i].fd;
}
}
rc = 0;
return fds;
}
timeval *zmq_poll_select_set_timeout_ (
long timeout_, bool first_pass, uint64_t now, uint64_t end, timeval &timeout)
{
timeval *ptimeout;
if (first_pass) {
timeout.tv_sec = 0;
timeout.tv_usec = 0;
ptimeout = &timeout;
} else if (timeout_ < 0)
ptimeout = NULL;
else {
timeout.tv_sec = static_cast<long> ((end - now) / 1000);
timeout.tv_usec = static_cast<long> ((end - now) % 1000 * 1000);
ptimeout = &timeout;
}
return ptimeout;
}
timespec *zmq_poll_select_set_timeout_ (
long timeout_, bool first_pass, uint64_t now, uint64_t end, timespec &timeout)
{
timespec *ptimeout;
if (first_pass) {
timeout.tv_sec = 0;
timeout.tv_nsec = 0;
ptimeout = &timeout;
} else if (timeout_ < 0)
ptimeout = NULL;
else {
timeout.tv_sec = static_cast<long> ((end - now) / 1000);
timeout.tv_nsec = static_cast<long> ((end - now) % 1000 * 1000000);
ptimeout = &timeout;
}
return ptimeout;
}
int zmq_poll_select_check_events_ (zmq_pollitem_t *items_,
int nitems_,
zmq_poll_select_fds_t_ &fds,
int &nevents)
{
// Check for the events.
for (int i = 0; i != nitems_; i++) {
items_[i].revents = 0;
// The poll item is a 0MQ socket. Retrieve pending events
// using the ZMQ_EVENTS socket option.
if (items_[i].socket) {
size_t zmq_events_size = sizeof (uint32_t);
uint32_t zmq_events;
if (zmq_getsockopt (items_[i].socket, ZMQ_EVENTS, &zmq_events,
&zmq_events_size)
== -1)
return -1;
if ((items_[i].events & ZMQ_POLLOUT) && (zmq_events & ZMQ_POLLOUT))
items_[i].revents |= ZMQ_POLLOUT;
if ((items_[i].events & ZMQ_POLLIN) && (zmq_events & ZMQ_POLLIN))
items_[i].revents |= ZMQ_POLLIN;
}
// Else, the poll item is a raw file descriptor, simply convert
// the events to zmq_pollitem_t-style format.
else {
if (FD_ISSET (items_[i].fd, fds.inset.get ()))
items_[i].revents |= ZMQ_POLLIN;
if (FD_ISSET (items_[i].fd, fds.outset.get ()))
items_[i].revents |= ZMQ_POLLOUT;
if (FD_ISSET (items_[i].fd, fds.errset.get ()))
items_[i].revents |= ZMQ_POLLERR;
}
if (items_[i].revents)
nevents++;
}
return 0;
}
bool zmq_poll_must_break_loop_ (long timeout_,
int nevents,
bool &first_pass,
zmq::clock_t &clock,
uint64_t &now,
uint64_t &end)
{
// If timeout is zero, exit immediately whether there are events or not.
if (timeout_ == 0)
return true;
// If there are events to return, we can exit immediately.
if (nevents)
return true;
// At this point we are meant to wait for events but there are none.
// If timeout is infinite we can just loop until we get some events.
if (timeout_ < 0) {
if (first_pass)
first_pass = false;
return false;
}
// The timeout is finite and there are no events. In the first pass
// we get a timestamp of when the polling have begun. (We assume that
// first pass have taken negligible time). We also compute the time
// when the polling should time out.
if (first_pass) {
now = clock.now_ms ();
end = now + timeout_;
if (now == end)
return true;
first_pass = false;
return false;
}
// Find out whether timeout have expired.
now = clock.now_ms ();
if (now >= end)
return true;
// finally, in all other cases, we just continue
return false;
}
#endif // ZMQ_HAVE_PPOLL
#if !defined _WIN32
int zmq_ppoll (zmq_pollitem_t *items_,
int nitems_,
long timeout_,
const sigset_t *sigmask_)
#else
// Windows has no sigset_t
int zmq_ppoll (zmq_pollitem_t *items_,
int nitems_,
long timeout_,
const void *sigmask_)
#endif
{
#ifdef ZMQ_HAVE_PPOLL
int rc = zmq_poll_check_items_ (items_, nitems_, timeout_);
if (rc <= 0) {
return rc;
}
zmq::clock_t clock;
uint64_t now = 0;
uint64_t end = 0;
zmq_poll_select_fds_t_ fds =
zmq_poll_build_select_fds_ (items_, nitems_, rc);
if (rc == -1) {
return -1;
}
bool first_pass = true;
int nevents = 0;
while (true) {
// Compute the timeout for the subsequent poll.
timespec timeout;
timespec *ptimeout = zmq_poll_select_set_timeout_ (timeout_, first_pass,
now, end, timeout);
// Wait for events. Ignore interrupts if there's infinite timeout.
while (true) {
memcpy (fds.inset.get (), fds.pollset_in.get (),
zmq::valid_pollset_bytes (*fds.pollset_in.get ()));
memcpy (fds.outset.get (), fds.pollset_out.get (),
zmq::valid_pollset_bytes (*fds.pollset_out.get ()));
memcpy (fds.errset.get (), fds.pollset_err.get (),
zmq::valid_pollset_bytes (*fds.pollset_err.get ()));
int rc =
pselect (fds.maxfd + 1, fds.inset.get (), fds.outset.get (),
fds.errset.get (), ptimeout, sigmask_);
if (unlikely (rc == -1)) {
errno_assert (errno == EINTR || errno == EBADF);
return -1;
}
break;
}
rc = zmq_poll_select_check_events_ (items_, nitems_, fds, nevents);
if (rc < 0) {
return rc;
}
if (zmq_poll_must_break_loop_ (timeout_, nevents, first_pass, clock,
now, end)) {
break;
}
}
return nevents;
#else
errno = ENOTSUP;
return -1;
#endif // ZMQ_HAVE_PPOLL
}
// The poller functionality
void *zmq_poller_new (void)
{
zmq::socket_poller_t *poller = new (std::nothrow) zmq::socket_poller_t;
if (!poller) {
errno = ENOMEM;
}
return poller;
}
int zmq_poller_destroy (void **poller_p_)
{
if (poller_p_) {
const zmq::socket_poller_t *const poller =
static_cast<const zmq::socket_poller_t *> (*poller_p_);
if (poller && poller->check_tag ()) {
delete poller;
*poller_p_ = NULL;
return 0;
}
}
errno = EFAULT;
return -1;
}
static int check_poller (void *const poller_)
{
if (!poller_
|| !(static_cast<zmq::socket_poller_t *> (poller_))->check_tag ()) {
errno = EFAULT;
return -1;
}
return 0;
}
static int check_events (const short events_)
{
if (events_ & ~(ZMQ_POLLIN | ZMQ_POLLOUT | ZMQ_POLLERR | ZMQ_POLLPRI)) {
errno = EINVAL;
return -1;
}
return 0;
}
static int check_poller_registration_args (void *const poller_, void *const s_)
{
if (-1 == check_poller (poller_))
return -1;
if (!s_ || !(static_cast<zmq::socket_base_t *> (s_))->check_tag ()) {
errno = ENOTSOCK;
return -1;
}
return 0;
}
static int check_poller_fd_registration_args (void *const poller_,
const zmq::fd_t fd_)
{
if (-1 == check_poller (poller_))
return -1;
if (fd_ == zmq::retired_fd) {
errno = EBADF;
return -1;
}
return 0;
}
int zmq_poller_size (void *poller_)
{
if (-1 == check_poller (poller_))
return -1;
return (static_cast<zmq::socket_poller_t *> (poller_))->size ();
}
int zmq_poller_add (void *poller_, void *s_, void *user_data_, short events_)
{
if (-1 == check_poller_registration_args (poller_, s_)
|| -1 == check_events (events_))
return -1;
zmq::socket_base_t *socket = static_cast<zmq::socket_base_t *> (s_);
return (static_cast<zmq::socket_poller_t *> (poller_))
->add (socket, user_data_, events_);
}
int zmq_poller_add_fd (void *poller_,
zmq::fd_t fd_,
void *user_data_,
short events_)
{
if (-1 == check_poller_fd_registration_args (poller_, fd_)
|| -1 == check_events (events_))
return -1;
return (static_cast<zmq::socket_poller_t *> (poller_))
->add_fd (fd_, user_data_, events_);
}
int zmq_poller_modify (void *poller_, void *s_, short events_)
{
if (-1 == check_poller_registration_args (poller_, s_)
|| -1 == check_events (events_))
return -1;
const zmq::socket_base_t *const socket =
static_cast<const zmq::socket_base_t *> (s_);
return (static_cast<zmq::socket_poller_t *> (poller_))
->modify (socket, events_);
}
int zmq_poller_modify_fd (void *poller_, zmq::fd_t fd_, short events_)
{
if (-1 == check_poller_fd_registration_args (poller_, fd_)
|| -1 == check_events (events_))
return -1;
return (static_cast<zmq::socket_poller_t *> (poller_))
->modify_fd (fd_, events_);
}
int zmq_poller_remove (void *poller_, void *s_)
{
if (-1 == check_poller_registration_args (poller_, s_))
return -1;
zmq::socket_base_t *socket = static_cast<zmq::socket_base_t *> (s_);
return (static_cast<zmq::socket_poller_t *> (poller_))->remove (socket);
}
int zmq_poller_remove_fd (void *poller_, zmq::fd_t fd_)
{
if (-1 == check_poller_fd_registration_args (poller_, fd_))
return -1;
return (static_cast<zmq::socket_poller_t *> (poller_))->remove_fd (fd_);
}
int zmq_poller_wait (void *poller_, zmq_poller_event_t *event_, long timeout_)
{
const int rc = zmq_poller_wait_all (poller_, event_, 1, timeout_);
if (rc < 0 && event_) {
event_->socket = NULL;
event_->fd = zmq::retired_fd;
event_->user_data = NULL;
event_->events = 0;
}
// wait_all returns number of events, but we return 0 for any success
return rc >= 0 ? 0 : rc;
}
int zmq_poller_wait_all (void *poller_,
zmq_poller_event_t *events_,
int n_events_,
long timeout_)
{
if (-1 == check_poller (poller_))
return -1;
if (!events_) {
errno = EFAULT;
return -1;
}
if (n_events_ < 0) {
errno = EINVAL;
return -1;
}
const int rc =
(static_cast<zmq::socket_poller_t *> (poller_))
->wait (reinterpret_cast<zmq::socket_poller_t::event_t *> (events_),
n_events_, timeout_);
return rc;
}
int zmq_poller_fd (void *poller_, zmq_fd_t *fd_)
{
if (!poller_
|| !(static_cast<zmq::socket_poller_t *> (poller_)->check_tag ())) {
errno = EFAULT;
return -1;
}
return static_cast<zmq::socket_poller_t *> (poller_)->signaler_fd (fd_);
}
// Peer-specific state
int zmq_socket_get_peer_state (void *s_,
const void *routing_id_,
size_t routing_id_size_)
{
const zmq::socket_base_t *const s = as_socket_base_t (s_);
if (!s)
return -1;
return s->get_peer_state (routing_id_, routing_id_size_);
}
// Timers
void *zmq_timers_new (void)
{
zmq::timers_t *timers = new (std::nothrow) zmq::timers_t;
alloc_assert (timers);
return timers;
}
int zmq_timers_destroy (void **timers_p_)
{
void *timers = *timers_p_;
if (!timers || !(static_cast<zmq::timers_t *> (timers))->check_tag ()) {
errno = EFAULT;
return -1;
}
delete (static_cast<zmq::timers_t *> (timers));
*timers_p_ = NULL;
return 0;
}
int zmq_timers_add (void *timers_,
size_t interval_,
zmq_timer_fn handler_,
void *arg_)
{
if (!timers_ || !(static_cast<zmq::timers_t *> (timers_))->check_tag ()) {
errno = EFAULT;
return -1;
}
return (static_cast<zmq::timers_t *> (timers_))
->add (interval_, handler_, arg_);
}
int zmq_timers_cancel (void *timers_, int timer_id_)
{
if (!timers_ || !(static_cast<zmq::timers_t *> (timers_))->check_tag ()) {
errno = EFAULT;
return -1;
}
return (static_cast<zmq::timers_t *> (timers_))->cancel (timer_id_);
}
int zmq_timers_set_interval (void *timers_, int timer_id_, size_t interval_)
{
if (!timers_ || !(static_cast<zmq::timers_t *> (timers_))->check_tag ()) {
errno = EFAULT;
return -1;
}
return (static_cast<zmq::timers_t *> (timers_))
->set_interval (timer_id_, interval_);
}
int zmq_timers_reset (void *timers_, int timer_id_)
{
if (!timers_ || !(static_cast<zmq::timers_t *> (timers_))->check_tag ()) {
errno = EFAULT;
return -1;
}
return (static_cast<zmq::timers_t *> (timers_))->reset (timer_id_);
}
long zmq_timers_timeout (void *timers_)
{
if (!timers_ || !(static_cast<zmq::timers_t *> (timers_))->check_tag ()) {
errno = EFAULT;
return -1;
}
return (static_cast<zmq::timers_t *> (timers_))->timeout ();
}
int zmq_timers_execute (void *timers_)
{
if (!timers_ || !(static_cast<zmq::timers_t *> (timers_))->check_tag ()) {
errno = EFAULT;
return -1;
}
return (static_cast<zmq::timers_t *> (timers_))->execute ();
}
// The proxy functionality
int zmq_proxy (void *frontend_, void *backend_, void *capture_)
{
if (!frontend_ || !backend_) {
errno = EFAULT;
return -1;
}
// Runs zmq::proxy_steerable with a NULL control_.
return zmq::proxy (static_cast<zmq::socket_base_t *> (frontend_),
static_cast<zmq::socket_base_t *> (backend_),
static_cast<zmq::socket_base_t *> (capture_));
}
int zmq_proxy_steerable (void *frontend_,
void *backend_,
void *capture_,
void *control_)
{
if (!frontend_ || !backend_) {
errno = EFAULT;
return -1;
}
return zmq::proxy_steerable (static_cast<zmq::socket_base_t *> (frontend_),
static_cast<zmq::socket_base_t *> (backend_),
static_cast<zmq::socket_base_t *> (capture_),
static_cast<zmq::socket_base_t *> (control_));
}
// The deprecated device functionality
int zmq_device (int /* type */, void *frontend_, void *backend_)
{
return zmq::proxy (static_cast<zmq::socket_base_t *> (frontend_),
static_cast<zmq::socket_base_t *> (backend_), NULL);
}
// Probe library capabilities; for now, reports on transport and security
int zmq_has (const char *capability_)
{
#if defined(ZMQ_HAVE_IPC)
if (strcmp (capability_, zmq::protocol_name::ipc) == 0)
return true;
#endif
#if defined(ZMQ_HAVE_OPENPGM)
if (strcmp (capability_, zmq::protocol_name::pgm) == 0)
return true;
#endif
#if defined(ZMQ_HAVE_TIPC)
if (strcmp (capability_, zmq::protocol_name::tipc) == 0)
return true;
#endif
#if defined(ZMQ_HAVE_NORM)
if (strcmp (capability_, zmq::protocol_name::norm) == 0)
return true;
#endif
#if defined(ZMQ_HAVE_CURVE)
if (strcmp (capability_, "curve") == 0)
return true;
#endif
#if defined(HAVE_LIBGSSAPI_KRB5)
if (strcmp (capability_, "gssapi") == 0)
return true;
#endif
#if defined(ZMQ_HAVE_VMCI)
if (strcmp (capability_, zmq::protocol_name::vmci) == 0)
return true;
#endif
#if defined(ZMQ_BUILD_DRAFT_API)
if (strcmp (capability_, "draft") == 0)
return true;
#endif
#if defined(ZMQ_HAVE_WS)
if (strcmp (capability_, "WS") == 0)
return true;
#endif
#if defined(ZMQ_HAVE_WSS)
if (strcmp (capability_, "WSS") == 0)
return true;
#endif
#if defined(ZMQ_HAVE_VSOCK)
if (strcmp (capability_, zmq::protocol_name::vsock) == 0)
return true;
#endif
// Whatever the application asked for, we don't have
return false;
}
int zmq_socket_monitor_pipes_stats (void *s_)
{
zmq::socket_base_t *s = as_socket_base_t (s_);
if (!s)
return -1;
return s->query_pipes_stats ();
}
↑ V595 The 'items_' pointer was utilized before it was verified against nullptr. Check lines: 845, 873.
↑ V522 There might be dereferencing of a potential null pointer 'events'.
↑ V547 Expression 'nitems_ < 0' is always false.
↑ V1048 The 'rc' variable was assigned the same value.
↑ V524 It is odd that the body of 'zmq_ctx_destroy' function is fully equivalent to the body of 'zmq_term' function.
↑ V524 It is odd that the body of 'zmq_disconnect' function is fully equivalent to the body of 'zmq_unbind' function.
↑ V601 The 'true' value is implicitly cast to the integer type.
↑ V601 The 'true' value is implicitly cast to the integer type.
↑ V601 The 'true' value is implicitly cast to the integer type.
↑ V601 The 'true' value is implicitly cast to the integer type.
↑ V601 The 'true' value is implicitly cast to the integer type.
↑ V601 The 'true' value is implicitly cast to the integer type.
↑ V601 The 'false' value is implicitly cast to the integer type.