/* Copyright Joyent, Inc. and other Node contributors. All rights reserved.

 *

 * Permission is hereby granted, free of charge, to any person obtaining a copy

 * of this software and associated documentation files (the "Software"), to

 * deal in the Software without restriction, including without limitation the

 * rights to use, copy, modify, merge, publish, distribute, sublicense, and/or

 * sell copies of the Software, and to permit persons to whom the Software is

 * furnished to do so, subject to the following conditions:

 *

 * The above copyright notice and this permission notice shall be included in

 * all copies or substantial portions of the Software.

 *

 * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR

 * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,

 * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE

 * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER

 * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING

 * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS

 * IN THE SOFTWARE.

 */

#include <assert.h>

#include <io.h>

#include <stdio.h>

#include <stdlib.h>

#include <string.h>

#include "handle-inl.h"

#include "internal.h"

#include "req-inl.h"

#include "stream-inl.h"

#include "uv-common.h"

#include "uv.h"

#include <aclapi.h>

#include <accctrl.h>

/* A zero-size buffer for use by uv_pipe_read */

static char uv_zero_[] = "";

/* Null uv_buf_t */

static const uv_buf_t uv_null_buf_ = { 0, NULL };

/* The timeout that the pipe will wait for the remote end to write data when

 * the local ends wants to shut it down. */

static const int64_t eof_timeout = 50; /* ms */

static const int default_pending_pipe_instances = 4;

/* Pipe prefix */

static char pipe_prefix[] = "\\\\?\\pipe";

static const int pipe_prefix_len = sizeof(pipe_prefix) - 1;

/* IPC incoming xfer queue item. */

typedef struct {

  uv__ipc_socket_xfer_type_t xfer_type;

  uv__ipc_socket_xfer_info_t xfer_info;

  QUEUE member;

} uv__ipc_xfer_queue_item_t;

/* IPC frame header flags. */

/* clang-format off */

enum {

  UV__IPC_FRAME_HAS_DATA                = 0x01,

  UV__IPC_FRAME_HAS_SOCKET_XFER         = 0x02,

  UV__IPC_FRAME_XFER_IS_TCP_CONNECTION  = 0x04,

  /* These are combinations of the flags above. */

  UV__IPC_FRAME_XFER_FLAGS              = 0x06,

  UV__IPC_FRAME_VALID_FLAGS             = 0x07

};

/* clang-format on */

/* IPC frame header. */

typedef struct {

  uint32_t flags;

  uint32_t reserved1;   /* Ignored. */

  uint32_t data_length; /* Must be zero if there is no data. */

  uint32_t reserved2;   /* Must be zero. */

} uv__ipc_frame_header_t;

/* To implement the IPC protocol correctly, these structures must have exactly

 * the right size. */

STATIC_ASSERT(sizeof(uv__ipc_frame_header_t) == 16);

STATIC_ASSERT(sizeof(uv__ipc_socket_xfer_info_t) == 632);

/* Coalesced write request. */

typedef struct {

  uv_write_t req;       /* Internal heap-allocated write request. */

  uv_write_t* user_req; /* Pointer to user-specified uv_write_t. */

} uv__coalesced_write_t;

static void eof_timer_init(uv_pipe_t* pipe);

static void eof_timer_start(uv_pipe_t* pipe);

static void eof_timer_stop(uv_pipe_t* pipe);

static void eof_timer_cb(uv_timer_t* timer);

static void eof_timer_destroy(uv_pipe_t* pipe);

static void eof_timer_close_cb(uv_handle_t* handle);

static void uv_unique_pipe_name(char* ptr, char* name, size_t size) {

  snprintf(name, size, "\\\\?\\pipe\\uv\\%p-%lu", ptr, GetCurrentProcessId());

}

int uv_pipe_init(uv_loop_t* loop, uv_pipe_t* handle, int ipc) {

  uv_stream_init(loop, (uv_stream_t*)handle, UV_NAMED_PIPE);

  handle->reqs_pending = 0;

  handle->handle = INVALID_HANDLE_VALUE;

  handle->name = NULL;

  handle->pipe.conn.ipc_remote_pid = 0;

  handle->pipe.conn.ipc_data_frame.payload_remaining = 0;

  QUEUE_INIT(&handle->pipe.conn.ipc_xfer_queue);

  handle->pipe.conn.ipc_xfer_queue_length = 0;

  handle->ipc = ipc;

  handle->pipe.conn.non_overlapped_writes_tail = NULL;

  return 0;

}

