/* 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 <string.h>

#include <stdio.h>

#include <stdlib.h>

#include "uv.h"

#include "internal.h"

#include "handle-inl.h"

#include "stream-inl.h"

#include "req-inl.h"

typedef struct uv__ipc_queue_item_s uv__ipc_queue_item_t;

struct uv__ipc_queue_item_s {

/*

* NOTE: It is important for socket_info_ex to be the first field,

* because we will we assigning it to the pending_ipc_info.socket_info

*/

uv__ipc_socket_info_ex socket_info_ex;

QUEUE member;

int tcp_connection;

};

/* 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 protocol flags. */

#define UV_IPC_RAW_DATA 0x0001

#define UV_IPC_TCP_SERVER 0x0002

#define UV_IPC_TCP_CONNECTION 0x0004

/* IPC frame header. */

typedef struct {

int flags;

uint64_t raw_data_length;

} uv_ipc_frame_header_t;

/* IPC frame, which contains an imported TCP socket stream. */

typedef struct {

uv_ipc_frame_header_t header;

uv__ipc_socket_info_ex socket_info_ex;

} uv_ipc_frame_uv_stream;

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-%u", 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_pid = 0;

handle->pipe.conn.remaining_ipc_rawdata_bytes = 0;

QUEUE_INIT(&handle->pipe.conn.pending_ipc_info.queue);

handle->pipe.conn.pending_ipc_info.queue_len = 0;

handle->ipc = ipc;

handle->pipe.conn.non_overlapped_writes_tail = NULL;

handle->pipe.conn.readfile_thread = NULL;

uv_req_init(loop, (uv_req_t*) &handle->pipe.conn.ipc_header_write_req);

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 (pCancelSynchronousIo &&

handle->flags & UV_HANDLE_NON_OVERLAPPED_PIPE) {

uv_mutex_init(&handle->pipe.conn.readfile_mutex);

handle->flags |= UV_HANDLE_PIPE_READ_CANCELABLE;

}

}

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;

}

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,

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,

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 (!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->flags |= duplex_flags;

return 0;

}

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_queue_item_t* item;

if (handle->flags & UV_HANDLE_PIPE_READ_CANCELABLE) {

handle->flags &= ~UV_HANDLE_PIPE_READ_CANCELABLE;

uv_mutex_destroy(&handle->pipe.conn.readfile_mutex);

}

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.pending_ipc_info.queue)) {

QUEUE* q;

SOCKET socket;

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

QUEUE_REMOVE(q);

item = QUEUE_DATA(q, uv__ipc_queue_item_t, member);

/* Materialize socket and close it */

socket = WSASocketW(FROM_PROTOCOL_INFO,

FROM_PROTOCOL_INFO,

FROM_PROTOCOL_INFO,

&item->socket_info_ex.socket_info,

0,

WSA_FLAG_OVERLAPPED);

uv__free(item);

if (socket != INVALID_SOCKET)

closesocket(socket);

}

handle->pipe.conn.pending_ipc_info.queue_len = 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) {

CloseHandle(handle->read_req.event_handle);

handle->read_req.event_handle = NULL;

}

}

}

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

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(loop, (uv_req_t*) req);

req->type = UV_ACCEPT;

req->data = handle;

req->pipeHandle = INVALID_HANDLE_VALUE;

req->next_pending = NULL;

}

/* Convert name to UTF16. */

nameSize = uv_utf8_to_utf16(name, NULL, 0) * sizeof(WCHAR);

handle->name = (WCHAR*)uv__malloc(nameSize);

if (!handle->name) {

uv_fatal_error(ERROR_OUTOFMEMORY, "uv__malloc");

}

