diff --git a/Makefile.am b/Makefile.am index e0c9414a..9a33f7e6 100644 --- a/Makefile.am +++ b/Makefile.am @@ -222,6 +222,7 @@ test_run_tests_SOURCES = test/blackhole-server.c \ test/test-pipe-close-stdout-read-stdin.c \ test/test-pipe-set-non-blocking.c \ test/test-pipe-set-fchmod.c \ + test/test-pipe-try-write.c \ test/test-platform-output.c \ test/test-poll.c \ test/test-poll-close.c \ diff --git a/src/win/internal.h b/src/win/internal.h index 89c72b8a..8a87743b 100644 --- a/src/win/internal.h +++ b/src/win/internal.h @@ -134,6 +134,8 @@ int uv__pipe_write(uv_loop_t* loop, uv_stream_t* send_handle, uv_write_cb cb); void uv__pipe_shutdown(uv_loop_t* loop, uv_pipe_t* handle, uv_shutdown_t* req); +int uv__pipe_try_write(uv_pipe_t* handle, const uv_buf_t bufs[], + unsigned int nbufs); void uv__process_pipe_read_req(uv_loop_t* loop, uv_pipe_t* handle, uv_req_t* req); diff --git a/src/win/pipe.c b/src/win/pipe.c index 787ba105..94233455 100644 --- a/src/win/pipe.c +++ b/src/win/pipe.c @@ -1782,6 +1782,78 @@ int uv__pipe_write(uv_loop_t* loop, } +int uv__pipe_try_write(uv_pipe_t* handle, + const uv_buf_t bufs[], + unsigned int nbufs) { + OVERLAPPED overlapped; + const uv_buf_t* buf; + int bytes_written; + unsigned int idx; + DWORD timeout; + DWORD err; + + if (handle->flags & UV_HANDLE_NON_OVERLAPPED_PIPE) { + return UV_EAGAIN; + } + + if (handle->stream.conn.write_reqs_pending > 0) { + return UV_EAGAIN; + } + + timeout = 0; + if (handle->flags & UV_HANDLE_BLOCKING_WRITES) { + timeout = INFINITE; + } + + memset(&overlapped, 0, sizeof(overlapped)); + + overlapped.hEvent = CreateEvent(NULL, FALSE, FALSE, NULL); + if (overlapped.hEvent == NULL) { + uv_fatal_error(GetLastError(), "CreateEvent"); + } + + bytes_written = 0; + for (err = 0, idx = 0; idx < nbufs; err = 0, idx += 1) { + buf = &bufs[idx]; + + if (WriteFile(handle->handle, buf->base, buf->len, NULL, &overlapped)) { + bytes_written += buf->len; + continue; + } + + err = GetLastError(); + if (err != ERROR_IO_PENDING) { + break; + } + + err = WaitForSingleObject(overlapped.hEvent, timeout); + if (err == WAIT_OBJECT_0) { + bytes_written += buf->len; + continue; + } + + if (err == WAIT_TIMEOUT && + CancelIo(handle->handle) && + GetOverlappedResult(handle->handle, &overlapped, &err, TRUE)) { + bytes_written += err; + err = WSAEWOULDBLOCK; /* Translates to UV_EAGAIN. */ + } else { + err = GetLastError(); + } + + break; + } + + CloseHandle(overlapped.hEvent); + + if (bytes_written == 0 && err != 0) { + return uv_translate_sys_error(err); + } + + return bytes_written; +} + + 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 diff --git a/src/win/stream.c b/src/win/stream.c index 7bf9ca38..7ccc79be 100644 --- a/src/win/stream.c +++ b/src/win/stream.c @@ -182,7 +182,7 @@ int uv_try_write(uv_stream_t* stream, case UV_TTY: return uv__tty_try_write((uv_tty_t*) stream, bufs, nbufs); case UV_NAMED_PIPE: - return UV_EAGAIN; + return uv__pipe_try_write((uv_pipe_t*) stream, bufs, nbufs); default: assert(0); return UV_ENOSYS; diff --git a/test/test-list.h b/test/test-list.h index ec23b344..00eda659 100644 --- a/test/test-list.h +++ b/test/test-list.h @@ -207,6 +207,9 @@ TEST_DECLARE (pipe_getsockname_blocking) TEST_DECLARE (pipe_pending_instances) TEST_DECLARE (pipe_sendmsg) TEST_DECLARE (pipe_server_close) +TEST_DECLARE (pipe_try_write_0) +TEST_DECLARE (pipe_try_write_1) +TEST_DECLARE (pipe_try_write_2) TEST_DECLARE (connection_fail) TEST_DECLARE (connection_fail_doesnt_auto_close) TEST_DECLARE (shutdown_close_tcp) @@ -587,6 +590,9 @@ TASK_LIST_START TEST_ENTRY (pipe_connect_on_prepare) TEST_ENTRY (pipe_server_close) + TEST_ENTRY (pipe_try_write_0) + TEST_ENTRY (pipe_try_write_1) + TEST_ENTRY (pipe_try_write_2) #ifndef _WIN32 TEST_ENTRY (pipe_close_stdout_read_stdin) #endif diff --git a/test/test-pipe-try-write.c b/test/test-pipe-try-write.c new file mode 100644 index 00000000..346a1c39 --- /dev/null +++ b/test/test-pipe-try-write.c @@ -0,0 +1,125 @@ +#include "uv.h" +#include "task.h" + +static void (*spam)(uv_pipe_t* handle); +static uv_pipe_t client_handle; +static uv_pipe_t peer_handle; +static uv_pipe_t server_handle; +static uv_write_t write_req; + + +static void write_cb(uv_write_t* req, int status) { + ASSERT(0 == status); +} + + +static void spam_0(uv_pipe_t* handle) { + uv_buf_t buf; + + buf = uv_buf_init("", 0); + ASSERT(0 == uv_try_write((uv_stream_t*) handle, &buf, 1)); + + /* Non-empty write to start the event loop moving. */ + buf = uv_buf_init("hello, world", sizeof("hello, world") - 1); + ASSERT(0 == uv_write(&write_req, (uv_stream_t*) handle, &buf, 1, write_cb)); +} + + +static void spam_1(uv_pipe_t* handle) { + uv_buf_t buf; + int rc; + + buf = uv_buf_init("hello, world", sizeof("hello, world") - 1); + do + rc = uv_try_write((uv_stream_t*) handle, &buf, 1); + while (rc > 0); + + ASSERT(rc == UV_EAGAIN); +} + + +static void spam_2(uv_pipe_t* handle) { + uv_buf_t bufs[2]; + int rc; + + bufs[0] = uv_buf_init("hello,", sizeof("hello,") - 1); + bufs[1] = uv_buf_init(" world", sizeof(" world") - 1); + + do + rc = uv_try_write((uv_stream_t*) handle, bufs, ARRAY_SIZE(bufs)); + while (rc > 0); + + ASSERT(rc == UV_EAGAIN); +} + + +static void alloc_cb(uv_handle_t* handle, size_t size, uv_buf_t* buf) { + static char base[256]; + + buf->base = base; + buf->len = sizeof(base); +} + + +static void read_cb(uv_stream_t* handle, ssize_t nread, const uv_buf_t* buf) { + if (spam == spam_0) { + ASSERT(nread > 0); /* Expect some bytes. */ + } else { + ASSERT(nread == (ssize_t) buf->len); /* Expect saturation. */ + } + + if (handle == (uv_stream_t*) &peer_handle) { + spam(&client_handle); + } else { + uv_close((uv_handle_t*) &peer_handle, NULL); + uv_close((uv_handle_t*) &client_handle, NULL); + uv_close((uv_handle_t*) &server_handle, NULL); + } +} + + +static void connection_cb(uv_stream_t* handle, int status) { + ASSERT(0 == status); + ASSERT(0 == uv_pipe_init(uv_default_loop(), &peer_handle, 0)); + ASSERT(0 == uv_accept((uv_stream_t*) &server_handle, + (uv_stream_t*) &peer_handle)); + ASSERT(0 == uv_read_start((uv_stream_t*) &peer_handle, alloc_cb, read_cb)); + spam(&peer_handle); +} + + +static void connect_cb(uv_connect_t* req, int status) { + ASSERT(0 == status); + ASSERT(0 == uv_read_start((uv_stream_t*) &client_handle, alloc_cb, read_cb)); +} + + +static int pipe_try_write(void (*spammer)(uv_pipe_t*)) { + uv_connect_t connect_req; + + spam = spammer; + ASSERT(0 == uv_pipe_init(uv_default_loop(), &client_handle, 0)); + ASSERT(0 == uv_pipe_init(uv_default_loop(), &server_handle, 0)); + ASSERT(0 == uv_pipe_bind(&server_handle, TEST_PIPENAME)); + ASSERT(0 == uv_listen((uv_stream_t*) &server_handle, 1, connection_cb)); + uv_pipe_connect(&connect_req, &client_handle, TEST_PIPENAME, connect_cb); + ASSERT(0 == uv_run(uv_default_loop(), UV_RUN_DEFAULT)); + + MAKE_VALGRIND_HAPPY(); + return 0; +} + + +TEST_IMPL(pipe_try_write_0) { + return pipe_try_write(spam_0); +} + + +TEST_IMPL(pipe_try_write_1) { + return pipe_try_write(spam_1); +} + + +TEST_IMPL(pipe_try_write_2) { + return pipe_try_write(spam_2); +}