global.c

/*
    Copyright (c) 2012-2014 Martin Sustrik  All rights reserved.
    Copyright (c) 2013 GoPivotal, Inc.  All rights reserved.
    Copyright 2016 Garrett D'Amore <garrett@damore.org>

    Permission is hereby granted, free of charge, to any person obtaining a copy
    of this software and associated documentation files (the "Software"),
    to deal in the Software without restriction, including without limitation
    the rights to use, copy, modify, merge, publish, distribute, sublicense,
    and/or sell copies of the Software, and to permit persons to whom
    the Software is furnished to do so, subject to the following conditions:

    The above copyright notice and this permission notice shall be included
    in all copies or substantial portions of the Software.

    THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
    IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
    FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL
    THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
    LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
    FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS
    IN THE SOFTWARE.
*/

#include "../nn.h"
#include "../transport.h"
#include "../protocol.h"

#include "global.h"
#include "sock.h"
#include "ep.h"

#include "../aio/pool.h"
#include "../aio/timer.h"

#include "../utils/err.h"
#include "../utils/alloc.h"
#include "../utils/mutex.h"
#include "../utils/condvar.h"
#include "../utils/once.h"
#include "../utils/list.h"
#include "../utils/cont.h"
#include "../utils/random.h"
#include "../utils/chunk.h"
#include "../utils/msg.h"
#include "../utils/attr.h"

#include "../pubsub.h"
#include "../pipeline.h"

#include <stddef.h>
#include <stdlib.h>
#include <string.h>
#include <time.h>

#if defined NN_HAVE_WINDOWS
#include "../utils/win.h"
#else
#include <unistd.h>
#endif

/*  Max number of concurrent SP sockets. Configureable at build time */
#ifndef NN_MAX_SOCKETS
#define NN_MAX_SOCKETS 512
#endif

/*  To save some space, list of unused socket slots uses uint16_t integers to
    refer to individual sockets. If there's a need to more that 0x10000 sockets,
    the type should be changed to uint32_t or int. */
CT_ASSERT (NN_MAX_SOCKETS <= 0x10000);

#define NN_CTX_FLAG_TERMED 1
#define NN_CTX_FLAG_TERMING 2
#define NN_CTX_FLAG_TERM (NN_CTX_FLAG_TERMED | NN_CTX_FLAG_TERMING)

#define NN_GLOBAL_SRC_STAT_TIMER 1

#define NN_GLOBAL_STATE_IDLE           1
#define NN_GLOBAL_STATE_ACTIVE         2
#define NN_GLOBAL_STATE_STOPPING_TIMER 3

/*  We could put these in an external header file, but there really is
    need to.  We are the only thing that needs them. */
extern struct nn_socktype nn_pair_socktype;
extern struct nn_socktype nn_xpair_socktype;
extern struct nn_socktype nn_pub_socktype;
extern struct nn_socktype nn_sub_socktype;
extern struct nn_socktype nn_xpub_socktype;
extern struct nn_socktype nn_xsub_socktype;
extern struct nn_socktype nn_rep_socktype;
extern struct nn_socktype nn_req_socktype;
extern struct nn_socktype nn_xrep_socktype;
extern struct nn_socktype nn_xreq_socktype;
extern struct nn_socktype nn_push_socktype;
extern struct nn_socktype nn_xpush_socktype;
extern struct nn_socktype nn_pull_socktype;
extern struct nn_socktype nn_xpull_socktype;
extern struct nn_socktype nn_respondent_socktype;
extern struct nn_socktype nn_surveyor_socktype;
extern struct nn_socktype nn_xrespondent_socktype;
extern struct nn_socktype nn_xsurveyor_socktype;
extern struct nn_socktype nn_bus_socktype;
extern struct nn_socktype nn_xbus_socktype;

/*  Array of known socket types. */
const struct nn_socktype *nn_socktypes[] = {
    &nn_pair_socktype,
    &nn_xpair_socktype,
    &nn_pub_socktype,
    &nn_sub_socktype,
    &nn_xpub_socktype,
    &nn_xsub_socktype,
    &nn_rep_socktype,
    &nn_req_socktype,
    &nn_xrep_socktype,
    &nn_xreq_socktype,
    &nn_push_socktype,
    &nn_xpush_socktype,
    &nn_pull_socktype,
    &nn_xpull_socktype,
    &nn_respondent_socktype,
    &nn_surveyor_socktype,
    &nn_xrespondent_socktype,
    &nn_xsurveyor_socktype,
    &nn_bus_socktype,
    &nn_xbus_socktype,
    NULL,
};

