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

void uv_pipe_connect(uv_connect_t* req,                     uv_pipe_t* handle,                     const char* name,                     uv_connect_cb cb) {    struct sockaddr_un saddr;    int saved_errno;    int new_sock;    int err;    int r;    saved_errno = errno;    new_sock = (handle->fd == -1);    err = -1;    if (new_sock)        if ((handle->fd = uv__socket(AF_UNIX, SOCK_STREAM, 0)) == -1)            goto out;    memset(&saddr, 0, sizeof saddr);    uv_strlcpy(saddr.sun_path, name, sizeof(saddr.sun_path));    saddr.sun_family = AF_UNIX;    /* We don't check for EINPROGRESS. Think about it: the socket     * is either there or not.     */    do {        r = connect(handle->fd, (struct sockaddr*)&saddr, sizeof saddr);    }    while (r == -1 && errno == EINTR);    if (r == -1)        goto out;    if (new_sock)        if (uv__stream_open((uv_stream_t*)handle,                            handle->fd,                            UV_STREAM_READABLE | UV_STREAM_WRITABLE))            goto out;    uv__io_start(handle->loop, &handle->read_watcher);    uv__io_start(handle->loop, &handle->write_watcher);    err = 0;out:    handle->delayed_error = err ? errno : 0; /* Passed to callback. */    handle->connect_req = req;    uv__req_init(handle->loop, req, UV_CONNECT);    req->handle = (uv_stream_t*)handle;    req->cb = cb;    ngx_queue_init(&req->queue);    /* Run callback on next tick. */    uv__io_feed(handle->loop, &handle->write_watcher, UV__IO_WRITE);    /* Mimic the Windows pipe implementation, always     * return 0 and let the callback handle errors.     */    errno = saved_errno;}

int uv_tcp_listen(uv_tcp_t* tcp, int backlog, uv_connection_cb cb) {  static int single_accept = -1;  if (tcp->delayed_error)    return uv__set_sys_error(tcp->loop, tcp->delayed_error);  if (single_accept == -1) {    const char* val = getenv("UV_TCP_SINGLE_ACCEPT");    single_accept = (val != NULL && atoi(val) != 0);  /* Off by default. */  }  if (single_accept)    tcp->flags |= UV_TCP_SINGLE_ACCEPT;  if (maybe_new_socket(tcp, AF_INET, UV_STREAM_READABLE))    return -1;  if (listen(tcp->io_watcher.fd, backlog))    return uv__set_sys_error(tcp->loop, errno);  tcp->connection_cb = cb;  /* Start listening for connections. */  tcp->io_watcher.cb = uv__server_io;  uv__io_start(tcp->loop, &tcp->io_watcher, UV__POLLIN);  return 0;}

int uv_shutdown(uv_shutdown_t* req, uv_stream_t* stream, uv_shutdown_cb cb) {  assert((stream->type == UV_TCP || stream->type == UV_NAMED_PIPE) &&         "uv_shutdown (unix) only supports uv_handle_t right now");  if (!(stream->flags & UV_STREAM_WRITABLE) ||      stream->flags & UV_STREAM_SHUT ||      stream->flags & UV_STREAM_SHUTTING ||      stream->flags & UV_CLOSED ||      stream->flags & UV_CLOSING) {    TDLOG("uv_shutdown: ENOTCONN");    return -ENOTCONN;  }  assert(uv__stream_fd(stream) >= 0);  /* Initialize request */  uv__req_init(stream->loop, req, UV_SHUTDOWN);  req->handle = stream;  req->cb = cb;  stream->shutdown_req = req;  stream->flags |= UV_STREAM_SHUTTING;  uv__io_start(stream->loop, &stream->io_watcher, UV__POLLOUT);  return 0;}

