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

void uv_process_tcp_connect_req(uv_loop_t* loop, uv_tcp_t* handle,    uv_connect_t* req) {  assert(handle->type == UV_TCP);  if (req->cb) {    if (REQ_SUCCESS(req)) {      if (setsockopt(handle->socket,                      SOL_SOCKET,                      SO_UPDATE_CONNECT_CONTEXT,                      NULL,                      0) == 0) {        uv_connection_init((uv_stream_t*)handle);        active_tcp_streams++;        ((uv_connect_cb)req->cb)(req, 0);      } else {        uv__set_sys_error(loop, WSAGetLastError());        ((uv_connect_cb)req->cb)(req, -1);      }    } else {      uv__set_sys_error(loop, GET_REQ_SOCK_ERROR(req));      ((uv_connect_cb)req->cb)(req, -1);    }  }  DECREASE_PENDING_REQ_COUNT(handle);}

void uv_process_tcp_write_req(uv_loop_t* loop, uv_tcp_t* handle,    uv_write_t* req) {  assert(handle->type == UV_TCP);  assert(handle->write_queue_size >= req->queued_bytes);  handle->write_queue_size -= req->queued_bytes;  if (handle->flags & UV_HANDLE_EMULATE_IOCP) {    if (req->wait_handle != INVALID_HANDLE_VALUE) {      UnregisterWait(req->wait_handle);    }    if (req->event_handle) {      CloseHandle(req->event_handle);    }  }  if (req->cb) {    uv__set_sys_error(loop, GET_REQ_SOCK_ERROR(req));    ((uv_write_cb)req->cb)(req, loop->last_err.code == UV_OK ? 0 : -1);  }  handle->write_reqs_pending--;  if (handle->flags & UV_HANDLE_SHUTTING &&      handle->write_reqs_pending == 0) {    uv_want_endgame(loop, (uv_handle_t*)handle);  }  DECREASE_PENDING_REQ_COUNT(handle);  uv_unref(loop);}

void uv_process_tcp_connect_req(uv_tcp_t* handle, uv_connect_t* req) {  assert(handle->type == UV_TCP);  if (req->cb) {    if (req->error.code == UV_OK) {      if (setsockopt(handle->socket,                      SOL_SOCKET,                      SO_UPDATE_CONNECT_CONTEXT,                      NULL,                      0) == 0) {        uv_connection_init((uv_stream_t*)handle);        active_tcp_streams++;        ((uv_connect_cb)req->cb)(req, 0);      } else {        uv_set_sys_error(WSAGetLastError());        ((uv_connect_cb)req->cb)(req, -1);      }    } else {      LOOP->last_error = req->error;      ((uv_connect_cb)req->cb)(req, -1);    }  }  DECREASE_PENDING_REQ_COUNT(handle);}

void uv_process_tcp_write_req(uv_loop_t* loop, uv_tcp_t* handle,    uv_write_t* req) {  int err;  assert(handle->type == UV_TCP);  assert(handle->write_queue_size >= req->queued_bytes);  handle->write_queue_size -= req->queued_bytes;  UNREGISTER_HANDLE_REQ(loop, handle, req);  if (handle->flags & UV_HANDLE_EMULATE_IOCP) {    if (req->wait_handle != INVALID_HANDLE_VALUE) {      pUnregisterWait(req->wait_handle);      req->wait_handle = INVALID_HANDLE_VALUE;    }    if (req->event_handle) {      CloseHandle(req->event_handle);      req->event_handle = NULL;    }  }  if (req->cb) {    err = GET_REQ_SOCK_ERROR(req);    req->cb(req, uv_translate_sys_error(err));  }  handle->write_reqs_pending--;  if (handle->shutdown_req != NULL &&      handle->write_reqs_pending == 0) {    uv_want_endgame(loop, (uv_handle_t*)handle);  }  DECREASE_PENDING_REQ_COUNT(handle);}

void uv_process_tcp_connect_req(uv_loop_t* loop, uv_tcp_t* handle,    uv_connect_t* req) {  int err;  assert(handle->type == UV_TCP);  UNREGISTER_HANDLE_REQ(loop, handle, req);  err = 0;  if (REQ_SUCCESS(req)) {    if (setsockopt(handle->socket,                    SOL_SOCKET,                    SO_UPDATE_CONNECT_CONTEXT,                    NULL,                    0) == 0) {      uv_connection_init((uv_stream_t*)handle);      handle->flags |= UV_HANDLE_READABLE | UV_HANDLE_WRITABLE;      loop->active_tcp_streams++;    } else {      err = WSAGetLastError();    }  } else {    err = GET_REQ_SOCK_ERROR(req);  }  req->cb(req, uv_translate_sys_error(err));  DECREASE_PENDING_REQ_COUNT(handle);}

void uv_process_pipe_accept_req(uv_loop_t* loop, uv_pipe_t* handle,    uv_req_t* raw_req) {  uv_pipe_accept_t* req = (uv_pipe_accept_t*) raw_req;  assert(handle->type == UV_NAMED_PIPE);  if (REQ_SUCCESS(req)) {    assert(req->pipeHandle != INVALID_HANDLE_VALUE);    req->next_pending = handle->pending_accepts;    handle->pending_accepts = req;    if (handle->connection_cb) {      handle->connection_cb((uv_stream_t*)handle, 0);    }  } else {    if (req->pipeHandle != INVALID_HANDLE_VALUE) {      CloseHandle(req->pipeHandle);      req->pipeHandle = INVALID_HANDLE_VALUE;    }    if (!(handle->flags & UV__HANDLE_CLOSING)) {      uv_pipe_queue_accept(loop, handle, req, FALSE);    }  }  DECREASE_PENDING_REQ_COUNT(handle);}