/* As with protocols, we could have these in a header file, but we are the
   only consumer, so just declare them inline. */

extern struct nn_transport nn_inproc;
extern struct nn_transport nn_ipc;
extern struct nn_transport nn_tcp;
extern struct nn_transport nn_ws;

const struct nn_transport *nn_transports[] = {
    &nn_inproc,
    &nn_ipc,
    &nn_tcp,
    &nn_ws,
    NULL,
};

struct nn_global {

    /*  The global table of existing sockets. The descriptor representing
        the socket is the index to this table. This pointer is also used to
        find out whether context is initialised. If it is NULL, context is
        uninitialised. */
    struct nn_sock **socks;

    /*  Stack of unused file descriptors. */
    uint16_t *unused;

    /*  Number of actual open sockets in the socket table. */
    size_t nsocks;

    /*  Combination of the flags listed above. */
    int flags;

    /*  Pool of worker threads. */
    struct nn_pool pool;

    /*  Timer and other machinery for submitting statistics  */
    int state;

    int print_errors;

    nn_mutex_t lock;
    nn_condvar_t cond;
};

/*  Singleton object containing the global state of the library. */
static struct nn_global self;
static nn_once_t once = NN_ONCE_INITIALIZER;


/*  Context creation- and termination-related private functions. */
static void nn_global_init (void);
static void nn_global_term (void);

/*  Private function that unifies nn_bind and nn_connect functionality.
    It returns the ID of the newly created endpoint. */
static int nn_global_create_ep (struct nn_sock *, const char *addr, int bind);

/*  Private socket creator which doesn't initialize global state and
    does no locking by itself */
static int nn_global_create_socket (int domain, int protocol);

/*  Socket holds. */
static int nn_global_hold_socket (struct nn_sock **sockp, int s);
static int nn_global_hold_socket_locked (struct nn_sock **sockp, int s);
static void nn_global_rele_socket(struct nn_sock *);

int nn_errno (void)
{
    return nn_err_errno ();
}

const char *nn_strerror (int errnum)
{
    return nn_err_strerror (errnum);
}

static void nn_global_init (void)
{
    int i;
    char *envvar;

#if defined NN_HAVE_WINDOWS
    int rc;
    WSADATA data;
#endif
    const struct nn_transport *tp;

    /*  Check whether the library was already initialised. If so, do nothing. */
    if (self.socks)
        return;

    /*  On Windows, initialise the socket library. */
#if defined NN_HAVE_WINDOWS
    rc = WSAStartup (MAKEWORD (2, 2), &data);
    nn_assert (rc == 0);
    nn_assert (LOBYTE (data.wVersion) == 2 &&
        HIBYTE (data.wVersion) == 2);
#endif

    /*  Initialise the memory allocation subsystem. */
    nn_alloc_init ();

    /*  Seed the pseudo-random number generator. */
    nn_random_seed ();

    /*  Allocate the global table of SP sockets. */
    self.socks = nn_alloc ((sizeof (struct nn_sock*) * NN_MAX_SOCKETS) +
        (sizeof (uint16_t) * NN_MAX_SOCKETS), "socket table");
    alloc_assert (self.socks);
    for (i = 0; i != NN_MAX_SOCKETS; ++i)
        self.socks [i] = NULL;
    self.nsocks = 0;
    self.flags = 0;

    /*  Print connection and accepting errors to the stderr  */
    envvar = getenv("NN_PRINT_ERRORS");
    /*  any non-empty string is true */
    self.print_errors = envvar && *envvar;

    /*  Allocate the stack of unused file descriptors. */
    self.unused = (uint16_t*) (self.socks + NN_MAX_SOCKETS);
    alloc_assert (self.unused);
    for (i = 0; i != NN_MAX_SOCKETS; ++i)
        self.unused [i] = NN_MAX_SOCKETS - i - 1;

    /*  Initialize transports if needed. */
    for (i = 0; (tp = nn_transports[i]) != NULL; i++) {
        if (tp->init != NULL) {
            tp->init ();
        }
    }

    /*  Start the worker threads. */
    nn_pool_init (&self.pool);
}