static void uv_pipe_connection_init(uv_pipe_t* handle) {

  uv_connection_init((uv_stream_t*) handle);

  handle->read_req.data = handle;

  handle->pipe.conn.eof_timer = NULL;

  assert(!(handle->flags & UV_HANDLE_PIPESERVER));

  if (handle->flags & UV_HANDLE_NON_OVERLAPPED_PIPE) {

    handle->pipe.conn.readfile_thread_handle = NULL;

    InitializeCriticalSection(&handle->pipe.conn.readfile_thread_lock);

  }

}

static HANDLE open_named_pipe(const WCHAR* name, DWORD* duplex_flags) {

  HANDLE pipeHandle;

  /*

   * Assume that we have a duplex pipe first, so attempt to

   * connect with GENERIC_READ | GENERIC_WRITE.

   */

  pipeHandle = CreateFileW(name,

                           GENERIC_READ | GENERIC_WRITE,

                           0,

                           NULL,

                           OPEN_EXISTING,

                           FILE_FLAG_OVERLAPPED,

                           NULL);

  if (pipeHandle != INVALID_HANDLE_VALUE) {

    *duplex_flags = UV_HANDLE_READABLE | UV_HANDLE_WRITABLE;

    return pipeHandle;

  }

  /*

   * If the pipe is not duplex CreateFileW fails with

   * ERROR_ACCESS_DENIED.  In that case try to connect

   * as a read-only or write-only.

   */

  if (GetLastError() == ERROR_ACCESS_DENIED) {

    pipeHandle = CreateFileW(name,

                             GENERIC_READ | FILE_WRITE_ATTRIBUTES,

                             0,

                             NULL,

                             OPEN_EXISTING,

                             FILE_FLAG_OVERLAPPED,

                             NULL);

    if (pipeHandle != INVALID_HANDLE_VALUE) {

      *duplex_flags = UV_HANDLE_READABLE;

      return pipeHandle;

    }

  }

  if (GetLastError() == ERROR_ACCESS_DENIED) {

    pipeHandle = CreateFileW(name,

                             GENERIC_WRITE | FILE_READ_ATTRIBUTES,

                             0,

                             NULL,

                             OPEN_EXISTING,

                             FILE_FLAG_OVERLAPPED,

                             NULL);

    if (pipeHandle != INVALID_HANDLE_VALUE) {

      *duplex_flags = UV_HANDLE_WRITABLE;

      return pipeHandle;

    }

  }

  return INVALID_HANDLE_VALUE;

}

static void close_pipe(uv_pipe_t* pipe) {

  assert(pipe->u.fd == -1 || pipe->u.fd > 2);

  if (pipe->u.fd == -1)

    CloseHandle(pipe->handle);

  else

    close(pipe->u.fd);

  pipe->u.fd = -1;

  pipe->handle = INVALID_HANDLE_VALUE;

}

int uv_stdio_pipe_server(uv_loop_t* loop, uv_pipe_t* handle, DWORD access,

    char* name, size_t nameSize) {

  HANDLE pipeHandle;

  int err;

  char* ptr = (char*)handle;

  for (;;) {

    uv_unique_pipe_name(ptr, name, nameSize);

    pipeHandle = CreateNamedPipeA(name,

      access | FILE_FLAG_OVERLAPPED | FILE_FLAG_FIRST_PIPE_INSTANCE | WRITE_DAC,

      PIPE_TYPE_BYTE | PIPE_READMODE_BYTE | PIPE_WAIT, 1, 65536, 65536, 0,

      NULL);

    if (pipeHandle != INVALID_HANDLE_VALUE) {

      /* No name collisions.  We're done. */

      break;

    }

    err = GetLastError();

    if (err != ERROR_PIPE_BUSY && err != ERROR_ACCESS_DENIED) {

      goto error;

    }

    /* Pipe name collision.  Increment the pointer and try again. */

    ptr++;

  }

  if (CreateIoCompletionPort(pipeHandle,

                             loop->iocp,

                             (ULONG_PTR)handle,

                             0) == NULL) {

    err = GetLastError();

    goto error;

  }

  uv_pipe_connection_init(handle);

  handle->handle = pipeHandle;

  return 0;

 error:

  if (pipeHandle != INVALID_HANDLE_VALUE)

    CloseHandle(pipeHandle);

  return err;

}