void uv_process_pipe_write_req(uv_loop_t* loop, uv_pipe_t* handle,    uv_write_t* req) {  assert(handle->type == UV_NAMED_PIPE);  assert(handle->write_queue_size >= req->queued_bytes);  handle->write_queue_size -= req->queued_bytes;  if (handle->flags & UV_HANDLE_EMULATE_IOCP) {    if (req->wait_handle != INVALID_HANDLE_VALUE) {      UnregisterWait(req->wait_handle);      req->wait_handle = INVALID_HANDLE_VALUE;    }    if (req->event_handle) {      CloseHandle(req->event_handle);      req->event_handle = NULL;    }  }  if (req->ipc_header) {    if (req == &handle->ipc_header_write_req) {      req->type = UV_UNKNOWN_REQ;    } else {      free(req);    }  } else {    if (req->cb) {      if (!REQ_SUCCESS(req)) {        uv__set_sys_error(loop, GET_REQ_ERROR(req));        ((uv_write_cb)req->cb)(req, -1);      } else {        ((uv_write_cb)req->cb)(req, 0);      }    }  }  handle->write_reqs_pending--;  if (handle->flags & UV_HANDLE_NON_OVERLAPPED_PIPE &&      handle->non_overlapped_writes_tail) {    assert(handle->write_reqs_pending > 0);    uv_queue_non_overlapped_write(handle);  }  if (handle->write_reqs_pending == 0) {    uv_unref(loop);  }  if (handle->write_reqs_pending == 0 &&      handle->flags & UV_HANDLE_SHUTTING) {    uv_want_endgame(loop, (uv_handle_t*)handle);  }  DECREASE_PENDING_REQ_COUNT(handle);}

void uv_process_udp_send_req(uv_udp_t* handle, uv_udp_send_t* req) {  assert(handle->type == UV_UDP);  if (req->cb) {    if (REQ_SUCCESS(req)) {      req->cb(req, 0);    } else {      LOOP->last_error = GET_REQ_UV_SOCK_ERROR(req);      req->cb(req, -1);    }  }  DECREASE_PENDING_REQ_COUNT(handle);}

void uv_process_pipe_connect_req(uv_pipe_t* handle, uv_connect_t* req) {  assert(handle->type == UV_NAMED_PIPE);  if (req->cb) {    if (req->error.code == UV_OK) {      uv_connection_init((uv_stream_t*)handle);      ((uv_connect_cb)req->cb)(req, 0);    } else {      LOOP->last_error = req->error;      ((uv_connect_cb)req->cb)(req, -1);    }  }  DECREASE_PENDING_REQ_COUNT(handle);}

void uv_process_udp_send_req(uv_loop_t* loop, uv_udp_t* handle,    uv_udp_send_t* req) {  assert(handle->type == UV_UDP);  if (req->cb) {    if (REQ_SUCCESS(req)) {      req->cb(req, 0);    } else {      uv__set_sys_error(loop, GET_REQ_SOCK_ERROR(req));      req->cb(req, -1);    }  }  DECREASE_PENDING_REQ_COUNT(handle);}

void uv_process_pipe_connect_req(uv_pipe_t* handle, uv_connect_t* req) {  assert(handle->type == UV_NAMED_PIPE);  if (req->cb) {    if (REQ_SUCCESS(req)) {      uv_pipe_connection_init(handle);      ((uv_connect_cb)req->cb)(req, 0);    } else {      LOOP->last_error = GET_REQ_UV_ERROR(req);      ((uv_connect_cb)req->cb)(req, -1);    }  }  DECREASE_PENDING_REQ_COUNT(handle);}

void uv_process_pipe_connect_req(uv_loop_t* loop, uv_pipe_t* handle,    uv_connect_t* req) {  assert(handle->type == UV_NAMED_PIPE);  if (req->cb) {    if (REQ_SUCCESS(req)) {      uv_pipe_connection_init(handle);      ((uv_connect_cb)req->cb)(req, 0);    } else {      uv__set_sys_error(loop, GET_REQ_ERROR(req));      ((uv_connect_cb)req->cb)(req, -1);    }  }  DECREASE_PENDING_REQ_COUNT(handle);}

void uv_process_tcp_accept_req(uv_loop_t* loop, uv_tcp_t* handle,    uv_req_t* raw_req) {  uv_tcp_accept_t* req = (uv_tcp_accept_t*) raw_req;  int err;  assert(handle->type == UV_TCP);  /* If handle->accepted_socket is not a valid socket, then */  /* uv_queue_accept must have failed. This is a serious error. We stop */  /* accepting connections and report this error to the connection */  /* callback. */  if (req->accept_socket == INVALID_SOCKET) {    if (handle->flags & UV_HANDLE_LISTENING) {      handle->flags &= ~UV_HANDLE_LISTENING;      DECREASE_ACTIVE_COUNT(loop, handle);      if (handle->connection_cb) {        err = GET_REQ_SOCK_ERROR(req);        handle->connection_cb((uv_stream_t*)handle,                              uv_translate_sys_error(err));      }    }  } else if (REQ_SUCCESS(req) &&      setsockopt(req->accept_socket,                  SOL_SOCKET,                  SO_UPDATE_ACCEPT_CONTEXT,                  (char*)&handle->socket,                  sizeof(handle->socket)) == 0) {    req->next_pending = handle->pending_accepts;    handle->pending_accepts = req;    /* Accept and SO_UPDATE_ACCEPT_CONTEXT were successful. */    if (handle->connection_cb) {      handle->connection_cb((uv_stream_t*)handle, 0);    }  } else {    /* Error related to accepted socket is ignored because the server */    /* socket may still be healthy. If the server socket is broken */    /* uv_queue_accept will detect it. */    closesocket(req->accept_socket);    req->accept_socket = INVALID_SOCKET;    if (handle->flags & UV_HANDLE_LISTENING) {      uv_tcp_queue_accept(handle, req);    }  }  DECREASE_PENDING_REQ_COUNT(handle);}