static void nn_global_term (void)
{
#if defined NN_HAVE_WINDOWS
    int rc;
#endif
    const struct nn_transport *tp;
    int i;

    /*  If there are no sockets remaining, uninitialise the global context. */
    nn_assert (self.socks);
    if (self.nsocks > 0)
        return;

    /*  Shut down the worker threads. */
    nn_pool_term (&self.pool);

    /*  Ask all the transport to deallocate their global resources. */
    for (i = 0; (tp = nn_transports[i]) != NULL; i++) {
        if (tp->term)
            tp->term ();
    }

    /*  Final deallocation of the nn_global object itself. */
    nn_free (self.socks);

    /*  This marks the global state as uninitialised. */
    self.socks = NULL;

    /*  Shut down the memory allocation subsystem. */
    nn_alloc_term ();

    /*  On Windows, uninitialise the socket library. */
#if defined NN_HAVE_WINDOWS
    rc = WSACleanup ();
    nn_assert (rc == 0);
#endif
}

void nn_term (void)
{
    int i;

    nn_mutex_lock (&self.lock);
    self.flags |= NN_CTX_FLAG_TERMING;
    nn_mutex_unlock (&self.lock);

    /* Make sure we really close resources, this will cause global
       resources to be freed too when the last socket is closed. */
    for (i = 0; i < NN_MAX_SOCKETS; i++) {
        (void) nn_close (i);
    }

    nn_mutex_lock (&self.lock);
    self.flags |= NN_CTX_FLAG_TERMED;
    self.flags &= ~NN_CTX_FLAG_TERMING;
    nn_condvar_broadcast(&self.cond);
    nn_mutex_unlock (&self.lock);
}

void nn_init (void)
{
    nn_mutex_lock (&self.lock);
    /*  Wait for any in progress term to complete. */
    while (self.flags & NN_CTX_FLAG_TERMING) {
        nn_condvar_wait (&self.cond, &self.lock, -1);
    }
    self.flags &= ~NN_CTX_FLAG_TERMED;
    nn_mutex_unlock (&self.lock);
}

void *nn_allocmsg (size_t size, int type)
{
    int rc;
    void *result;

    rc = nn_chunk_alloc (size, type, &result);
    if (rc == 0)
        return result;
    errno = -rc;
    return NULL;
}

void *nn_reallocmsg (void *msg, size_t size)
{
    int rc;

    rc = nn_chunk_realloc (size, &msg);
    if (rc == 0)
        return msg;
    errno = -rc;
    return NULL;
}

int nn_freemsg (void *msg)
{
    nn_chunk_free (msg);
    return 0;
}

struct nn_cmsghdr *nn_cmsg_nxthdr_ (const struct nn_msghdr *mhdr,
    const struct nn_cmsghdr *cmsg)
{
    char *data;
    size_t sz;
    struct nn_cmsghdr *next;
    size_t headsz;

    /*  Early return if no message is provided. */
    if (nn_slow (mhdr == NULL))
        return NULL;

    /*  Get the actual data. */
    if (mhdr->msg_controllen == NN_MSG) {
        data = *((void**) mhdr->msg_control);
        sz = nn_chunk_size (data);
    }
    else {
        data = (char*) mhdr->msg_control;
        sz = mhdr->msg_controllen;
    }

    /*  Ancillary data allocation was not even large enough for one element. */
    if (nn_slow (sz < NN_CMSG_SPACE (0)))
        return NULL;

    /*  If cmsg is set to NULL we are going to return first property.
        Otherwise move to the next property. */
    if (!cmsg)
        next = (struct nn_cmsghdr*) data;
    else
        next = (struct nn_cmsghdr*)
            (((char*) cmsg) + NN_CMSG_ALIGN_ (cmsg->cmsg_len));

    /*  If there's no space for next property, treat it as the end
        of the property list. */
    headsz = ((char*) next) - data;
    if (headsz + NN_CMSG_SPACE (0) > sz ||
          headsz + NN_CMSG_ALIGN_ (next->cmsg_len) > sz)
        return NULL;
    
    /*  Success. */
    return next;
}