int uv_pipe_listen(uv_pipe_t* handle, int backlog, uv_connection_cb cb) {  int saved_errno;  int status;  saved_errno = errno;  status = -1;  if (uv__stream_fd(handle) == -1) {    uv__set_artificial_error(handle->loop, UV_EINVAL);    goto out;  }  assert(uv__stream_fd(handle) >= 0);  if ((status = listen(uv__stream_fd(handle), backlog)) == -1) {    uv__set_sys_error(handle->loop, errno);  } else {    handle->connection_cb = cb;    handle->io_watcher.cb = uv__pipe_accept;    uv__io_start(handle->loop, &handle->io_watcher, UV__POLLIN);  }out:  errno = saved_errno;  return status;}

int uv_read_start(uv_stream_t* stream,                  uv_alloc_cb alloc_cb,                  uv_read_cb read_cb) {  assert(stream->type == UV_TCP || stream->type == UV_NAMED_PIPE ||         stream->type == UV_TTY);  if (stream->flags & UV_CLOSING)    return -EINVAL;  /* The UV_STREAM_READING flag is irrelevant of the state of the tcp - it just   * expresses the desired state of the user.   */  stream->flags |= UV_STREAM_READING;  /* TODO: try to do the read inline? */  /* TODO: keep track of tcp state. If we've gotten a EOF then we should   * not start the IO watcher.   */  assert(uv__stream_fd(stream) >= 0);  assert(alloc_cb);  stream->read_cb = read_cb;  stream->alloc_cb = alloc_cb;  uv__io_start(stream->loop, &stream->io_watcher, UV__POLLIN);  uv__handle_start(stream);  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(stream->fd >= 0);    getsockopt(stream->fd, SOL_SOCKET, SO_ERROR, &error, &errorsize);  }  if (error == EINPROGRESS)    return;  if (error == 0)    uv__io_start(stream->loop, &stream->read_watcher);  stream->connect_req = NULL;  uv__req_unregister(stream->loop, req);  if (req->cb) {    uv__set_sys_error(stream->loop, error);    req->cb(req, error ? -1 : 0);  }}

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;}

int uv_fs_event_init(uv_loop_t* loop,                     uv_fs_event_t* handle,                     const char* filename,                     uv_fs_event_cb cb,                     int flags) {  int portfd;  int first_run = 0;  if (loop->fs_fd == -1) {    if ((portfd = port_create()) == -1) {      uv__set_sys_error(loop, errno);      return -1;    }    loop->fs_fd = portfd;    first_run = 1;  }  uv__handle_init(loop, (uv_handle_t*)handle, UV_FS_EVENT);  uv__handle_start(handle); /* FIXME shouldn't start automatically */  handle->filename = strdup(filename);  handle->fd = PORT_UNUSED;  handle->cb = cb;  memset(&handle->fo, 0, sizeof handle->fo);  handle->fo.fo_name = handle->filename;  uv__fs_event_rearm(handle);  if (first_run) {    uv__io_init(&loop->fs_event_watcher, uv__fs_event_read, portfd);    uv__io_start(loop, &loop->fs_event_watcher, UV__POLLIN);  }  return 0;}

int uv_poll_start(uv_poll_t* handle, int pevents, uv_poll_cb poll_cb) {  int events;  assert((pevents & ~(UV_READABLE | UV_WRITABLE | UV_DISCONNECT)) == 0);  assert(!uv__is_closing(handle));  uv__poll_stop(handle);  if (pevents == 0)    return 0;  events = 0;  if (pevents & UV_READABLE)    events |= POLLIN;  if (pevents & UV_WRITABLE)    events |= POLLOUT;  if (pevents & UV_DISCONNECT)    events |= UV__POLLRDHUP;  uv__io_start(handle->loop, &handle->io_watcher, events);  uv__handle_start(handle);  handle->poll_cb = poll_cb;  return 0;}