void uv_tcp_endgame(uv_loop_t* loop, uv_tcp_t* handle) {  int status;  int sys_error;  if (handle->flags & UV_HANDLE_CONNECTION &&      handle->flags & UV_HANDLE_SHUTTING &&      !(handle->flags & UV_HANDLE_SHUT) &&      handle->write_reqs_pending == 0) {    if (shutdown(handle->socket, SD_SEND) != SOCKET_ERROR) {      status = 0;      handle->flags |= UV_HANDLE_SHUT;    } else {      status = -1;      sys_error = WSAGetLastError();    }    if (handle->shutdown_req->cb) {      if (status == -1) {        uv__set_sys_error(loop, sys_error);      }      handle->shutdown_req->cb(handle->shutdown_req, status);    }    DECREASE_PENDING_REQ_COUNT(handle);    return;  }  if (handle->flags & UV_HANDLE_CLOSING &&      handle->reqs_pending == 0) {    assert(!(handle->flags & UV_HANDLE_CLOSED));    handle->flags |= UV_HANDLE_CLOSED;    if (!(handle->flags & UV_HANDLE_CONNECTION) && handle->accept_reqs) {      free(handle->accept_reqs);      handle->accept_reqs = NULL;    }    if (handle->close_cb) {      handle->close_cb((uv_handle_t*)handle);    }    active_tcp_streams--;    uv_unref(loop);  }}

void uv_process_tcp_write_req(uv_tcp_t* handle, uv_write_t* req) {  assert(handle->type == UV_TCP);  handle->write_queue_size -= req->queued_bytes;  if (req->cb) {    LOOP->last_error = GET_REQ_UV_SOCK_ERROR(req);    ((uv_write_cb)req->cb)(req, LOOP->last_error.code == UV_OK ? 0 : -1);  }  handle->write_reqs_pending--;  if (handle->flags & UV_HANDLE_SHUTTING &&      handle->write_reqs_pending == 0) {    uv_want_endgame((uv_handle_t*)handle);  }  DECREASE_PENDING_REQ_COUNT(handle);}

void uv_process_pipe_write_req(uv_pipe_t* handle, uv_write_t* req) {  assert(handle->type == UV_NAMED_PIPE);  handle->write_queue_size -= req->queued_bytes;  if (req->cb) {    LOOP->last_error = req->error;    ((uv_write_cb)req->cb)(req, LOOP->last_error.code == UV_OK ? 0 : -1);  }  handle->write_reqs_pending--;  if (handle->write_reqs_pending == 0 &&      handle->flags & UV_HANDLE_SHUTTING) {    uv_want_endgame((uv_handle_t*)handle);  }  DECREASE_PENDING_REQ_COUNT(handle);}

void uv_process_udp_send_req(uv_loop_t* loop, uv_udp_t* handle,    uv_udp_send_t* req) {  int err;  assert(handle->type == UV_UDP);  UNREGISTER_HANDLE_REQ(loop, handle, req);  if (req->cb) {    err = 0;    if (!REQ_SUCCESS(req)) {      err = GET_REQ_SOCK_ERROR(req);    }    req->cb(req, uv_translate_sys_error(err));  }  DECREASE_PENDING_REQ_COUNT(handle);}

void uv_process_tty_read_line_req(uv_loop_t* loop, uv_tty_t* handle,    uv_req_t* req) {  uv_buf_t buf;  assert(handle->type == UV_TTY);  buf = handle->read_line_buffer;  handle->flags &= ~UV_HANDLE_READ_PENDING;  handle->read_line_buffer = uv_null_buf_;  if (!REQ_SUCCESS(req)) {    /* Read was not successful */    if ((handle->flags & UV_HANDLE_READING) &&        !(handle->flags & UV_HANDLE_TTY_RAW)) {      /* Real error */      handle->flags &= ~UV_HANDLE_READING;      loop->last_error = GET_REQ_UV_ERROR(req);      handle->read_cb((uv_stream_t*) handle, -1, buf);    } else {      /* The read was cancelled, or whatever we don't care */      uv_set_sys_error(loop, WSAEWOULDBLOCK); /* maps to UV_EAGAIN */      handle->read_cb((uv_stream_t*) handle, 0, buf);    }  } else {    /* Read successful */    /* TODO: read unicode, convert to utf-8 */    DWORD bytes = req->overlapped.InternalHigh;    if (bytes == 0) {      uv_set_sys_error(loop, WSAEWOULDBLOCK); /* maps to UV_EAGAIN */    }    handle->read_cb((uv_stream_t*) handle, bytes, buf);  }  /* Wait for more input events. */  if ((handle->flags & UV_HANDLE_READING) &&      !(handle->flags & UV_HANDLE_READ_PENDING)) {    uv_tty_queue_read(loop, handle);  }  DECREASE_PENDING_REQ_COUNT(handle);}

void uv_process_pipe_shutdown_req(uv_pipe_t* handle, uv_shutdown_t* req) {  assert(handle->type == UV_NAMED_PIPE);  /* Initialize and optionally start the eof timer. */  /* This makes no sense if we've already seen EOF. */  if (!(handle->flags & UV_HANDLE_EOF)) {    eof_timer_init(handle);    /* If reading start the timer right now. */    /* Otherwise uv_pipe_queue_read will start it. */    if (handle->flags & UV_HANDLE_READ_PENDING) {      eof_timer_start(handle);    }  }  if (req->cb) {    req->cb(req, 0);  }  DECREASE_PENDING_REQ_COUNT(handle);}