int nn_global_create_socket (int domain, int protocol)
{
    int rc;
    int s;
    int i;
    const struct nn_socktype *socktype;
    struct nn_sock *sock;

    /* The function is called with lock held */

    /*  Only AF_SP and AF_SP_RAW domains are supported. */
    if (domain != AF_SP && domain != AF_SP_RAW) {
        return -EAFNOSUPPORT;
    }

    /*  If socket limit was reached, report error. */
    if (self.nsocks >= NN_MAX_SOCKETS) {
        return -EMFILE;
    }

    /*  Find an empty socket slot. */
    s = self.unused [NN_MAX_SOCKETS - self.nsocks - 1];

    /*  Find the appropriate socket type. */
    for (i = 0; (socktype = nn_socktypes[i]) != NULL; i++) {
        if (socktype->domain == domain && socktype->protocol == protocol) {

            /*  Instantiate the socket. */
            if ((sock = nn_alloc (sizeof (struct nn_sock), "sock")) == NULL)
                return -ENOMEM;
            rc = nn_sock_init (sock, socktype, s);
            if (rc < 0)
                return rc;

            /*  Adjust the global socket table. */
            self.socks [s] = sock;
            ++self.nsocks;
            return s;
        }
    }
    /*  Specified socket type wasn't found. */
    return -EINVAL;
}

static void nn_lib_init(void)
{
    /*  This function is executed once to initialize global locks. */
    nn_mutex_init (&self.lock);
    nn_condvar_init (&self.cond);
}

int nn_socket (int domain, int protocol)
{
    int rc;

    nn_do_once (&once, nn_lib_init);

    nn_mutex_lock (&self.lock);

    /*  If nn_term() was already called, return ETERM. */
    if (nn_slow (self.flags & NN_CTX_FLAG_TERM)) {
        nn_mutex_unlock (&self.lock);
        errno = ETERM;
        return -1;
    }

    /*  Make sure that global state is initialised. */
    nn_global_init ();

    rc = nn_global_create_socket (domain, protocol);

    if (rc < 0) {
        nn_global_term ();
        nn_mutex_unlock (&self.lock);
        errno = -rc;
        return -1;
    }

    nn_mutex_unlock (&self.lock);

    return rc;
}

int nn_close (int s)
{
    int rc;
    struct nn_sock *sock;

    nn_mutex_lock (&self.lock);
    rc = nn_global_hold_socket_locked (&sock, s);
    if (nn_slow (rc < 0)) {
        nn_mutex_unlock (&self.lock);
        errno = -rc;
        return -1;
    }

    /*  Start the shutdown process on the socket.  This will cause
        all other socket users, as well as endpoints, to begin cleaning up.
        This is done with the lock held to ensure that two instances
        of nn_close can't access the same socket. */
    nn_sock_stop (sock);

    /*  We have to drop both the hold we just acquired, as well as
        the original hold, in order for nn_sock_term to complete. */
    nn_sock_rele (sock);
    nn_sock_rele (sock);
    nn_mutex_unlock (&self.lock);

    /*  Now clean up.  The termination routine below will block until
        all other consumers of the socket have dropped their holds, and
        all endpoints have cleanly exited. */
    rc = nn_sock_term (sock);
    if (nn_slow (rc == -EINTR)) {
        nn_global_rele_socket (sock);
        errno = EINTR;
        return -1;
    }

    /*  Remove the socket from the socket table, add it to unused socket
        table. */
    nn_mutex_lock (&self.lock);
    self.socks [s] = NULL;
    self.unused [NN_MAX_SOCKETS - self.nsocks] = s;
    --self.nsocks;
    nn_free (sock);

    /*  Destroy the global context if there's no socket remaining. */
    nn_global_term ();

    nn_mutex_unlock (&self.lock);

    return 0;
}

int nn_setsockopt (int s, int level, int option, const void *optval,
    size_t optvallen)
{
    int rc;
    struct nn_sock *sock;

    rc = nn_global_hold_socket (&sock, s);
    if (nn_slow (rc < 0)) {
        errno = -rc;
        return -1;
    }

    if (nn_slow (!optval && optvallen)) {
        rc = -EFAULT;
        goto fail;
    }

    rc = nn_sock_setopt (sock, level, option, optval, optvallen);
    if (nn_slow (rc < 0))
        goto fail;
    errnum_assert (rc == 0, -rc);
    nn_global_rele_socket (sock);
    return 0;

fail:
    nn_global_rele_socket (sock);
    errno = -rc;
    return -1;
}