if (!uv_utf8_to_utf16(name, 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.

*/

handle->pipe.serv.accept_reqs[0].pipeHandle = CreateNamedPipeW(handle->name,

PIPE_ACCESS_DUPLEX | FILE_FLAG_OVERLAPPED |

FILE_FLAG_FIRST_PIPE_INSTANCE,

PIPE_TYPE_BYTE | PIPE_READMODE_BYTE | PIPE_WAIT,

PIPE_UNLIMITED_INSTANCES, 65536, 65536, 0, NULL);

if (handle->pipe.serv.accept_reqs[0].pipeHandle == INVALID_HANDLE_VALUE) {

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;

}

if (uv_set_pipe_handle(loop,

handle,

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

0)) {

err = GetLastError();

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;

}

if (handle->pipe.serv.accept_reqs[0].pipeHandle != INVALID_HANDLE_VALUE) {

CloseHandle(handle->pipe.serv.accept_reqs[0].pipeHandle);

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

}

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, 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(loop, (uv_req_t*) req);

req->type = UV_CONNECT;

req->handle = (uv_stream_t*) handle;

req->cb = cb;

/* Convert name to UTF16. */

nameSize = uv_utf8_to_utf16(name, NULL, 0) * sizeof(WCHAR);

handle->name = (WCHAR*)uv__malloc(nameSize);

if (!handle->name) {

uv_fatal_error(ERROR_OUTOFMEMORY, "uv__malloc");

}

if (!uv_utf8_to_utf16(name, 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,

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_pause_read(uv_pipe_t* handle) {

if (handle->flags & UV_HANDLE_PIPE_READ_CANCELABLE) {

/* Pause the ReadFile task briefly, to work

around the Windows kernel bug that causes

any access to a NamedPipe to deadlock if

any process has called ReadFile */

HANDLE h;

uv_mutex_lock(&handle->pipe.conn.readfile_mutex);

h = handle->pipe.conn.readfile_thread;

while (h) {

/* spinlock: we expect this to finish quickly,

or we are probably about to deadlock anyways

(in the kernel), so it doesn't matter */

pCancelSynchronousIo(h);

SwitchToThread(); /* yield thread control briefly */

h = handle->pipe.conn.readfile_thread;

}

}

}

void uv__pipe_unpause_read(uv_pipe_t* handle) {

if (handle->flags & UV_HANDLE_PIPE_READ_CANCELABLE) {

uv_mutex_unlock(&handle->pipe.conn.readfile_mutex);

}

}

void uv__pipe_stop_read(uv_pipe_t* handle) {

handle->flags &= ~UV_HANDLE_READING;

uv__pipe_pause_read((uv_pipe_t*)handle);

uv__pipe_unpause_read((uv_pipe_t*)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_stop_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;

}

}

}

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

CloseHandle(handle->handle);

handle->handle = INVALID_HANDLE_VALUE;

}

}

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

assert(req->pipeHandle == INVALID_HANDLE_VALUE);

req->pipeHandle = CreateNamedPipeW(handle->name,

PIPE_ACCESS_DUPLEX | FILE_FLAG_OVERLAPPED,

PIPE_TYPE_BYTE | PIPE_READMODE_BYTE | PIPE_WAIT,

PIPE_UNLIMITED_INSTANCES, 65536, 65536, 0, NULL);

if (req->pipeHandle == INVALID_HANDLE_VALUE) {

SET_REQ_ERROR(req, GetLastError());

uv_insert_pending_req(loop, (uv_req_t*) req);

handle->reqs_pending++;

return;

}

if (uv_set_pipe_handle(loop, handle, req->pipeHandle, 0)) {

CloseHandle(req->pipeHandle);

req->pipeHandle = INVALID_HANDLE_VALUE;

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;

}

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;

uv__ipc_queue_item_t* item;

int err;

if (server->ipc) {

if (QUEUE_EMPTY(&server->pipe.conn.pending_ipc_info.queue)) {

/* No valid pending sockets. */

return WSAEWOULDBLOCK;

}

q = QUEUE_HEAD(&server->pipe.conn.pending_ipc_info.queue);

QUEUE_REMOVE(q);

server->pipe.conn.pending_ipc_info.queue_len--;

item = QUEUE_DATA(q, uv__ipc_queue_item_t, member);

err = uv_tcp_import((uv_tcp_t*)client,

&item->socket_info_ex,

item->tcp_connection);

if (err != 0)

return err;

uv__free(item);

} else {

pipe_client = (uv_pipe_t*)client;

/* Find a connection instance that has been connected, but not yet */

/* accepted. */

req = server->pipe.serv.pending_accepts;

if (!req) {

/* No valid connections found, so we error out. */

return WSAEWOULDBLOCK;

}

/* Initialize the client handle and copy the pipeHandle to the client */

uv_pipe_connection_init(pipe_client);

pipe_client->handle = req->pipeHandle;

pipe_client->flags |= UV_HANDLE_READABLE | UV_HANDLE_WRITABLE;

/* Prepare the req to pick up a new connection */

server->pipe.serv.pending_accepts = req->next_pending;

req->next_pending = NULL;

req->pipeHandle = INVALID_HANDLE_VALUE;

if (!(server->flags & UV__HANDLE_CLOSING)) {

uv_pipe_queue_accept(loop, server, req, FALSE);

}

}

return 0;

}