void uv_process_pipe_write_req(uv_pipe_t* handle, uv_write_t* req) {  assert(handle->type == UV_NAMED_PIPE);  handle->write_queue_size -= req->queued_bytes;  if (req->cb) {    if (!REQ_SUCCESS(req)) {      LOOP->last_error = GET_REQ_UV_ERROR(req);      ((uv_write_cb)req->cb)(req, -1);    } else {      ((uv_write_cb)req->cb)(req, 0);    }  }  handle->write_reqs_pending--;  if (handle->write_reqs_pending == 0 &&      handle->flags & UV_HANDLE_SHUTTING) {    uv_want_endgame((uv_handle_t*)handle);  }  DECREASE_PENDING_REQ_COUNT(handle);}

void uv_process_tcp_write_req(uv_loop_t* loop, uv_tcp_t* handle,    uv_write_t* req) {  int err;  assert(handle->type == UV_TCP);  assert(handle->write_queue_size >= req->u.io.queued_bytes);  handle->write_queue_size -= req->u.io.queued_bytes;  UNREGISTER_HANDLE_REQ(loop, handle, req);  if (handle->flags & UV_HANDLE_EMULATE_IOCP) {    if (req->wait_handle != INVALID_HANDLE_VALUE) {      UnregisterWait(req->wait_handle);      req->wait_handle = INVALID_HANDLE_VALUE;    }    if (req->event_handle) {      CloseHandle(req->event_handle);      req->event_handle = NULL;    }  }  if (req->cb) {    err = uv_translate_sys_error(GET_REQ_SOCK_ERROR(req));    if (err == UV_ECONNABORTED) {      /* use UV_ECANCELED for consistency with Unix */      err = UV_ECANCELED;    }    req->cb(req, err);  }  handle->stream.conn.write_reqs_pending--;  if (handle->stream.conn.shutdown_req != NULL &&      handle->stream.conn.write_reqs_pending == 0) {    uv_want_endgame(loop, (uv_handle_t*)handle);  }  DECREASE_PENDING_REQ_COUNT(handle);}

//.........这里部分代码省略.........  int partial;  assert(handle->type == UV_UDP);  handle->flags &= ~UV_HANDLE_READ_PENDING;  if (!REQ_SUCCESS(req)) {    DWORD err = GET_REQ_SOCK_ERROR(req);    if (err == WSAEMSGSIZE) {      /* Not a real error, it just indicates that the received packet */      /* was bigger than the receive buffer. */    } else if (err == WSAECONNRESET || err == WSAENETRESET) {      /* A previous sendto operation failed; ignore this error. If */      /* zero-reading we need to call WSARecv/WSARecvFrom _without_ the */      /* MSG_PEEK flag to clear out the error queue. For nonzero reads, */      /* immediately queue a new receive. */      if (!(handle->flags & UV_HANDLE_ZERO_READ)) {        goto done;      }    } else {      /* A real error occurred. Report the error to the user only if we're */      /* currently reading. */      if (handle->flags & UV_HANDLE_READING) {        uv__set_sys_error(loop, err);        uv_udp_recv_stop(handle);        buf = (handle->flags & UV_HANDLE_ZERO_READ) ?              uv_buf_init(NULL, 0) : handle->recv_buffer;        handle->recv_cb(handle, -1, buf, NULL, 0);      }      goto done;    }  }  if (!(handle->flags & UV_HANDLE_ZERO_READ)) {    /* Successful read */    partial = !REQ_SUCCESS(req);    handle->recv_cb(handle,                    req->overlapped.InternalHigh,                    handle->recv_buffer,                    (struct sockaddr*) &handle->recv_from,                    partial ? UV_UDP_PARTIAL : 0);  } else if (handle->flags & UV_HANDLE_READING) {    DWORD bytes, err, flags;    struct sockaddr_storage from;    int from_len;    /* Do a nonblocking receive */    /* TODO: try to read multiple datagrams at once. FIONREAD maybe? */    buf = handle->alloc_cb((uv_handle_t*) handle, 65536);    assert(buf.len > 0);    memset(&from, 0, sizeof from);    from_len = sizeof from;    flags = 0;    if (WSARecvFrom(handle->socket,                    (WSABUF*)&buf,                    1,                    &bytes,                    &flags,                    (struct sockaddr*) &from,                    &from_len,                    NULL,                    NULL) != SOCKET_ERROR) {      /* Message received */      handle->recv_cb(handle, bytes, buf, (struct sockaddr*) &from, 0);    } else {      err = WSAGetLastError();      if (err == WSAEMSGSIZE) {        /* Message truncated */        handle->recv_cb(handle,                        bytes,                        buf,                        (struct sockaddr*) &from,                        UV_UDP_PARTIAL);      } if (err == WSAEWOULDBLOCK) {        /* Kernel buffer empty */        uv__set_sys_error(loop, WSAEWOULDBLOCK);        handle->recv_cb(handle, 0, buf, NULL, 0);      } else if (err != WSAECONNRESET && err != WSAENETRESET) {        /* Serious error. WSAECONNRESET/WSANETRESET is ignored because this */        /* just indicates that a previous sendto operation failed. */        uv_udp_recv_stop(handle);        uv__set_sys_error(loop, err);        handle->recv_cb(handle, -1, buf, NULL, 0);      }    }  }done:  /* Post another read if still reading and not closing. */  if ((handle->flags & UV_HANDLE_READING) &&      !(handle->flags & UV_HANDLE_READ_PENDING)) {    uv_udp_queue_recv(loop, handle);  }  DECREASE_PENDING_REQ_COUNT(handle);}