int nn_getsockopt (int s, int level, int option, void *optval,
    size_t *optvallen)
{
    int rc;
    struct nn_sock *sock;

    rc = nn_global_hold_socket (&sock, s);
    if (nn_slow (rc < 0)) {
        errno = -rc;
        return -1;
    }

    if (nn_slow (!optval && optvallen)) {
        rc = -EFAULT;
        goto fail;
    }

    rc = nn_sock_getopt (sock, level, option, optval, optvallen);
    if (nn_slow (rc < 0))
        goto fail;
    errnum_assert (rc == 0, -rc);
    nn_global_rele_socket (sock);
    return 0;

fail:
    nn_global_rele_socket (sock);
    errno = -rc;
    return -1;
}

int nn_bind (int s, const char *addr)
{
    int rc;
    struct nn_sock *sock;

    rc = nn_global_hold_socket (&sock, s);
    if (rc < 0) {
        errno = -rc;
        return -1;
    }

    rc = nn_global_create_ep (sock, addr, 1);
    if (nn_slow (rc < 0)) {
        nn_global_rele_socket (sock);
        errno = -rc;
        return -1;
    }

    nn_global_rele_socket (sock);
    return rc;
}

int nn_connect (int s, const char *addr)
{
    int rc;
    struct nn_sock *sock;

    rc = nn_global_hold_socket (&sock, s);
    if (nn_slow (rc < 0)) {
        errno = -rc;
        return -1;
    }

    rc = nn_global_create_ep (sock, addr, 0);
    if (rc < 0) {
        nn_global_rele_socket (sock);
        errno = -rc;
        return -1;
    }

    nn_global_rele_socket (sock);
    return rc;
}

int nn_shutdown (int s, int how)
{
    int rc;
    struct nn_sock *sock;

    rc = nn_global_hold_socket (&sock, s);
    if (nn_slow (rc < 0)) {
        errno = -rc;
        return -1;
    }

    rc = nn_sock_rm_ep (sock, how);
    if (nn_slow (rc < 0)) {
        nn_global_rele_socket (sock);
        errno = -rc;
        return -1;
    }
    nn_assert (rc == 0);

    nn_global_rele_socket (sock);
    return 0;
}

int nn_send (int s, const void *buf, size_t len, int flags)
{
    struct nn_iovec iov;
    struct nn_msghdr hdr;

    iov.iov_base = (void*) buf;
    iov.iov_len = len;

    hdr.msg_iov = &iov;
    hdr.msg_iovlen = 1;
    hdr.msg_control = NULL;
    hdr.msg_controllen = 0;

    return nn_sendmsg (s, &hdr, flags);
}

int nn_recv (int s, void *buf, size_t len, int flags)
{
    struct nn_iovec iov;
    struct nn_msghdr hdr;

    iov.iov_base = buf;
    iov.iov_len = len;

    hdr.msg_iov = &iov;
    hdr.msg_iovlen = 1;
    hdr.msg_control = NULL;
    hdr.msg_controllen = 0;

    return nn_recvmsg (s, &hdr, flags);
}

