luo980/libuv
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

win,pipe: fix race with concurrent readers (#4470)

Browse Source 
This fixes a race condition if multiple threads are reading from the
same NamedPipe, which could previously lead to a deadlock situation. We
also substantially improve performance now also, since the PeekFile
call is unnecessary overhead with this change. This API was added in
Windows Vista.

Related to #4467, though doesn't address any of the problems there. I
believe that someone could now implement uv__pipe_try_write using
this same code pattern however.
This commit is contained in:
Jameson Nash 2024-08-02 10:50:32 -04:00 committed by GitHub
parent 4e310d0f90
commit 727ee7237e
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
1 changed files with 114 additions and 93 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

207
src/win/pipe.c
View File

@ -667,15 +667,10 @@ void uv__pipe_endgame(uv_loop_t* loop, uv_pipe_t* handle) {
} }
handle->pipe.conn.ipc_xfer_queue_length = 0; handle->pipe.conn.ipc_xfer_queue_length = 0;
if (handle->flags & UV_HANDLE_EMULATE_IOCP) { assert(handle->read_req.wait_handle == INVALID_HANDLE_VALUE);
if (handle->read_req.wait_handle != INVALID_HANDLE_VALUE) { if (handle->read_req.event_handle != NULL) {
UnregisterWait(handle->read_req.wait_handle); CloseHandle(handle->read_req.event_handle);
handle->read_req.wait_handle = INVALID_HANDLE_VALUE; handle->read_req.event_handle = NULL;
}
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) if (handle->flags & UV_HANDLE_NON_OVERLAPPED_PIPE)
@ -1417,13 +1412,12 @@ static void uv__pipe_queue_read(uv_loop_t* loop, uv_pipe_t* handle) {
} }
if (handle->flags & UV_HANDLE_EMULATE_IOCP) { if (handle->flags & UV_HANDLE_EMULATE_IOCP) {
if (req->wait_handle == INVALID_HANDLE_VALUE) { assert(req->wait_handle == INVALID_HANDLE_VALUE);
if (!RegisterWaitForSingleObject(&req->wait_handle, if (!RegisterWaitForSingleObject(&req->wait_handle,
req->event_handle, post_completion_read_wait, (void*) req, req->event_handle, post_completion_read_wait, (void*) req,
INFINITE, WT_EXECUTEINWAITTHREAD)) { INFINITE, WT_EXECUTEINWAITTHREAD | WT_EXECUTEONLYONCE)) {
SET_REQ_ERROR(req, GetLastError()); SET_REQ_ERROR(req, GetLastError());
goto error; goto error;
}
} }
} }
} }
@ -1451,16 +1445,16 @@ int uv__pipe_read_start(uv_pipe_t* handle,
handle->read_cb = read_cb; handle->read_cb = read_cb;
handle->alloc_cb = alloc_cb; handle->alloc_cb = alloc_cb;
if (handle->read_req.event_handle == NULL) {
handle->read_req.event_handle = CreateEvent(NULL, 0, 0, NULL);
if (handle->read_req.event_handle == NULL) {
uv_fatal_error(GetLastError(), "CreateEvent");
}
}
/* If reading was stopped and then started again, there could still be a read /* If reading was stopped and then started again, there could still be a read
* request pending. */ * request pending. */
if (!(handle->flags & UV_HANDLE_READ_PENDING)) { if (!(handle->flags & UV_HANDLE_READ_PENDING)) {
if (handle->flags & UV_HANDLE_EMULATE_IOCP &&
handle->read_req.event_handle == NULL) {
handle->read_req.event_handle = CreateEvent(NULL, 0, 0, NULL);
if (handle->read_req.event_handle == NULL) {
uv_fatal_error(GetLastError(), "CreateEvent");
}
}
uv__pipe_queue_read(loop, handle); uv__pipe_queue_read(loop, handle);
} }
@ -1713,7 +1707,7 @@ static int uv__pipe_write_data(uv_loop_t* loop,
if (handle->flags & UV_HANDLE_EMULATE_IOCP) { if (handle->flags & UV_HANDLE_EMULATE_IOCP) {
if (!RegisterWaitForSingleObject(&req->wait_handle, if (!RegisterWaitForSingleObject(&req->wait_handle,
req->event_handle, post_completion_write_wait, (void*) req, req->event_handle, post_completion_write_wait, (void*) req,
INFINITE, WT_EXECUTEINWAITTHREAD)) { INFINITE, WT_EXECUTEINWAITTHREAD | WT_EXECUTEONLYONCE)) {
return GetLastError(); return GetLastError();
} }
} }
@ -1889,7 +1883,7 @@ static void uv__pipe_read_error(uv_loop_t* loop, uv_pipe_t* handle, int error,
static void uv__pipe_read_error_or_eof(uv_loop_t* loop, uv_pipe_t* handle, static void uv__pipe_read_error_or_eof(uv_loop_t* loop, uv_pipe_t* handle,
int error, uv_buf_t buf) { DWORD error, uv_buf_t buf) {
if (error == ERROR_BROKEN_PIPE) { if (error == ERROR_BROKEN_PIPE) {
uv__pipe_read_eof(loop, handle, buf); uv__pipe_read_eof(loop, handle, buf);
} else { } else {
@ -1919,17 +1913,25 @@ static void uv__pipe_queue_ipc_xfer_info(
/* Read an exact number of bytes from a pipe. If an error or end-of-file is /* 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 * encountered before the requested number of bytes are read, an error is
* returned. */ * returned. */
static int uv__pipe_read_exactly(HANDLE h, void* buffer, DWORD count) { static DWORD uv__pipe_read_exactly(uv_pipe_t* handle, void* buffer, DWORD count) {
DWORD bytes_read, bytes_read_now; uv_read_t* req;
DWORD bytes_read;
DWORD bytes_read_now;
bytes_read = 0; bytes_read = 0;
while (bytes_read < count) { while (bytes_read < count) {
if (!ReadFile(h, req = &handle->read_req;
memset(&req->u.io.overlapped, 0, sizeof(req->u.io.overlapped));
req->u.io.overlapped.hEvent = (HANDLE) ((uintptr_t) req->event_handle | 1);
if (!ReadFile(handle->handle,
(char*) buffer + bytes_read, (char*) buffer + bytes_read,
count - bytes_read, count - bytes_read,
&bytes_read_now, &bytes_read_now,
NULL)) { &req->u.io.overlapped)) {
return GetLastError(); if (GetLastError() != ERROR_IO_PENDING)
return GetLastError();
if (!GetOverlappedResult(handle->handle, &req->u.io.overlapped, &bytes_read_now, TRUE))
return GetLastError();
} }
bytes_read += bytes_read_now; bytes_read += bytes_read_now;
@ -1940,16 +1942,17 @@ static int uv__pipe_read_exactly(HANDLE h, void* buffer, DWORD count) {
} }
static DWORD uv__pipe_read_data(uv_loop_t* loop, static int uv__pipe_read_data(uv_loop_t* loop,
uv_pipe_t* handle, uv_pipe_t* handle,
DWORD suggested_bytes, DWORD* bytes_read,
DWORD max_bytes) { DWORD max_bytes) {
DWORD bytes_read;
uv_buf_t buf; uv_buf_t buf;
uv_read_t* req;
DWORD r;
/* Ask the user for a buffer to read data into. */ /* Ask the user for a buffer to read data into. */
buf = uv_buf_init(NULL, 0); buf = uv_buf_init(NULL, 0);
handle->alloc_cb((uv_handle_t*) handle, suggested_bytes, &buf); handle->alloc_cb((uv_handle_t*) handle, *bytes_read, &buf);
if (buf.base == NULL || buf.len == 0) { if (buf.base == NULL || buf.len == 0) {
handle->read_cb((uv_stream_t*) handle, UV_ENOBUFS, &buf); handle->read_cb((uv_stream_t*) handle, UV_ENOBUFS, &buf);
return 0; /* Break out of read loop. */ return 0; /* Break out of read loop. */
@ -1962,29 +1965,50 @@ static DWORD uv__pipe_read_data(uv_loop_t* loop,
if (max_bytes > buf.len) if (max_bytes > buf.len)
max_bytes = buf.len; max_bytes = buf.len;
/* Read into the user buffer. */ /* Read into the user buffer.
if (!ReadFile(handle->handle, buf.base, max_bytes, &bytes_read, NULL)) { * Prepare an Event so that we can cancel if it doesn't complete immediately.
uv__pipe_read_error_or_eof(loop, handle, GetLastError(), buf); */
return 0; /* Break out of read loop. */ req = &handle->read_req;
memset(&req->u.io.overlapped, 0, sizeof(req->u.io.overlapped));
req->u.io.overlapped.hEvent = (HANDLE) ((uintptr_t) req->event_handle | 1);
if (ReadFile(handle->handle, buf.base, max_bytes, bytes_read, &req->u.io.overlapped)) {
r = ERROR_SUCCESS;
} else {
r = GetLastError();
*bytes_read = 0;
if (r == ERROR_IO_PENDING) {
r = CancelIoEx(handle->handle, &req->u.io.overlapped);
assert(r || GetLastError() == ERROR_NOT_FOUND);
if (!GetOverlappedResult(handle->handle, &req->u.io.overlapped, bytes_read, TRUE)) {
r = GetLastError();
*bytes_read = 0;
}
}
} }
/* Call the read callback. */ /* Call the read callback. */
handle->read_cb((uv_stream_t*) handle, bytes_read, &buf); if (r == ERROR_SUCCESS || r == ERROR_OPERATION_ABORTED)
handle->read_cb((uv_stream_t*) handle, *bytes_read, &buf);
else
uv__pipe_read_error_or_eof(loop, handle, r, buf);
return bytes_read; return *bytes_read == max_bytes;
} }
static DWORD uv__pipe_read_ipc(uv_loop_t* loop, uv_pipe_t* handle) { static int uv__pipe_read_ipc(uv_loop_t* loop, uv_pipe_t* handle) {
uint32_t* data_remaining = &handle->pipe.conn.ipc_data_frame.payload_remaining; uint32_t* data_remaining;
int err; DWORD err;
DWORD more;
DWORD bytes_read;
data_remaining = &handle->pipe.conn.ipc_data_frame.payload_remaining;
if (*data_remaining > 0) { if (*data_remaining > 0) {
/* Read frame data payload. */ /* Read frame data payload. */
DWORD bytes_read = bytes_read = *data_remaining;
uv__pipe_read_data(loop, handle, *data_remaining, *data_remaining); more = uv__pipe_read_data(loop, handle, &bytes_read, bytes_read);
*data_remaining -= bytes_read; *data_remaining -= bytes_read;
return bytes_read;
} else { } else {
/* Start of a new IPC frame. */ /* Start of a new IPC frame. */
@ -1995,7 +2019,7 @@ static DWORD uv__pipe_read_ipc(uv_loop_t* loop, uv_pipe_t* handle) {
/* Read the IPC frame header. */ /* Read the IPC frame header. */
err = uv__pipe_read_exactly( err = uv__pipe_read_exactly(
handle->handle, &frame_header, sizeof frame_header); handle, &frame_header, sizeof frame_header);
if (err) if (err)
goto error; goto error;
@ -2031,21 +2055,28 @@ static DWORD uv__pipe_read_ipc(uv_loop_t* loop, uv_pipe_t* handle) {
/* If no socket xfer info follows, return here. Data will be read in a /* If no socket xfer info follows, return here. Data will be read in a
* subsequent invocation of uv__pipe_read_ipc(). */ * subsequent invocation of uv__pipe_read_ipc(). */
if (xfer_type == UV__IPC_SOCKET_XFER_NONE) 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, &xfer_info, sizeof xfer_info);
if (err)
goto error;
/* Read transferred socket information. */ /* Store the pending socket info. */
err = uv__pipe_read_exactly(handle->handle, &xfer_info, sizeof xfer_info); uv__pipe_queue_ipc_xfer_info(handle, xfer_type, &xfer_info);
if (err) }
goto error;
/* Store the pending socket info. */ more = 1;
uv__pipe_queue_ipc_xfer_info(handle, xfer_type, &xfer_info);
/* Return number of bytes read. */
return sizeof frame_header + sizeof xfer_info;
} }
/* Return whether the caller should immediately try another read call to get
* more data. */
if (more && *data_remaining == 0) {
/* TODO: use PeekNamedPipe to see if it is really worth trying to do
* another ReadFile call. */
}
return more;
invalid: invalid:
/* Invalid frame. */ /* Invalid frame. */
err = WSAECONNABORTED; /* Maps to UV_ECONNABORTED. */ err = WSAECONNABORTED; /* Maps to UV_ECONNABORTED. */
@ -2059,12 +2090,20 @@ error:
void uv__process_pipe_read_req(uv_loop_t* loop, void uv__process_pipe_read_req(uv_loop_t* loop,
uv_pipe_t* handle, uv_pipe_t* handle,
uv_req_t* req) { uv_req_t* req) {
DWORD err;
DWORD more;
DWORD bytes_requested;
assert(handle->type == UV_NAMED_PIPE); assert(handle->type == UV_NAMED_PIPE);
handle->flags &= ~(UV_HANDLE_READ_PENDING | UV_HANDLE_CANCELLATION_PENDING); handle->flags &= ~(UV_HANDLE_READ_PENDING | UV_HANDLE_CANCELLATION_PENDING);
DECREASE_PENDING_REQ_COUNT(handle); DECREASE_PENDING_REQ_COUNT(handle);
eof_timer_stop(handle); eof_timer_stop(handle);
if (handle->read_req.wait_handle != INVALID_HANDLE_VALUE) {
UnregisterWait(handle->read_req.wait_handle);
handle->read_req.wait_handle = INVALID_HANDLE_VALUE;
}
/* At this point, we're done with bookkeeping. If the user has stopped /* 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 * reading the pipe in the meantime, there is nothing left to do, since there
* is no callback that we can call. */ * is no callback that we can call. */
@ -2073,7 +2112,7 @@ void uv__process_pipe_read_req(uv_loop_t* loop,
if (!REQ_SUCCESS(req)) { if (!REQ_SUCCESS(req)) {
/* An error occurred doing the zero-read. */ /* An error occurred doing the zero-read. */
DWORD err = GET_REQ_ERROR(req); err = GET_REQ_ERROR(req);
/* If the read was cancelled by uv__pipe_interrupt_read(), the request may /* 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 * indicate an ERROR_OPERATION_ABORTED error. This error isn't relevant to
@ -2084,34 +2123,18 @@ void uv__process_pipe_read_req(uv_loop_t* loop,
} else { } else {
/* The zero-read completed without error, indicating there is data /* The zero-read completed without error, indicating there is data
* available in the kernel buffer. */ * available in the kernel buffer. */
DWORD avail; while (handle->flags & UV_HANDLE_READING) {
bytes_requested = 65536;
/* 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. */ /* Depending on the type of pipe, read either IPC frames or raw data. */
DWORD bytes_read = if (handle->ipc)
handle->ipc ? uv__pipe_read_ipc(loop, handle) more = uv__pipe_read_ipc(loop, handle);
: uv__pipe_read_data(loop, handle, avail, (DWORD) -1); else
more = uv__pipe_read_data(loop, handle, &bytes_requested, INT32_MAX);
/* If no bytes were read, treat this as an indication that an error /* If no bytes were read, treat this as an indication that an error
* occurred, and break out of the read loop. */ * occurred, and break out of the read loop. */
if (bytes_read == 0) if (more == 0)
break; 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;
} }
} }
@ -2134,15 +2157,13 @@ void uv__process_pipe_write_req(uv_loop_t* loop, uv_pipe_t* handle,
UNREGISTER_HANDLE_REQ(loop, handle); UNREGISTER_HANDLE_REQ(loop, handle);
if (handle->flags & UV_HANDLE_EMULATE_IOCP) { if (req->wait_handle != INVALID_HANDLE_VALUE) {
if (req->wait_handle != INVALID_HANDLE_VALUE) { UnregisterWait(req->wait_handle);
UnregisterWait(req->wait_handle); req->wait_handle = INVALID_HANDLE_VALUE;
req->wait_handle = INVALID_HANDLE_VALUE; }
} if (req->event_handle) {
if (req->event_handle) { CloseHandle(req->event_handle);
CloseHandle(req->event_handle); req->event_handle = NULL;
req->event_handle = NULL;
}
} }
err = GET_REQ_ERROR(req); err = GET_REQ_ERROR(req);