//.........这里部分代码省略.........                                         NULL,                                         NULL);        }        /* Whatever happened, the last character wasn't a high surrogate. */        handle->last_utf16_high_surrogate = 0;        /* If the utf16 character(s) couldn't be converted something must */        /* be wrong. */        if (!char_len) {          uv_set_sys_error(loop, GetLastError());          handle->flags &= ~UV_HANDLE_READING;          handle->read_cb((uv_stream_t*) handle, -1, buf);          goto out;        }        handle->last_key_len = (unsigned char) (prefix_len + char_len);        handle->last_key_offset = 0;        continue;      } else {        /* Function key pressed */        const char* vt100;        size_t prefix_len, vt100_len;        vt100 = get_vt100_fn_key(KEV.wVirtualKeyCode,                                  !!(KEV.dwControlKeyState & SHIFT_PRESSED),                                  !!(KEV.dwControlKeyState & (                                    LEFT_CTRL_PRESSED |                                    RIGHT_CTRL_PRESSED)),                                  &vt100_len);        /* If we were unable to map to a vt100 sequence, just ignore. */        if (!vt100) {          continue;        }        /* Prefix with /x033 when the alt key was held. */        if (KEV.dwControlKeyState & (LEFT_ALT_PRESSED | RIGHT_ALT_PRESSED)) {          handle->last_key[0] = '/033';          prefix_len = 1;        } else {          prefix_len = 0;        }        /* Copy the vt100 sequence to the handle buffer. */        assert(prefix_len + vt100_len < sizeof handle->last_key);        memcpy(&handle->last_key[prefix_len], vt100, vt100_len);        handle->last_key_len = (unsigned char) (prefix_len + vt100_len);        handle->last_key_offset = 0;        continue;      }    } else {      /* Copy any bytes left from the last keypress to the user buffer. */      if (handle->last_key_offset < handle->last_key_len) {        /* Allocate a buffer if needed */        if (buf_used == 0) {          buf = handle->alloc_cb((uv_handle_t*) handle, 1024);        }        buf.base[buf_used++] = handle->last_key[handle->last_key_offset++];        /* If the buffer is full, emit it */        if (buf_used == buf.len) {          handle->read_cb((uv_stream_t*) handle, buf_used, buf);          buf = uv_null_buf_;          buf_used = 0;        }        continue;      }      /* Apply dwRepeat from the last input record. */      if (--KEV.wRepeatCount > 0) {        handle->last_key_offset = 0;        continue;      }      handle->last_key_len = 0;      continue;    }  }  /* Send the buffer back to the user */  if (buf_used > 0) {    handle->read_cb((uv_stream_t*) handle, buf_used, buf);  } out:  /* Wait for more input events. */  if ((handle->flags & UV_HANDLE_READING) &&      !(handle->flags & UV_HANDLE_READ_PENDING)) {    uv_tty_queue_read(loop, handle);  }  DECREASE_PENDING_REQ_COUNT(handle);#undef KEV}

//.........这里部分代码省略.........        err = WSAECONNRESET;      }      handle->read_cb((uv_stream_t*)handle,                      uv_translate_sys_error(err),                      &buf);    }  } else {    if (!(handle->flags & UV_HANDLE_ZERO_READ)) {      /* The read was done with a non-zero buffer length. */      if (req->overlapped.InternalHigh > 0) {        /* Successful read */        handle->read_cb((uv_stream_t*)handle,                        req->overlapped.InternalHigh,                        &handle->read_buffer);        /* Read again only if bytes == buf.len */        if (req->overlapped.InternalHigh < handle->read_buffer.len) {          goto done;        }      } else {        /* Connection closed */        if (handle->flags & UV_HANDLE_READING) {          handle->flags &= ~UV_HANDLE_READING;          DECREASE_ACTIVE_COUNT(loop, handle);        }        handle->flags &= ~UV_HANDLE_READABLE;        buf.base = 0;        buf.len = 0;        handle->read_cb((uv_stream_t*)handle, UV_EOF, &handle->read_buffer);        goto done;      }    }    /* Do nonblocking reads until the buffer is empty */    while (handle->flags & UV_HANDLE_READING) {      handle->alloc_cb((uv_handle_t*) handle, 65536, &buf);      if (buf.len == 0) {        handle->read_cb((uv_stream_t*) handle, UV_ENOBUFS, &buf);        break;      }      assert(buf.base != NULL);      flags = 0;      if (WSARecv(handle->socket,                  (WSABUF*)&buf,                  1,                  &bytes,                  &flags,                  NULL,                  NULL) != SOCKET_ERROR) {        if (bytes > 0) {          /* Successful read */          handle->read_cb((uv_stream_t*)handle, bytes, &buf);          /* Read again only if bytes == buf.len */          if (bytes < buf.len) {            break;          }        } else {          /* Connection closed */          handle->flags &= ~(UV_HANDLE_READING | UV_HANDLE_READABLE);          DECREASE_ACTIVE_COUNT(loop, handle);          handle->read_cb((uv_stream_t*)handle, UV_EOF, &buf);          break;        }      } else {        err = WSAGetLastError();        if (err == WSAEWOULDBLOCK) {          /* Read buffer was completely empty, report a 0-byte read. */          handle->read_cb((uv_stream_t*)handle, 0, &buf);        } else {          /* Ouch! serious error. */          handle->flags &= ~UV_HANDLE_READING;          DECREASE_ACTIVE_COUNT(loop, handle);          if (err == WSAECONNABORTED) {            /* Turn WSAECONNABORTED into UV_ECONNRESET to be consistent with */            /* Unix. */            err = WSAECONNRESET;          }          handle->read_cb((uv_stream_t*)handle,                          uv_translate_sys_error(err),                          &buf);        }        break;      }    }done:    /* Post another read if still reading and not closing. */    if ((handle->flags & UV_HANDLE_READING) &&        !(handle->flags & UV_HANDLE_READ_PENDING)) {      uv_tcp_queue_read(loop, handle);    }  }  DECREASE_PENDING_REQ_COUNT(handle);}