/* Starts listening for connections for the given pipe. */

int uv_pipe_listen(uv_pipe_t* handle, int backlog, uv_connection_cb cb) {

uv_loop_t* loop = handle->loop;

int i;

if (handle->flags & UV_HANDLE_LISTENING) {

handle->stream.serv.connection_cb = cb;

}

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

return WSAEINVAL;

}

if (handle->flags & UV_HANDLE_READING) {

return WSAEISCONN;

}

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

return ERROR_NOT_SUPPORTED;

}

handle->flags |= UV_HANDLE_LISTENING;

INCREASE_ACTIVE_COUNT(loop, handle);

handle->stream.serv.connection_cb = cb;

/* First pipe handle should have already been created in uv_pipe_bind */

assert(handle->pipe.serv.accept_reqs[0].pipeHandle != INVALID_HANDLE_VALUE);

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

uv_pipe_queue_accept(loop, handle, &handle->pipe.serv.accept_reqs[i], i == 0);

}

return 0;

}

static DWORD WINAPI uv_pipe_zero_readfile_thread_proc(void* parameter) {

int result;

DWORD bytes;

uv_read_t* req = (uv_read_t*) parameter;

uv_pipe_t* handle = (uv_pipe_t*) req->data;

uv_loop_t* loop = handle->loop;

HANDLE hThread = NULL;

DWORD err;

uv_mutex_t *m = &handle->pipe.conn.readfile_mutex;

assert(req != NULL);

assert(req->type == UV_READ);

assert(handle->type == UV_NAMED_PIPE);

if (handle->flags & UV_HANDLE_PIPE_READ_CANCELABLE) {

uv_mutex_lock(m); /* mutex controls *setting* of readfile_thread */

if (DuplicateHandle(GetCurrentProcess(), GetCurrentThread(),

GetCurrentProcess(), &hThread,

0, TRUE, DUPLICATE_SAME_ACCESS)) {

handle->pipe.conn.readfile_thread = hThread;

} else {

hThread = NULL;

}

uv_mutex_unlock(m);

}

restart_readfile:

result = ReadFile(handle->handle,

&uv_zero_,

0,

&bytes,

NULL);

if (!result) {

err = GetLastError();

if (err == ERROR_OPERATION_ABORTED &&

handle->flags & UV_HANDLE_PIPE_READ_CANCELABLE) {

if (handle->flags & UV_HANDLE_READING) {

/* just a brief break to do something else */

handle->pipe.conn.readfile_thread = NULL;

/* resume after it is finished */

uv_mutex_lock(m);

handle->pipe.conn.readfile_thread = hThread;

uv_mutex_unlock(m);

goto restart_readfile;

} else {

result = 1; /* successfully stopped reading */

}

}

}

if (hThread) {

assert(hThread == handle->pipe.conn.readfile_thread);

/* mutex does not control clearing readfile_thread */

handle->pipe.conn.readfile_thread = NULL;

uv_mutex_lock(m);

/* only when we hold the mutex lock is it safe to

open or close the handle */

CloseHandle(hThread);

uv_mutex_unlock(m);

}

if (!result) {

SET_REQ_ERROR(req, err);

}

POST_COMPLETION_FOR_REQ(loop, req);

return 0;

}

static DWORD WINAPI uv_pipe_writefile_thread_proc(void* parameter) {

int result;

DWORD bytes;

uv_write_t* req = (uv_write_t*) parameter;

uv_pipe_t* handle = (uv_pipe_t*) req->handle;

uv_loop_t* loop = handle->loop;

assert(req != NULL);

assert(req->type == UV_WRITE);

assert(handle->type == UV_NAMED_PIPE);

assert(req->write_buffer.base);

result = WriteFile(handle->handle,

req->write_buffer.base,

req->write_buffer.len,

&bytes,