/* 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;

}

static int uv__pipe_server(

HANDLE* pipeHandle_ptr, DWORD access,

char* name, size_t nameSize, char* random) {

HANDLE pipeHandle;

int err;

for (;;) {

uv_unique_pipe_name(random, name, nameSize);

pipeHandle = CreateNamedPipeA(name,

access | FILE_FLAG_FIRST_PIPE_INSTANCE,

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 random number and try again. */

random++;

}

*pipeHandle_ptr = pipeHandle;

return 0;

error:

if (pipeHandle != INVALID_HANDLE_VALUE)

CloseHandle(pipeHandle);

return err;

}

static int uv__create_pipe_pair(

HANDLE* server_pipe_ptr, HANDLE* client_pipe_ptr,

unsigned int server_flags, unsigned int client_flags,

int inherit_client, char* random) {

/* allowed flags are: UV_READABLE_PIPE | UV_WRITABLE_PIPE | UV_NONBLOCK_PIPE */

char pipe_name[64];

SECURITY_ATTRIBUTES sa;

DWORD server_access;

DWORD client_access;

HANDLE server_pipe;

HANDLE client_pipe;

int err;

server_pipe = INVALID_HANDLE_VALUE;

client_pipe = INVALID_HANDLE_VALUE;

server_access = 0;

if (server_flags & UV_READABLE_PIPE)

server_access |= PIPE_ACCESS_INBOUND;

if (server_flags & UV_WRITABLE_PIPE)

server_access |= PIPE_ACCESS_OUTBOUND;

if (server_flags & UV_NONBLOCK_PIPE)

server_access |= FILE_FLAG_OVERLAPPED;

server_access |= WRITE_DAC;

client_access = 0;

if (client_flags & UV_READABLE_PIPE)

client_access |= GENERIC_READ;

else

client_access |= FILE_READ_ATTRIBUTES;

if (client_flags & UV_WRITABLE_PIPE)

client_access |= GENERIC_WRITE;

else

client_access |= FILE_WRITE_ATTRIBUTES;

client_access |= WRITE_DAC;

/* Create server pipe handle. */

err = uv__pipe_server(&server_pipe,

server_access,

pipe_name,

sizeof(pipe_name),

random);

if (err)

goto error;

/* Create client pipe handle. */

sa.nLength = sizeof sa;

sa.lpSecurityDescriptor = NULL;

sa.bInheritHandle = inherit_client;

client_pipe = CreateFileA(pipe_name,

client_access,

0,

&sa,

OPEN_EXISTING,

(client_flags & UV_NONBLOCK_PIPE) ? FILE_FLAG_OVERLAPPED : 0,

NULL);

if (client_pipe == INVALID_HANDLE_VALUE) {

err = GetLastError();

goto error;

}

#ifndef NDEBUG

/* Validate that the pipe was opened in the right mode. */

{

DWORD mode;

BOOL r;

r = GetNamedPipeHandleState(client_pipe, &mode, NULL, NULL, NULL, NULL, 0);

if (r == TRUE) {

assert(mode == (PIPE_READMODE_BYTE | PIPE_WAIT));

} else {

fprintf(stderr, "libuv assertion failure: GetNamedPipeHandleState failed\n");

}

}

#endif

/* Do a blocking ConnectNamedPipe. This should not block because we have

* both ends of the pipe created. */

if (!ConnectNamedPipe(server_pipe, NULL)) {

if (GetLastError() != ERROR_PIPE_CONNECTED) {

err = GetLastError();

goto error;

}

}

*client_pipe_ptr = client_pipe;

*server_pipe_ptr = server_pipe;

return 0;

error:

if (server_pipe != INVALID_HANDLE_VALUE)

CloseHandle(server_pipe);

if (client_pipe != INVALID_HANDLE_VALUE)

CloseHandle(client_pipe);

return err;

}

int uv_pipe(uv_file fds[2], int read_flags, int write_flags) {

uv_file temp[2];

int err;

HANDLE readh;

HANDLE writeh;

/* Make the server side the inbound (read) end, */

/* so that both ends will have FILE_READ_ATTRIBUTES permission. */

/* TODO: better source of local randomness than &fds? */

read_flags |= UV_READABLE_PIPE;

write_flags |= UV_WRITABLE_PIPE;

err = uv__create_pipe_pair(&readh, &writeh, read_flags, write_flags, 0, (char*) &fds[0]);

if (err != 0)

return err;

temp[0] = _open_osfhandle((intptr_t) readh, 0);

if (temp[0] == -1) {

if (errno == UV_EMFILE)

err = UV_EMFILE;

else

err = UV_UNKNOWN;

CloseHandle(readh);

CloseHandle(writeh);

return err;

}

temp[1] = _open_osfhandle((intptr_t) writeh, 0);

if (temp[1] == -1) {

if (errno == UV_EMFILE)

err = UV_EMFILE;

else

err = UV_UNKNOWN;

_close(temp[0]);

CloseHandle(writeh);

return err;

}

fds[0] = temp[0];

fds[1] = temp[1];

return 0;

}

int uv__create_stdio_pipe_pair(uv_loop_t* loop,

uv_pipe_t* parent_pipe, HANDLE* child_pipe_ptr, unsigned int flags) {

/* The parent_pipe is always the server_pipe and kept by libuv.

* The child_pipe is always the client_pipe and is passed to the child.

* The flags are specified with respect to their usage in the child. */

HANDLE server_pipe;

HANDLE client_pipe;

unsigned int server_flags;

unsigned int client_flags;

int err;

server_pipe = INVALID_HANDLE_VALUE;

client_pipe = INVALID_HANDLE_VALUE;

server_flags = 0;

client_flags = 0;

if (flags & UV_READABLE_PIPE) {

/* The server needs inbound (read) access too, otherwise CreateNamedPipe()

* won't give us the FILE_READ_ATTRIBUTES permission. We need that to probe

* the state of the write buffer when we're trying to shutdown the pipe. */

server_flags |= UV_READABLE_PIPE | UV_WRITABLE_PIPE;

client_flags |= UV_READABLE_PIPE;

}

if (flags & UV_WRITABLE_PIPE) {

server_flags |= UV_READABLE_PIPE;

client_flags |= UV_WRITABLE_PIPE;

}

server_flags |= UV_NONBLOCK_PIPE;

if (flags & UV_NONBLOCK_PIPE || parent_pipe->ipc) {

client_flags |= UV_NONBLOCK_PIPE;

}

err = uv__create_pipe_pair(&server_pipe, &client_pipe,

server_flags, client_flags, 1, (char*) server_pipe);

if (err)

goto error;

if (CreateIoCompletionPort(server_pipe,

loop->iocp,

(ULONG_PTR) parent_pipe,

0) == NULL) {

err = GetLastError();

goto error;

}

uv_pipe_connection_init(parent_pipe);

parent_pipe->handle = server_pipe;

*child_pipe_ptr = client_pipe;

/* The server end is now readable and/or writable. */

if (flags & UV_READABLE_PIPE)

parent_pipe->flags |= UV_HANDLE_WRITABLE;

if (flags & UV_WRITABLE_PIPE)

parent_pipe->flags |= UV_HANDLE_READABLE;

return 0;

error:

if (server_pipe != INVALID_HANDLE_VALUE)

CloseHandle(server_pipe);

if (client_pipe != INVALID_HANDLE_VALUE)

CloseHandle(client_pipe);

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;

}