static int uv_set_pipe_handle(uv_loop_t* loop,

                              uv_pipe_t* handle,

                              HANDLE pipeHandle,

                              int fd,

                              DWORD duplex_flags) {

  NTSTATUS nt_status;

  IO_STATUS_BLOCK io_status;

  FILE_MODE_INFORMATION mode_info;

  DWORD mode = PIPE_READMODE_BYTE | PIPE_WAIT;

  DWORD current_mode = 0;

  DWORD err = 0;

  if (handle->flags & UV_HANDLE_PIPESERVER)

    return UV_EINVAL;

  if (handle->handle != INVALID_HANDLE_VALUE)

    return UV_EBUSY;

  if (!SetNamedPipeHandleState(pipeHandle, &mode, NULL, NULL)) {

    err = GetLastError();

    if (err == ERROR_ACCESS_DENIED) {

      /*

       * SetNamedPipeHandleState can fail if the handle doesn't have either

       * GENERIC_WRITE  or FILE_WRITE_ATTRIBUTES.

       * But if the handle already has the desired wait and blocking modes

       * we can continue.

       */

      if (!GetNamedPipeHandleState(pipeHandle, &current_mode, NULL, NULL,

                                   NULL, NULL, 0)) {

        return -1;

      } else if (current_mode & PIPE_NOWAIT) {

        SetLastError(ERROR_ACCESS_DENIED);

        return -1;

      }

    } else {

      /* If this returns ERROR_INVALID_PARAMETER we probably opened

       * something that is not a pipe. */

      if (err == ERROR_INVALID_PARAMETER) {

        SetLastError(WSAENOTSOCK);

      }

      return -1;

    }

  }

  /* Check if the pipe was created with FILE_FLAG_OVERLAPPED. */

  nt_status = pNtQueryInformationFile(pipeHandle,

                                      &io_status,

                                      &mode_info,

                                      sizeof(mode_info),

                                      FileModeInformation);

  if (nt_status != STATUS_SUCCESS) {

    return -1;

  }

  if (mode_info.Mode & FILE_SYNCHRONOUS_IO_ALERT ||

      mode_info.Mode & FILE_SYNCHRONOUS_IO_NONALERT) {

    /* Non-overlapped pipe. */

    handle->flags |= UV_HANDLE_NON_OVERLAPPED_PIPE;

  } else {

    /* Overlapped pipe.  Try to associate with IOCP. */

    if (CreateIoCompletionPort(pipeHandle,

                               loop->iocp,

                               (ULONG_PTR) handle,

                               0) == NULL) {

      handle->flags |= UV_HANDLE_EMULATE_IOCP;

    }

  }

  handle->handle = pipeHandle;

  handle->u.fd = fd;

  handle->flags |= duplex_flags;

  return 0;

}

static int pipe_alloc_accept(uv_loop_t* loop, uv_pipe_t* handle,

                             uv_pipe_accept_t* req, BOOL firstInstance) {

  assert(req->pipeHandle == INVALID_HANDLE_VALUE);

  req->pipeHandle =

      CreateNamedPipeW(handle->name,

                       PIPE_ACCESS_DUPLEX | FILE_FLAG_OVERLAPPED | WRITE_DAC |

                         (firstInstance ? FILE_FLAG_FIRST_PIPE_INSTANCE : 0),

                       PIPE_TYPE_BYTE | PIPE_READMODE_BYTE | PIPE_WAIT,

                       PIPE_UNLIMITED_INSTANCES, 65536, 65536, 0, NULL);

  if (req->pipeHandle == INVALID_HANDLE_VALUE) {

    return 0;

  }

  /* Associate it with IOCP so we can get events. */

  if (CreateIoCompletionPort(req->pipeHandle,

                             loop->iocp,

                             (ULONG_PTR) handle,

                             0) == NULL) {

    uv_fatal_error(GetLastError(), "CreateIoCompletionPort");

  }

  /* Stash a handle in the server object for use from places such as

   * getsockname and chmod. As we transfer ownership of these to client

   * objects, we'll allocate new ones here. */

  handle->handle = req->pipeHandle;

  return 1;

}

