Alien-uv
view release on metacpan or search on metacpan
libuv/src/win/pipe.c view on Meta::CPAN
#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;
}
libuv/src/win/pipe.c view on Meta::CPAN
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) {
assert(req->pipeHandle == INVALID_HANDLE_VALUE);
req->pipeHandle = CreateNamedPipeW(handle->name,
PIPE_ACCESS_DUPLEX | FILE_FLAG_OVERLAPPED | WRITE_DAC,
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, -1, 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);
libuv/src/win/pipe.c view on Meta::CPAN
}
static void uv_pipe_queue_read(uv_loop_t* loop, uv_pipe_t* handle) {
uv_read_t* req;
int result;
assert(handle->flags & UV_HANDLE_READING);
assert(!(handle->flags & UV_HANDLE_READ_PENDING));
assert(handle->handle != INVALID_HANDLE_VALUE);
req = &handle->read_req;
if (handle->flags & UV_HANDLE_NON_OVERLAPPED_PIPE) {
handle->pipe.conn.readfile_thread_handle = NULL; /* Reset cancellation. */
if (!QueueUserWorkItem(&uv_pipe_zero_readfile_thread_proc,
req,
WT_EXECUTELONGFUNCTION)) {
/* Make this req pending reporting an error. */
SET_REQ_ERROR(req, GetLastError());
goto error;
}
} else {
memset(&req->u.io.overlapped, 0, sizeof(req->u.io.overlapped));
if (handle->flags & UV_HANDLE_EMULATE_IOCP) {
req->u.io.overlapped.hEvent = (HANDLE) ((uintptr_t) req->event_handle | 1);
}
/* Do 0-read */
result = ReadFile(handle->handle,
&uv_zero_,
0,
NULL,
&req->u.io.overlapped);
if (!result && GetLastError() != ERROR_IO_PENDING) {
/* Make this req pending reporting an error. */
SET_REQ_ERROR(req, GetLastError());
goto error;
}
if (handle->flags & UV_HANDLE_EMULATE_IOCP) {
if (!req->event_handle) {
req->event_handle = CreateEvent(NULL, 0, 0, NULL);
if (!req->event_handle) {
uv_fatal_error(GetLastError(), "CreateEvent");
}
}
if (req->wait_handle == INVALID_HANDLE_VALUE) {
if (!RegisterWaitForSingleObject(&req->wait_handle,
req->u.io.overlapped.hEvent, post_completion_read_wait, (void*) req,
INFINITE, WT_EXECUTEINWAITTHREAD)) {
SET_REQ_ERROR(req, GetLastError());
goto error;
}
}
}
}
/* Start the eof timer if there is one */
eof_timer_start(handle);
handle->flags |= UV_HANDLE_READ_PENDING;
handle->reqs_pending++;
return;
error:
uv_insert_pending_req(loop, (uv_req_t*)req);
handle->flags |= UV_HANDLE_READ_PENDING;
handle->reqs_pending++;
}
int uv_pipe_read_start(uv_pipe_t* handle,
uv_alloc_cb alloc_cb,
uv_read_cb read_cb) {
uv_loop_t* loop = handle->loop;
handle->flags |= UV_HANDLE_READING;
INCREASE_ACTIVE_COUNT(loop, handle);
handle->read_cb = read_cb;
handle->alloc_cb = alloc_cb;
/* If reading was stopped and then started again, there could still be a read
* request pending. */
if (!(handle->flags & UV_HANDLE_READ_PENDING))
uv_pipe_queue_read(loop, handle);
return 0;
}
static void uv_insert_non_overlapped_write_req(uv_pipe_t* handle,
uv_write_t* req) {
req->next_req = NULL;
if (handle->pipe.conn.non_overlapped_writes_tail) {
req->next_req =
handle->pipe.conn.non_overlapped_writes_tail->next_req;
handle->pipe.conn.non_overlapped_writes_tail->next_req = (uv_req_t*)req;
handle->pipe.conn.non_overlapped_writes_tail = req;
} else {
req->next_req = (uv_req_t*)req;
handle->pipe.conn.non_overlapped_writes_tail = req;
}
}
static uv_write_t* uv_remove_non_overlapped_write_req(uv_pipe_t* handle) {
uv_write_t* req;
if (handle->pipe.conn.non_overlapped_writes_tail) {
req = (uv_write_t*)handle->pipe.conn.non_overlapped_writes_tail->next_req;
if (req == handle->pipe.conn.non_overlapped_writes_tail) {
handle->pipe.conn.non_overlapped_writes_tail = NULL;
} else {
handle->pipe.conn.non_overlapped_writes_tail->next_req =
req->next_req;
}
return req;
} else {
libuv/src/win/pipe.c view on Meta::CPAN
if (send_handle != NULL) {
/* Add frame header flags. */
switch (xfer_type) {
case UV__IPC_SOCKET_XFER_TCP_CONNECTION:
frame_header.flags |= UV__IPC_FRAME_HAS_SOCKET_XFER |
UV__IPC_FRAME_XFER_IS_TCP_CONNECTION;
break;
case UV__IPC_SOCKET_XFER_TCP_SERVER:
frame_header.flags |= UV__IPC_FRAME_HAS_SOCKET_XFER;
break;
default:
assert(0); /* Unreachable. */
}
/* Add xfer info buffer. */
bufs[buf_index++] = uv_buf_init((char*) &xfer_info, sizeof xfer_info);
}
if (data_length > 0) {
/* Update frame header. */
frame_header.flags |= UV__IPC_FRAME_HAS_DATA;
frame_header.data_length = (uint32_t) data_length;
/* Add data buffers to buffers list. */
for (i = 0; i < data_buf_count; i++)
bufs[buf_index++] = data_bufs[i];
}
/* Write buffers. We set the `always_copy` flag, so it is not a problem that
* some of the written data lives on the stack. */
err = uv__pipe_write_data(loop, req, handle, bufs, buf_count, cb, 1);
/* If we had to heap-allocate the bufs array, free it now. */
if (bufs != stack_bufs) {
uv__free(bufs);
}
return err;
}
int uv__pipe_write(uv_loop_t* loop,
uv_write_t* req,
uv_pipe_t* handle,
const uv_buf_t bufs[],
size_t nbufs,
uv_stream_t* send_handle,
uv_write_cb cb) {
if (handle->ipc) {
/* IPC pipe write: use framing protocol. */
return uv__pipe_write_ipc(loop, req, handle, bufs, nbufs, send_handle, cb);
} else {
/* Non-IPC pipe write: put data on the wire directly. */
assert(send_handle == NULL);
return uv__pipe_write_data(loop, req, handle, bufs, nbufs, cb, 0);
}
}
static void uv_pipe_read_eof(uv_loop_t* loop, uv_pipe_t* handle,
uv_buf_t buf) {
/* If there is an eof timer running, we don't need it any more, so discard
* it. */
eof_timer_destroy(handle);
handle->flags &= ~UV_HANDLE_READABLE;
uv_read_stop((uv_stream_t*) handle);
handle->read_cb((uv_stream_t*) handle, UV_EOF, &buf);
}
static void uv_pipe_read_error(uv_loop_t* loop, uv_pipe_t* handle, int error,
uv_buf_t buf) {
/* If there is an eof timer running, we don't need it any more, so discard
* it. */
eof_timer_destroy(handle);
uv_read_stop((uv_stream_t*) handle);
handle->read_cb((uv_stream_t*)handle, uv_translate_sys_error(error), &buf);
}
static void uv_pipe_read_error_or_eof(uv_loop_t* loop, uv_pipe_t* handle,
int error, uv_buf_t buf) {
if (error == ERROR_BROKEN_PIPE) {
uv_pipe_read_eof(loop, handle, buf);
} else {
uv_pipe_read_error(loop, handle, error, buf);
}
}
static void uv__pipe_queue_ipc_xfer_info(
uv_pipe_t* handle,
uv__ipc_socket_xfer_type_t xfer_type,
uv__ipc_socket_xfer_info_t* xfer_info) {
uv__ipc_xfer_queue_item_t* item;
item = (uv__ipc_xfer_queue_item_t*) uv__malloc(sizeof(*item));
if (item == NULL)
uv_fatal_error(ERROR_OUTOFMEMORY, "uv__malloc");
item->xfer_type = xfer_type;
item->xfer_info = *xfer_info;
QUEUE_INSERT_TAIL(&handle->pipe.conn.ipc_xfer_queue, &item->member);
handle->pipe.conn.ipc_xfer_queue_length++;
}
/* Read an exact number of bytes from a pipe. If an error or end-of-file is
* encountered before the requested number of bytes are read, an error is
* returned. */
static int uv__pipe_read_exactly(HANDLE h, void* buffer, DWORD count) {
DWORD bytes_read, bytes_read_now;
bytes_read = 0;
while (bytes_read < count) {
if (!ReadFile(h,
(char*) buffer + bytes_read,
count - bytes_read,
&bytes_read_now,
NULL)) {
return GetLastError();
}
bytes_read += bytes_read_now;
}
assert(bytes_read == count);
return 0;
}
static DWORD uv__pipe_read_data(uv_loop_t* loop,
libuv/src/win/pipe.c view on Meta::CPAN
if (frame_header.reserved2 != 0)
goto invalid;
/* Parse xfer flags. */
xfer_flags = frame_header.flags & UV__IPC_FRAME_XFER_FLAGS;
if (xfer_flags & UV__IPC_FRAME_HAS_SOCKET_XFER) {
/* Socket coming -- determine the type. */
xfer_type = xfer_flags & UV__IPC_FRAME_XFER_IS_TCP_CONNECTION
? UV__IPC_SOCKET_XFER_TCP_CONNECTION
: UV__IPC_SOCKET_XFER_TCP_SERVER;
} else if (xfer_flags == 0) {
/* No socket. */
xfer_type = UV__IPC_SOCKET_XFER_NONE;
} else {
/* Invalid flags. */
goto invalid;
}
/* Parse data frame information. */
if (frame_header.flags & UV__IPC_FRAME_HAS_DATA) {
*data_remaining = frame_header.data_length;
} else if (frame_header.data_length != 0) {
/* Data length greater than zero but data flag not set -- invalid. */
goto invalid;
}
/* If no socket xfer info follows, return here. Data will be read in a
* subsequent invocation of uv__pipe_read_ipc(). */
if (xfer_type == UV__IPC_SOCKET_XFER_NONE)
return sizeof frame_header; /* Number of bytes read. */
/* Read transferred socket information. */
err = uv__pipe_read_exactly(handle->handle, &xfer_info, sizeof xfer_info);
if (err)
goto error;
/* Store the pending socket info. */
uv__pipe_queue_ipc_xfer_info(handle, xfer_type, &xfer_info);
/* Return number of bytes read. */
return sizeof frame_header + sizeof xfer_info;
}
invalid:
/* Invalid frame. */
err = WSAECONNABORTED; /* Maps to UV_ECONNABORTED. */
error:
uv_pipe_read_error_or_eof(loop, handle, err, uv_null_buf_);
return 0; /* Break out of read loop. */
}
void uv_process_pipe_read_req(uv_loop_t* loop,
uv_pipe_t* handle,
uv_req_t* req) {
assert(handle->type == UV_NAMED_PIPE);
handle->flags &= ~(UV_HANDLE_READ_PENDING | UV_HANDLE_CANCELLATION_PENDING);
DECREASE_PENDING_REQ_COUNT(handle);
eof_timer_stop(handle);
/* At this point, we're done with bookkeeping. If the user has stopped
* reading the pipe in the meantime, there is nothing left to do, since there
* is no callback that we can call. */
if (!(handle->flags & UV_HANDLE_READING))
return;
if (!REQ_SUCCESS(req)) {
/* An error occurred doing the zero-read. */
DWORD err = GET_REQ_ERROR(req);
/* If the read was cancelled by uv__pipe_interrupt_read(), the request may
* indicate an ERROR_OPERATION_ABORTED error. This error isn't relevant to
* the user; we'll start a new zero-read at the end of this function. */
if (err != ERROR_OPERATION_ABORTED)
uv_pipe_read_error_or_eof(loop, handle, err, uv_null_buf_);
} else {
/* The zero-read completed without error, indicating there is data
* available in the kernel buffer. */
DWORD avail;
/* Get the number of bytes available. */
avail = 0;
if (!PeekNamedPipe(handle->handle, NULL, 0, NULL, &avail, NULL))
uv_pipe_read_error_or_eof(loop, handle, GetLastError(), uv_null_buf_);
/* Read until we've either read all the bytes available, or the 'reading'
* flag is cleared. */
while (avail > 0 && handle->flags & UV_HANDLE_READING) {
/* Depending on the type of pipe, read either IPC frames or raw data. */
DWORD bytes_read =
handle->ipc ? uv__pipe_read_ipc(loop, handle)
: uv__pipe_read_data(loop, handle, avail, (DWORD) -1);
/* If no bytes were read, treat this as an indication that an error
* occurred, and break out of the read loop. */
if (bytes_read == 0)
break;
/* It is possible that more bytes were read than we thought were
* available. To prevent `avail` from underflowing, break out of the loop
* if this is the case. */
if (bytes_read > avail)
break;
/* Recompute the number of bytes available. */
avail -= bytes_read;
}
}
/* Start another zero-read request if necessary. */
if ((handle->flags & UV_HANDLE_READING) &&
!(handle->flags & UV_HANDLE_READ_PENDING)) {
uv_pipe_queue_read(loop, handle);
}
}
void uv_process_pipe_write_req(uv_loop_t* loop, uv_pipe_t* handle,
libuv/src/win/pipe.c view on Meta::CPAN
assert(handle->type == UV_NAMED_PIPE);
if (handle->flags & UV_HANDLE_CLOSING) {
/* The req->pipeHandle should be freed already in uv_pipe_cleanup(). */
assert(req->pipeHandle == INVALID_HANDLE_VALUE);
DECREASE_PENDING_REQ_COUNT(handle);
return;
}
if (REQ_SUCCESS(req)) {
assert(req->pipeHandle != INVALID_HANDLE_VALUE);
req->next_pending = handle->pipe.serv.pending_accepts;
handle->pipe.serv.pending_accepts = req;
if (handle->stream.serv.connection_cb) {
handle->stream.serv.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_connect_req(uv_loop_t* loop, uv_pipe_t* handle,
uv_connect_t* req) {
int err;
assert(handle->type == UV_NAMED_PIPE);
UNREGISTER_HANDLE_REQ(loop, handle, req);
if (req->cb) {
err = 0;
if (REQ_SUCCESS(req)) {
uv_pipe_connection_init(handle);
} else {
err = GET_REQ_ERROR(req);
}
req->cb(req, uv_translate_sys_error(err));
}
DECREASE_PENDING_REQ_COUNT(handle);
}
void uv_process_pipe_shutdown_req(uv_loop_t* loop, uv_pipe_t* handle,
uv_shutdown_t* req) {
assert(handle->type == UV_NAMED_PIPE);
UNREGISTER_HANDLE_REQ(loop, handle, req);
if (handle->flags & UV_HANDLE_READABLE) {
/* Initialize and optionally start the eof timer. Only do this if the pipe
* is readable and we haven't seen EOF come in ourselves. */
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);
}
} else {
/* This pipe is not readable. We can just close it to let the other end
* know that we're done writing. */
close_pipe(handle);
}
if (req->cb) {
req->cb(req, 0);
}
DECREASE_PENDING_REQ_COUNT(handle);
}
static void eof_timer_init(uv_pipe_t* pipe) {
int r;
assert(pipe->pipe.conn.eof_timer == NULL);
assert(pipe->flags & UV_HANDLE_CONNECTION);
pipe->pipe.conn.eof_timer = (uv_timer_t*) uv__malloc(sizeof *pipe->pipe.conn.eof_timer);
r = uv_timer_init(pipe->loop, pipe->pipe.conn.eof_timer);
assert(r == 0); /* timers can't fail */
pipe->pipe.conn.eof_timer->data = pipe;
uv_unref((uv_handle_t*) pipe->pipe.conn.eof_timer);
}
static void eof_timer_start(uv_pipe_t* pipe) {
assert(pipe->flags & UV_HANDLE_CONNECTION);
if (pipe->pipe.conn.eof_timer != NULL) {
uv_timer_start(pipe->pipe.conn.eof_timer, eof_timer_cb, eof_timeout, 0);
}
}
static void eof_timer_stop(uv_pipe_t* pipe) {
assert(pipe->flags & UV_HANDLE_CONNECTION);
if (pipe->pipe.conn.eof_timer != NULL) {
uv_timer_stop(pipe->pipe.conn.eof_timer);
}
}
static void eof_timer_cb(uv_timer_t* timer) {
uv_pipe_t* pipe = (uv_pipe_t*) timer->data;
uv_loop_t* loop = timer->loop;
assert(pipe->type == UV_NAMED_PIPE);
/* This should always be true, since we start the timer only in
* uv_pipe_queue_read after successfully calling ReadFile, or in
* uv_process_pipe_shutdown_req if a read is pending, and we always
* immediately stop the timer in uv_process_pipe_read_req. */
assert(pipe->flags & UV_HANDLE_READ_PENDING);
/* If there are many packets coming off the iocp then the timer callback may
* be called before the read request is coming off the queue. Therefore we
* check here if the read request has completed but will be processed later.
*/
if ((pipe->flags & UV_HANDLE_READ_PENDING) &&
HasOverlappedIoCompleted(&pipe->read_req.u.io.overlapped)) {
return;
}
/* Force both ends off the pipe. */
close_pipe(pipe);
/* Stop reading, so the pending read that is going to fail will not be
* reported to the user. */
uv_read_stop((uv_stream_t*) pipe);
/* Report the eof and update flags. This will get reported even if the user
* stopped reading in the meantime. TODO: is that okay? */
uv_pipe_read_eof(loop, pipe, uv_null_buf_);
}
static void eof_timer_destroy(uv_pipe_t* pipe) {
assert(pipe->flags & UV_HANDLE_CONNECTION);
if (pipe->pipe.conn.eof_timer) {
uv_close((uv_handle_t*) pipe->pipe.conn.eof_timer, eof_timer_close_cb);
pipe->pipe.conn.eof_timer = NULL;
}
}
static void eof_timer_close_cb(uv_handle_t* handle) {
assert(handle->type == UV_TIMER);
uv__free(handle);
}
int uv_pipe_open(uv_pipe_t* pipe, uv_file file) {
HANDLE os_handle = uv__get_osfhandle(file);
NTSTATUS nt_status;
IO_STATUS_BLOCK io_status;
FILE_ACCESS_INFORMATION access;
DWORD duplex_flags = 0;
if (os_handle == INVALID_HANDLE_VALUE)
return UV_EBADF;
uv__once_init();
/* In order to avoid closing a stdio file descriptor 0-2, duplicate the
* underlying OS handle and forget about the original fd.
* We could also opt to use the original OS handle and just never close it,
* but then there would be no reliable way to cancel pending read operations
* upon close.
*/
if (file <= 2) {
if (!DuplicateHandle(INVALID_HANDLE_VALUE,
os_handle,
INVALID_HANDLE_VALUE,
&os_handle,
0,
FALSE,
DUPLICATE_SAME_ACCESS))
return uv_translate_sys_error(GetLastError());
file = -1;
}
/* Determine what kind of permissions we have on this handle.
* Cygwin opens the pipe in message mode, but we can support it,
* just query the access flags and set the stream flags accordingly.
*/
nt_status = pNtQueryInformationFile(os_handle,
&io_status,
&access,
sizeof(access),
FileAccessInformation);
if (nt_status != STATUS_SUCCESS)
return UV_EINVAL;
if (pipe->ipc) {
if (!(access.AccessFlags & FILE_WRITE_DATA) ||
!(access.AccessFlags & FILE_READ_DATA)) {
return UV_EINVAL;
}
}
if (access.AccessFlags & FILE_WRITE_DATA)
duplex_flags |= UV_HANDLE_WRITABLE;
if (access.AccessFlags & FILE_READ_DATA)
duplex_flags |= UV_HANDLE_READABLE;
if (os_handle == INVALID_HANDLE_VALUE ||
uv_set_pipe_handle(pipe->loop,
( run in 0.502 second using v1.01-cache-2.11-cpan-02777c243ea )