void uv_tcp_endgame(uv_loop_t* loop, uv_tcp_t* handle) {  int err;  unsigned int i;  uv_tcp_accept_t* req;  if (handle->flags & UV_HANDLE_CONNECTION &&      handle->shutdown_req != NULL &&      handle->write_reqs_pending == 0) {    UNREGISTER_HANDLE_REQ(loop, handle, handle->shutdown_req);    err = 0;    if (handle->flags & UV__HANDLE_CLOSING) {      err = ERROR_OPERATION_ABORTED;    } else if (shutdown(handle->socket, SD_SEND) == SOCKET_ERROR) {      err = WSAGetLastError();    }    if (handle->shutdown_req->cb) {      handle->shutdown_req->cb(handle->shutdown_req,                               uv_translate_sys_error(err));    }    handle->shutdown_req = NULL;    DECREASE_PENDING_REQ_COUNT(handle);    return;  }  if (handle->flags & UV__HANDLE_CLOSING &&      handle->reqs_pending == 0) {    assert(!(handle->flags & UV_HANDLE_CLOSED));    if (!(handle->flags & UV_HANDLE_TCP_SOCKET_CLOSED)) {      closesocket(handle->socket);      handle->flags |= UV_HANDLE_TCP_SOCKET_CLOSED;    }    if (!(handle->flags & UV_HANDLE_CONNECTION) && handle->accept_reqs) {      if (handle->flags & UV_HANDLE_EMULATE_IOCP) {        for (i = 0; i < uv_simultaneous_server_accepts; i++) {          req = &handle->accept_reqs[i];          if (req->wait_handle != INVALID_HANDLE_VALUE) {            pUnregisterWait(req->wait_handle);            req->wait_handle = INVALID_HANDLE_VALUE;          }          if (req->event_handle) {            CloseHandle(req->event_handle);            req->event_handle = NULL;          }        }      }      free(handle->accept_reqs);      handle->accept_reqs = NULL;    }    if (handle->flags & UV_HANDLE_CONNECTION &&        handle->flags & UV_HANDLE_EMULATE_IOCP) {      if (handle->read_req.wait_handle != INVALID_HANDLE_VALUE) {        pUnregisterWait(handle->read_req.wait_handle);        handle->read_req.wait_handle = INVALID_HANDLE_VALUE;      }      if (handle->read_req.event_handle) {        CloseHandle(handle->read_req.event_handle);        handle->read_req.event_handle = NULL;      }    }    uv__handle_close(handle);    loop->active_tcp_streams--;  }}

void uv_process_pipe_read_req(uv_pipe_t* handle, uv_req_t* req) {  DWORD bytes, avail;  uv_buf_t buf;  assert(handle->type == UV_NAMED_PIPE);  handle->flags &= ~UV_HANDLE_READ_PENDING;  if (req->error.code != UV_OK) {    /* An error occurred doing the 0-read. */    if (handle->flags & UV_HANDLE_READING) {      /* Stop reading and report error. */      handle->flags &= ~UV_HANDLE_READING;      LOOP->last_error = req->error;      buf.base = 0;      buf.len = 0;      handle->read_cb((uv_stream_t*)handle, -1, buf);    }  } else {    /* Do non-blocking reads until the buffer is empty */    while (handle->flags & UV_HANDLE_READING) {      if (!PeekNamedPipe(handle->handle,                         NULL,                         0,                         NULL,                         &avail,                         NULL)) {        uv_set_sys_error(GetLastError());        buf.base = 0;        buf.len = 0;        handle->read_cb((uv_stream_t*)handle, -1, buf);        break;      }      /* TODO: do we need to check avail > 0? */      buf = handle->alloc_cb((uv_stream_t*)handle, avail);      assert(buf.len > 0);      if (ReadFile(handle->handle,                   buf.base,                   buf.len,                   &bytes,                   NULL)) {        if (bytes > 0) {          /* Successful read */          handle->read_cb((uv_stream_t*)handle, bytes, buf);          /* Read again only if bytes == buf.len */          if (bytes <= buf.len) {            break;          }        } else {          /* Connection closed */          handle->flags &= ~UV_HANDLE_READING;          handle->flags |= UV_HANDLE_EOF;          LOOP->last_error.code = UV_EOF;          LOOP->last_error.sys_errno_ = ERROR_SUCCESS;          handle->read_cb((uv_stream_t*)handle, -1, buf);          break;        }      } else {        /* Ouch! serious error. */        uv_set_sys_error(GetLastError());        handle->read_cb((uv_stream_t*)handle, -1, buf);        break;      }    }    /* Post another 0-read if still reading and not closing. */    if ((handle->flags & UV_HANDLE_READING) &&        !(handle->flags & UV_HANDLE_READ_PENDING)) {      uv_pipe_queue_read(handle);    }  }  DECREASE_PENDING_REQ_COUNT(handle);}