static DWORD WINAPI pipe_shutdown_thread_proc(void* parameter) {

  uv_loop_t* loop;

  uv_pipe_t* handle;

  uv_shutdown_t* req;

  req = (uv_shutdown_t*) parameter;

  assert(req);

  handle = (uv_pipe_t*) req->handle;

  assert(handle);

  loop = handle->loop;

  assert(loop);

  FlushFileBuffers(handle->handle);

  /* Post completed */

  POST_COMPLETION_FOR_REQ(loop, req);

  return 0;

}

void uv_pipe_endgame(uv_loop_t* loop, uv_pipe_t* handle) {

  int err;

  DWORD result;

  uv_shutdown_t* req;

  NTSTATUS nt_status;

  IO_STATUS_BLOCK io_status;

  FILE_PIPE_LOCAL_INFORMATION pipe_info;

  uv__ipc_xfer_queue_item_t* xfer_queue_item;

  if ((handle->flags & UV_HANDLE_CONNECTION) &&

      handle->stream.conn.shutdown_req != NULL &&

      handle->stream.conn.write_reqs_pending == 0) {

    req = handle->stream.conn.shutdown_req;

    /* Clear the shutdown_req field so we don't go here again. */

    handle->stream.conn.shutdown_req = NULL;

    if (handle->flags & UV_HANDLE_CLOSING) {

      UNREGISTER_HANDLE_REQ(loop, handle, req);

      /* Already closing. Cancel the shutdown. */

      if (req->cb) {

        req->cb(req, UV_ECANCELED);

      }

      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) {

        err = pRtlNtStatusToDosError(nt_status);

        req->cb(req, uv_translate_sys_error(err));

      }

      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) {

        err = GetLastError();

        req->cb(req, uv_translate_sys_error(err));

      }

      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) {

      /* Free pending sockets */

      while (!QUEUE_EMPTY(&handle->pipe.conn.ipc_xfer_queue)) {

        QUEUE* q;

        SOCKET socket;

        q = QUEUE_HEAD(&handle->pipe.conn.ipc_xfer_queue);

        QUEUE_REMOVE(q);

        xfer_queue_item = QUEUE_DATA(q, uv__ipc_xfer_queue_item_t, member);

        /* Materialize socket and close it */

        socket = WSASocketW(FROM_PROTOCOL_INFO,

                            FROM_PROTOCOL_INFO,

                            FROM_PROTOCOL_INFO,

                            &xfer_queue_item->xfer_info.socket_info,

                            0,

                            WSA_FLAG_OVERLAPPED);

        uv__free(xfer_queue_item);

        if (socket != INVALID_SOCKET)

          closesocket(socket);

      }

      handle->pipe.conn.ipc_xfer_queue_length = 0;

      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 != NULL) {

          CloseHandle(handle->read_req.event_handle);

          handle->read_req.event_handle = NULL;

        }

      }

      if (handle->flags & UV_HANDLE_NON_OVERLAPPED_PIPE)

        DeleteCriticalSection(&handle->pipe.conn.readfile_thread_lock);

    }

    if (handle->flags & UV_HANDLE_PIPESERVER) {

      assert(handle->pipe.serv.accept_reqs);

      uv__free(handle->pipe.serv.accept_reqs);

      handle->pipe.serv.accept_reqs = NULL;

    }

    uv__handle_close(handle);

  }

}

void uv_pipe_pending_instances(uv_pipe_t* handle, int count) {

  if (handle->flags & UV_HANDLE_BOUND)

    return;

  handle->pipe.serv.pending_instances = count;

  handle->flags |= UV_HANDLE_PIPESERVER;

}

/* Creates a pipe server. */