static int uv__send(uv_udp_send_t* req,                    uv_udp_t* handle,                    uv_buf_t bufs[],                    int bufcnt,                    struct sockaddr* addr,                    socklen_t addrlen,                    uv_udp_send_cb send_cb) {  assert(bufcnt > 0);  if (uv__udp_maybe_deferred_bind(handle, addr->sa_family))    return -1;  uv__req_init(handle->loop, req, UV_UDP_SEND);  assert(addrlen <= sizeof(req->addr));  memcpy(&req->addr, addr, addrlen);  req->send_cb = send_cb;  req->handle = handle;  req->bufcnt = bufcnt;  if (bufcnt <= (int) ARRAY_SIZE(req->bufsml)) {    req->bufs = req->bufsml;  }  else if ((req->bufs = malloc(bufcnt * sizeof(bufs[0]))) == NULL) {    uv__set_sys_error(handle->loop, ENOMEM);    return -1;  }  memcpy(req->bufs, bufs, bufcnt * sizeof(bufs[0]));  ngx_queue_insert_tail(&handle->write_queue, &req->queue);  uv__io_start(handle->loop, &handle->io_watcher, UV__POLLOUT);  uv__handle_start(handle);  return 0;}

int uv_fs_event_start(uv_fs_event_t* handle,                      uv_fs_event_cb cb,                      const char* filename,                      unsigned int flags) {  int portfd;  int first_run;  if (uv__is_active(handle))    return -EINVAL;  first_run = 0;  if (handle->loop->fs_fd == -1) {    portfd = port_create();    if (portfd == -1)      return -errno;    handle->loop->fs_fd = portfd;    first_run = 1;  }  uv__handle_start(handle);  handle->filename = strdup(filename);  handle->fd = PORT_UNUSED;  handle->cb = cb;  memset(&handle->fo, 0, sizeof handle->fo);  handle->fo.fo_name = handle->filename;  uv__fs_event_rearm(handle);  /* FIXME(bnoordhuis) Check return code. */  if (first_run) {    uv__io_init(&handle->loop->fs_event_watcher, uv__fs_event_read, portfd);    uv__io_start(handle->loop, &handle->loop->fs_event_watcher, UV__POLLIN);  }  return 0;}

void uv_pipe_connect(uv_connect_t* req,                    uv_pipe_t* handle,                    const char* name,                    uv_connect_cb cb) {  struct sockaddr_un saddr;  int saved_errno;  int new_sock;  int err;  int r;  saved_errno = errno;  new_sock = (uv__stream_fd(handle) == -1);  err = -1;  if (new_sock)    if ((handle->io_watcher.fd = uv__socket(AF_UNIX, SOCK_STREAM, 0)) == -1)      goto out;  memset(&saddr, 0, sizeof saddr);  uv_strlcpy(saddr.sun_path, name, sizeof(saddr.sun_path));  saddr.sun_family = AF_UNIX;  do {    r = connect(uv__stream_fd(handle),                (struct sockaddr*)&saddr, sizeof saddr);  }  while (r == -1 && errno == EINTR);  if (r == -1)    if (errno != EINPROGRESS)      goto out;  if (new_sock)    if (uv__stream_open((uv_stream_t*)handle,                        uv__stream_fd(handle),                        UV_STREAM_READABLE | UV_STREAM_WRITABLE))      goto out;  uv__io_start(handle->loop, &handle->io_watcher, UV__POLLIN | UV__POLLOUT);  err = 0;out:  handle->delayed_error = err ? errno : 0; /* Passed to callback. */  handle->connect_req = req;  uv__req_init(handle->loop, req, UV_CONNECT);  req->handle = (uv_stream_t*)handle;  req->cb = cb;  QUEUE_INIT(&req->queue);  /* Force callback to run on next tick in case of error. */  if (err != 0)    uv__io_feed(handle->loop, &handle->io_watcher);  /* Mimic the Windows pipe implementation, always   * return 0 and let the callback handle errors.   */  errno = saved_errno;}

static void uv__next_accept(uv_idle_t* idle, int status) {  uv_stream_t* stream = idle->data;  uv_idle_stop(idle);  if (stream->accepted_fd == -1)    uv__io_start(stream->loop, &stream->read_watcher);}

