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

int uv_tty_init(uv_loop_t* loop, uv_tty_t* tty, int fd, int readable) {  uv__stream_init(loop, (uv_stream_t*)tty, UV_TTY);  if (readable) {    uv__nonblock(fd, 1);    uv__stream_open((uv_stream_t*)tty, fd, UV_STREAM_READABLE);  } else {    /* Note: writable tty we set to blocking mode. */    uv__stream_open((uv_stream_t*)tty, fd, UV_STREAM_WRITABLE);    tty->flags |= UV_STREAM_BLOCKING;  }  tty->mode = 0;  return 0;}

int uv_device_open(uv_loop_t* loop,                   uv_device_t* device,                   uv_os_fd_t fd,                   int flags) {    int err;    int stream_flags = 0;    assert(device);    if (flags != O_RDONLY && flags != O_WRONLY && flags != O_RDWR) {        return -EINVAL;    }    uv__stream_init(loop, (uv_stream_t*) device, UV_DEVICE);    if (flags & O_RDONLY) {        stream_flags |= UV_STREAM_READABLE;    } else if (flags & O_WRONLY) {        stream_flags |= UV_STREAM_WRITABLE;    } else if (flags & O_RDWR) {        stream_flags |= UV_STREAM_READABLE | UV_STREAM_WRITABLE;    }    err = uv__nonblock(fd, 1);    if (err) {        return err;    }    err = uv__stream_open((uv_stream_t*) device, fd, stream_flags);    if (err) {        return err;    }    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 = (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_pipe_open(uv_pipe_t* handle, uv_os_fd_t fd) {  int flags;  int mode;  int err;  flags = 0;  if (uv__fd_exists(handle->loop, fd))    return UV_EEXIST;  do    mode = fcntl(fd, F_GETFL);  while (mode == -1 && errno == EINTR);  if (mode == -1)    return UV__ERR(errno); /* according to docs, must be EBADF */  err = uv__nonblock(fd, 1);  if (err)    return err;#if defined(__APPLE__)  err = uv__stream_try_select((uv_stream_t*) handle, &fd);  if (err)    return err;#endif /* defined(__APPLE__) */  mode &= O_ACCMODE;  if (mode != O_WRONLY)    flags |= UV_HANDLE_READABLE;  if (mode != O_RDONLY)    flags |= UV_HANDLE_WRITABLE;  return uv__stream_open((uv_stream_t*)handle, fd, flags);}

static int uv__process_open_stream(uv_stdio_container_t* container,                                   int pipefds[2],                                   int writable) {  int flags;  if (!(container->flags & UV_CREATE_PIPE) || pipefds[0] < 0)    return 0;  if (uv__close(pipefds[1]))    if (errno != EINTR && errno != EINPROGRESS)      abort();  pipefds[1] = -1;  uv__nonblock(pipefds[0], 1);  if (container->data.stream->type == UV_NAMED_PIPE &&      ((uv_pipe_t*)container->data.stream)->ipc)    flags = UV_STREAM_READABLE | UV_STREAM_WRITABLE;  else if (writable)    flags = UV_STREAM_WRITABLE;  else    flags = UV_STREAM_READABLE;  return uv__stream_open(container->data.stream, pipefds[0], flags);}

int uv_tty_init(uv_loop_t* loop, uv_tty_t* tty, int fd, int readable) {  uv__stream_init(loop, (uv_stream_t*)tty, UV_TTY);  if (readable) {    uv__nonblock(fd, 1);    uv__stream_open((uv_stream_t*)tty, fd, UV_READABLE);  } else {    /* Note: writable tty we set to blocking mode. */    uv__stream_open((uv_stream_t*)tty, fd, UV_WRITABLE);    tty->blocking = 1;  }  loop->counters.tty_init++;  tty->mode = 0;  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;}

int uv_pipe_open(uv_pipe_t* handle, uv_file fd) {#if defined(__APPLE__)  if (uv__stream_try_select((uv_stream_t*) handle, &fd))    return -1;#endif /* defined(__APPLE__) */  return uv__stream_open((uv_stream_t*)handle,                         fd,                         UV_STREAM_READABLE | UV_STREAM_WRITABLE);}

int uv_tcp_open(uv_tcp_t* handle, uv_os_sock_t sock) {  int err;  err = uv__nonblock(sock, 1);  if (err)    return err;  return uv__stream_open((uv_stream_t*)handle,                         sock,                         UV_STREAM_READABLE | UV_STREAM_WRITABLE);}

int uv_tty_init(uv_loop_t* loop, uv_tty_t* tty, int fd, int readable) {  uv__stream_init(loop, (uv_stream_t*)tty, UV_TTY);#if defined(__APPLE__)  if (uv__stream_try_select((uv_stream_t*) tty, &fd))    return -1;#endif /* defined(__APPLE__) */  if (readable) {    uv__nonblock(fd, 1);    uv__stream_open((uv_stream_t*)tty, fd, UV_STREAM_READABLE);  } else {    /* Note: writable tty we set to blocking mode. */    uv__stream_open((uv_stream_t*)tty, fd, UV_STREAM_WRITABLE);    tty->flags |= UV_STREAM_BLOCKING;  }  tty->mode = 0;  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;}

int uv_tty_init(uv_loop_t* loop, uv_tty_t* tty, int fd, int readable) {  int flags;  int newfd;  int r;  uv__stream_init(loop, (uv_stream_t*)tty, UV_TTY);  /* Reopen the file descriptor when it refers to a tty. This lets us put the   * tty in non-blocking mode without affecting other processes that share it   * with us.   *   * Example: `node | cat` - if we put our fd 0 in non-blocking mode, it also   * affects fd 1 of `cat` because both file descriptors refer to the same   * struct file in the kernel. When we reopen our fd 0, it points to a   * different struct file, hence changing its properties doesn't affect   * other processes.   */  if (isatty(fd)) {    newfd = open("/dev/tty", O_RDWR);    if (newfd == -1)      return uv__set_sys_error(loop, errno);    do      r = dup2(newfd, fd);    while (r == -1 && (errno == EINTR || errno == EBUSY));    /* EINVAL means newfd == fd which could conceivably happen if another     * thread called close(fd) between our calls to isatty() and open().     * That's a rather unlikely event but let's handle it anyway.     */    if (r == -1 && errno != EINVAL) {      close(newfd);      return uv__set_sys_error(loop, errno);    }    fd = newfd;  }  if (readable)    flags = UV_STREAM_READABLE;  else    flags = UV_STREAM_WRITABLE;  uv__nonblock(fd, 1);  uv__stream_open((uv_stream_t*)tty, fd, flags);  tty->mode = 0;  return 0;}

int uv_pipe_open(uv_pipe_t* handle, uv_os_fd_t fd) {  int err;  err = uv__nonblock(fd, 1);  if (err)    return err;#if defined(__APPLE__)  err = uv__stream_try_select((uv_stream_t*) handle, &fd);  if (err)    return err;#endif /* defined(__APPLE__) */  return uv__stream_open((uv_stream_t*)handle,                         fd,                         UV_STREAM_READABLE | UV_STREAM_WRITABLE);}

static int maybe_new_socket(uv_tcp_t* handle, int domain, int flags) {  int sockfd;  if (uv__stream_fd(handle) != -1)    return 0;  sockfd = uv__socket(domain, SOCK_STREAM, 0);  if (sockfd == -1)    return uv__set_sys_error(handle->loop, errno);  if (uv__stream_open((uv_stream_t*)handle, sockfd, flags)) {    close(sockfd);    return -1;  }  return 0;}

static int uv__bind(uv_tcp_t* tcp,                    int domain,                    struct sockaddr* addr,                    int addrsize) {  int saved_errno;  int status;  saved_errno = errno;  status = -1;  if (tcp->fd < 0) {    if ((tcp->fd = uv__socket(domain, SOCK_STREAM, 0)) == -1) {      uv__set_sys_error(tcp->loop, errno);      goto out;    }    if (uv__stream_open((uv_stream_t*)tcp,                        tcp->fd,                        UV_STREAM_READABLE | UV_STREAM_WRITABLE)) {      close(tcp->fd);      tcp->fd = -1;      status = -2;      goto out;    }  }  assert(tcp->fd >= 0);  tcp->delayed_error = 0;  if (bind(tcp->fd, addr, addrsize) == -1) {    if (errno == EADDRINUSE) {      tcp->delayed_error = errno;    } else {      uv__set_sys_error(tcp->loop, errno);      goto out;    }  }  status = 0;out:  errno = saved_errno;  return status;}

static int uv__process_open_stream(uv_stdio_container_t* container, int fds[2],                                   int writable) {  int fd = fds[writable ? 1 : 0];  int child_fd = fds[writable ? 0 : 1];  int flags;  /* No need to create stream */  if (!(container->flags & UV_CREATE_PIPE) || fd < 0) {    return 0;  }  assert(child_fd >= 0);  close(child_fd);  uv__nonblock(fd, 1);  flags = uv__process_stdio_flags(container, writable);  return uv__stream_open((uv_stream_t*)container->data.stream, fd, flags);}

static int maybe_new_socket(uv_tcp_t* handle, int domain, int flags) {  int sockfd;  int err;  if (uv__stream_fd(handle) != -1)    return 0;  err = uv__socket(domain, SOCK_STREAM, 0);  if (err < 0)    return err;  sockfd = err;  err = uv__stream_open((uv_stream_t*) handle, sockfd, flags);  if (err) {    close(sockfd);    return err;  }  return 0;}

int uv_tcp_listen(uv_tcp_t* tcp, int backlog, uv_connection_cb cb) {  int r;  if (tcp->delayed_error) {    uv__set_sys_error(tcp->loop, tcp->delayed_error);    return -1;  }  if (tcp->fd < 0) {    if ((tcp->fd = uv__socket(AF_INET, SOCK_STREAM, 0)) == -1) {      uv__set_sys_error(tcp->loop, errno);      return -1;    }    if (uv__stream_open((uv_stream_t*)tcp, tcp->fd, UV_READABLE)) {      uv__close(tcp->fd);      tcp->fd = -1;      return -1;    }  }  assert(tcp->fd >= 0);  r = listen(tcp->fd, backlog);  if (r < 0) {    uv__set_sys_error(tcp->loop, errno);    return -1;  }  tcp->connection_cb = cb;  /* Start listening for connections. */  uv__io_watcher_start(      (uv_handle_t*)tcp,      &tcp->io,      &tcp->io.read_watcher,      uv__server_io);  return 0;}

static int uv__process_open_stream(uv_stdio_container_t* container,                                   int pipefds[2]) {  int flags;  int err;  if (!(container->flags & UV_CREATE_PIPE) || pipefds[0] < 0)    return 0;  err = uv__close(pipefds[1]);  if (err != 0)    abort();  pipefds[1] = -1;  uv__nonblock(pipefds[0], 1);  flags = 0;  if (container->flags & UV_WRITABLE_PIPE)    flags |= UV_HANDLE_READABLE;  if (container->flags & UV_READABLE_PIPE)    flags |= UV_HANDLE_WRITABLE;  return uv__stream_open(container->data.stream, pipefds[0], flags);}

int uv_device_init(uv_loop_t* loop,                   uv_device_t* device,                   const char*path,                   int flags) {  int fd, err;  int stream_flags;  if (flags != O_RDONLY && flags != O_WRONLY && flags != O_RDWR)    return -EINVAL;  uv__stream_init(loop, (uv_stream_t*) device, UV_DEVICE);  fd = open(path, flags);   if (fd < 0)    return -errno;  stream_flags = 0;  if (flags & O_RDONLY)     stream_flags |= UV_STREAM_READABLE;  else if (flags & O_WRONLY)    stream_flags |= UV_STREAM_WRITABLE;  else if (flags & O_RDWR)    stream_flags |= UV_STREAM_READABLE | UV_STREAM_WRITABLE;  err = uv__nonblock(fd, 1);  if (err) {    close(fd);    return err;  }  err = uv__stream_open((uv_stream_t*)device, fd, stream_flags);  if (err) {    close(fd);    return err;  }  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_err_new(server->loop, EAGAIN);    goto out;  }  if (uv__stream_open(streamClient, streamServer->accepted_fd,        UV_READABLE | UV_WRITABLE)) {    /* TODO handle error */    streamServer->accepted_fd = -1;    uv__close(streamServer->accepted_fd);    goto out;  }  ev_io_start(streamServer->loop->ev, &streamServer->read_watcher);  streamServer->accepted_fd = -1;  status = 0;out:  errno = saved_errno;  return status;}

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;  }  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;  }  uv__io_start(streamServer->loop, &streamServer->io_watcher, UV__POLLIN);  streamServer->accepted_fd = -1;  status = 0;out:  errno = saved_errno;  return status;}

static int new_socket(uv_tcp_t* handle, int domain, unsigned long flags) {  struct sockaddr_storage saddr;  socklen_t slen;  int sockfd;  int err;  err = uv__socket(domain, SOCK_STREAM, 0);  if (err < 0)    return err;  sockfd = err;  err = uv__stream_open((uv_stream_t*) handle, sockfd, flags);  if (err) {    uv__close(sockfd);    return err;  }  if (flags & UV_HANDLE_BOUND) {    /* Bind this new socket to an arbitrary port */    slen = sizeof(saddr);    memset(&saddr, 0, sizeof(saddr));    err = getsockname(uv__stream_fd(handle), (struct sockaddr*) &saddr, &slen);    if (err) {      uv__close(sockfd);      return err;    }    err = bind(uv__stream_fd(handle), (struct sockaddr*) &saddr, slen);    if (err) {      uv__close(sockfd);      return err;    }  }  return 0;}

int uv_tty_init(uv_loop_t* loop, uv_tty_t* tty, int fd, int readable) {  int flags;  int newfd;  int r;  flags = 0;  newfd = -1;  uv__stream_init(loop, (uv_stream_t*) tty, UV_TTY);  /* Reopen the file descriptor when it refers to a tty. This lets us put the   * tty in non-blocking mode without affecting other processes that share it   * with us.   *   * Example: `node | cat` - if we put our fd 0 in non-blocking mode, it also   * affects fd 1 of `cat` because both file descriptors refer to the same   * struct file in the kernel. When we reopen our fd 0, it points to a   * different struct file, hence changing its properties doesn't affect   * other processes.   */  if (isatty(fd)) {    r = uv__open_cloexec("/dev/tty", O_RDWR);    if (r < 0) {      /* fallback to using blocking writes */      if (!readable)        flags |= UV_STREAM_BLOCKING;      goto skip;    }    newfd = r;    r = uv__dup2_cloexec(newfd, fd);    if (r < 0 && r != -EINVAL) {      /* EINVAL means newfd == fd which could conceivably happen if another       * thread called close(fd) between our calls to isatty() and open().       * That's a rather unlikely event but let's handle it anyway.       */      uv__close(newfd);      return r;    }    fd = newfd;  }skip:#if defined(__APPLE__)  r = uv__stream_try_select((uv_stream_t*) tty, &fd);  if (r) {    if (newfd != -1)      uv__close(newfd);    return r;  }#endif  if (readable)    flags |= UV_STREAM_READABLE;  else    flags |= UV_STREAM_WRITABLE;  if (!(flags & UV_STREAM_BLOCKING))    uv__nonblock(fd, 1);  uv__stream_open((uv_stream_t*) tty, fd, flags);  tty->mode = UV_TTY_MODE_NORMAL;  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 new_sock;  int err;  int r;  size_t name_len;  name_len = strlen(name);    if (name_len > sizeof(saddr.sun_path) - 1) {    err = -ENAMETOOLONG;    goto out;  }  new_sock = (uv__stream_fd(handle) == -1);  if (new_sock) {    err = uv__socket(AF_UNIX, SOCK_STREAM, 0);    if (err < 0)      goto out;    handle->io_watcher.fd = err;  }  memset(&saddr, 0, sizeof saddr);  memcpy(saddr.sun_path, name, name_len);  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 && errno != EINPROGRESS) {    err = -errno;#if defined(__CYGWIN__) || defined(__MSYS__)    /* EBADF is supposed to mean that the socket fd is bad, but       Cygwin reports EBADF instead of ENOTSOCK when the file is       not a socket.  We do not expect to see a bad fd here       (e.g. due to new_sock), so translate the error.  */    if (err == -EBADF)      err = -ENOTSOCK;#endif    goto out;  }  err = 0;  if (new_sock) {    err = uv__stream_open((uv_stream_t*)handle,                          uv__stream_fd(handle),                          UV_STREAM_READABLE | UV_STREAM_WRITABLE);  }  if (err == 0)    uv__io_start(handle->loop, &handle->io_watcher, POLLIN | POLLOUT);out:  handle->delayed_error = err;  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)    uv__io_feed(handle->loop, &handle->io_watcher);}

int uv_pipe_connect(uv_connect_t* req,                    uv_pipe_t* handle,                    const char* name,                    uv_connect_cb cb) {  struct sockaddr_un sun;  int saved_errno;  int sockfd;  int status;  int r;  saved_errno = errno;  sockfd = -1;  status = -1;  if ((sockfd = uv__socket(AF_UNIX, SOCK_STREAM, 0)) == -1) {    uv_err_new(handle->loop, errno);    goto out;  }  memset(&sun, 0, sizeof sun);  uv__strlcpy(sun.sun_path, name, sizeof(sun.sun_path));  sun.sun_family = AF_UNIX;  /* We don't check for EINPROGRESS. Think about it: the socket   * is either there or not.   */  do {    r = connect(sockfd, (struct sockaddr*)&sun, sizeof sun);  }  while (r == -1 && errno == EINTR);  if (r == -1) {    uv_err_new(handle->loop, errno);    uv__close(sockfd);    goto out;  }  uv__stream_open((uv_stream_t*)handle, sockfd, UV_READABLE | UV_WRITABLE);  ev_io_start(handle->loop->ev, &handle->read_watcher);  ev_io_start(handle->loop->ev, &handle->write_watcher);  status = 0;out:  handle->delayed_error = status; /* Passed to callback. */  handle->connect_req = req;  req->handle = (uv_stream_t*)handle;  req->type = UV_CONNECT;  req->cb = cb;  ngx_queue_init(&req->queue);  /* Run callback on next tick. */  ev_feed_event(handle->loop->ev, &handle->read_watcher, EV_CUSTOM);  assert(ev_is_pending(&handle->read_watcher));  /* Mimic the Windows pipe implementation, always   * return 0 and let the callback handle errors.   */  errno = saved_errno;  return 0;}

int uv_tcp_open(uv_tcp_t* handle, uv_os_sock_t sock) {  return uv__stream_open((uv_stream_t*)handle,                         sock,                         UV_STREAM_READABLE | UV_STREAM_WRITABLE);}

/* creates and starts a thread and associated watcher */int uv_thread_create(uv_loop_t *loop, uv_thread_t *thread, uv_process_options_t options) {  /* initialise the thread handle */  uv__handle_init(loop, (uv_handle_t*)thread, UV_THREAD);  loop->counters.thread_init++;  thread->exit_cb = options.exit_cb;    /* initialise a watcher for this thread */  ev_async_init((ev_async *)&thread->thread_watcher, uv__thread_exit);  ev_async_start(loop->ev, (ev_async *)&thread->thread_watcher);  thread->thread_watcher.data = thread;    /* initialise the shared handle */  uv_thread_shared_t *hnd = (uv_thread_shared_t *)calloc(sizeof(uv_thread_shared_t), 1);  if(!hnd)    goto error;    thread->thread_shared = hnd;  pthread_mutex_init(&hnd->mtx, NULL);  pthread_cond_init(&hnd->cond, NULL);  hnd->thread_handle = thread;  hnd->options = &options;  hnd->thread_arg = options.thread_arg;  hnd->stdin_fd = hnd->stdout_fd = hnd->stderr_fd = -1;  /* FIXME: take ownership of args, but better way sought */  hnd->args = options.args;  hnd->env = options.env;  options.args = 0;  options.env = 0;  /* set up pipes to caller where requested */  int r = 0, flags;  int stdin_pipe[2] = { -1, -1 };  int stdout_pipe[2] = { -1, -1 };  int stderr_pipe[2] = { -1, -1 };  if(options.stdin_stream) {    r = uv__process_init_pipe(options.stdin_stream, stdin_pipe, 0);    if(r)      goto error;    if(stdin_pipe[0] >= 0) {      hnd->stdin_fd = stdin_pipe[0];      uv__nonblock(stdin_pipe[1], 1);      flags = UV_WRITABLE | (options.stdin_stream->ipc ? UV_READABLE : 0);      uv__stream_open((uv_stream_t*)options.stdin_stream, stdin_pipe[1],                      flags);    }  }  if(options.stdout_stream) {    r = uv__process_init_pipe(options.stdout_stream, stdout_pipe, 0);    if(r)      goto error;    if(stdout_pipe[1] >= 0) {      hnd->stdout_fd = stdout_pipe[1];      uv__nonblock(stdout_pipe[0], 1);      flags = UV_READABLE | (options.stdout_stream->ipc ? UV_WRITABLE : 0);      uv__stream_open((uv_stream_t*)options.stdout_stream, stdout_pipe[0],                      flags);    }  }  if(options.stderr_stream) {    r = uv__process_init_pipe(options.stderr_stream, stderr_pipe, 0);    if(r)      goto error;    if(stderr_pipe[1] >= 0) {      hnd->stderr_fd = stderr_pipe[1];      uv__nonblock(stderr_pipe[0], 1);      flags = UV_READABLE | (options.stderr_stream->ipc ? UV_WRITABLE : 0);      uv__stream_open((uv_stream_t*)options.stderr_stream, stderr_pipe[0],                      flags);    }  }    /* synchronously create the thread */  pthread_mutex_lock(&hnd->mtx);  pthread_create(&hnd->thread_id, 0, uv__thread_run, hnd);  pthread_cond_wait(&hnd->cond, &hnd->mtx);  pthread_mutex_unlock(&hnd->mtx);  return 0;error:  uv__set_sys_error(loop, errno);  uv__close(stdin_pipe[0]);  uv__close(stdin_pipe[1]);  uv__close(stdout_pipe[0]);  uv__close(stdout_pipe[1]);  uv__close(stderr_pipe[0]);  uv__close(stderr_pipe[1]);  return -1;}

//.........这里部分代码省略.........    if (stdout_pipe[1] >= 0) {      uv__close(stdout_pipe[0]);      dup2(stdout_pipe[1], STDOUT_FILENO);    } else {      /* Reset flags that might be set by Node */      uv__cloexec(STDOUT_FILENO, 0);      uv__nonblock(STDOUT_FILENO, 0);    }    if (stderr_pipe[1] >= 0) {      uv__close(stderr_pipe[0]);      dup2(stderr_pipe[1], STDERR_FILENO);    } else {      /* Reset flags that might be set by Node */      uv__cloexec(STDERR_FILENO, 0);      uv__nonblock(STDERR_FILENO, 0);    }    if (options.cwd && chdir(options.cwd)) {      perror("chdir()");      _exit(127);    }    environ = options.env;    execvp(options.file, options.args);    perror("execvp()");    _exit(127);    /* Execution never reaches here. */  }  /* Parent. */  /* Restore environment. */  environ = save_our_env;#if SPAWN_WAIT_EXEC  /* POLLHUP signals child has exited or execve()'d. */  uv__close(signal_pipe[1]);  do {    pfd.fd = signal_pipe[0];    pfd.events = POLLIN|POLLHUP;    pfd.revents = 0;    errno = 0, status = poll(&pfd, 1, -1);  }  while (status == -1 && (errno == EINTR || errno == ENOMEM));  assert((status == 1) && "poll() on pipe read end failed");  uv__close(signal_pipe[0]);#endif  process->pid = pid;  ev_child_init(&process->child_watcher, uv__chld, pid, 0);  ev_child_start(process->loop->ev, &process->child_watcher);  process->child_watcher.data = process;  if (stdin_pipe[1] >= 0) {    assert(options.stdin_stream);    assert(stdin_pipe[0] >= 0);    uv__close(stdin_pipe[0]);    uv__nonblock(stdin_pipe[1], 1);    flags = UV_WRITABLE | (options.stdin_stream->ipc ? UV_READABLE : 0);    uv__stream_open((uv_stream_t*)options.stdin_stream, stdin_pipe[1],        flags);  }  if (stdout_pipe[0] >= 0) {    assert(options.stdout_stream);    assert(stdout_pipe[1] >= 0);    uv__close(stdout_pipe[1]);    uv__nonblock(stdout_pipe[0], 1);    flags = UV_READABLE | (options.stdout_stream->ipc ? UV_WRITABLE : 0);    uv__stream_open((uv_stream_t*)options.stdout_stream, stdout_pipe[0],        flags);  }  if (stderr_pipe[0] >= 0) {    assert(options.stderr_stream);    assert(stderr_pipe[1] >= 0);    uv__close(stderr_pipe[1]);    uv__nonblock(stderr_pipe[0], 1);    flags = UV_READABLE | (options.stderr_stream->ipc ? UV_WRITABLE : 0);    uv__stream_open((uv_stream_t*)options.stderr_stream, stderr_pipe[0],        flags);  }  return 0;error:  uv__set_sys_error(process->loop, errno);  uv__close(stdin_pipe[0]);  uv__close(stdin_pipe[1]);  uv__close(stdout_pipe[0]);  uv__close(stdout_pipe[1]);  uv__close(stderr_pipe[0]);  uv__close(stderr_pipe[1]);  return -1;}