int uv_pipe_bind(uv_pipe_t* handle, const char* name) {

  uv_loop_t* loop = handle->loop;

  int i, err, nameSize;

  uv_pipe_accept_t* req;

  if (handle->flags & UV_HANDLE_BOUND) {

    return UV_EINVAL;

  }

  if (!name) {

    return UV_EINVAL;

  }

  if (!(handle->flags & UV_HANDLE_PIPESERVER)) {

    handle->pipe.serv.pending_instances = default_pending_pipe_instances;

  }

  handle->pipe.serv.accept_reqs = (uv_pipe_accept_t*)

    uv__malloc(sizeof(uv_pipe_accept_t) * handle->pipe.serv.pending_instances);

  if (!handle->pipe.serv.accept_reqs) {

    uv_fatal_error(ERROR_OUTOFMEMORY, "uv__malloc");

  }

  for (i = 0; i < handle->pipe.serv.pending_instances; i++) {

    req = &handle->pipe.serv.accept_reqs[i];

    UV_REQ_INIT(req, UV_ACCEPT);

    req->data = handle;

    req->pipeHandle = INVALID_HANDLE_VALUE;

    req->next_pending = NULL;

  }

  /* Convert name to UTF16. */

  nameSize = MultiByteToWideChar(CP_UTF8, 0, name, -1, NULL, 0) * sizeof(WCHAR);

  handle->name = uv__malloc(nameSize);

  if (!handle->name) {

    uv_fatal_error(ERROR_OUTOFMEMORY, "uv__malloc");

  }

  if (!MultiByteToWideChar(CP_UTF8,

                           0,

                           name,

                           -1,

                           handle->name,

                           nameSize / sizeof(WCHAR))) {

    err = GetLastError();

    goto error;

  }

  /*

   * Attempt to create the first pipe with FILE_FLAG_FIRST_PIPE_INSTANCE.

   * If this fails then there's already a pipe server for the given pipe name.

   */

  if (!pipe_alloc_accept(loop,

                         handle,

                         &handle->pipe.serv.accept_reqs[0],

                         TRUE)) {

    err = GetLastError();

    if (err == ERROR_ACCESS_DENIED) {

      err = WSAEADDRINUSE;  /* Translates to UV_EADDRINUSE. */

    } else if (err == ERROR_PATH_NOT_FOUND || err == ERROR_INVALID_NAME) {

      err = WSAEACCES;  /* Translates to UV_EACCES. */

    }

    goto error;

  }

  handle->pipe.serv.pending_accepts = NULL;

  handle->flags |= UV_HANDLE_PIPESERVER;

  handle->flags |= UV_HANDLE_BOUND;

  return 0;

error:

  if (handle->name) {

    uv__free(handle->name);

    handle->name = NULL;

  }

  return uv_translate_sys_error(err);

}

static DWORD WINAPI pipe_connect_thread_proc(void* parameter) {

  uv_loop_t* loop;

  uv_pipe_t* handle;

  uv_connect_t* req;

  HANDLE pipeHandle = INVALID_HANDLE_VALUE;

  DWORD duplex_flags;

  req = (uv_connect_t*) parameter;

  assert(req);

  handle = (uv_pipe_t*) req->handle;

  assert(handle);

  loop = handle->loop;

  assert(loop);

  /* We're here because CreateFile on a pipe returned ERROR_PIPE_BUSY. We wait

   * for the pipe to become available with WaitNamedPipe. */

  while (WaitNamedPipeW(handle->name, 30000)) {

    /* The pipe is now available, try to connect. */

    pipeHandle = open_named_pipe(handle->name, &duplex_flags);

    if (pipeHandle != INVALID_HANDLE_VALUE)

      break;

    SwitchToThread();

  }

  if (pipeHandle != INVALID_HANDLE_VALUE &&

      !uv_set_pipe_handle(loop, handle, pipeHandle, -1, duplex_flags)) {

    SET_REQ_SUCCESS(req);

  } else {

    SET_REQ_ERROR(req, GetLastError());

  }

  /* Post completed */

  POST_COMPLETION_FOR_REQ(loop, req);

  return 0;

}