int nn_sendmsg (int s, const struct nn_msghdr *msghdr, int flags)
{
    int rc;
    size_t sz;
    size_t spsz;
    int i;
    struct nn_iovec *iov;
    struct nn_msg msg;
    void *chunk;
    int nnmsg;
    struct nn_cmsghdr *cmsg;
    struct nn_sock *sock;

    rc = nn_global_hold_socket (&sock, s);
    if (nn_slow (rc < 0)) {
        errno = -rc;
        return -1;
    }

    if (nn_slow (!msghdr)) {
        rc = -EINVAL;
        goto fail;
    }

    if (nn_slow (msghdr->msg_iovlen < 0)) {
        rc = -EMSGSIZE;
        goto fail;
    }

    if (msghdr->msg_iovlen == 1 && msghdr->msg_iov [0].iov_len == NN_MSG) {
        chunk = *(void**) msghdr->msg_iov [0].iov_base;
        if (nn_slow (chunk == NULL)) {
            rc = -EFAULT;
            goto fail;
        }
        sz = nn_chunk_size (chunk);
        nn_msg_init_chunk (&msg, chunk);
        nnmsg = 1;
    }
    else {

        /*  Compute the total size of the message. */
        sz = 0;
        for (i = 0; i != msghdr->msg_iovlen; ++i) {
            iov = &msghdr->msg_iov [i];
            if (nn_slow (iov->iov_len == NN_MSG)) {
               rc = -EINVAL;
               goto fail;
            }
            if (nn_slow (!iov->iov_base && iov->iov_len)) {
                rc = -EFAULT;
                goto fail;
            }
            if (nn_slow (sz + iov->iov_len < sz)) {
                rc = -EINVAL;
                goto fail;
            }
            sz += iov->iov_len;
        }

        /*  Create a message object from the supplied scatter array. */
        nn_msg_init (&msg, sz);
        sz = 0;
        for (i = 0; i != msghdr->msg_iovlen; ++i) {
            iov = &msghdr->msg_iov [i];
            memcpy (((uint8_t*) nn_chunkref_data (&msg.body)) + sz,
                iov->iov_base, iov->iov_len);
            sz += iov->iov_len;
        }

        nnmsg = 0;
    }

    /*  Add ancillary data to the message. */
    if (msghdr->msg_control) {

        /*  Copy all headers. */
        /*  TODO: SP_HDR should not be copied here! */
        if (msghdr->msg_controllen == NN_MSG) {
            chunk = *((void**) msghdr->msg_control);
            nn_chunkref_term (&msg.hdrs);
            nn_chunkref_init_chunk (&msg.hdrs, chunk);
        }
        else {
            nn_chunkref_term (&msg.hdrs);
            nn_chunkref_init (&msg.hdrs, msghdr->msg_controllen);
            memcpy (nn_chunkref_data (&msg.hdrs),
                msghdr->msg_control, msghdr->msg_controllen);
        }

        /* Search for SP_HDR property. */
        cmsg = NN_CMSG_FIRSTHDR (msghdr);
        while (cmsg) {
            if (cmsg->cmsg_level == PROTO_SP && cmsg->cmsg_type == SP_HDR) {
                unsigned char *ptr = NN_CMSG_DATA (cmsg);
                size_t clen = cmsg->cmsg_len - NN_CMSG_SPACE (0);
                if (clen > sizeof (size_t)) {
                    spsz = *(size_t *)(void *)ptr;
                    if (spsz <= (clen - sizeof (size_t))) {
                        /*  Copy body of SP_HDR property into 'sphdr'. */
                        nn_chunkref_term (&msg.sphdr);
                        nn_chunkref_init (&msg.sphdr, spsz);
                         memcpy (nn_chunkref_data (&msg.sphdr),
                             ptr + sizeof (size_t), spsz);
                    }
                }
                break;
            }
            cmsg = NN_CMSG_NXTHDR (msghdr, cmsg);
        }
    }

    /*  Send it further down the stack. */
    rc = nn_sock_send (sock, &msg, flags);
    if (nn_slow (rc < 0)) {

        /*  If we are dealing with user-supplied buffer, detach it from
            the message object. */
        if (nnmsg)
            nn_chunkref_init (&msg.body, 0);

        nn_msg_term (&msg);
        goto fail;
    }

    /*  Adjust the statistics. */
    nn_sock_stat_increment (sock, NN_STAT_MESSAGES_SENT, 1);
    nn_sock_stat_increment (sock, NN_STAT_BYTES_SENT, sz);

    nn_global_rele_socket (sock);

    return (int) sz;

fail:
    nn_global_rele_socket (sock);

    errno = -rc;
    return -1;
}