int uv__tcp_connect(uv_connect_t* req,                    uv_tcp_t* handle,                    const struct sockaddr* addr,                    unsigned int addrlen,                    uv_connect_cb cb) {  int err;  int r;  assert(handle->type == UV_TCP);  if (handle->connect_req != NULL)    return -EALREADY;  /* FIXME(bnoordhuis) -EINVAL or maybe -EBUSY. */  err = maybe_new_socket(handle,                         addr->sa_family,                         UV_STREAM_READABLE | UV_STREAM_WRITABLE);  if (err)    return err;  handle->delayed_error = 0;  do {    errno = 0;    r = connect(uv__stream_fd(handle), addr, addrlen);  } while (r == -1 && errno == EINTR);  /* We not only check the return value, but also check the errno != 0.   * Because in rare cases connect() will return -1 but the errno   * is 0 (for example, on Android 4.3, OnePlus phone A0001_12_150227)   * and actually the tcp three-way handshake is completed.   */  if (r == -1 && errno != 0) {    if (errno == EINPROGRESS)      ; /* not an error */    else if (errno == ECONNREFUSED)    /* If we get a ECONNREFUSED wait until the next tick to report the     * error. Solaris wants to report immediately--other unixes want to     * wait.     */      handle->delayed_error = -errno;    else      return -errno;  }  uv__req_init(handle->loop, req, UV_CONNECT);  req->cb = cb;  req->handle = (uv_stream_t*) handle;  QUEUE_INIT(&req->queue);  handle->connect_req = req;  uv__io_start(handle->loop, &handle->io_watcher, POLLOUT);  if (handle->delayed_error)    uv__io_feed(handle->loop, &handle->io_watcher);  return 0;}

static void uv__udp_start_watcher(uv_udp_t* handle,                                  uv__io_t* w,                                  uv__io_cb cb,                                  int events) {  if (uv__io_active(w)) return;  uv__io_init(w, cb, handle->fd, events);  uv__io_start(handle->loop, w);  uv__handle_start(handle);}