void uv_pipe_connect(uv_connect_t* req, uv_pipe_t* handle,

    const char* name, uv_connect_cb cb) {

  uv_loop_t* loop = handle->loop;

  int err, nameSize;

  HANDLE pipeHandle = INVALID_HANDLE_VALUE;

  DWORD duplex_flags;

  UV_REQ_INIT(req, UV_CONNECT);

  req->handle = (uv_stream_t*) handle;

  req->cb = cb;

  /* Convert name to UTF16. */

  nameSize = MultiByteToWideChar(CP_UTF8, 0, name, -1, NULL, 0) * sizeof(WCHAR);

  handle->name = uv__malloc(nameSize);

  if (!handle->name) {

    uv_fatal_error(ERROR_OUTOFMEMORY, "uv__malloc");

  }

  if (!MultiByteToWideChar(CP_UTF8,

                           0,

                           name,

                           -1,

                           handle->name,

                           nameSize / sizeof(WCHAR))) {

    err = GetLastError();

    goto error;

  }

  pipeHandle = open_named_pipe(handle->name, &duplex_flags);

  if (pipeHandle == INVALID_HANDLE_VALUE) {

    if (GetLastError() == ERROR_PIPE_BUSY) {

      /* Wait for the server to make a pipe instance available. */

      if (!QueueUserWorkItem(&pipe_connect_thread_proc,

                             req,

                             WT_EXECUTELONGFUNCTION)) {

        err = GetLastError();

        goto error;

      }

      REGISTER_HANDLE_REQ(loop, handle, req);

      handle->reqs_pending++;

      return;

    }

    err = GetLastError();

    goto error;

  }

  assert(pipeHandle != INVALID_HANDLE_VALUE);

  if (uv_set_pipe_handle(loop,

                         (uv_pipe_t*) req->handle,

                         pipeHandle,

                         -1,

                         duplex_flags)) {

    err = GetLastError();

    goto error;

  }

  SET_REQ_SUCCESS(req);

  uv_insert_pending_req(loop, (uv_req_t*) req);

  handle->reqs_pending++;

  REGISTER_HANDLE_REQ(loop, handle, req);

  return;

error:

  if (handle->name) {

    uv__free(handle->name);

    handle->name = NULL;

  }

  if (pipeHandle != INVALID_HANDLE_VALUE) {

    CloseHandle(pipeHandle);

  }

  /* Make this req pending reporting an error. */

  SET_REQ_ERROR(req, err);

  uv_insert_pending_req(loop, (uv_req_t*) req);

  handle->reqs_pending++;

  REGISTER_HANDLE_REQ(loop, handle, req);

  return;

}

void uv__pipe_interrupt_read(uv_pipe_t* handle) {

  BOOL r;

  if (!(handle->flags & UV_HANDLE_READ_PENDING))

    return; /* No pending reads. */

  if (handle->flags & UV_HANDLE_CANCELLATION_PENDING)

    return; /* Already cancelled. */

  if (handle->handle == INVALID_HANDLE_VALUE)

    return; /* Pipe handle closed. */

  if (!(handle->flags & UV_HANDLE_NON_OVERLAPPED_PIPE)) {

    /* Cancel asynchronous read. */

    r = CancelIoEx(handle->handle, &handle->read_req.u.io.overlapped);

    assert(r || GetLastError() == ERROR_NOT_FOUND);

  } else {

    /* Cancel synchronous read (which is happening in the thread pool). */

    HANDLE thread;

    volatile HANDLE* thread_ptr = &handle->pipe.conn.readfile_thread_handle;

    EnterCriticalSection(&handle->pipe.conn.readfile_thread_lock);

    thread = *thread_ptr;

    if (thread == NULL) {

      /* The thread pool thread has not yet reached the point of blocking, we

       * can pre-empt it by setting thread_handle to INVALID_HANDLE_VALUE. */

      *thread_ptr = INVALID_HANDLE_VALUE;

    } else {

      /* Spin until the thread has acknowledged (by setting the thread to

       * INVALID_HANDLE_VALUE) that it is past the point of blocking. */

      while (thread != INVALID_HANDLE_VALUE) {

        r = CancelSynchronousIo(thread);

        assert(r || GetLastError() == ERROR_NOT_FOUND);

        SwitchToThread(); /* Yield thread. */

        thread = *thread_ptr;

      }

    }

    LeaveCriticalSection(&handle->pipe.conn.readfile_thread_lock);

  }

  /* Set flag to indicate that read has been cancelled. */

  handle->flags |= UV_HANDLE_CANCELLATION_PENDING;

}