void uv_tcp_endgame(uv_loop_t* loop, uv_tcp_t* handle) {  int status;  int sys_error;  unsigned int i;  uv_tcp_accept_t* req;  if (handle->flags & UV_HANDLE_CONNECTION &&      handle->flags & UV_HANDLE_SHUTTING &&      !(handle->flags & UV_HANDLE_SHUT) &&      handle->write_reqs_pending == 0) {    if (shutdown(handle->socket, SD_SEND) != SOCKET_ERROR) {      status = 0;      handle->flags |= UV_HANDLE_SHUT;    } else {      status = -1;      sys_error = WSAGetLastError();    }    if (handle->shutdown_req->cb) {      if (status == -1) {        uv__set_sys_error(loop, sys_error);      }      handle->shutdown_req->cb(handle->shutdown_req, status);    }    DECREASE_PENDING_REQ_COUNT(handle);    return;  }  if (handle->flags & UV_HANDLE_CLOSING &&      handle->reqs_pending == 0) {    assert(!(handle->flags & UV_HANDLE_CLOSED));    handle->flags |= UV_HANDLE_CLOSED;    if (!(handle->flags & UV_HANDLE_CONNECTION) && handle->accept_reqs) {      if (handle->flags & UV_HANDLE_EMULATE_IOCP) {        for (i = 0; i < uv_simultaneous_server_accepts; i++) {          req = &handle->accept_reqs[i];          if (req->wait_handle != INVALID_HANDLE_VALUE) {            UnregisterWait(req->wait_handle);            req->wait_handle = INVALID_HANDLE_VALUE;          }          if (req->event_handle) {            CloseHandle(req->event_handle);            req->event_handle = NULL;          }        }      }      free(handle->accept_reqs);      handle->accept_reqs = NULL;    }    if (handle->close_cb) {      handle->close_cb((uv_handle_t*)handle);    }    active_tcp_streams--;    uv_unref(loop);  }}