int nn_recvmsg (int s, struct nn_msghdr *msghdr, int flags)
{
    int rc;
    struct nn_msg msg;
    uint8_t *data;
    size_t sz;
    int i;
    struct nn_iovec *iov;
    void *chunk;
    size_t hdrssz;
    void *ctrl;
    size_t ctrlsz;
    size_t spsz;
    size_t sptotalsz;
    struct nn_cmsghdr *chdr;
    struct nn_sock *sock;

    rc = nn_global_hold_socket (&sock, s);
    if (nn_slow (rc < 0)) {
        errno = -rc;
        return -1;
    }

    if (nn_slow (!msghdr)) {
        rc = -EINVAL;
        goto fail;
    }

    if (nn_slow (msghdr->msg_iovlen < 0)) {
        rc = -EMSGSIZE;
        goto fail;
    }

    /*  Get a message. */
    rc = nn_sock_recv (sock, &msg, flags);
    if (nn_slow (rc < 0)) {
        goto fail;
    }

    if (msghdr->msg_iovlen == 1 && msghdr->msg_iov [0].iov_len == NN_MSG) {
        chunk = nn_chunkref_getchunk (&msg.body);
        *(void**) (msghdr->msg_iov [0].iov_base) = chunk;
        sz = nn_chunk_size (chunk);
    }
    else {

        /*  Copy the message content into the supplied gather array. */
        data = nn_chunkref_data (&msg.body);
        sz = nn_chunkref_size (&msg.body);
        for (i = 0; i != msghdr->msg_iovlen; ++i) {
            iov = &msghdr->msg_iov [i];
            if (nn_slow (iov->iov_len == NN_MSG)) {
                nn_msg_term (&msg);
                rc = -EINVAL;
                goto fail;
            }
            if (iov->iov_len > sz) {
                memcpy (iov->iov_base, data, sz);
                break;
            }
            memcpy (iov->iov_base, data, iov->iov_len);
            data += iov->iov_len;
            sz -= iov->iov_len;
        }
        sz = nn_chunkref_size (&msg.body);
    }

    /*  Retrieve the ancillary data from the message. */
    if (msghdr->msg_control) {

        spsz = nn_chunkref_size (&msg.sphdr);
        sptotalsz = NN_CMSG_SPACE (spsz+sizeof (size_t));
        ctrlsz = sptotalsz + nn_chunkref_size (&msg.hdrs);

        if (msghdr->msg_controllen == NN_MSG) {

            /* Allocate the buffer. */
            rc = nn_chunk_alloc (ctrlsz, 0, &ctrl);
            errnum_assert (rc == 0, -rc);

            /* Set output parameters. */
            *((void**) msghdr->msg_control) = ctrl;
        }
        else {

            /* Just use the buffer supplied by the user. */
            ctrl = msghdr->msg_control;
            ctrlsz = msghdr->msg_controllen;
        }

        /* If SP header alone won't fit into the buffer, return no ancillary
           properties. */
        if (ctrlsz >= sptotalsz) {
            char *ptr;

            /*  Fill in SP_HDR ancillary property. */
            chdr = (struct nn_cmsghdr*) ctrl;
            chdr->cmsg_len = sptotalsz;
            chdr->cmsg_level = PROTO_SP;
            chdr->cmsg_type = SP_HDR;
            ptr = (void *)chdr;
            ptr += sizeof (*chdr);
            *(size_t *)(void *)ptr = spsz;
            ptr += sizeof (size_t);
            memcpy (ptr, nn_chunkref_data (&msg.sphdr), spsz);

            /*  Fill in as many remaining properties as possible.
                Truncate the trailing properties if necessary. */
            hdrssz = nn_chunkref_size (&msg.hdrs);
            if (hdrssz > ctrlsz - sptotalsz)
                hdrssz = ctrlsz - sptotalsz;
            memcpy (((char*) ctrl) + sptotalsz,
                nn_chunkref_data (&msg.hdrs), hdrssz);
        }
    }

    nn_msg_term (&msg);

    /*  Adjust the statistics. */
    nn_sock_stat_increment (sock, NN_STAT_MESSAGES_RECEIVED, 1);
    nn_sock_stat_increment (sock, NN_STAT_BYTES_RECEIVED, sz);

    nn_global_rele_socket (sock);

    return (int) sz;

fail:
    nn_global_rele_socket (sock);

    errno = -rc;
    return -1;
}