void uv__pipe_read_stop(uv_pipe_t* handle) {

  handle->flags &= ~UV_HANDLE_READING;

  DECREASE_ACTIVE_COUNT(handle->loop, handle);

  uv__pipe_interrupt_read(handle);

}

/* Cleans up uv_pipe_t (server or connection) and all resources associated with

 * it. */

void uv_pipe_cleanup(uv_loop_t* loop, uv_pipe_t* handle) {

  int i;

  HANDLE pipeHandle;

  uv__pipe_interrupt_read(handle);

  if (handle->name) {

    uv__free(handle->name);

    handle->name = NULL;

  }

  if (handle->flags & UV_HANDLE_PIPESERVER) {

    for (i = 0; i < handle->pipe.serv.pending_instances; i++) {

      pipeHandle = handle->pipe.serv.accept_reqs[i].pipeHandle;

      if (pipeHandle != INVALID_HANDLE_VALUE) {

        CloseHandle(pipeHandle);

        handle->pipe.serv.accept_reqs[i].pipeHandle = INVALID_HANDLE_VALUE;

      }

    }

    handle->handle = INVALID_HANDLE_VALUE;

  }

  if (handle->flags & UV_HANDLE_CONNECTION) {

    handle->flags &= ~UV_HANDLE_WRITABLE;

    eof_timer_destroy(handle);

  }

  if ((handle->flags & UV_HANDLE_CONNECTION)

      && handle->handle != INVALID_HANDLE_VALUE)

    close_pipe(handle);

}

void uv_pipe_close(uv_loop_t* loop, uv_pipe_t* handle) {

  if (handle->flags & UV_HANDLE_READING) {

    handle->flags &= ~UV_HANDLE_READING;

    DECREASE_ACTIVE_COUNT(loop, handle);

  }

  if (handle->flags & UV_HANDLE_LISTENING) {

    handle->flags &= ~UV_HANDLE_LISTENING;

    DECREASE_ACTIVE_COUNT(loop, handle);

  }

  uv_pipe_cleanup(loop, handle);

  if (handle->reqs_pending == 0) {

    uv_want_endgame(loop, (uv_handle_t*) handle);

  }

  handle->flags &= ~(UV_HANDLE_READABLE | UV_HANDLE_WRITABLE);

  uv__handle_closing(handle);

}

static void uv_pipe_queue_accept(uv_loop_t* loop, uv_pipe_t* handle,

    uv_pipe_accept_t* req, BOOL firstInstance) {

  assert(handle->flags & UV_HANDLE_LISTENING);

  if (!firstInstance && !pipe_alloc_accept(loop, handle, req, FALSE)) {

    SET_REQ_ERROR(req, GetLastError());

    uv_insert_pending_req(loop, (uv_req_t*) req);

    handle->reqs_pending++;

    return;

  }

  assert(req->pipeHandle != INVALID_HANDLE_VALUE);

  /* Prepare the overlapped structure. */

  memset(&(req->u.io.overlapped), 0, sizeof(req->u.io.overlapped));

  if (!ConnectNamedPipe(req->pipeHandle, &req->u.io.overlapped) &&

      GetLastError() != ERROR_IO_PENDING) {

    if (GetLastError() == ERROR_PIPE_CONNECTED) {

      SET_REQ_SUCCESS(req);

    } else {

      CloseHandle(req->pipeHandle);

      req->pipeHandle = INVALID_HANDLE_VALUE;

      /* Make this req pending reporting an error. */

      SET_REQ_ERROR(req, GetLastError());

    }

    uv_insert_pending_req(loop, (uv_req_t*) req);

    handle->reqs_pending++;

    return;

  }

  /* Wait for completion via IOCP */

  handle->reqs_pending++;

}

int uv_pipe_accept(uv_pipe_t* server, uv_stream_t* client) {

  uv_loop_t* loop = server->loop;

  uv_pipe_t* pipe_client;

  uv_pipe_accept_t* req;

  QUEUE* q;