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

int uv_poll_start(uv_poll_t* handle, int pevents, uv_poll_cb poll_cb) {  int events;  assert((pevents & ~(UV_READABLE | UV_WRITABLE)) == 0);  assert(!(handle->flags & (UV_CLOSING | UV_CLOSED)));  if (pevents == 0) {    uv__poll_stop(handle);    return 0;  }  events = 0;  if (pevents & UV_READABLE)    events |= UV__IO_READ;  if (pevents & UV_WRITABLE)    events |= UV__IO_WRITE;  uv__io_stop(handle->loop, &handle->io_watcher);  uv__io_set(&handle->io_watcher, uv__poll_io, handle->fd, events);  uv__io_start(handle->loop, &handle->io_watcher);  handle->poll_cb = poll_cb;  uv__handle_start(handle);  return 0;}

void uv__io_close(uv_loop_t* loop, uv__io_t* w) {  uv__io_stop(loop, w, UV__POLLIN | UV__POLLOUT);  ngx_queue_remove(&w->pending_queue);  /* Remove stale events for this file descriptor */  uv__platform_invalidate_fd(loop, w->fd);}

static void uv__poll_stop(uv_poll_t* handle) {  uv__io_stop(handle->loop,              &handle->io_watcher,              POLLIN | POLLOUT | UV__POLLRDHUP | UV__POLLPRI);  uv__handle_stop(handle);  uv__platform_invalidate_fd(handle->loop, handle->io_watcher.fd);}

void uv__io_close(uv_loop_t* loop, uv__io_t* w) {  uv__io_stop(loop, w, POLLIN | POLLOUT | UV__POLLRDHUP);  QUEUE_REMOVE(&w->pending_queue);  /* Remove stale events for this file descriptor */  uv__platform_invalidate_fd(loop, w->fd);}

static void uv__drain(uv_stream_t* stream) {  uv_shutdown_t* req;  assert(!uv_write_queue_head(stream));  assert(stream->write_queue_size == 0);  uv__io_stop(stream->loop, &stream->io_watcher, UV__POLLOUT);  /* Shutdown? */  if ((stream->flags & UV_STREAM_SHUTTING) &&      !(stream->flags & UV_CLOSING) &&      !(stream->flags & UV_STREAM_SHUT)) {    assert(stream->shutdown_req);    req = stream->shutdown_req;    stream->shutdown_req = NULL;    uv__req_unregister(stream->loop, req);    if (shutdown(uv__stream_fd(stream), SHUT_WR)) {      /* Error. Report it. User should call uv_close(). */      uv__set_sys_error(stream->loop, errno);      if (req->cb) {        req->cb(req, -1);      }    } else {      uv__set_sys_error(stream->loop, 0);      ((uv_handle_t*) stream)->flags |= UV_STREAM_SHUT;      if (req->cb) {        req->cb(req, 0);      }    }  }}

static void uv__udp_sendmsg(uv_loop_t* loop,                            uv__io_t* w,                            unsigned int revents) {  uv_udp_t* handle;  handle = container_of(w, uv_udp_t, io_watcher);  assert(handle->type == UV_UDP);  assert(revents & UV__POLLOUT);  assert(!ngx_queue_empty(&handle->write_queue)      || !ngx_queue_empty(&handle->write_completed_queue));  /* Write out pending data first. */  uv__udp_run_pending(handle);  /* Drain 'request completed' queue. */  uv__udp_run_completed(handle);  if (!ngx_queue_empty(&handle->write_completed_queue)) {    /* Schedule completion callbacks. */    uv__io_feed(handle->loop, &handle->io_watcher);  }  else if (ngx_queue_empty(&handle->write_queue)) {    /* Pending queue and completion queue empty, stop watcher. */    uv__io_stop(loop, &handle->io_watcher, UV__POLLOUT);    if (!uv__io_active(&handle->io_watcher, UV__POLLIN))      uv__handle_stop(handle);  }}

