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

void xs::pipe_t::process_pipe_term_ack (){    //  Notify the user that all the references to the pipe should be dropped.    xs_assert (sink);    sink->terminated (this);    //  In terminating and double_terminated states there's nothing to do.    //  Simply deallocate the pipe. In terminated state we have to ack the    //  peer before deallocating this side of the pipe. All the other states    //  are invalid.    if (state == terminating) ;    else if (state == double_terminated);    else if (state == terminated) {        outpipe = NULL;        send_pipe_term_ack (peer);    }    else        xs_assert (false);    //  We'll deallocate the inbound pipe, the peer will deallocate the outbound    //  pipe (which is an inbound pipe from its point of view).    //  First, delete all the unread messages in the pipe. We have to do it by    //  hand because msg_t doesn't have automatic destructor. Then deallocate    //  the ypipe itself.    msg_t msg;    while (inpipe->read (&msg)) {       int rc = msg.close ();       errno_assert (rc == 0);    }    delete inpipe;    //  Deallocate the pipe object    delete this;}

void xs::pgm_receiver_t::activate_in (){    //  It is possible that the most recently used decoder    //  processed the whole buffer but failed to write    //  the last message into the pipe.    if (pending_bytes == 0) {        if (mru_decoder != NULL) {            mru_decoder->process_buffer (NULL, 0);            session->flush ();        }        //  Resume polling.        set_pollin (pipe_handle);        set_pollin (socket_handle);        return;    }    xs_assert (mru_decoder != NULL);    xs_assert (pending_ptr != NULL);    //  Ask the decoder to process remaining data.    size_t n = mru_decoder->process_buffer (pending_ptr, pending_bytes);    pending_bytes -= n;    session->flush ();    if (pending_bytes > 0)        return;    //  Resume polling.    set_pollin (pipe_handle);    set_pollin (socket_handle);    in_event (retired_fd);}

void xs::signaler_recv (xs::signaler_t *self_){    //  Attempt to read a signal.#if defined XS_HAVE_EVENTFD    uint64_t dummy;    ssize_t sz = read (self_->r, &dummy, sizeof (dummy));    errno_assert (sz == sizeof (dummy));    //  If we accidentally grabbed the next signal along with the current    //  one, return it back to the eventfd object.    if (unlikely (dummy == 2)) {        const uint64_t inc = 1;        ssize_t sz = write (self_->w, &inc, sizeof (inc));        errno_assert (sz == sizeof (inc));        return;    }    xs_assert (dummy == 1);#else    unsigned char dummy;#if defined XS_HAVE_WINDOWS    int nbytes = ::recv (self_->r, (char*) &dummy, sizeof (dummy), 0);    wsa_assert (nbytes != SOCKET_ERROR);#else    ssize_t nbytes = ::recv (self_->r, &dummy, sizeof (dummy), 0);    errno_assert (nbytes >= 0);#endif    xs_assert (nbytes == sizeof (dummy));    xs_assert (dummy == 0);#endif}

void xs::signaler_send (xs::signaler_t *self_){#if defined XS_HAVE_EVENTFD    const uint64_t inc = 1;    ssize_t sz = write (self_->w, &inc, sizeof (inc));    errno_assert (sz == sizeof (inc));#elif defined XS_HAVE_WINDOWS    unsigned char dummy = 0;    int nbytes = ::send (self_->w, (char*) &dummy, sizeof (dummy), 0);    wsa_assert (nbytes != SOCKET_ERROR);    xs_assert (nbytes == sizeof (dummy));#else    unsigned char dummy = 0;    while (true) {#if defined MSG_NOSIGNAL        ssize_t nbytes = ::send (self_->w, &dummy, sizeof (dummy),            MSG_NOSIGNAL);#else        ssize_t nbytes = ::send (self_->w, &dummy, sizeof (dummy), 0);#endif        if (unlikely (nbytes == -1 && errno == EINTR))            continue;        xs_assert (nbytes == sizeof (dummy));        break;    }#endif}

void xs::xrep_t::xwrite_activated (pipe_t *pipe_){    for (outpipes_t::iterator it = outpipes.begin ();          it != outpipes.end (); ++it) {        if (it->second.pipe == pipe_) {            xs_assert (!it->second.active);            it->second.active = true;            return;        }    }    xs_assert (false);}

int xs::signaler_wait (xs::signaler_t *self_, int timeout_){#ifdef XS_USE_SYNC_POLL    struct pollfd pfd;    pfd.fd = self_->r;    pfd.events = POLLIN;    int rc = poll (&pfd, 1, timeout_);    if (unlikely (rc < 0)) {        errno_assert (errno == EINTR);        return -1;    }    else if (unlikely (rc == 0)) {        errno = EAGAIN;        return -1;    }    xs_assert (rc == 1);    xs_assert (pfd.revents & POLLIN);    return 0;#elif defined XS_USE_SYNC_SELECT    FD_SET (self_->r, &self_->fds);    struct timeval timeout;    if (timeout_ >= 0) {        timeout.tv_sec = timeout_ / 1000;        timeout.tv_usec = timeout_ % 1000 * 1000;    }#ifdef XS_HAVE_WINDOWS    int rc = select (0, &self_->fds, NULL, NULL,        timeout_ >= 0 ? &timeout : NULL);    wsa_assert (rc != SOCKET_ERROR);#else    int rc = select (self_->r + 1, &self_->fds, NULL, NULL,        timeout_ >= 0 ? &timeout : NULL);    if (unlikely (rc < 0)) {        errno_assert (errno == EINTR);        return -1;    }#endif    if (unlikely (rc == 0)) {        errno = EAGAIN;        return -1;    }    xs_assert (rc == 1);    return 0;#else#error    return -1;#endif}

size_t xs::msg_t::size (){    //  Check the validity of the message.    xs_assert (check ());    switch (u.base.type) {    case type_vsm:        return u.vsm.size;    case type_lmsg:        return u.lmsg.content->size;    default:        xs_assert (false);        return 0;    }}

void xs::stream_engine_t::error (){    xs_assert (session);    session->detach ();    unplug ();    delete this;}

void xs::ipc_listener_t::in_event (fd_t fd_){    fd_t fd = accept ();    //  If connection was reset by the peer in the meantime, just ignore it.    //  TODO: Handle specific errors like ENFILE/EMFILE etc.    if (fd == retired_fd)        return;    //  Create the engine object for this connection.    stream_engine_t *engine = new (std::nothrow) stream_engine_t (fd, options);    alloc_assert (engine);    //  Choose I/O thread to run connecter in. Given that we are already    //  running in an I/O thread, there must be at least one available.    io_thread_t *thread = choose_io_thread (options.affinity);    xs_assert (thread);    //  Create and launch a session object.     session_base_t *session = session_base_t::create (thread, false, socket,        options, NULL, NULL);    errno_assert (session);    session->inc_seqnum ();    launch_child (session);    send_attach (session, engine, false);}

xs::socket_base_t::~socket_base_t (){    xs_assert (destroyed);    if (initialised)        mailbox_close (&mailbox);}

int xs::sub_t::xsetsockopt (int option_, const void *optval_,    size_t optvallen_){    if (option_ != XS_SUBSCRIBE && option_ != XS_UNSUBSCRIBE) {        errno = EINVAL;        return -1;    }    if (optvallen_ > 0 && !optval_) {        errno = EFAULT;        return -1;    }    //  Find the relevant filter.    filters_t::iterator it;    for (it = filters.begin (); it != filters.end (); ++it)        if (it->type->id (NULL) == options.filter)            break;    //  Process the subscription. If the filter of the specified type does not    //  exist yet, create it.    if (option_ == XS_SUBSCRIBE) {        if (it == filters.end ()) {            filter_t f;            f.type = get_filter (options.filter);            xs_assert (f.type);            f.instance = f.type->sf_create ((void*) (core_t*) this);            xs_assert (f.instance);            filters.push_back (f);            it = filters.end () - 1;        }        int rc = it->type->sf_subscribe ((void*) (core_t*) this, it->instance,            (const unsigned char*) optval_, optvallen_);        errno_assert (rc == 0);        return 0;    }    else if (option_ == XS_UNSUBSCRIBE) {        xs_assert (it != filters.end ());        int rc = it->type->sf_unsubscribe ((void*) (core_t*) this, it->instance,            (const unsigned char*) optval_, optvallen_);        errno_assert (rc == 0);        return 0;    }    xs_assert (false);    return -1;}

void xs::pgm_sender_t::out_event (fd_t fd_){    //  POLLOUT event from send socket. If write buffer is empty,     //  try to read new data from the encoder.    if (write_size == 0) {        //  First two bytes (sizeof uint16_t) are used to store message         //  offset in following steps. Note that by passing our buffer to        //  the get data function we prevent it from returning its own buffer.        unsigned char *bf = out_buffer + sizeof (uint16_t);        size_t bfsz = out_buffer_size - sizeof (uint16_t);        int offset = -1;        encoder.get_data (&bf, &bfsz, &offset);        //  If there are no data to write stop polling for output.        if (!bfsz) {            reset_pollout (handle);            return;        }        //  Put offset information in the buffer.        write_size = bfsz + sizeof (uint16_t);        put_uint16 (out_buffer, offset == -1 ? 0xffff : (uint16_t) offset);    }    if (tx_timer) {        rm_timer (tx_timer);        tx_timer = NULL;    }    //  Send the data.    size_t nbytes = pgm_socket.send (out_buffer, write_size);    //  We can write either all data or 0 which means rate limit reached.    if (nbytes == write_size) {        write_size = 0;    } else {        xs_assert (nbytes == 0);        if (errno == ENOMEM) {            const long timeout = pgm_socket.get_tx_timeout ();            xs_assert (!tx_timer);            tx_timer = add_timer (timeout);        } else            errno_assert (errno == EBUSY);    }}

void xs::stream_engine_t::out_event (fd_t fd_){    bool more_data = true;    //  If protocol header was not yet sent...    if (unlikely (!options.legacy_protocol && !header_sent)) {        int hbytes = write (out_header, sizeof out_header);        //  It should always be possible to write the full protocol header to a        //  freshly connected TCP socket. Therefore, if we get an error or        //  partial write here the peer has disconnected.        if (hbytes != sizeof out_header) {            error ();            return;        }        header_sent = true;    }    //  If write buffer is empty, try to read new data from the encoder.    if (!outsize) {        outpos = NULL;        more_data = encoder.get_data (&outpos, &outsize);        //  If IO handler has unplugged engine, flush transient IO handler.        if (unlikely (!plugged)) {            xs_assert (leftover_session);            leftover_session->flush ();            return;        }        //  If there is no data to send, stop polling for output.        if (outsize == 0) {            reset_pollout (handle);            return;        }    }    //  If there are any data to write in write buffer, write as much as    //  possible to the socket. Note that amount of data to write can be    //  arbitratily large. However, we assume that underlying TCP layer has    //  limited transmission buffer and thus the actual number of bytes    //  written should be reasonably modest.    int nbytes = write (outpos, outsize);    //  Handle problems with the connection.    if (nbytes == -1) {        error ();        return;    }    outpos += nbytes;    outsize -= nbytes;    //  If the encoder reports that there are no more data to get from it    //  we can stop polling for POLLOUT immediately.    if (!more_data && !outsize)        reset_pollout (handle);}

void xs::own_t::unregister_term_ack (){    xs_assert (term_acks > 0);    term_acks--;    //  This may be a last ack we are waiting for before termination...    check_term_acks (); }

void xs::xrespondent_t::xattach_pipe (pipe_t *pipe_, bool icanhasall_){    xs_assert (pipe_);    //  Add the pipe to the map out outbound pipes.    outpipe_t outpipe = {pipe_, true};    bool ok = outpipes.insert (outpipes_t::value_type (        next_peer_id, outpipe)).second;    xs_assert (ok);    //  Add the pipe to the list of inbound pipes.    blob_t identity (4, 0);    put_uint32 ((unsigned char*) identity.data (), next_peer_id);    pipe_->set_identity (identity);    fq.attach (pipe_);    //  Generate a new unique peer identity.    ++next_peer_id;    }

void xs::pair_t::xattach_pipe (pipe_t *pipe_, bool icanhasall_){    xs_assert (pipe_ != NULL);    //  XS_PAIR socket can only be connected to a single peer.    //  The socket rejects any further connection requests.    if (pipe == NULL)        pipe = pipe_;    else        pipe_->terminate (false);}

void xs::stream_engine_t::in_event (fd_t fd_){    bool disconnection = false;    //  If there's no data to process in the buffer...    if (!insize) {        //  Retrieve the buffer and read as much data as possible.        //  Note that buffer can be arbitrarily large. However, we assume        //  the underlying TCP layer has fixed buffer size and thus the        //  number of bytes read will be always limited.        decoder.get_buffer (&inpos, &insize);        insize = read (inpos, insize);        //  Check whether the peer has closed the connection.        if (insize == (size_t) -1) {            insize = 0;            disconnection = true;        }    }    //  Push the data to the decoder.    size_t processed = decoder.process_buffer (inpos, insize);    if (unlikely (processed == (size_t) -1)) {        disconnection = true;    }    else {        //  Stop polling for input if we got stuck.        if (processed < insize) {            //  This may happen if queue limits are in effect.            if (plugged)                reset_pollin (handle);        }        //  Adjust the buffer.        inpos += processed;        insize -= processed;    }    //  Flush all messages the decoder may have produced.    //  If IO handler has unplugged engine, flush transient IO handler.    if (unlikely (!plugged)) {        xs_assert (leftover_session);        leftover_session->flush ();    } else {        session->flush ();    }    if (session && disconnection)        error ();}

void xs::pgm_sender_t::timer_event (handle_t handle_){    //  Timer cancels on return by io_thread.    if (handle_ == rx_timer) {        rx_timer = NULL;        in_event (retired_fd);    } else if (handle_ == tx_timer) {        tx_timer = NULL;        out_event (retired_fd);    } else        xs_assert (false);}

void xs::pipe_t::rollback (){    //  Remove incomplete message from the outbound pipe.    msg_t msg;    if (outpipe) {		while (outpipe->unwrite (&msg)) {		    xs_assert (msg.flags () & msg_t::more);		    int rc = msg.close ();		    errno_assert (rc == 0);		}    }}

bool xs::sub_t::xhas_in (){    //  There are subsequent parts of the partly-read message available.    if (more)        return true;    //  If there's already a message prepared by a previous call to xs_poll,    //  return straight ahead.    if (has_message)        return true;    //  TODO: This can result in infinite loop in the case of continuous    //  stream of non-matching messages.    while (true) {        //  Get a message using fair queueing algorithm.        int rc = xsub_t::xrecv (&message, XS_DONTWAIT);        //  If there's no message available, return immediately.        //  The same when error occurs.        if (rc != 0) {            xs_assert (errno == EAGAIN);            return false;        }        //  Check whether the message matches at least one subscription.        if (match (&message)) {            has_message = true;            return true;        }        //  Message doesn't match. Pop any remaining parts of the message        //  from the pipe.        while (message.flags () & msg_t::more) {            rc = xsub_t::xrecv (&message, XS_DONTWAIT);            xs_assert (rc == 0);        }    }}

bool xs::xreq_t::xhas_in (){    //  We may already have a message pre-fetched.    if (prefetched)        return true;    //  Try to read the next message to the pre-fetch buffer.    int rc = xreq_t::xrecv (&prefetched_msg, XS_DONTWAIT);    if (rc != 0 && errno == EAGAIN)        return false;    xs_assert (rc == 0);    prefetched = true;    return true;}