void uv_pipe_endgame(uv_loop_t* loop, uv_pipe_t* handle) {  DWORD result;  uv_shutdown_t* req;  NTSTATUS nt_status;  IO_STATUS_BLOCK io_status;  FILE_PIPE_LOCAL_INFORMATION pipe_info;  if ((handle->flags & UV_HANDLE_CONNECTION) &&      handle->shutdown_req != NULL &&      handle->write_reqs_pending == 0) {    req = handle->shutdown_req;    /* Clear the shutdown_req field so we don't go here again. */    handle->shutdown_req = NULL;    if (handle->flags & UV__HANDLE_CLOSING) {      UNREGISTER_HANDLE_REQ(loop, handle, req);      /* Already closing. Cancel the shutdown. */      if (req->cb) {        uv__set_artificial_error(loop, UV_ECANCELED);        req->cb(req, -1);      }      DECREASE_PENDING_REQ_COUNT(handle);      return;    }    /* Try to avoid flushing the pipe buffer in the thread pool. */    nt_status = pNtQueryInformationFile(handle->handle,                                        &io_status,                                        &pipe_info,                                        sizeof pipe_info,                                        FilePipeLocalInformation);    if (nt_status != STATUS_SUCCESS) {      /* Failure */      UNREGISTER_HANDLE_REQ(loop, handle, req);      handle->flags |= UV_HANDLE_WRITABLE; /* Questionable */      if (req->cb) {        uv__set_sys_error(loop, pRtlNtStatusToDosError(nt_status));        req->cb(req, -1);      }      DECREASE_PENDING_REQ_COUNT(handle);      return;    }    if (pipe_info.OutboundQuota == pipe_info.WriteQuotaAvailable) {      /* Short-circuit, no need to call FlushFileBuffers. */      uv_insert_pending_req(loop, (uv_req_t*) req);      return;    }    /* Run FlushFileBuffers in the thread pool. */    result = QueueUserWorkItem(pipe_shutdown_thread_proc,                               req,                               WT_EXECUTELONGFUNCTION);    if (result) {      return;    } else {      /* Failure. */      UNREGISTER_HANDLE_REQ(loop, handle, req);      handle->flags |= UV_HANDLE_WRITABLE; /* Questionable */      if (req->cb) {        uv__set_sys_error(loop, GetLastError());        req->cb(req, -1);      }      DECREASE_PENDING_REQ_COUNT(handle);      return;    }  }  if (handle->flags & UV__HANDLE_CLOSING &&      handle->reqs_pending == 0) {    assert(!(handle->flags & UV_HANDLE_CLOSED));    if (handle->flags & UV_HANDLE_CONNECTION) {      if (handle->pending_ipc_info.socket_info) {        free(handle->pending_ipc_info.socket_info);        handle->pending_ipc_info.socket_info = NULL;      }      if (handle->flags & UV_HANDLE_EMULATE_IOCP) {        if (handle->read_req.wait_handle != INVALID_HANDLE_VALUE) {          UnregisterWait(handle->read_req.wait_handle);          handle->read_req.wait_handle = INVALID_HANDLE_VALUE;        }        if (handle->read_req.event_handle) {          CloseHandle(handle->read_req.event_handle);          handle->read_req.event_handle = NULL;        }      }    }    if (handle->flags & UV_HANDLE_PIPESERVER) {//.........这里部分代码省略.........

void uv_process_tcp_read_req(uv_loop_t* loop, uv_tcp_t* handle,    uv_req_t* req) {  DWORD bytes, flags, err;  uv_buf_t buf;  assert(handle->type == UV_TCP);  handle->flags &= ~UV_HANDLE_READ_PENDING;  if (!REQ_SUCCESS(req)) {    /* An error occurred doing the read. */    if ((handle->flags & UV_HANDLE_READING)) {      handle->flags &= ~UV_HANDLE_READING;      uv__set_sys_error(loop, GET_REQ_SOCK_ERROR(req));      buf = (handle->flags & UV_HANDLE_ZERO_READ) ?            uv_buf_init(NULL, 0) : handle->read_buffer;      handle->read_cb((uv_stream_t*)handle, -1, buf);    }  } else {    if (!(handle->flags & UV_HANDLE_ZERO_READ)) {      /* The read was done with a non-zero buffer length. */      if (req->overlapped.InternalHigh > 0) {        /* Successful read */        handle->read_cb((uv_stream_t*)handle,                        req->overlapped.InternalHigh,                        handle->read_buffer);        /* Read again only if bytes == buf.len */        if (req->overlapped.InternalHigh < handle->read_buffer.len) {          goto done;        }      } else {        /* Connection closed */        handle->flags &= ~UV_HANDLE_READING;        handle->flags |= UV_HANDLE_EOF;        uv__set_error(loop, UV_EOF, ERROR_SUCCESS);        buf.base = 0;        buf.len = 0;        handle->read_cb((uv_stream_t*)handle, -1, handle->read_buffer);        goto done;      }    }    /* Do nonblocking reads until the buffer is empty */    while (handle->flags & UV_HANDLE_READING) {      buf = handle->alloc_cb((uv_handle_t*) handle, 65536);      assert(buf.len > 0);      flags = 0;      if (WSARecv(handle->socket,                  (WSABUF*)&buf,                  1,                  &bytes,                  &flags,                  NULL,                  NULL) != SOCKET_ERROR) {        if (bytes > 0) {          /* Successful read */          handle->read_cb((uv_stream_t*)handle, bytes, buf);          /* Read again only if bytes == buf.len */          if (bytes < buf.len) {            break;          }        } else {          /* Connection closed */          handle->flags &= ~UV_HANDLE_READING;          handle->flags |= UV_HANDLE_EOF;          uv__set_error(loop, UV_EOF, ERROR_SUCCESS);          handle->read_cb((uv_stream_t*)handle, -1, buf);          break;        }      } else {        err = WSAGetLastError();        if (err == WSAEWOULDBLOCK) {          /* Read buffer was completely empty, report a 0-byte read. */          uv__set_sys_error(loop, WSAEWOULDBLOCK);          handle->read_cb((uv_stream_t*)handle, 0, buf);        } else {          /* Ouch! serious error. */          uv__set_sys_error(loop, err);          handle->flags &= ~UV_HANDLE_READING;          handle->read_cb((uv_stream_t*)handle, -1, buf);        }        break;      }    }done:    /* Post another read if still reading and not closing. */    if ((handle->flags & UV_HANDLE_READING) &&        !(handle->flags & UV_HANDLE_READ_PENDING)) {      uv_tcp_queue_read(loop, handle);    }  }  DECREASE_PENDING_REQ_COUNT(handle);}

//.........这里部分代码省略.........                        &ipc_frame.header,                        sizeof(ipc_frame.header),                        &bytes,                        NULL)) {            uv_pipe_read_error_or_eof(loop, handle, GetLastError(),              uv_null_buf_);            break;          }          assert(bytes == sizeof(ipc_frame.header));          assert(ipc_frame.header.flags <= (UV_IPC_TCP_SERVER | UV_IPC_RAW_DATA |            UV_IPC_TCP_CONNECTION));          if (ipc_frame.header.flags & UV_IPC_TCP_SERVER) {            assert(avail - sizeof(ipc_frame.header) >=              sizeof(ipc_frame.socket_info));            /* Read the TCP socket info. */            if (!ReadFile(handle->handle,                          &ipc_frame.socket_info,                          sizeof(ipc_frame) - sizeof(ipc_frame.header),                          &bytes,                          NULL)) {              uv_pipe_read_error_or_eof(loop, handle, GetLastError(),                uv_null_buf_);              break;            }            assert(bytes == sizeof(ipc_frame) - sizeof(ipc_frame.header));            /* Store the pending socket info. */            assert(!handle->pending_ipc_info.socket_info);            handle->pending_ipc_info.socket_info =              (WSAPROTOCOL_INFOW*)malloc(sizeof(*(handle->pending_ipc_info.socket_info)));            if (!handle->pending_ipc_info.socket_info) {              uv_fatal_error(ERROR_OUTOFMEMORY, "malloc");            }            *(handle->pending_ipc_info.socket_info) = ipc_frame.socket_info;            handle->pending_ipc_info.tcp_connection =              ipc_frame.header.flags & UV_IPC_TCP_CONNECTION;          }          if (ipc_frame.header.flags & UV_IPC_RAW_DATA) {            handle->remaining_ipc_rawdata_bytes =              ipc_frame.header.raw_data_length;            continue;          }        } else {          avail = min(avail, (DWORD)handle->remaining_ipc_rawdata_bytes);        }      }      buf = handle->alloc_cb((uv_handle_t*) handle, avail);      assert(buf.len > 0);      if (ReadFile(handle->handle,                   buf.base,                   buf.len,                   &bytes,                   NULL)) {        /* Successful read */        if (handle->ipc) {          assert(handle->remaining_ipc_rawdata_bytes >= bytes);          handle->remaining_ipc_rawdata_bytes =            handle->remaining_ipc_rawdata_bytes - bytes;          if (handle->read2_cb) {            handle->read2_cb(handle, bytes, buf,              handle->pending_ipc_info.socket_info ? UV_TCP : UV_UNKNOWN_HANDLE);          } else if (handle->read_cb) {            handle->read_cb((uv_stream_t*)handle, bytes, buf);          }          if (handle->pending_ipc_info.socket_info) {            free(handle->pending_ipc_info.socket_info);            handle->pending_ipc_info.socket_info = NULL;          }        } else {          handle->read_cb((uv_stream_t*)handle, bytes, buf);        }        /* Read again only if bytes == buf.len */        if (bytes <= buf.len) {          break;        }      } else {        uv_pipe_read_error_or_eof(loop, handle, GetLastError(), uv_null_buf_);        break;      }    }    /* Post another 0-read if still reading and not closing. */    if ((handle->flags & UV_HANDLE_READING) &&        !(handle->flags & UV_HANDLE_READ_PENDING)) {      uv_pipe_queue_read(loop, handle);    }  }  DECREASE_PENDING_REQ_COUNT(handle);}