static void uv__drain(uv_stream_t* stream) {  uv_shutdown_t* req;  int err;  assert(QUEUE_EMPTY(&stream->write_queue));  uv__io_stop(stream->loop, &stream->io_watcher, UV__POLLOUT);  /* Shutdown? */  if ((stream->flags & UV_STREAM_SHUTTING) &&      !(stream->flags & UV_CLOSING) &&      !(stream->flags & UV_STREAM_SHUT)) {    assert(stream->shutdown_req);    req = stream->shutdown_req;    stream->shutdown_req = NULL;    stream->flags &= ~UV_STREAM_SHUTTING;    uv__req_unregister(stream->loop, req);    err = 0;    if (tuvp_shutdown(uv__stream_fd(stream), SHUT_WR))      err = -errno;    if (err == 0)      stream->flags |= UV_STREAM_SHUT;    if (req->cb != NULL)      req->cb(req, err);  }}

void uv__stream_close(uv_stream_t* handle) {#if defined(__APPLE__)  /* Terminate select loop first */  if (handle->select != NULL) {    uv__stream_select_t* s;    s = handle->select;    uv_sem_post(&s->sem);    uv_thread_join(&s->thread);    uv_sem_destroy(&s->sem);    uv_mutex_destroy(&s->mutex);    close(s->fake_fd);    uv_close((uv_handle_t*) &s->async, uv__stream_osx_cb_close);    handle->select = NULL;  }#endif /* defined(__APPLE__) */  uv_read_stop(handle);  uv__io_stop(handle->loop, &handle->write_watcher);  close(handle->fd);  handle->fd = -1;  if (handle->accepted_fd >= 0) {    close(handle->accepted_fd);    handle->accepted_fd = -1;  }  assert(!uv__io_active(&handle->read_watcher));  assert(!uv__io_active(&handle->write_watcher));}

void uv__fs_event_close(uv_fs_event_t* handle) {#if defined(__APPLE__)  if (uv__fsevents_close(handle))    uv__io_stop(handle->loop, &handle->event_watcher, UV__POLLIN);#else  uv__io_stop(handle->loop, &handle->event_watcher, UV__POLLIN);#endif /* defined(__APPLE__) */  uv__handle_stop(handle);  free(handle->filename);  handle->filename = NULL;  close(handle->event_watcher.fd);  handle->event_watcher.fd = -1;}

static void uv__stream_eof(uv_stream_t* stream, const uv_buf_t* buf) {  stream->flags |= UV_STREAM_READ_EOF;  uv__io_stop(stream->loop, &stream->io_watcher, UV__POLLIN);  if (!uv__io_active(&stream->io_watcher, UV__POLLOUT))    uv__handle_stop(stream);  stream->read_cb(stream, UV_EOF, buf);  stream->flags &= ~UV_STREAM_READING;}

int uv_read_stop(uv_stream_t* stream) {  uv__io_stop(stream->loop, &stream->read_watcher);  uv__handle_stop(stream);  stream->flags &= ~UV_STREAM_READING;  stream->read_cb = NULL;  stream->read2_cb = NULL;  stream->alloc_cb = NULL;  return 0;}

void uv__server_io(uv_loop_t* loop, uv__io_t* w, int events) {  int fd;  uv_stream_t* stream = container_of(w, uv_stream_t, read_watcher);  assert(events == UV__IO_READ);  assert(!(stream->flags & UV_CLOSING));  if (stream->accepted_fd >= 0) {    uv__io_stop(loop, &stream->read_watcher);    return;  }  /* connection_cb can close the server socket while we're   * in the loop so check it on each iteration.   */  while (stream->fd != -1) {    assert(stream->accepted_fd < 0);    fd = uv__accept(stream->fd);    if (fd < 0) {      if (errno == EAGAIN || errno == EWOULDBLOCK) {        /* No problem. */        return;      } else if (errno == EMFILE) {        /* TODO special trick. unlock reserved socket, accept, close. */        return;      } else if (errno == ECONNABORTED) {        /* ignore */        continue;      } else {        uv__set_sys_error(stream->loop, errno);        stream->connection_cb((uv_stream_t*)stream, -1);      }    } else {      stream->accepted_fd = fd;      stream->connection_cb((uv_stream_t*)stream, 0);      if (stream->accepted_fd >= 0) {        /* The user hasn't yet accepted called uv_accept() */        uv__io_stop(stream->loop, &stream->read_watcher);        return;      }    }  }}

static void uv__udp_stop_watcher(uv_udp_t* handle, uv__io_t* w) {  if (!uv__io_active(w)) return;  uv__io_stop(handle->loop, w);  if (!uv__io_active(&handle->read_watcher) &&      !uv__io_active(&handle->write_watcher))  {    uv__handle_stop(handle);  }}

int uv__udp_recv_stop(uv_udp_t* handle) {  uv__io_stop(handle->loop, &handle->io_watcher, UV__POLLIN);  if (!uv__io_active(&handle->io_watcher, UV__POLLOUT))    uv__handle_stop(handle);  handle->alloc_cb = NULL;  handle->recv_cb = NULL;  return 0;}

int uv_read_stop(uv_stream_t* stream) {  if (!(stream->flags & UV_STREAM_READING))    return 0;  stream->flags &= ~UV_STREAM_READING;  uv__io_stop(stream->loop, &stream->io_watcher, UV__POLLIN);  if (!uv__io_active(&stream->io_watcher, UV__POLLOUT))    uv__handle_stop(stream);  stream->read_cb = NULL;  stream->alloc_cb = NULL;  return 0;}

void uv__loop_delete(uv_loop_t* loop) {  ev_loop_destroy(loop->ev);#if __linux__  if (loop->inotify_fd != -1) {    uv__io_stop(loop, &loop->inotify_read_watcher);    close(loop->inotify_fd);    loop->inotify_fd = -1;  }#endif#if HAVE_PORTS_FS  if (loop->fs_fd != -1)    close(loop->fs_fd);#endif}

void uv__async_stop(uv_loop_t* loop) {  if (loop->async_io_watcher.fd == -1)    return;  if (loop->async_wfd != -1) {    if (loop->async_wfd != loop->async_io_watcher.fd)      uv__close(loop->async_wfd);    loop->async_wfd = -1;  }  uv__io_stop(loop, &loop->async_io_watcher, POLLIN);  uv__close(loop->async_io_watcher.fd);  loop->async_io_watcher.fd = -1;}

void uv__async_stop(uv_loop_t* loop, struct uv__async* wa) {    if (wa->io_watcher.fd == -1)        return;    if (wa->wfd != -1) {        if (wa->wfd != wa->io_watcher.fd)            uv__close(wa->wfd);        wa->wfd = -1;    }    uv__io_stop(loop, &wa->io_watcher, POLLIN);    uv__close(wa->io_watcher.fd);    wa->io_watcher.fd = -1;}

void uv__stream_close(uv_stream_t* handle) {  uv_read_stop(handle);  uv__io_stop(handle->loop, &handle->write_watcher);  close(handle->fd);  handle->fd = -1;  if (handle->accepted_fd >= 0) {    close(handle->accepted_fd);    handle->accepted_fd = -1;  }  assert(!uv__io_active(&handle->read_watcher));  assert(!uv__io_active(&handle->write_watcher));}

int uv_read_stop(uv_stream_t* stream) {  uv__io_stop(stream->loop, &stream->io_watcher, UV__POLLIN);  uv__handle_stop(stream);  stream->flags &= ~UV_STREAM_READING;#if defined(__APPLE__)  /* Notify select() thread about state change */  if (stream->select != NULL)    uv__stream_osx_interrupt_select(stream);#endif /* defined(__APPLE__) */  stream->read_cb = NULL;  stream->read2_cb = NULL;  stream->alloc_cb = NULL;  return 0;}

/** * We get called here from directly following a call to connect(2). * In order to determine if we've errored out or succeeded must call * getsockopt. */static void uv__stream_connect(uv_stream_t* stream) {  int error;  uv_connect_t* req = stream->connect_req;  socklen_t errorsize = sizeof(int);  assert(stream->type == UV_TCP || stream->type == UV_NAMED_PIPE);  assert(req);  if (stream->delayed_error) {    /* To smooth over the differences between unixes errors that     * were reported synchronously on the first connect can be delayed     * until the next tick--which is now.     */    error = stream->delayed_error;    stream->delayed_error = 0;  } else {    /* Normal situation: we need to get the socket error from the kernel. */    assert(uv__stream_fd(stream) >= 0);    tuvp_getsockopt(uv__stream_fd(stream),                    SOL_SOCKET,                    SO_ERROR,                    &error,                    &errorsize);    error = -error;  }  if (error == -EINPROGRESS)    return;  stream->connect_req = NULL;  uv__req_unregister(stream->loop, req);  if (error < 0 || QUEUE_EMPTY(&stream->write_queue)) {    uv__io_stop(stream->loop, &stream->io_watcher, UV__POLLOUT);  }  if (req->cb)    req->cb(req, error);  if (uv__stream_fd(stream) == -1)    return;  if (error < 0) {    uv__stream_flush_write_queue(stream, -ECANCELED);    uv__write_callbacks(stream);  }}

int uv_try_write(uv_stream_t* stream,                 const uv_buf_t bufs[],                 unsigned int nbufs) {  int r;  int has_pollout;  size_t written;  size_t req_size;  uv_write_t req;  /* Connecting or already writing some data */  if (stream->connect_req != NULL || stream->write_queue_size != 0)    return -EAGAIN;  has_pollout = uv__io_active(&stream->io_watcher, UV__POLLOUT);  r = uv_write(&req, stream, bufs, nbufs, uv_try_write_cb);  if (r != 0)    return r;  /* Remove not written bytes from write queue size */  written = uv__count_bufs(bufs, nbufs);  if (req.bufs != NULL)    req_size = uv__write_req_size(&req);  else    req_size = 0;  written -= req_size;  stream->write_queue_size -= req_size;  /* Unqueue request, regardless of immediateness */  QUEUE_REMOVE(&req.queue);  uv__req_unregister(stream->loop, &req);  if (req.bufs != req.bufsml)    free(req.bufs);  req.bufs = NULL;  /* Do not poll for writable, if we wasn't before calling this */  if (!has_pollout) {    uv__io_stop(stream->loop, &stream->io_watcher, UV__POLLOUT);  }  if (written == 0)    return -EAGAIN;  else    return written;}

void uv__server_io(uv_loop_t* loop, uv__io_t* w, unsigned int events) {  uv_stream_t* stream;  int err;  stream = container_of(w, uv_stream_t, io_watcher);  assert(events == UV__POLLIN);  assert(stream->accepted_fd == -1);  assert(!(stream->flags & UV_CLOSING));  uv__io_start(stream->loop, &stream->io_watcher, UV__POLLIN);  /* connection_cb can close the server socket while we're   * in the loop so check it on each iteration.   */  while (uv__stream_fd(stream) != -1) {    assert(stream->accepted_fd == -1);    set_errno(0);    err = uv__accept(uv__stream_fd(stream));    if (err < 0) {      if (err == -EAGAIN || err == -EWOULDBLOCK)        return;  /* Not an error. */      if (err == -ECONNABORTED)        continue;  /* Ignore. Nothing we can do about that. */      stream->connection_cb(stream, err);      continue;    }    stream->accepted_fd = err;    stream->connection_cb(stream, 0);    if (stream->accepted_fd != -1) {      /* The user hasn't yet accepted called uv_accept() */      uv__io_stop(loop, &stream->io_watcher, UV__POLLIN);      return;    }    /* done accept for mbed */    break;  }}

static void uv__udp_run_completed(uv_udp_t* handle) {  uv_udp_send_t* req;  QUEUE* q;  assert(!(handle->flags & UV_UDP_PROCESSING));  handle->flags |= UV_UDP_PROCESSING;  while (!QUEUE_EMPTY(&handle->write_completed_queue)) {    q = QUEUE_HEAD(&handle->write_completed_queue);    QUEUE_REMOVE(q);    req = QUEUE_DATA(q, uv_udp_send_t, queue);    uv__req_unregister(handle->loop, req);    handle->send_queue_size -= uv__count_bufs(req->bufs, req->nbufs);    handle->send_queue_count--;    if (req->bufs != req->bufsml)      uv__free(req->bufs);    req->bufs = NULL;    if (req->send_cb == NULL)      continue;    /* req->status >= 0 == bytes written     * req->status <  0 == errno     */    if (req->status >= 0)      req->send_cb(req, 0);    else      req->send_cb(req, req->status);  }  if (QUEUE_EMPTY(&handle->write_queue)) {    /* Pending queue and completion queue empty, stop watcher. */    uv__io_stop(handle->loop, &handle->io_watcher, UV__POLLOUT);    if (!uv__io_active(&handle->io_watcher, UV__POLLIN))      uv__handle_stop(handle);  }  handle->flags &= ~UV_UDP_PROCESSING;}

static void uv__poll_io(uv_loop_t* loop, uv__io_t* w, unsigned int events) {  uv_poll_t* handle;  int pevents;  handle = container_of(w, uv_poll_t, io_watcher);  if (events & UV__POLLERR) {    uv__io_stop(loop, w, UV__POLLIN | UV__POLLOUT);    uv__handle_stop(handle);    uv__set_sys_error(handle->loop, EBADF);    handle->poll_cb(handle, -1, 0);    return;  }  pevents = 0;  if (events & UV__POLLIN) pevents |= UV_READABLE;  if (events & UV__POLLOUT) pevents |= UV_WRITABLE;  handle->poll_cb(handle, 0, pevents);}

static void uv__poll_io(uv_loop_t* loop, uv__io_t* w, unsigned int events) {  uv_poll_t* handle;  int pevents;  handle = container_of(w, uv_poll_t, io_watcher);  if (events & POLLERR) {    uv__io_stop(loop, w, POLLIN | POLLOUT | UV__POLLRDHUP);    uv__handle_stop(handle);    handle->poll_cb(handle, -EBADF, 0);    return;  }  pevents = 0;  if (events & POLLIN)    pevents |= UV_READABLE;  if (events & POLLOUT)    pevents |= UV_WRITABLE;  if (events & UV__POLLRDHUP)    pevents |= UV_DISCONNECT;  handle->poll_cb(handle, 0, pevents);}

static void uv__poll_io(uv_loop_t* loop, uv__io_t* w, unsigned int events) {  uv_poll_t* handle;  int pevents;  handle = container_of(w, uv_poll_t, io_watcher);  /*   * As documented in the kernel source fs/kernfs/file.c #780   * poll will return POLLERR|POLLPRI in case of sysfs   * polling. This does not happen in case of out-of-band   * TCP messages.   *   * The above is the case on (at least) FreeBSD and Linux.   *   * So to properly determine a POLLPRI or a POLLERR we need   * to check for both.   */  if ((events & POLLERR) && !(events & UV__POLLPRI)) {    uv__io_stop(loop, w, POLLIN | POLLOUT | UV__POLLRDHUP | UV__POLLPRI);    uv__handle_stop(handle);    handle->poll_cb(handle, UV_EBADF, 0);    return;  }  pevents = 0;  if (events & POLLIN)    pevents |= UV_READABLE;  if (events & UV__POLLPRI)    pevents |= UV_PRIORITIZED;  if (events & POLLOUT)    pevents |= UV_WRITABLE;  if (events & UV__POLLRDHUP)    pevents |= UV_DISCONNECT;  handle->poll_cb(handle, 0, pevents);}

static void uv__read(uv_stream_t* stream) {  uv_buf_t buf;  ssize_t nread;  struct msghdr msg;  struct cmsghdr* cmsg;  char cmsg_space[64];  int count;  /* Prevent loop starvation when the data comes in as fast as (or faster than)   * we can read it. XXX Need to rearm fd if we switch to edge-triggered I/O.   */  count = 32;  /* XXX: Maybe instead of having UV_STREAM_READING we just test if   * tcp->read_cb is NULL or not?   */  while ((stream->read_cb || stream->read2_cb)      && (stream->flags & UV_STREAM_READING)      && (count-- > 0)) {    assert(stream->alloc_cb);    buf = stream->alloc_cb((uv_handle_t*)stream, 64 * 1024);    assert(buf.len > 0);    assert(buf.base);    assert(uv__stream_fd(stream) >= 0);    if (stream->read_cb) {      do {        nread = read(uv__stream_fd(stream), buf.base, buf.len);      }      while (nread < 0 && errno == EINTR);    } else {      assert(stream->read2_cb);      /* read2_cb uses recvmsg */      msg.msg_flags = 0;      msg.msg_iov = (struct iovec*) &buf;      msg.msg_iovlen = 1;      msg.msg_name = NULL;      msg.msg_namelen = 0;      /* Set up to receive a descriptor even if one isn't in the message */      msg.msg_controllen = 64;      msg.msg_control = (void *) cmsg_space;      do {        nread = recvmsg(uv__stream_fd(stream), &msg, 0);      }      while (nread < 0 && errno == EINTR);    }    if (nread < 0) {      /* Error */      if (errno == EAGAIN || errno == EWOULDBLOCK) {        /* Wait for the next one. */        if (stream->flags & UV_STREAM_READING) {          uv__io_start(stream->loop, &stream->io_watcher, UV__POLLIN);        }        uv__set_sys_error(stream->loop, EAGAIN);        if (stream->read_cb) {          stream->read_cb(stream, 0, buf);        } else {          stream->read2_cb((uv_pipe_t*)stream, 0, buf, UV_UNKNOWN_HANDLE);        }        return;      } else {        /* Error. User should call uv_close(). */        uv__set_sys_error(stream->loop, errno);        if (stream->read_cb) {          stream->read_cb(stream, -1, buf);        } else {          stream->read2_cb((uv_pipe_t*)stream, -1, buf, UV_UNKNOWN_HANDLE);        }        assert(!uv__io_active(&stream->io_watcher, UV__POLLIN));        return;      }    } else if (nread == 0) {      /* EOF */      uv__set_artificial_error(stream->loop, UV_EOF);      uv__io_stop(stream->loop, &stream->io_watcher, UV__POLLIN);      if (!uv__io_active(&stream->io_watcher, UV__POLLOUT))        uv__handle_stop(stream);      if (stream->read_cb) {        stream->read_cb(stream, -1, buf);      } else {        stream->read2_cb((uv_pipe_t*)stream, -1, buf, UV_UNKNOWN_HANDLE);      }      return;    } else {      /* Successful read */      ssize_t buflen = buf.len;      if (stream->read_cb) {        stream->read_cb(stream, nread, buf);//.........这里部分代码省略.........

void uv__platform_loop_delete(uv_loop_t* loop) {  if (loop->inotify_fd == -1) return;  uv__io_stop(loop, &loop->inotify_read_watcher, UV__POLLIN);  uv__close(loop->inotify_fd);  loop->inotify_fd = -1;}