uint64_t nn_get_statistic (int s, int statistic)
{
    int rc;
    struct nn_sock *sock;
    uint64_t val;

    rc = nn_global_hold_socket (&sock, s);
    if (nn_slow (rc < 0)) {
        errno = -rc;
        return (uint64_t)-1;
    }

    switch (statistic) {
    case NN_STAT_ESTABLISHED_CONNECTIONS:
        val = sock->statistics.established_connections;
        break;
    case NN_STAT_ACCEPTED_CONNECTIONS:
        val = sock->statistics.accepted_connections;
        break;
    case NN_STAT_DROPPED_CONNECTIONS:
        val = sock->statistics.dropped_connections;
        break;
    case NN_STAT_BROKEN_CONNECTIONS:
        val = sock->statistics.broken_connections;
        break;
    case NN_STAT_CONNECT_ERRORS:
        val = sock->statistics.connect_errors;
        break;
    case NN_STAT_BIND_ERRORS:
        val = sock->statistics.bind_errors;
        break;
    case NN_STAT_ACCEPT_ERRORS:
        val = sock->statistics.bind_errors;
        break;
    case NN_STAT_MESSAGES_SENT:
        val = sock->statistics.messages_sent;
        break;
    case NN_STAT_MESSAGES_RECEIVED:
        val = sock->statistics.messages_received;
        break;
    case NN_STAT_BYTES_SENT:
        val = sock->statistics.bytes_sent;
        break;
    case NN_STAT_BYTES_RECEIVED:
        val = sock->statistics.bytes_received;
        break;
    case NN_STAT_CURRENT_CONNECTIONS:
        val = sock->statistics.current_connections;
        break;
    case NN_STAT_INPROGRESS_CONNECTIONS:
        val = sock->statistics.inprogress_connections;
        break;
    case NN_STAT_CURRENT_SND_PRIORITY:
        val = sock->statistics.current_snd_priority;
        break;
    case NN_STAT_CURRENT_EP_ERRORS:
        val = sock->statistics.current_ep_errors;
        break;
    default:
        val = (uint64_t)-1;
        errno = EINVAL;
        break;
    }

    nn_global_rele_socket (sock);
    return val;
}

static int nn_global_create_ep (struct nn_sock *sock, const char *addr,
    int bind)
{
    int rc;
    const char *proto;
    const char *delim;
    size_t protosz;
    const struct nn_transport *tp;
    int i;

    /*  Check whether address is valid. */
    if (!addr)
        return -EINVAL;
    if (strlen (addr) >= NN_SOCKADDR_MAX)
        return -ENAMETOOLONG;

    /*  Separate the protocol and the actual address. */
    proto = addr;
    delim = strchr (addr, ':');
    if (!delim)
        return -EINVAL;
    if (delim [1] != '/' || delim [2] != '/')
        return -EINVAL;
    protosz = delim - addr;
    addr += protosz + 3;

    /*  Find the specified protocol. */
    tp = NULL;
    for (i = 0; ((tp = nn_transports[i]) != NULL); i++) {
        if (strlen (tp->name) == protosz &&
              memcmp (tp->name, proto, protosz) == 0)
            break;
    }

    /*  The protocol specified doesn't match any known protocol. */
    if (tp == NULL) {
        return -EPROTONOSUPPORT;
    }

    /*  Ask the socket to create the endpoint. */
    rc = nn_sock_add_ep (sock, tp, bind, addr);
    return rc;
}

const struct nn_transport *nn_global_transport (int id)
{
    const struct nn_transport *tp;
    int i;

    for (i = 0; (tp = nn_transports[i]) != NULL; i++) {
        if (tp->id == id)
            return tp;
    }
    return NULL;
}

struct nn_pool *nn_global_getpool ()
{
    return &self.pool;
}

int nn_global_print_errors ()
{
    return self.print_errors;
}

/*  Get the socket structure for a socket id.  This must be called under
    the global lock (self.lock.)  The socket itself will not be freed
    while the hold is active. */
int nn_global_hold_socket_locked(struct nn_sock **sockp, int s)
{
    struct nn_sock *sock;

    if (nn_slow (s < 0 || s >= NN_MAX_SOCKETS || self.socks == NULL))
        return -EBADF;

    sock = self.socks[s];
    if (nn_slow (sock == NULL)) {
        return -EBADF;
    }

    if (nn_slow (nn_sock_hold (sock) != 0)) {
        return -EBADF;
    }
    *sockp = sock;
    return 0;
}

int nn_global_hold_socket(struct nn_sock **sockp, int s)
{
    int rc;
    nn_mutex_lock(&self.lock);
    rc = nn_global_hold_socket_locked(sockp, s);
    nn_mutex_unlock(&self.lock);
    return rc;
}

void nn_global_rele_socket(struct nn_sock *sock)
{
    nn_mutex_lock(&self.lock);
    nn_sock_rele(sock);
    nn_mutex_unlock(&self.lock);
}