int uv_fs_event_start(uv_fs_event_t* handle,                      uv_fs_event_cb cb,                      const char* filename,                      unsigned int flags) {#ifdef HAVE_SYS_AHAFS_EVPRODS_H  int  fd, rc, str_offset = 0;  char cwd[PATH_MAX];  char absolute_path[PATH_MAX];  char readlink_cwd[PATH_MAX];  /* Figure out whether filename is absolute or not */  if (filename[0] == '/') {    /* We have absolute pathname */    snprintf(absolute_path, sizeof(absolute_path), "%s", filename);  } else {    /* We have a relative pathname, compose the absolute pathname */    snprintf(cwd, sizeof(cwd), "/proc/%lu/cwd", (unsigned long) getpid());    rc = readlink(cwd, readlink_cwd, sizeof(readlink_cwd) - 1);    if (rc < 0)      return rc;    /* readlink does not null terminate our string */    readlink_cwd[rc] = '/0';    if (filename[0] == '.' && filename[1] == '/')      str_offset = 2;    snprintf(absolute_path, sizeof(absolute_path), "%s%s", readlink_cwd,             filename + str_offset);  }  if (uv__is_ahafs_mounted() < 0)  /* /aha checks failed */    return UV_ENOSYS;  /* Setup ahafs */  rc = uv__setup_ahafs((const char *)absolute_path, &fd);  if (rc != 0)    return rc;  /* Setup/Initialize all the libuv routines */  uv__handle_start(handle);  uv__io_init(&handle->event_watcher, uv__ahafs_event, fd);  handle->path = uv__strdup(filename);  handle->cb = cb;  handle->dir_filename = NULL;  uv__io_start(handle->loop, &handle->event_watcher, POLLIN);  return 0;#else  return -ENOSYS;#endif}

int uv_pipe_listen(uv_pipe_t* handle, int backlog, uv_connection_cb cb) {  if (uv__stream_fd(handle) == -1)    return -EINVAL;  if (listen(uv__stream_fd(handle), backlog))    return -errno;  handle->connection_cb = cb;  handle->io_watcher.cb = uv__server_io;  uv__io_start(handle->loop, &handle->io_watcher, UV__POLLIN);  return 0;}

int uv_accept(uv_stream_t* server, uv_stream_t* client) {  uv_stream_t* streamServer;  uv_stream_t* streamClient;  int saved_errno;  int status;  /* TODO document this */  assert(server->loop == client->loop);  saved_errno = errno;  status = -1;  streamServer = (uv_stream_t*)server;  streamClient = (uv_stream_t*)client;  if (streamServer->accepted_fd < 0) {    uv__set_sys_error(server->loop, EAGAIN);    goto out;  }  switch (streamClient->type) {    case UV_NAMED_PIPE:    case UV_TCP:      if (uv__stream_open(streamClient, streamServer->accepted_fd,            UV_STREAM_READABLE | UV_STREAM_WRITABLE)) {        /* TODO handle error */        close(streamServer->accepted_fd);        streamServer->accepted_fd = -1;        goto out;      }      break;    case UV_UDP:      if (uv_udp_open((uv_udp_t*) client, streamServer->accepted_fd)) {        close(streamServer->accepted_fd);        streamServer->accepted_fd = -1;        goto out;      }      break;    default:      assert(0);  }  uv__io_start(streamServer->loop, &streamServer->io_watcher, UV__POLLIN);  streamServer->accepted_fd = -1;  status = 0;out:  errno = saved_errno;  return status;}

static int uv__connect(uv_connect_t* req,                       uv_tcp_t* handle,                       struct sockaddr* addr,                       socklen_t addrlen,                       uv_connect_cb cb) {  int err;  int r;  assert(handle->type == UV_TCP);  if (handle->connect_req != NULL)    return -EALREADY;  /* FIXME(bnoordhuis) -EINVAL or maybe -EBUSY. */  err = maybe_new_socket(handle,                         addr->sa_family,                         UV_STREAM_READABLE | UV_STREAM_WRITABLE);  if (err)    return err;  handle->delayed_error = 0;  do    r = connect(uv__stream_fd(handle), addr, addrlen);  while (r == -1 && errno == EINTR);  if (r == -1) {    if (errno == EINPROGRESS)      ; /* not an error */    else if (errno == ECONNREFUSED)    /* If we get a ECONNREFUSED wait until the next tick to report the     * error. Solaris wants to report immediately--other unixes want to     * wait.     */      handle->delayed_error = -errno;    else      return -errno;  }  uv__req_init(handle->loop, req, UV_CONNECT);  req->cb = cb;  req->handle = (uv_stream_t*) handle;  QUEUE_INIT(&req->queue);  handle->connect_req = req;  uv__io_start(handle->loop, &handle->io_watcher, UV__POLLOUT);  if (handle->delayed_error)    uv__io_feed(handle->loop, &handle->io_watcher);  return 0;}

static int uv__connect(uv_connect_t* req,                       uv_tcp_t* handle,                       struct sockaddr* addr,                       socklen_t addrlen,                       uv_connect_cb cb) {  int r;  assert(handle->type == UV_TCP);  if (handle->connect_req)    return uv__set_sys_error(handle->loop, EALREADY);  if (maybe_new_socket(handle,                       addr->sa_family,                       UV_STREAM_READABLE|UV_STREAM_WRITABLE)) {    return -1;  }  handle->delayed_error = 0;  do    r = connect(handle->fd, addr, addrlen);  while (r == -1 && errno == EINTR);  if (r == -1) {    if (errno == EINPROGRESS)      ; /* not an error */    else if (errno == ECONNREFUSED)    /* If we get a ECONNREFUSED wait until the next tick to report the     * error. Solaris wants to report immediately--other unixes want to     * wait.     */      handle->delayed_error = errno;    else      return uv__set_sys_error(handle->loop, errno);  }  uv__req_init(handle->loop, req, UV_CONNECT);  req->cb = cb;  req->handle = (uv_stream_t*) handle;  ngx_queue_init(&req->queue);  handle->connect_req = req;  uv__io_start(handle->loop, &handle->write_watcher);  if (handle->delayed_error)    uv__io_feed(handle->loop, &handle->write_watcher, UV__IO_WRITE);  return 0;}

int uv__udp_send(uv_udp_send_t* req,                 uv_udp_t* handle,                 const uv_buf_t bufs[],                 unsigned int nbufs,                 const struct sockaddr* addr,                 unsigned int addrlen,                 uv_udp_send_cb send_cb) {  int err;  int empty_queue;  assert(nbufs > 0);  err = uv__udp_maybe_deferred_bind(handle, addr->sa_family, 0);  if (err)    return err;  /* It's legal for send_queue_count > 0 even when the write_queue is empty;   * it means there are error-state requests in the write_completed_queue that   * will touch up send_queue_size/count later.   */  empty_queue = (handle->send_queue_count == 0);  uv__req_init(handle->loop, req, UV_UDP_SEND);  assert(addrlen <= sizeof(req->addr));  memcpy(&req->addr, addr, addrlen);  req->send_cb = send_cb;  req->handle = handle;  req->nbufs = nbufs;  req->bufs = req->bufsml;  if (nbufs > ARRAY_SIZE(req->bufsml))    req->bufs = uv__malloc(nbufs * sizeof(bufs[0]));  if (req->bufs == NULL)    return -ENOMEM;  memcpy(req->bufs, bufs, nbufs * sizeof(bufs[0]));  handle->send_queue_size += uv__count_bufs(req->bufs, req->nbufs);  handle->send_queue_count++;  QUEUE_INSERT_TAIL(&handle->write_queue, &req->queue);  uv__handle_start(handle);  if (empty_queue && !(handle->flags & UV_UDP_PROCESSING)) {    uv__udp_sendmsg(handle);  } else {    uv__io_start(handle->loop, &handle->io_watcher, UV__POLLOUT);  }  return 0;}

int uv_fs_event_start(uv_fs_event_t* handle,                      uv_fs_event_cb cb,                      const char* path,                      unsigned int flags) {#if defined(__APPLE__)  struct stat statbuf;#endif /* defined(__APPLE__) */  int fd;  if (uv__is_active(handle))    return -EINVAL;  /* TODO open asynchronously - but how do we report back errors? */  fd = open(path, O_RDONLY);  if (fd == -1)    return -errno;  uv__handle_start(handle);  uv__io_init(&handle->event_watcher, uv__fs_event, fd);  handle->path = uv__strdup(path);  handle->cb = cb;#if defined(__APPLE__)  /* Nullify field to perform checks later */  handle->cf_cb = NULL;  handle->realpath = NULL;  handle->realpath_len = 0;  handle->cf_flags = flags;  if (fstat(fd, &statbuf))    goto fallback;  /* FSEvents works only with directories */  if (!(statbuf.st_mode & S_IFDIR))    goto fallback;  /* The fallback fd is no longer needed */  uv__close(fd);  handle->event_watcher.fd = -1;  return uv__fsevents_init(handle);fallback:#endif /* defined(__APPLE__) */  uv__io_start(handle->loop, &handle->event_watcher, POLLIN);  return 0;}

static int init_inotify(uv_loop_t* loop) {  int err;  if (loop->inotify_fd != -1)    return 0;  err = new_inotify_fd();  if (err < 0)    return err;  loop->inotify_fd = err;  uv__io_init(&loop->inotify_read_watcher, uv__inotify_read, loop->inotify_fd);  uv__io_start(loop, &loop->inotify_read_watcher, UV__POLLIN);  return 0;}

int uv__async_start(uv_loop_t* loop, struct uv__async* wa, uv__async_cb cb) {    int pipefd[2];    int err;    if (wa->io_watcher.fd != -1)        return 0;    err = uv__async_eventfd();    if (err >= 0) {        pipefd[0] = err;        pipefd[1] = -1;    }    else if (err == -ENOSYS) {        err = uv__make_pipe(pipefd, UV__F_NONBLOCK);#if defined(__linux__)        /* Save a file descriptor by opening one of the pipe descriptors as         * read/write through the procfs.  That file descriptor can then         * function as both ends of the pipe.         */        if (err == 0) {            char buf[32];            int fd;            snprintf(buf, sizeof(buf), "/proc/self/fd/%d", pipefd[0]);            fd = uv__open_cloexec(buf, O_RDWR);            if (fd >= 0) {                uv__close(pipefd[0]);                uv__close(pipefd[1]);                pipefd[0] = fd;                pipefd[1] = fd;            }        }#endif    }    if (err < 0)        return err;    uv__io_init(&wa->io_watcher, uv__async_io, pipefd[0]);    uv__io_start(loop, &wa->io_watcher, POLLIN);    wa->wfd = pipefd[1];    wa->cb = cb;    return 0;}

int uv_tcp_listen(uv_tcp_t* tcp, int backlog, uv_connection_cb cb) {  if (tcp->delayed_error)    return uv__set_sys_error(tcp->loop, tcp->delayed_error);  if (maybe_new_socket(tcp, AF_INET, UV_STREAM_READABLE))    return -1;  if (listen(tcp->fd, backlog))    return uv__set_sys_error(tcp->loop, errno);  tcp->connection_cb = cb;  /* Start listening for connections. */  uv__io_set(&tcp->read_watcher, uv__server_io, tcp->fd, UV__IO_READ);  uv__io_start(tcp->loop, &tcp->read_watcher);  return 0;}

int uv_fs_event_init(uv_loop_t* loop,                     uv_fs_event_t* handle,                     const char* filename,                     uv_fs_event_cb cb,                     int flags) {#if defined(__APPLE__)  struct stat statbuf;#endif /* defined(__APPLE__) */  int fd;  /* TODO open asynchronously - but how do we report back errors? */  if ((fd = open(filename, O_RDONLY)) == -1) {    uv__set_sys_error(loop, errno);    return -1;  }  uv__handle_init(loop, (uv_handle_t*)handle, UV_FS_EVENT);  uv__handle_start(handle); /* FIXME shouldn't start automatically */  uv__io_init(&handle->event_watcher, uv__fs_event, fd);  handle->filename = strdup(filename);  handle->cb = cb;#if defined(__APPLE__)  /* Nullify field to perform checks later */  handle->cf_eventstream = NULL;  handle->realpath = NULL;  handle->realpath_len = 0;  handle->cf_flags = flags;  if (fstat(fd, &statbuf))    goto fallback;  /* FSEvents works only with directories */  if (!(statbuf.st_mode & S_IFDIR))    goto fallback;  return uv__fsevents_init(handle);fallback:#endif /* defined(__APPLE__) */  uv__io_start(loop, &handle->event_watcher, UV__POLLIN);  return 0;}

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;  }}

int uv_pipe_listen(uv_pipe_t* handle, int backlog, uv_connection_cb cb) {  if (uv__stream_fd(handle) == -1)    return -EINVAL;#if defined(__MVS__)  /* On zOS, backlog=0 has undefined behaviour */  if (backlog == 0)    backlog = 1;  else if (backlog < 0)    backlog = SOMAXCONN;#endif  if (listen(uv__stream_fd(handle), backlog))    return -errno;  handle->connection_cb = cb;  handle->io_watcher.cb = uv__server_io;  uv__io_start(handle->loop, &handle->io_watcher, POLLIN);  return 0;}

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)));  uv__poll_stop(handle);  if (pevents == 0) return 0;  events = 0;  if (pevents & UV_READABLE) events |= UV__POLLIN;  if (pevents & UV_WRITABLE) events |= UV__POLLOUT;  uv__io_start(handle->loop, &handle->io_watcher, events);  uv__handle_start(handle);  handle->poll_cb = poll_cb;  return 0;}