自学教程：C++ zmq_assert函数代码示例

zmq::server_t::~server_t (){        zmq_assert (outpipes.empty ());}

bool zmq::trie_t::rm (unsigned char *prefix_, size_t size_){    //  TODO: Shouldn't an error be reported if the key does not exist?    if (!size_) {        if (!refcnt)            return false;        refcnt--;        return refcnt == 0;    }    unsigned char c = *prefix_;    if (!count || c < min || c >= min + count)        return false;    trie_t *next_node =        count == 1 ? next.node : next.table [c - min];    if (!next_node)        return false;    bool ret = next_node->rm (prefix_ + 1, size_ - 1);    //  Prune redundant nodes    if (next_node->is_redundant ()) {        delete next_node;        zmq_assert (count > 0);        if (count == 1) {            //  The just pruned node is was the only live node            next.node = 0;            count = 0;            --live_nodes;            zmq_assert (live_nodes == 0);        }        else {            next.table [c - min] = 0;            zmq_assert (live_nodes > 1);            --live_nodes;            //  Compact the table if possible            if (live_nodes == 1) {                //  We can switch to using the more compact single-node                //  representation since the table only contains one live node                trie_t *node = 0;                //  Since we always compact the table the pruned node must                //  either be the left-most or right-most ptr in the node                //  table                if (c == min) {                    //  The pruned node is the left-most node ptr in the                    //  node table => keep the right-most node                    node = next.table [count - 1];                    min += count - 1;                }                else                if (c == min + count - 1) {                    //  The pruned node is the right-most node ptr in the                    //  node table => keep the left-most node                    node = next.table [0];                }                zmq_assert (node);                free (next.table);                next.node = node;                count = 1;            }            else            if (c == min) {                //  We can compact the table "from the left".                //  Find the left-most non-null node ptr, which we'll use as                //  our new min                unsigned char new_min = min;                for (unsigned short i = 1; i < count; ++i) {                    if (next.table [i]) {                        new_min = i + min;                        break;                    }                }                zmq_assert (new_min != min);                trie_t **old_table = next.table;                zmq_assert (new_min > min);                zmq_assert (count > new_min - min);                count = count - (new_min - min);                next.table = (trie_t**) malloc (sizeof (trie_t*) * count);                alloc_assert (next.table);                memmove (next.table, old_table + (new_min - min),                        sizeof (trie_t*) * count);                free (old_table);                min = new_min;            }            else            if (c == min + count - 1) {                //  We can compact the table "from the right".                //  Find the right-most non-null node ptr, which we'll use to                //  determine the new table size                unsigned short new_count = count;                for (unsigned short i = 1; i < count; ++i) {                    if (next.table [count - 1 - i]) {                        new_count = count - i;//.........这里部分代码省略.........

int zmq::resolve_ip_interface (sockaddr_storage* addr_, socklen_t *addr_len_,    char const *interface_){    //  Find the ':' at end that separates NIC name from service.    const char *delimiter = strrchr (interface_, ':');    if (!delimiter) {        errno = EINVAL;        return -1;    }    //  Separate the name/port.    std::string iface (interface_, delimiter - interface_);    std::string service (delimiter + 1);    //  Initialize the output parameter.    memset (addr_, 0, sizeof (*addr_));    //  Initialise IPv4-format family/port.    sockaddr_in ip4_addr;    memset (&ip4_addr, 0, sizeof (ip4_addr));    ip4_addr.sin_family = AF_INET;    ip4_addr.sin_port = htons ((uint16_t) atoi (service.c_str()));    //  Initialize temporary output pointers with ip4_addr    sockaddr *out_addr = (sockaddr *) &ip4_addr;    size_t out_addrlen = sizeof (ip4_addr);    //  0 is not a valid port.    if (!ip4_addr.sin_port) {        errno = EINVAL;        return -1;    }    //  * resolves to INADDR_ANY.    if (iface.compare("*") == 0) {        ip4_addr.sin_addr.s_addr = htonl (INADDR_ANY);        zmq_assert (out_addrlen <= sizeof (*addr_));        memcpy (addr_, out_addr, out_addrlen);        *addr_len_ = out_addrlen;        return 0;    }    //  Try to resolve the string as a NIC name.    int rc = resolve_nic_name (&ip4_addr.sin_addr, iface.c_str());    if (rc != 0 && errno != ENODEV)        return rc;    if (rc == 0) {        zmq_assert (out_addrlen <= sizeof (*addr_));        memcpy (addr_, out_addr, out_addrlen);        *addr_len_ = out_addrlen;        return 0;    }    //  There's no such interface name. Assume literal address.#if defined ZMQ_HAVE_OPENVMS && defined __ia64    __addrinfo64 *res = NULL;    __addrinfo64 req;#else    addrinfo *res = NULL;    addrinfo req;#endif    memset (&req, 0, sizeof (req));    //  We only support IPv4 addresses for now.    req.ai_family = AF_INET;    //  Arbitrary, not used in the output, but avoids duplicate results.    req.ai_socktype = SOCK_STREAM;    //  Restrict hostname/service to literals to avoid any DNS lookups or    //  service-name irregularity due to indeterminate socktype.    req.ai_flags = AI_PASSIVE | AI_NUMERICHOST | AI_NUMERICSERV;    //  Resolve the literal address. Some of the error info is lost in case    //  of error, however, there's no way to report EAI errors via errno.    rc = getaddrinfo (iface.c_str(), service.c_str(), &req, &res);    if (rc) {        errno = ENODEV;        return -1;    }    //  Use the first result.    zmq_assert ((size_t) (res->ai_addrlen) <= sizeof (*addr_));    memcpy (addr_, res->ai_addr, res->ai_addrlen);    *addr_len_ = res->ai_addrlen;    //  Cleanup getaddrinfo after copying the possibly referenced result.    if (res)        freeaddrinfo (res);    return 0;}

void zmq::socket_base_t::xhiccuped (pipe_t *pipe_){    zmq_assert (false);}

int zmq::router_t::xsend (msg_t *msg_, int flags_){    //  If this is the first part of the message it's the ID of the    //  peer to send the message to.    if (!more_out) {        zmq_assert (!current_out);        int retval = 0;        //  If we have malformed message (prefix with no subsequent message)        //  then just silently ignore it.        //  TODO: The connections should be killed instead.        if (msg_->flags () & msg_t::more) {            more_out = true;            //  Find the pipe associated with the identity stored in the prefix.            //  If there's no such pipe just silently ignore the message, unless            //  fail_unreachable is set.            blob_t identity ((unsigned char*) msg_->data (), msg_->size ());            outpipes_t::iterator it = outpipes.find (identity);            if (it != outpipes.end ()) {                current_out = it->second.pipe;                if (!current_out->check_write ()) {                    it->second.active = false;                    more_out = false;                    current_out = NULL;                }            } else if(fail_unroutable) {                more_out = false;                errno = EHOSTUNREACH;                retval = -1;            }        }        int rc = msg_->close ();        errno_assert (rc == 0);        rc = msg_->init ();        errno_assert (rc == 0);        return retval;    }    //  Check whether this is the last part of the message.    more_out = msg_->flags () & msg_t::more ? true : false;    //  Push the message into the pipe. If there's no out pipe, just drop it.    if (current_out) {        bool ok = current_out->write (msg_);        if (unlikely (!ok))            current_out = NULL;        else if (!more_out) {            current_out->flush ();            current_out = NULL;        }    }    else {        int rc = msg_->close ();        errno_assert (rc == 0);    }    //  Detach the message from the data buffer.    int rc = msg_->init ();    errno_assert (rc == 0);    return 0;}

int zmq::curve_server_t::decode (msg_t *msg_){    zmq_assert (state == connected);    if (msg_->size () < 33) {        //  Temporary support for security debugging        puts ("CURVE I: invalid CURVE client, sent malformed command");        errno = EPROTO;        return -1;    }    const uint8_t *message = static_cast <uint8_t *> (msg_->data ());    if (memcmp (message, "/x07MESSAGE", 8)) {        //  Temporary support for security debugging        puts ("CURVE I: invalid CURVE client, did not send MESSAGE");        errno = EPROTO;        return -1;    }    uint8_t message_nonce [crypto_box_NONCEBYTES];    memcpy (message_nonce, "CurveZMQMESSAGEC", 16);    memcpy (message_nonce + 16, message + 8, 8);    uint64_t nonce = get_uint64(message + 8);    if (nonce <= cn_peer_nonce) {        errno = EPROTO;        return -1;    }    cn_peer_nonce = nonce;    const size_t clen = crypto_box_BOXZEROBYTES + msg_->size () - 16;    uint8_t *message_plaintext = static_cast <uint8_t *> (malloc (clen));    alloc_assert (message_plaintext);    uint8_t *message_box = static_cast <uint8_t *> (malloc (clen));    alloc_assert (message_box);    memset (message_box, 0, crypto_box_BOXZEROBYTES);    memcpy (message_box + crypto_box_BOXZEROBYTES,            message + 16, msg_->size () - 16);    int rc = crypto_box_open_afternm (message_plaintext, message_box,                                      clen, message_nonce, cn_precom);    if (rc == 0) {        rc = msg_->close ();        zmq_assert (rc == 0);        rc = msg_->init_size (clen - 1 - crypto_box_ZEROBYTES);        zmq_assert (rc == 0);        const uint8_t flags = message_plaintext [crypto_box_ZEROBYTES];        if (flags & 0x01)            msg_->set_flags (msg_t::more);        if (flags & 0x02)            msg_->set_flags (msg_t::command);        memcpy (msg_->data (),                message_plaintext + crypto_box_ZEROBYTES + 1,                msg_->size ());    }    else {        //  Temporary support for security debugging        puts ("CURVE I: connection key used for MESSAGE is wrong");        errno = EPROTO;    }    free (message_plaintext);    free (message_box);    return rc;}

//.........这里部分代码省略.........        puts ("CURVE I: cannot open client INITIATE cookie");        errno = EPROTO;        return -1;    }    //  Check cookie plain text is as expected [C' + s']    if (memcmp (cookie_plaintext + crypto_secretbox_ZEROBYTES, cn_client, 32)    ||  memcmp (cookie_plaintext + crypto_secretbox_ZEROBYTES + 32, cn_secret, 32)) {        //  Temporary support for security debugging        puts ("CURVE I: client INITIATE cookie is not valid");        errno = EPROTO;        return -1;    }    const size_t clen = (msg_->size () - 113) + crypto_box_BOXZEROBYTES;    uint8_t initiate_nonce [crypto_box_NONCEBYTES];    uint8_t initiate_plaintext [crypto_box_ZEROBYTES + 128 + 256];    uint8_t initiate_box [crypto_box_BOXZEROBYTES + 144 + 256];    //  Open Box [C + vouch + metadata](C'->S')    memset (initiate_box, 0, crypto_box_BOXZEROBYTES);    memcpy (initiate_box + crypto_box_BOXZEROBYTES,            initiate + 113, clen - crypto_box_BOXZEROBYTES);    memcpy (initiate_nonce, "CurveZMQINITIATE", 16);    memcpy (initiate_nonce + 16, initiate + 105, 8);    cn_peer_nonce = get_uint64(initiate + 105);    rc = crypto_box_open (initiate_plaintext, initiate_box,                          clen, initiate_nonce, cn_client, cn_secret);    if (rc != 0) {        //  Temporary support for security debugging        puts ("CURVE I: cannot open client INITIATE");        errno = EPROTO;        return -1;    }    const uint8_t *client_key = initiate_plaintext + crypto_box_ZEROBYTES;    uint8_t vouch_nonce [crypto_box_NONCEBYTES];    uint8_t vouch_plaintext [crypto_box_ZEROBYTES + 64];    uint8_t vouch_box [crypto_box_BOXZEROBYTES + 80];    //  Open Box Box [C',S](C->S') and check contents    memset (vouch_box, 0, crypto_box_BOXZEROBYTES);    memcpy (vouch_box + crypto_box_BOXZEROBYTES,            initiate_plaintext + crypto_box_ZEROBYTES + 48, 80);    memcpy (vouch_nonce, "VOUCH---", 8);    memcpy (vouch_nonce + 8,            initiate_plaintext + crypto_box_ZEROBYTES + 32, 16);    rc = crypto_box_open (vouch_plaintext, vouch_box,                          sizeof vouch_box,                          vouch_nonce, client_key, cn_secret);    if (rc != 0) {        //  Temporary support for security debugging        puts ("CURVE I: cannot open client INITIATE vouch");        errno = EPROTO;        return -1;    }    //  What we decrypted must be the client's short-term public key    if (memcmp (vouch_plaintext + crypto_box_ZEROBYTES, cn_client, 32)) {        //  Temporary support for security debugging        puts ("CURVE I: invalid handshake from client (public key)");        errno = EPROTO;        return -1;    }    //  Precompute connection secret from client key    rc = crypto_box_beforenm (cn_precom, cn_client, cn_secret);    zmq_assert (rc == 0);	puts("zmq::curve_server_t::process_initiate before zap_connect ");	//  Use ZAP protocol (RFC 27) to authenticate the user.    rc = session->zap_connect ();    if (rc == 0) {        send_zap_request (client_key);        rc = receive_and_process_zap_reply ();        if (rc == 0)            state = status_code == "200"                ? send_ready                : send_error;        else        if (errno == EAGAIN)            state = expect_zap_reply;        else            return -1;    }    else        state = send_ready;	puts("zmq::curve_server_t::process_initiate end");    return parse_metadata (initiate_plaintext + crypto_box_ZEROBYTES + 128,                           clen - crypto_box_ZEROBYTES - 128);}

zmq::socks_connecter_t::~socks_connecter_t (){    zmq_assert (s == retired_fd);    LIBZMQ_DELETE(proxy_addr);}

void zmq::devpoll_t::devpoll_ctl (fd_t fd_, short events_){    struct pollfd pfd = {fd_, events_, 0};    ssize_t rc = write (devpoll_fd, &pfd, sizeof pfd);    zmq_assert (rc == sizeof pfd);}

//.........这里部分代码省略.........        }        break;#endif    case ZMQ_CONFLATE:        if (is_int) {            *value = conflate;            return 0;        }        break;        //  If libgssapi isn't installed, these options provoke EINVAL#ifdef HAVE_LIBGSSAPI_KRB5    case ZMQ_GSSAPI_SERVER:        if (is_int) {            *value = as_server && mechanism == ZMQ_GSSAPI;            return 0;        }        break;    case ZMQ_GSSAPI_PRINCIPAL:        if (*optvallen_ >= gss_principal.size () + 1) {            memcpy (optval_, gss_principal.c_str (), gss_principal.size () + 1);            *optvallen_ = gss_principal.size () + 1;            return 0;        }        break;    case ZMQ_GSSAPI_SERVICE_PRINCIPAL:        if (*optvallen_ >= gss_service_principal.size () + 1) {            memcpy (optval_, gss_service_principal.c_str (), gss_service_principal.size () + 1);            *optvallen_ = gss_service_principal.size () + 1;            return 0;        }        break;    case ZMQ_GSSAPI_PLAINTEXT:        if (is_int) {            *value = gss_plaintext;            return 0;        }        break;#endif    case ZMQ_HANDSHAKE_IVL:        if (is_int) {            *value = handshake_ivl;            return 0;        }        break;    case ZMQ_INVERT_MATCHING:        if (is_int) {            *value = invert_matching;            return 0;        }        break;    case ZMQ_HEARTBEAT_IVL:        if (is_int) {            *value = heartbeat_interval;            return 0;        }        break;    case ZMQ_HEARTBEAT_TTL:        if (is_int) {            // Convert the internal deciseconds value to milliseconds            *value = heartbeat_ttl * 100;            return 0;        }        break;    case ZMQ_HEARTBEAT_TIMEOUT:        if (is_int) {            *value = heartbeat_timeout;            return 0;        }        break;    case ZMQ_USE_FD:        if (is_int) {            *value = use_fd;            return 0;        }        break;    default:#if defined (ZMQ_ACT_MILITANT)        malformed = false;#endif        break;    }#if defined (ZMQ_ACT_MILITANT)    if (malformed)        zmq_assert (false);#endif    errno = EINVAL;    return -1;}

int zmq_poll (zmq_pollitem_t *items_, int nitems_, long timeout_){#if defined ZMQ_HAVE_LINUX || defined ZMQ_HAVE_FREEBSD ||/    defined ZMQ_HAVE_OPENBSD || defined ZMQ_HAVE_SOLARIS ||/    defined ZMQ_HAVE_OSX || defined ZMQ_HAVE_QNXNTO ||/    defined ZMQ_HAVE_HPUX || defined ZMQ_HAVE_AIX ||/    defined ZMQ_HAVE_NETBSD    pollfd *pollfds = (pollfd*) malloc (nitems_ * sizeof (pollfd));    zmq_assert (pollfds);    int npollfds = 0;    int nsockets = 0;    zmq::app_thread_t *app_thread = NULL;    for (int i = 0; i != nitems_; i++) {        //  0MQ sockets.        if (items_ [i].socket) {            //  Get the app_thread the socket is living in. If there are two            //  sockets in the same pollset with different app threads, fail.            zmq::socket_base_t *s = (zmq::socket_base_t*) items_ [i].socket;            if (app_thread) {                if (app_thread != s->get_thread ()) {                    free (pollfds);                    errno = EFAULT;                    return -1;                }            }            else                app_thread = s->get_thread ();            nsockets++;            continue;        }        //  Raw file descriptors.        pollfds [npollfds].fd = items_ [i].fd;        pollfds [npollfds].events =            (items_ [i].events & ZMQ_POLLIN ? POLLIN : 0) |            (items_ [i].events & ZMQ_POLLOUT ? POLLOUT : 0);        npollfds++;    }    //  If there's at least one 0MQ socket in the pollset we have to poll    //  for 0MQ commands. If ZMQ_POLL was not set, fail.    if (nsockets) {        pollfds [npollfds].fd = app_thread->get_signaler ()->get_fd ();        if (pollfds [npollfds].fd == zmq::retired_fd) {            free (pollfds);            errno = ENOTSUP;            return -1;        }        pollfds [npollfds].events = POLLIN;        npollfds++;    }    //  First iteration just check for events, don't block. Waiting would    //  prevent exiting on any events that may already been signaled on    //  0MQ sockets.    int rc = poll (pollfds, npollfds, 0);    if (rc == -1 && errno == EINTR && timeout_ >= 0) {        free (pollfds);        return 0;    }    errno_assert (rc >= 0 || (rc == -1 && errno == EINTR));    int timeout = timeout_ > 0 ? timeout_ / 1000 : -1;    int nevents = 0;    while (true) {        //  Process 0MQ commands if needed.        if (nsockets && pollfds [npollfds -1].revents & POLLIN)            app_thread->process_commands (false, false);        //  Check for the events.        int pollfd_pos = 0;        for (int i = 0; i != nitems_; i++) {            //  If the poll item is a raw file descriptor, simply convert            //  the events to zmq_pollitem_t-style format.            if (!items_ [i].socket) {                items_ [i].revents = 0;                if (pollfds [pollfd_pos].revents & POLLIN)                    items_ [i].revents |= ZMQ_POLLIN;                if (pollfds [pollfd_pos].revents & POLLOUT)                    items_ [i].revents |= ZMQ_POLLOUT;                if (pollfds [pollfd_pos].revents & ~(POLLIN | POLLOUT))                    items_ [i].revents |= ZMQ_POLLERR;                if (items_ [i].revents)                    nevents++;                pollfd_pos++;                continue;            }            //  The poll item is a 0MQ socket.            zmq::socket_base_t *s = (zmq::socket_base_t*) items_ [i].socket;//.........这里部分代码省略.........

void zmq::generic_mtrie_t<T>::rm_helper (value_t *pipe_,                                         unsigned char **buff_,                                         size_t buffsize_,                                         size_t maxbuffsize_,                                         void (*func_) (prefix_t data_,                                                        size_t size_,                                                        Arg arg_),                                         Arg arg_,                                         bool call_on_uniq_){    //  Remove the subscription from this node.    if (pipes && pipes->erase (pipe_)) {        if (!call_on_uniq_ || pipes->empty ()) {            func_ (*buff_, buffsize_, arg_);        }        if (pipes->empty ()) {            LIBZMQ_DELETE (pipes);        }    }    //  Adjust the buffer.    if (buffsize_ >= maxbuffsize_) {        maxbuffsize_ = buffsize_ + 256;        *buff_ = (unsigned char *) realloc (*buff_, maxbuffsize_);        alloc_assert (*buff_);    }    //  If there are no subnodes in the trie, return.    if (count == 0)        return;    //  If there's one subnode (optimisation).    if (count == 1) {        (*buff_)[buffsize_] = min;        buffsize_++;        next.node->rm_helper (pipe_, buff_, buffsize_, maxbuffsize_, func_,                              arg_, call_on_uniq_);        //  Prune the node if it was made redundant by the removal        if (next.node->is_redundant ()) {            LIBZMQ_DELETE (next.node);            count = 0;            --live_nodes;            zmq_assert (live_nodes == 0);        }        return;    }    //  If there are multiple subnodes.    //    //  New min non-null character in the node table after the removal    unsigned char new_min = min + count - 1;    //  New max non-null character in the node table after the removal    unsigned char new_max = min;    for (unsigned short c = 0; c != count; c++) {        (*buff_)[buffsize_] = min + c;        if (next.table[c]) {            next.table[c]->rm_helper (pipe_, buff_, buffsize_ + 1, maxbuffsize_,                                      func_, arg_, call_on_uniq_);            //  Prune redundant nodes from the mtrie            if (next.table[c]->is_redundant ()) {                LIBZMQ_DELETE (next.table[c]);                zmq_assert (live_nodes > 0);                --live_nodes;            } else {                //  The node is not redundant, so it's a candidate for being                //  the new min/max node.                //                //  We loop through the node array from left to right, so the                //  first non-null, non-redundant node encountered is the new                //  minimum index. Conversely, the last non-redundant, non-null                //  node encountered is the new maximum index.                if (c + min < new_min)                    new_min = c + min;                if (c + min > new_max)                    new_max = c + min;            }        }    }    zmq_assert (count > 1);    //  Free the node table if it's no longer used.    if (live_nodes == 0) {        free (next.table);        next.table = NULL;        count = 0;    }    //  Compact the node table if possible    else if (live_nodes == 1) {        //  If there's only one live node in the table we can        //  switch to using the more compact single-node        //  representation        zmq_assert (new_min == new_max);        zmq_assert (new_min >= min && new_min < min + count);        generic_mtrie_t *node = next.table[new_min - min];        zmq_assert (node);//.........这里部分代码省略.........

int zmq::router_t::xrecv (msg_t *msg_, int flags_){    //  if there is a prefetched identity, return it.    if (prefetched == 2)    {        int rc = msg_->init_size (prefetched_id.size ());        errno_assert (rc == 0);        memcpy (msg_->data (), prefetched_id.data (), prefetched_id.size ());        msg_->set_flags (msg_t::more);        prefetched = 1;        return 0;    }    //  If there is a prefetched message, return it.    if (prefetched == 1) {        int rc = msg_->move (prefetched_msg);        errno_assert (rc == 0);        more_in = msg_->flags () & msg_t::more ? true : false;        prefetched = 0;        return 0;    }    pipe_t *pipe = NULL;    while (true) {        //  Get next message part.        int rc = fq.recvpipe (msg_, flags_, &pipe);        if (rc != 0)            return -1;        //  If identity is received, change the key assigned to the pipe.        if (likely (!(msg_->flags () & msg_t::identity)))            break;        zmq_assert (!more_in);        //  Empty identity means we can preserve the auto-generated identity        if (msg_->size ()) {            blob_t identity ((unsigned char*) msg_->data (), msg_->size ());            outpipes_t::iterator it = outpipes.find (identity);            if (it == outpipes.end ()) {                //  Find the pipe and change its identity                bool changed = false;                it = outpipes.begin ();                while (it != outpipes.end ()) {                    if (it->second.pipe == pipe) {                        pipe->set_identity (identity);                        outpipes.erase (it);                        outpipe_t outpipe = {pipe, true};                        if (!outpipes.insert (                            outpipes_t::value_type (identity, outpipe)).second)                            zmq_assert (false);                        changed = true;                        break;                    }                    ++it;                }                zmq_assert (changed);            }        }    }    //  If we are in the middle of reading a message, just return the next part.    if (more_in) {        more_in = msg_->flags () & msg_t::more ? true : false;        return 0;    }     //  We are at the beginning of a new message. Move the message part we    //  have to the prefetched and return the ID of the peer instead.    int rc = prefetched_msg.move (*msg_);    errno_assert (rc == 0);    prefetched = 1;    rc = msg_->close ();    errno_assert (rc == 0);    blob_t identity = pipe->get_identity ();    rc = msg_->init_size (identity.size ());    errno_assert (rc == 0);    memcpy (msg_->data (), identity.data (), identity.size ());    msg_->set_flags (msg_t::more);    return 0;}

typename zmq::generic_mtrie_t<T>::rm_result zmq::generic_mtrie_t<T>::rm_helper (  prefix_t prefix_, size_t size_, value_t *pipe_){    if (!size_) {        if (!pipes)            return not_found;        typename pipes_t::size_type erased = pipes->erase (pipe_);        if (pipes->empty ()) {            zmq_assert (erased == 1);            LIBZMQ_DELETE (pipes);            return last_value_removed;        }        return (erased == 1) ? values_remain : not_found;    }    unsigned char c = *prefix_;    if (!count || c < min || c >= min + count)        return not_found;    generic_mtrie_t *next_node = count == 1 ? next.node : next.table[c - min];    if (!next_node)        return not_found;    rm_result ret = next_node->rm_helper (prefix_ + 1, size_ - 1, pipe_);    if (next_node->is_redundant ()) {        LIBZMQ_DELETE (next_node);        zmq_assert (count > 0);        if (count == 1) {            next.node = 0;            count = 0;            --live_nodes;            zmq_assert (live_nodes == 0);        } else {            next.table[c - min] = 0;            zmq_assert (live_nodes > 1);            --live_nodes;            //  Compact the table if possible            if (live_nodes == 1) {                //  If there's only one live node in the table we can                //  switch to using the more compact single-node                //  representation                unsigned short i;                for (i = 0; i < count; ++i)                    if (next.table[i])                        break;                zmq_assert (i < count);                min += i;                count = 1;                generic_mtrie_t *oldp = next.table[i];                free (next.table);                next.node = oldp;            } else if (c == min) {                //  We can compact the table "from the left"                unsigned short i;                for (i = 1; i < count; ++i)                    if (next.table[i])                        break;                zmq_assert (i < count);                min += i;                count -= i;                generic_mtrie_t **old_table = next.table;                next.table = (generic_mtrie_t **) malloc (                  sizeof (generic_mtrie_t *) * count);                alloc_assert (next.table);                memmove (next.table, old_table + i,                         sizeof (generic_mtrie_t *) * count);                free (old_table);            } else if (c == min + count - 1) {                //  We can compact the table "from the right"                unsigned short i;                for (i = 1; i < count; ++i)                    if (next.table[count - 1 - i])                        break;                zmq_assert (i < count);                count -= i;                generic_mtrie_t **old_table = next.table;                next.table = (generic_mtrie_t **) malloc (                  sizeof (generic_mtrie_t *) * count);                alloc_assert (next.table);                memmove (next.table, old_table,                         sizeof (generic_mtrie_t *) * count);                free (old_table);            }        }    }    return ret;}

void zmq::stream_engine_t::plug (io_thread_t *io_thread_,    session_base_t *session_){    zmq_assert (!plugged);    plugged = true;    //  Connect to session object.    zmq_assert (!session);    zmq_assert (session_);    session = session_;    socket = session-> get_socket ();    //  Connect to I/O threads poller object.    io_object_t::plug (io_thread_);    handle = add_fd (s);    io_error = false;    if (options.raw_socket) {        // no handshaking for raw sock, instantiate raw encoder and decoders        encoder = new (std::nothrow) raw_encoder_t (options.tcp_send_buffer_size);        alloc_assert (encoder);        decoder = new (std::nothrow) raw_decoder_t (options.tcp_recv_buffer_size);        alloc_assert (decoder);        // disable handshaking for raw socket        handshaking = false;        next_msg = &stream_engine_t::pull_msg_from_session;        process_msg = &stream_engine_t::push_raw_msg_to_session;        properties_t properties;        if (init_properties(properties)) {            //  Compile metadata.            zmq_assert (metadata == NULL);            metadata = new (std::nothrow) metadata_t (properties);        }        if (options.raw_notify) {            //  For raw sockets, send an initial 0-length message to the            // application so that it knows a peer has connected.            msg_t connector;            connector.init();            push_raw_msg_to_session (&connector);            connector.close();            session->flush ();        }    }    else {        // start optional timer, to prevent handshake hanging on no input        set_handshake_timer ();        //  Send the 'length' and 'flags' fields of the identity message.        //  The 'length' field is encoded in the long format.        outpos = greeting_send;        outpos [outsize++] = 0xff;        put_uint64 (&outpos [outsize], options.identity_size + 1);        outsize += 8;        outpos [outsize++] = 0x7f;    }    set_pollin (handle);    set_pollout (handle);    //  Flush all the data that may have been already received downstream.    in_event ();}

//.........这里部分代码省略.........        //  Find out whether timeout have expired.        now = clock.now_ms ();        if (now >= end)            break;    }    if (pollfds != spollfds)        free (pollfds);    return nevents;#elif 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;#else        return usleep (timeout_ * 1000);#endif    }    zmq::clock_t clock;    uint64_t now = 0;    uint64_t end = 0;    //  Ensure we do not attempt to select () on more than FD_SETSIZE    //  file descriptors.    zmq_assert (nitems_ <= FD_SETSIZE);    fd_set pollset_in;    FD_ZERO (&pollset_in);    fd_set pollset_out;    FD_ZERO (&pollset_out);    fd_set pollset_err;    FD_ZERO (&pollset_err);    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, &notify_fd,                &zmq_fd_size) == -1)                return -1;            if (items_ [i].events) {                FD_SET (notify_fd, &pollset_in);                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);

bool zmq::trie_t::add (unsigned char *prefix_, size_t size_){    //  We are at the node corresponding to the prefix. We are done.    if (!size_) {        ++refcnt;        return refcnt == 1;    }    unsigned char c = *prefix_;    if (c < min || c >= min + count) {        //  The character is out of range of currently handled        //  charcters. We have to extend the table.        if (!count) {            min = c;            count = 1;            next.node = NULL;        }        else        if (count == 1) {            unsigned char oldc = min;            trie_t *oldp = next.node;            count = (min < c ? c - min : min - c) + 1;            next.table = (trie_t**)                malloc (sizeof (trie_t*) * count);            alloc_assert (next.table);            for (unsigned short i = 0; i != count; ++i)                next.table [i] = 0;            min = std::min (min, c);            next.table [oldc - min] = oldp;        }        else        if (min < c) {            //  The new character is above the current character range.            unsigned short old_count = count;            count = c - min + 1;            next.table = (trie_t**) realloc ((void*) next.table,                sizeof (trie_t*) * count);            zmq_assert (next.table);            for (unsigned short i = old_count; i != count; i++)                next.table [i] = NULL;        }        else {            //  The new character is below the current character range.            unsigned short old_count = count;            count = (min + old_count) - c;            next.table = (trie_t**) realloc ((void*) next.table,                sizeof (trie_t*) * count);            zmq_assert (next.table);            memmove (next.table + min - c, next.table,                old_count * sizeof (trie_t*));            for (unsigned short i = 0; i != min - c; i++)                next.table [i] = NULL;            min = c;        }    }    //  If next node does not exist, create one.    if (count == 1) {        if (!next.node) {            next.node = new (std::nothrow) trie_t;            alloc_assert (next.node);            ++live_nodes;            zmq_assert (live_nodes == 1);        }        return next.node->add (prefix_ + 1, size_ - 1);    }    else {        if (!next.table [c - min]) {            next.table [c - min] = new (std::nothrow) trie_t;            alloc_assert (next.table [c - min]);            ++live_nodes;            zmq_assert (live_nodes > 1);        }        return next.table [c - min]->add (prefix_ + 1, size_ - 1);    }}

void zmq::xpub_t::xattach_pipes (class reader_t *inpipe_,    class writer_t *outpipe_, const blob_t &peer_identity_){    zmq_assert (!inpipe_ && outpipe_);    dist.attach (outpipe_);}

void zmq::own_t::set_owner (own_t *owner_){    zmq_assert (!owner);    owner = owner_;}