1 /* Copyright Joyent, Inc. and other Node contributors. All rights reserved.
3 * Permission is hereby granted, free of charge, to any person obtaining a copy
4 * of this software and associated documentation files (the "Software"), to
5 * deal in the Software without restriction, including without limitation the
6 * rights to use, copy, modify, merge, publish, distribute, sublicense, and/or
7 * sell copies of the Software, and to permit persons to whom the Software is
8 * furnished to do so, subject to the following conditions:
10 * The above copyright notice and this permission notice shall be included in
11 * all copies or substantial portions of the Software.
13 * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
14 * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
15 * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
16 * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
17 * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
18 * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS
29 #include "handle-inl.h"
30 #include "stream-inl.h"
34 /* A zero-size buffer for use by uv_pipe_read */
35 static char uv_zero_[] = "";
38 static const uv_buf_t uv_null_buf_ = { 0, NULL };
40 /* The timeout that the pipe will wait for the remote end to write data */
41 /* when the local ends wants to shut it down. */
42 static const int64_t eof_timeout = 50; /* ms */
44 static const int default_pending_pipe_instances = 4;
46 /* IPC protocol flags. */
47 #define UV_IPC_RAW_DATA 0x0001
48 #define UV_IPC_TCP_SERVER 0x0002
49 #define UV_IPC_TCP_CONNECTION 0x0004
51 /* IPC frame header. */
54 uint64_t raw_data_length;
55 } uv_ipc_frame_header_t;
57 /* IPC frame, which contains an imported TCP socket stream. */
59 uv_ipc_frame_header_t header;
60 WSAPROTOCOL_INFOW socket_info;
61 } uv_ipc_frame_uv_stream;
63 static void eof_timer_init(uv_pipe_t* pipe);
64 static void eof_timer_start(uv_pipe_t* pipe);
65 static void eof_timer_stop(uv_pipe_t* pipe);
66 static void eof_timer_cb(uv_timer_t* timer, int status);
67 static void eof_timer_destroy(uv_pipe_t* pipe);
68 static void eof_timer_close_cb(uv_handle_t* handle);
71 static void uv_unique_pipe_name(char* ptr, char* name, size_t size) {
72 _snprintf(name, size, "\\\\.\\pipe\\uv\\%p-%d", ptr, GetCurrentProcessId());
76 int uv_pipe_init(uv_loop_t* loop, uv_pipe_t* handle, int ipc) {
77 uv_stream_init(loop, (uv_stream_t*)handle, UV_NAMED_PIPE);
79 handle->reqs_pending = 0;
80 handle->handle = INVALID_HANDLE_VALUE;
83 handle->remaining_ipc_rawdata_bytes = 0;
84 handle->pending_ipc_info.socket_info = NULL;
85 handle->pending_ipc_info.tcp_connection = 0;
87 handle->non_overlapped_writes_tail = NULL;
89 uv_req_init(loop, (uv_req_t*) &handle->ipc_header_write_req);
95 static void uv_pipe_connection_init(uv_pipe_t* handle) {
96 uv_connection_init((uv_stream_t*) handle);
97 handle->read_req.data = handle;
98 handle->eof_timer = NULL;
102 static HANDLE open_named_pipe(WCHAR* name, DWORD* duplex_flags) {
106 * Assume that we have a duplex pipe first, so attempt to
107 * connect with GENERIC_READ | GENERIC_WRITE.
109 pipeHandle = CreateFileW(name,
110 GENERIC_READ | GENERIC_WRITE,
114 FILE_FLAG_OVERLAPPED,
116 if (pipeHandle != INVALID_HANDLE_VALUE) {
117 *duplex_flags = UV_HANDLE_READABLE | UV_HANDLE_WRITABLE;
122 * If the pipe is not duplex CreateFileW fails with
123 * ERROR_ACCESS_DENIED. In that case try to connect
124 * as a read-only or write-only.
126 if (GetLastError() == ERROR_ACCESS_DENIED) {
127 pipeHandle = CreateFileW(name,
128 GENERIC_READ | FILE_WRITE_ATTRIBUTES,
132 FILE_FLAG_OVERLAPPED,
135 if (pipeHandle != INVALID_HANDLE_VALUE) {
136 *duplex_flags = UV_HANDLE_READABLE;
141 if (GetLastError() == ERROR_ACCESS_DENIED) {
142 pipeHandle = CreateFileW(name,
143 GENERIC_WRITE | FILE_READ_ATTRIBUTES,
147 FILE_FLAG_OVERLAPPED,
150 if (pipeHandle != INVALID_HANDLE_VALUE) {
151 *duplex_flags = UV_HANDLE_WRITABLE;
156 return INVALID_HANDLE_VALUE;
160 uv_err_t uv_stdio_pipe_server(uv_loop_t* loop, uv_pipe_t* handle, DWORD access,
161 char* name, size_t nameSize) {
165 char* ptr = (char*)handle;
168 uv_unique_pipe_name(ptr, name, nameSize);
170 pipeHandle = CreateNamedPipeA(name,
171 access | FILE_FLAG_OVERLAPPED | FILE_FLAG_FIRST_PIPE_INSTANCE,
172 PIPE_TYPE_BYTE | PIPE_READMODE_BYTE | PIPE_WAIT, 1, 65536, 65536, 0,
175 if (pipeHandle != INVALID_HANDLE_VALUE) {
176 /* No name collisions. We're done. */
180 errorno = GetLastError();
181 if (errorno != ERROR_PIPE_BUSY && errorno != ERROR_ACCESS_DENIED) {
182 err = uv__new_sys_error(errorno);
186 /* Pipe name collision. Increment the pointer and try again. */
190 if (CreateIoCompletionPort(pipeHandle,
194 err = uv__new_sys_error(GetLastError());
198 uv_pipe_connection_init(handle);
199 handle->handle = pipeHandle;
204 if (pipeHandle != INVALID_HANDLE_VALUE) {
205 CloseHandle(pipeHandle);
212 static int uv_set_pipe_handle(uv_loop_t* loop, uv_pipe_t* handle,
213 HANDLE pipeHandle, DWORD duplex_flags) {
215 IO_STATUS_BLOCK io_status;
216 FILE_MODE_INFORMATION mode_info;
217 DWORD mode = PIPE_TYPE_BYTE | PIPE_READMODE_BYTE | PIPE_WAIT;
219 if (!SetNamedPipeHandleState(pipeHandle, &mode, NULL, NULL)) {
220 /* If this returns ERROR_INVALID_PARAMETER we probably opened something */
221 /* that is not a pipe. */
222 if (GetLastError() == ERROR_INVALID_PARAMETER) {
223 SetLastError(WSAENOTSOCK);
228 /* Check if the pipe was created with FILE_FLAG_OVERLAPPED. */
229 nt_status = pNtQueryInformationFile(pipeHandle,
233 FileModeInformation);
234 if (nt_status != STATUS_SUCCESS) {
238 if (mode_info.Mode & FILE_SYNCHRONOUS_IO_ALERT ||
239 mode_info.Mode & FILE_SYNCHRONOUS_IO_NONALERT) {
240 /* Non-overlapped pipe. */
241 handle->flags |= UV_HANDLE_NON_OVERLAPPED_PIPE;
243 /* Overlapped pipe. Try to associate with IOCP. */
244 if (CreateIoCompletionPort(pipeHandle,
248 handle->flags |= UV_HANDLE_EMULATE_IOCP;
252 handle->handle = pipeHandle;
253 handle->flags |= duplex_flags;
259 static DWORD WINAPI pipe_shutdown_thread_proc(void* parameter) {
264 req = (uv_shutdown_t*) parameter;
266 handle = (uv_pipe_t*) req->handle;
271 FlushFileBuffers(handle->handle);
274 POST_COMPLETION_FOR_REQ(loop, req);
280 void uv_pipe_endgame(uv_loop_t* loop, uv_pipe_t* handle) {
284 IO_STATUS_BLOCK io_status;
285 FILE_PIPE_LOCAL_INFORMATION pipe_info;
287 if ((handle->flags & UV_HANDLE_CONNECTION) &&
288 handle->shutdown_req != NULL &&
289 handle->write_reqs_pending == 0) {
290 req = handle->shutdown_req;
292 /* Clear the shutdown_req field so we don't go here again. */
293 handle->shutdown_req = NULL;
295 if (handle->flags & UV__HANDLE_CLOSING) {
296 UNREGISTER_HANDLE_REQ(loop, handle, req);
298 /* Already closing. Cancel the shutdown. */
300 uv__set_artificial_error(loop, UV_ECANCELED);
304 DECREASE_PENDING_REQ_COUNT(handle);
308 /* Try to avoid flushing the pipe buffer in the thread pool. */
309 nt_status = pNtQueryInformationFile(handle->handle,
313 FilePipeLocalInformation);
315 if (nt_status != STATUS_SUCCESS) {
317 UNREGISTER_HANDLE_REQ(loop, handle, req);
319 handle->flags |= UV_HANDLE_WRITABLE; /* Questionable */
321 uv__set_sys_error(loop, pRtlNtStatusToDosError(nt_status));
325 DECREASE_PENDING_REQ_COUNT(handle);
329 if (pipe_info.OutboundQuota == pipe_info.WriteQuotaAvailable) {
330 /* Short-circuit, no need to call FlushFileBuffers. */
331 uv_insert_pending_req(loop, (uv_req_t*) req);
335 /* Run FlushFileBuffers in the thread pool. */
336 result = QueueUserWorkItem(pipe_shutdown_thread_proc,
338 WT_EXECUTELONGFUNCTION);
344 UNREGISTER_HANDLE_REQ(loop, handle, req);
346 handle->flags |= UV_HANDLE_WRITABLE; /* Questionable */
348 uv__set_sys_error(loop, GetLastError());
352 DECREASE_PENDING_REQ_COUNT(handle);
357 if (handle->flags & UV__HANDLE_CLOSING &&
358 handle->reqs_pending == 0) {
359 assert(!(handle->flags & UV_HANDLE_CLOSED));
361 if (handle->flags & UV_HANDLE_CONNECTION) {
362 if (handle->pending_ipc_info.socket_info) {
363 free(handle->pending_ipc_info.socket_info);
364 handle->pending_ipc_info.socket_info = NULL;
367 if (handle->flags & UV_HANDLE_EMULATE_IOCP) {
368 if (handle->read_req.wait_handle != INVALID_HANDLE_VALUE) {
369 UnregisterWait(handle->read_req.wait_handle);
370 handle->read_req.wait_handle = INVALID_HANDLE_VALUE;
372 if (handle->read_req.event_handle) {
373 CloseHandle(handle->read_req.event_handle);
374 handle->read_req.event_handle = NULL;
379 if (handle->flags & UV_HANDLE_PIPESERVER) {
380 assert(handle->accept_reqs);
381 free(handle->accept_reqs);
382 handle->accept_reqs = NULL;
385 uv__handle_close(handle);
390 void uv_pipe_pending_instances(uv_pipe_t* handle, int count) {
391 handle->pending_instances = count;
392 handle->flags |= UV_HANDLE_PIPESERVER;
396 /* Creates a pipe server. */
397 int uv_pipe_bind(uv_pipe_t* handle, const char* name) {
398 uv_loop_t* loop = handle->loop;
399 int i, errorno, nameSize;
400 uv_pipe_accept_t* req;
402 if (handle->flags & UV_HANDLE_BOUND) {
403 uv__set_sys_error(loop, WSAEINVAL);
408 uv__set_sys_error(loop, WSAEINVAL);
412 if (!(handle->flags & UV_HANDLE_PIPESERVER)) {
413 handle->pending_instances = default_pending_pipe_instances;
416 handle->accept_reqs = (uv_pipe_accept_t*)
417 malloc(sizeof(uv_pipe_accept_t) * handle->pending_instances);
418 if (!handle->accept_reqs) {
419 uv_fatal_error(ERROR_OUTOFMEMORY, "malloc");
422 for (i = 0; i < handle->pending_instances; i++) {
423 req = &handle->accept_reqs[i];
424 uv_req_init(loop, (uv_req_t*) req);
425 req->type = UV_ACCEPT;
427 req->pipeHandle = INVALID_HANDLE_VALUE;
428 req->next_pending = NULL;
431 /* Convert name to UTF16. */
432 nameSize = uv_utf8_to_utf16(name, NULL, 0) * sizeof(WCHAR);
433 handle->name = (WCHAR*)malloc(nameSize);
435 uv_fatal_error(ERROR_OUTOFMEMORY, "malloc");
438 if (!uv_utf8_to_utf16(name, handle->name, nameSize / sizeof(WCHAR))) {
439 uv__set_sys_error(loop, GetLastError());
444 * Attempt to create the first pipe with FILE_FLAG_FIRST_PIPE_INSTANCE.
445 * If this fails then there's already a pipe server for the given pipe name.
447 handle->accept_reqs[0].pipeHandle = CreateNamedPipeW(handle->name,
448 PIPE_ACCESS_DUPLEX | FILE_FLAG_OVERLAPPED |
449 FILE_FLAG_FIRST_PIPE_INSTANCE,
450 PIPE_TYPE_BYTE | PIPE_READMODE_BYTE | PIPE_WAIT,
451 PIPE_UNLIMITED_INSTANCES, 65536, 65536, 0, NULL);
453 if (handle->accept_reqs[0].pipeHandle == INVALID_HANDLE_VALUE) {
454 errorno = GetLastError();
455 if (errorno == ERROR_ACCESS_DENIED) {
456 uv__set_error(loop, UV_EADDRINUSE, errorno);
457 } else if (errorno == ERROR_PATH_NOT_FOUND || errorno == ERROR_INVALID_NAME) {
458 uv__set_error(loop, UV_EACCES, errorno);
460 uv__set_sys_error(loop, errorno);
465 if (uv_set_pipe_handle(loop, handle, handle->accept_reqs[0].pipeHandle, 0)) {
466 uv__set_sys_error(loop, GetLastError());
470 handle->pending_accepts = NULL;
471 handle->flags |= UV_HANDLE_PIPESERVER;
472 handle->flags |= UV_HANDLE_BOUND;
482 if (handle->accept_reqs[0].pipeHandle != INVALID_HANDLE_VALUE) {
483 CloseHandle(handle->accept_reqs[0].pipeHandle);
484 handle->accept_reqs[0].pipeHandle = INVALID_HANDLE_VALUE;
491 static DWORD WINAPI pipe_connect_thread_proc(void* parameter) {
495 HANDLE pipeHandle = INVALID_HANDLE_VALUE;
498 req = (uv_connect_t*) parameter;
500 handle = (uv_pipe_t*) req->handle;
505 /* We're here because CreateFile on a pipe returned ERROR_PIPE_BUSY. */
506 /* We wait for the pipe to become available with WaitNamedPipe. */
507 while (WaitNamedPipeW(handle->name, 30000)) {
508 /* The pipe is now available, try to connect. */
509 pipeHandle = open_named_pipe(handle->name, &duplex_flags);
510 if (pipeHandle != INVALID_HANDLE_VALUE) {
517 if (pipeHandle != INVALID_HANDLE_VALUE &&
518 !uv_set_pipe_handle(loop, handle, pipeHandle, duplex_flags)) {
519 SET_REQ_SUCCESS(req);
521 SET_REQ_ERROR(req, GetLastError());
525 POST_COMPLETION_FOR_REQ(loop, req);
531 void uv_pipe_connect(uv_connect_t* req, uv_pipe_t* handle,
532 const char* name, uv_connect_cb cb) {
533 uv_loop_t* loop = handle->loop;
534 int errorno, nameSize;
535 HANDLE pipeHandle = INVALID_HANDLE_VALUE;
538 uv_req_init(loop, (uv_req_t*) req);
539 req->type = UV_CONNECT;
540 req->handle = (uv_stream_t*) handle;
543 /* Convert name to UTF16. */
544 nameSize = uv_utf8_to_utf16(name, NULL, 0) * sizeof(WCHAR);
545 handle->name = (WCHAR*)malloc(nameSize);
547 uv_fatal_error(ERROR_OUTOFMEMORY, "malloc");
550 if (!uv_utf8_to_utf16(name, handle->name, nameSize / sizeof(WCHAR))) {
551 errorno = GetLastError();
555 pipeHandle = open_named_pipe(handle->name, &duplex_flags);
556 if (pipeHandle == INVALID_HANDLE_VALUE) {
557 if (GetLastError() == ERROR_PIPE_BUSY) {
558 /* Wait for the server to make a pipe instance available. */
559 if (!QueueUserWorkItem(&pipe_connect_thread_proc,
561 WT_EXECUTELONGFUNCTION)) {
562 errorno = GetLastError();
566 REGISTER_HANDLE_REQ(loop, handle, req);
567 handle->reqs_pending++;
572 errorno = GetLastError();
576 assert(pipeHandle != INVALID_HANDLE_VALUE);
578 if (uv_set_pipe_handle(loop,
579 (uv_pipe_t*) req->handle,
582 errorno = GetLastError();
586 SET_REQ_SUCCESS(req);
587 uv_insert_pending_req(loop, (uv_req_t*) req);
588 handle->reqs_pending++;
589 REGISTER_HANDLE_REQ(loop, handle, req);
598 if (pipeHandle != INVALID_HANDLE_VALUE) {
599 CloseHandle(pipeHandle);
602 /* Make this req pending reporting an error. */
603 SET_REQ_ERROR(req, errorno);
604 uv_insert_pending_req(loop, (uv_req_t*) req);
605 handle->reqs_pending++;
606 REGISTER_HANDLE_REQ(loop, handle, req);
611 /* Cleans up uv_pipe_t (server or connection) and all resources associated */
613 void uv_pipe_cleanup(uv_loop_t* loop, uv_pipe_t* handle) {
622 if (handle->flags & UV_HANDLE_PIPESERVER) {
623 for (i = 0; i < handle->pending_instances; i++) {
624 pipeHandle = handle->accept_reqs[i].pipeHandle;
625 if (pipeHandle != INVALID_HANDLE_VALUE) {
626 CloseHandle(pipeHandle);
627 handle->accept_reqs[i].pipeHandle = INVALID_HANDLE_VALUE;
632 if (handle->flags & UV_HANDLE_CONNECTION) {
633 handle->flags &= ~UV_HANDLE_WRITABLE;
634 eof_timer_destroy(handle);
637 if ((handle->flags & UV_HANDLE_CONNECTION)
638 && handle->handle != INVALID_HANDLE_VALUE) {
639 CloseHandle(handle->handle);
640 handle->handle = INVALID_HANDLE_VALUE;
645 void uv_pipe_close(uv_loop_t* loop, uv_pipe_t* handle) {
646 if (handle->flags & UV_HANDLE_READING) {
647 handle->flags &= ~UV_HANDLE_READING;
648 DECREASE_ACTIVE_COUNT(loop, handle);
651 if (handle->flags & UV_HANDLE_LISTENING) {
652 handle->flags &= ~UV_HANDLE_LISTENING;
653 DECREASE_ACTIVE_COUNT(loop, handle);
656 uv_pipe_cleanup(loop, handle);
658 if (handle->reqs_pending == 0) {
659 uv_want_endgame(loop, (uv_handle_t*) handle);
662 handle->flags &= ~(UV_HANDLE_READABLE | UV_HANDLE_WRITABLE);
663 uv__handle_closing(handle);
667 static void uv_pipe_queue_accept(uv_loop_t* loop, uv_pipe_t* handle,
668 uv_pipe_accept_t* req, BOOL firstInstance) {
669 assert(handle->flags & UV_HANDLE_LISTENING);
671 if (!firstInstance) {
672 assert(req->pipeHandle == INVALID_HANDLE_VALUE);
674 req->pipeHandle = CreateNamedPipeW(handle->name,
675 PIPE_ACCESS_DUPLEX | FILE_FLAG_OVERLAPPED,
676 PIPE_TYPE_BYTE | PIPE_READMODE_BYTE | PIPE_WAIT,
677 PIPE_UNLIMITED_INSTANCES, 65536, 65536, 0, NULL);
679 if (req->pipeHandle == INVALID_HANDLE_VALUE) {
680 SET_REQ_ERROR(req, GetLastError());
681 uv_insert_pending_req(loop, (uv_req_t*) req);
682 handle->reqs_pending++;
686 if (uv_set_pipe_handle(loop, handle, req->pipeHandle, 0)) {
687 CloseHandle(req->pipeHandle);
688 req->pipeHandle = INVALID_HANDLE_VALUE;
689 SET_REQ_ERROR(req, GetLastError());
690 uv_insert_pending_req(loop, (uv_req_t*) req);
691 handle->reqs_pending++;
696 assert(req->pipeHandle != INVALID_HANDLE_VALUE);
698 /* Prepare the overlapped structure. */
699 memset(&(req->overlapped), 0, sizeof(req->overlapped));
701 if (!ConnectNamedPipe(req->pipeHandle, &req->overlapped) &&
702 GetLastError() != ERROR_IO_PENDING) {
703 if (GetLastError() == ERROR_PIPE_CONNECTED) {
704 SET_REQ_SUCCESS(req);
706 CloseHandle(req->pipeHandle);
707 req->pipeHandle = INVALID_HANDLE_VALUE;
708 /* Make this req pending reporting an error. */
709 SET_REQ_ERROR(req, GetLastError());
711 uv_insert_pending_req(loop, (uv_req_t*) req);
712 handle->reqs_pending++;
716 handle->reqs_pending++;
720 int uv_pipe_accept(uv_pipe_t* server, uv_stream_t* client) {
721 uv_loop_t* loop = server->loop;
722 uv_pipe_t* pipe_client;
723 uv_pipe_accept_t* req;
726 if (!server->pending_ipc_info.socket_info) {
727 /* No valid pending sockets. */
728 uv__set_sys_error(loop, WSAEWOULDBLOCK);
732 return uv_tcp_import((uv_tcp_t*)client, server->pending_ipc_info.socket_info,
733 server->pending_ipc_info.tcp_connection);
735 pipe_client = (uv_pipe_t*)client;
737 /* Find a connection instance that has been connected, but not yet */
739 req = server->pending_accepts;
742 /* No valid connections found, so we error out. */
743 uv__set_sys_error(loop, WSAEWOULDBLOCK);
747 /* Initialize the client handle and copy the pipeHandle to the client */
748 uv_pipe_connection_init(pipe_client);
749 pipe_client->handle = req->pipeHandle;
750 pipe_client->flags |= UV_HANDLE_READABLE | UV_HANDLE_WRITABLE;
752 /* Prepare the req to pick up a new connection */
753 server->pending_accepts = req->next_pending;
754 req->next_pending = NULL;
755 req->pipeHandle = INVALID_HANDLE_VALUE;
757 if (!(server->flags & UV__HANDLE_CLOSING)) {
758 uv_pipe_queue_accept(loop, server, req, FALSE);
766 /* Starts listening for connections for the given pipe. */
767 int uv_pipe_listen(uv_pipe_t* handle, int backlog, uv_connection_cb cb) {
768 uv_loop_t* loop = handle->loop;
771 if (handle->flags & UV_HANDLE_LISTENING) {
772 handle->connection_cb = cb;
775 if (!(handle->flags & UV_HANDLE_BOUND)) {
776 uv__set_artificial_error(loop, UV_EINVAL);
780 if (handle->flags & UV_HANDLE_READING) {
781 uv__set_artificial_error(loop, UV_EISCONN);
785 if (!(handle->flags & UV_HANDLE_PIPESERVER)) {
786 uv__set_artificial_error(loop, UV_ENOTSUP);
790 handle->flags |= UV_HANDLE_LISTENING;
791 INCREASE_ACTIVE_COUNT(loop, handle);
792 handle->connection_cb = cb;
794 /* First pipe handle should have already been created in uv_pipe_bind */
795 assert(handle->accept_reqs[0].pipeHandle != INVALID_HANDLE_VALUE);
797 for (i = 0; i < handle->pending_instances; i++) {
798 uv_pipe_queue_accept(loop, handle, &handle->accept_reqs[i], i == 0);
805 static DWORD WINAPI uv_pipe_zero_readfile_thread_proc(void* parameter) {
808 uv_read_t* req = (uv_read_t*) parameter;
809 uv_pipe_t* handle = (uv_pipe_t*) req->data;
810 uv_loop_t* loop = handle->loop;
813 assert(req->type == UV_READ);
814 assert(handle->type == UV_NAMED_PIPE);
816 result = ReadFile(handle->handle,
823 SET_REQ_ERROR(req, GetLastError());
826 POST_COMPLETION_FOR_REQ(loop, req);
831 static DWORD WINAPI uv_pipe_writefile_thread_proc(void* parameter) {
834 uv_write_t* req = (uv_write_t*) parameter;
835 uv_pipe_t* handle = (uv_pipe_t*) req->handle;
836 uv_loop_t* loop = handle->loop;
839 assert(req->type == UV_WRITE);
840 assert(handle->type == UV_NAMED_PIPE);
841 assert(req->write_buffer.base);
843 result = WriteFile(handle->handle,
844 req->write_buffer.base,
845 req->write_buffer.len,
850 SET_REQ_ERROR(req, GetLastError());
853 POST_COMPLETION_FOR_REQ(loop, req);
858 static void CALLBACK post_completion_read_wait(void* context, BOOLEAN timed_out) {
862 req = (uv_read_t*) context;
864 handle = (uv_tcp_t*)req->data;
865 assert(handle != NULL);
868 if (!PostQueuedCompletionStatus(handle->loop->iocp,
869 req->overlapped.InternalHigh,
872 uv_fatal_error(GetLastError(), "PostQueuedCompletionStatus");
877 static void CALLBACK post_completion_write_wait(void* context, BOOLEAN timed_out) {
881 req = (uv_write_t*) context;
883 handle = (uv_tcp_t*)req->handle;
884 assert(handle != NULL);
887 if (!PostQueuedCompletionStatus(handle->loop->iocp,
888 req->overlapped.InternalHigh,
891 uv_fatal_error(GetLastError(), "PostQueuedCompletionStatus");
896 static void uv_pipe_queue_read(uv_loop_t* loop, uv_pipe_t* handle) {
900 assert(handle->flags & UV_HANDLE_READING);
901 assert(!(handle->flags & UV_HANDLE_READ_PENDING));
903 assert(handle->handle != INVALID_HANDLE_VALUE);
905 req = &handle->read_req;
907 if (handle->flags & UV_HANDLE_NON_OVERLAPPED_PIPE) {
908 if (!QueueUserWorkItem(&uv_pipe_zero_readfile_thread_proc,
910 WT_EXECUTELONGFUNCTION)) {
911 /* Make this req pending reporting an error. */
912 SET_REQ_ERROR(req, GetLastError());
916 memset(&req->overlapped, 0, sizeof(req->overlapped));
917 if (handle->flags & UV_HANDLE_EMULATE_IOCP) {
918 req->overlapped.hEvent = (HANDLE) ((uintptr_t) req->event_handle | 1);
922 result = ReadFile(handle->handle,
928 if (!result && GetLastError() != ERROR_IO_PENDING) {
929 /* Make this req pending reporting an error. */
930 SET_REQ_ERROR(req, GetLastError());
934 if (handle->flags & UV_HANDLE_EMULATE_IOCP) {
935 if (!req->event_handle) {
936 req->event_handle = CreateEvent(NULL, 0, 0, NULL);
937 if (!req->event_handle) {
938 uv_fatal_error(GetLastError(), "CreateEvent");
941 if (req->wait_handle == INVALID_HANDLE_VALUE) {
942 if (!RegisterWaitForSingleObject(&req->wait_handle,
943 req->overlapped.hEvent, post_completion_read_wait, (void*) req,
944 INFINITE, WT_EXECUTEINWAITTHREAD)) {
945 SET_REQ_ERROR(req, GetLastError());
952 /* Start the eof timer if there is one */
953 eof_timer_start(handle);
954 handle->flags |= UV_HANDLE_READ_PENDING;
955 handle->reqs_pending++;
959 uv_insert_pending_req(loop, (uv_req_t*)req);
960 handle->flags |= UV_HANDLE_READ_PENDING;
961 handle->reqs_pending++;
965 static int uv_pipe_read_start_impl(uv_pipe_t* handle, uv_alloc_cb alloc_cb,
966 uv_read_cb read_cb, uv_read2_cb read2_cb) {
967 uv_loop_t* loop = handle->loop;
969 handle->flags |= UV_HANDLE_READING;
970 INCREASE_ACTIVE_COUNT(loop, handle);
971 handle->read_cb = read_cb;
972 handle->read2_cb = read2_cb;
973 handle->alloc_cb = alloc_cb;
975 /* If reading was stopped and then started again, there could still be a */
976 /* read request pending. */
977 if (!(handle->flags & UV_HANDLE_READ_PENDING))
978 uv_pipe_queue_read(loop, handle);
984 int uv_pipe_read_start(uv_pipe_t* handle, uv_alloc_cb alloc_cb,
985 uv_read_cb read_cb) {
986 return uv_pipe_read_start_impl(handle, alloc_cb, read_cb, NULL);
990 int uv_pipe_read2_start(uv_pipe_t* handle, uv_alloc_cb alloc_cb,
991 uv_read2_cb read_cb) {
992 return uv_pipe_read_start_impl(handle, alloc_cb, NULL, read_cb);
996 static void uv_insert_non_overlapped_write_req(uv_pipe_t* handle,
998 req->next_req = NULL;
999 if (handle->non_overlapped_writes_tail) {
1001 handle->non_overlapped_writes_tail->next_req;
1002 handle->non_overlapped_writes_tail->next_req = (uv_req_t*)req;
1003 handle->non_overlapped_writes_tail = req;
1005 req->next_req = (uv_req_t*)req;
1006 handle->non_overlapped_writes_tail = req;
1011 static uv_write_t* uv_remove_non_overlapped_write_req(uv_pipe_t* handle) {
1014 if (handle->non_overlapped_writes_tail) {
1015 req = (uv_write_t*)handle->non_overlapped_writes_tail->next_req;
1017 if (req == handle->non_overlapped_writes_tail) {
1018 handle->non_overlapped_writes_tail = NULL;
1020 handle->non_overlapped_writes_tail->next_req =
1032 static void uv_queue_non_overlapped_write(uv_pipe_t* handle) {
1033 uv_write_t* req = uv_remove_non_overlapped_write_req(handle);
1035 if (!QueueUserWorkItem(&uv_pipe_writefile_thread_proc,
1037 WT_EXECUTELONGFUNCTION)) {
1038 uv_fatal_error(GetLastError(), "QueueUserWorkItem");
1044 static int uv_pipe_write_impl(uv_loop_t* loop, uv_write_t* req,
1045 uv_pipe_t* handle, uv_buf_t bufs[], int bufcnt,
1046 uv_stream_t* send_handle, uv_write_cb cb) {
1048 uv_tcp_t* tcp_send_handle;
1049 uv_write_t* ipc_header_req;
1050 uv_ipc_frame_uv_stream ipc_frame;
1052 if (bufcnt != 1 && (bufcnt != 0 || !send_handle)) {
1053 uv__set_artificial_error(loop, UV_ENOTSUP);
1057 /* Only TCP handles are supported for sharing. */
1058 if (send_handle && ((send_handle->type != UV_TCP) ||
1059 (!(send_handle->flags & UV_HANDLE_BOUND) &&
1060 !(send_handle->flags & UV_HANDLE_CONNECTION)))) {
1061 uv__set_artificial_error(loop, UV_ENOTSUP);
1065 assert(handle->handle != INVALID_HANDLE_VALUE);
1067 uv_req_init(loop, (uv_req_t*) req);
1068 req->type = UV_WRITE;
1069 req->handle = (uv_stream_t*) handle;
1071 req->ipc_header = 0;
1072 req->event_handle = NULL;
1073 req->wait_handle = INVALID_HANDLE_VALUE;
1074 memset(&req->overlapped, 0, sizeof(req->overlapped));
1077 assert(!(handle->flags & UV_HANDLE_NON_OVERLAPPED_PIPE));
1078 ipc_frame.header.flags = 0;
1080 /* Use the IPC framing protocol. */
1082 tcp_send_handle = (uv_tcp_t*)send_handle;
1084 if (uv_tcp_duplicate_socket(tcp_send_handle, handle->ipc_pid,
1085 &ipc_frame.socket_info)) {
1088 ipc_frame.header.flags |= UV_IPC_TCP_SERVER;
1090 if (tcp_send_handle->flags & UV_HANDLE_CONNECTION) {
1091 ipc_frame.header.flags |= UV_IPC_TCP_CONNECTION;
1096 ipc_frame.header.flags |= UV_IPC_RAW_DATA;
1097 ipc_frame.header.raw_data_length = bufs[0].len;
1101 * Use the provided req if we're only doing a single write.
1102 * If we're doing multiple writes, use ipc_header_write_req to do
1103 * the first write, and then use the provided req for the second write.
1105 if (!(ipc_frame.header.flags & UV_IPC_RAW_DATA)) {
1106 ipc_header_req = req;
1109 * Try to use the preallocated write req if it's available.
1110 * Otherwise allocate a new one.
1112 if (handle->ipc_header_write_req.type != UV_WRITE) {
1113 ipc_header_req = (uv_write_t*)&handle->ipc_header_write_req;
1115 ipc_header_req = (uv_write_t*)malloc(sizeof(uv_write_t));
1116 if (!ipc_header_req) {
1117 uv_fatal_error(ERROR_OUTOFMEMORY, "malloc");
1121 uv_req_init(loop, (uv_req_t*) ipc_header_req);
1122 ipc_header_req->type = UV_WRITE;
1123 ipc_header_req->handle = (uv_stream_t*) handle;
1124 ipc_header_req->cb = NULL;
1125 ipc_header_req->ipc_header = 1;
1128 /* Write the header or the whole frame. */
1129 memset(&ipc_header_req->overlapped, 0, sizeof(ipc_header_req->overlapped));
1131 result = WriteFile(handle->handle,
1133 ipc_frame.header.flags & UV_IPC_TCP_SERVER ?
1134 sizeof(ipc_frame) : sizeof(ipc_frame.header),
1136 &ipc_header_req->overlapped);
1137 if (!result && GetLastError() != ERROR_IO_PENDING) {
1138 uv__set_sys_error(loop, GetLastError());
1143 /* Request completed immediately. */
1144 ipc_header_req->queued_bytes = 0;
1146 /* Request queued by the kernel. */
1147 ipc_header_req->queued_bytes = ipc_frame.header.flags & UV_IPC_TCP_SERVER ?
1148 sizeof(ipc_frame) : sizeof(ipc_frame.header);
1149 handle->write_queue_size += ipc_header_req->queued_bytes;
1152 REGISTER_HANDLE_REQ(loop, handle, ipc_header_req);
1153 handle->reqs_pending++;
1154 handle->write_reqs_pending++;
1156 /* If we don't have any raw data to write - we're done. */
1157 if (!(ipc_frame.header.flags & UV_IPC_RAW_DATA)) {
1162 if (handle->flags & UV_HANDLE_NON_OVERLAPPED_PIPE) {
1163 req->write_buffer = bufs[0];
1164 uv_insert_non_overlapped_write_req(handle, req);
1165 if (handle->write_reqs_pending == 0) {
1166 uv_queue_non_overlapped_write(handle);
1169 /* Request queued by the kernel. */
1170 req->queued_bytes = uv_count_bufs(bufs, bufcnt);
1171 handle->write_queue_size += req->queued_bytes;
1173 result = WriteFile(handle->handle,
1179 if (!result && GetLastError() != ERROR_IO_PENDING) {
1180 uv__set_sys_error(loop, GetLastError());
1185 /* Request completed immediately. */
1186 req->queued_bytes = 0;
1188 /* Request queued by the kernel. */
1189 req->queued_bytes = uv_count_bufs(bufs, bufcnt);
1190 handle->write_queue_size += req->queued_bytes;
1193 if (handle->flags & UV_HANDLE_EMULATE_IOCP) {
1194 req->event_handle = CreateEvent(NULL, 0, 0, NULL);
1195 if (!req->event_handle) {
1196 uv_fatal_error(GetLastError(), "CreateEvent");
1198 if (!RegisterWaitForSingleObject(&req->wait_handle,
1199 req->overlapped.hEvent, post_completion_write_wait, (void*) req,
1200 INFINITE, WT_EXECUTEINWAITTHREAD)) {
1201 uv__set_sys_error(loop, GetLastError());
1207 REGISTER_HANDLE_REQ(loop, handle, req);
1208 handle->reqs_pending++;
1209 handle->write_reqs_pending++;
1215 int uv_pipe_write(uv_loop_t* loop, uv_write_t* req, uv_pipe_t* handle,
1216 uv_buf_t bufs[], int bufcnt, uv_write_cb cb) {
1217 return uv_pipe_write_impl(loop, req, handle, bufs, bufcnt, NULL, cb);
1221 int uv_pipe_write2(uv_loop_t* loop, uv_write_t* req, uv_pipe_t* handle,
1222 uv_buf_t bufs[], int bufcnt, uv_stream_t* send_handle, uv_write_cb cb) {
1224 uv__set_artificial_error(loop, UV_EINVAL);
1228 return uv_pipe_write_impl(loop, req, handle, bufs, bufcnt, send_handle, cb);
1232 static void uv_pipe_read_eof(uv_loop_t* loop, uv_pipe_t* handle,
1234 /* If there is an eof timer running, we don't need it any more, */
1235 /* so discard it. */
1236 eof_timer_destroy(handle);
1238 handle->flags &= ~UV_HANDLE_READABLE;
1239 uv_read_stop((uv_stream_t*) handle);
1241 uv__set_artificial_error(loop, UV_EOF);
1242 if (handle->read2_cb) {
1243 handle->read2_cb(handle, -1, uv_null_buf_, UV_UNKNOWN_HANDLE);
1245 handle->read_cb((uv_stream_t*) handle, -1, uv_null_buf_);
1250 static void uv_pipe_read_error(uv_loop_t* loop, uv_pipe_t* handle, int error,
1252 /* If there is an eof timer running, we don't need it any more, */
1253 /* so discard it. */
1254 eof_timer_destroy(handle);
1256 uv_read_stop((uv_stream_t*) handle);
1258 uv__set_sys_error(loop, error);
1259 if (handle->read2_cb) {
1260 handle->read2_cb(handle, -1, buf, UV_UNKNOWN_HANDLE);
1262 handle->read_cb((uv_stream_t*)handle, -1, buf);
1267 static void uv_pipe_read_error_or_eof(uv_loop_t* loop, uv_pipe_t* handle,
1268 int error, uv_buf_t buf) {
1269 if (error == ERROR_BROKEN_PIPE) {
1270 uv_pipe_read_eof(loop, handle, buf);
1272 uv_pipe_read_error(loop, handle, error, buf);
1277 void uv_process_pipe_read_req(uv_loop_t* loop, uv_pipe_t* handle,
1281 uv_ipc_frame_uv_stream ipc_frame;
1283 assert(handle->type == UV_NAMED_PIPE);
1285 handle->flags &= ~UV_HANDLE_READ_PENDING;
1286 eof_timer_stop(handle);
1288 if (!REQ_SUCCESS(req)) {
1289 /* An error occurred doing the 0-read. */
1290 if (handle->flags & UV_HANDLE_READING) {
1291 uv_pipe_read_error_or_eof(loop,
1297 /* Do non-blocking reads until the buffer is empty */
1298 while (handle->flags & UV_HANDLE_READING) {
1299 if (!PeekNamedPipe(handle->handle,
1305 uv_pipe_read_error_or_eof(loop, handle, GetLastError(), uv_null_buf_);
1310 /* There is nothing to read after all. */
1315 /* Use the IPC framing protocol to read the incoming data. */
1316 if (handle->remaining_ipc_rawdata_bytes == 0) {
1317 /* We're reading a new frame. First, read the header. */
1318 assert(avail >= sizeof(ipc_frame.header));
1320 if (!ReadFile(handle->handle,
1322 sizeof(ipc_frame.header),
1325 uv_pipe_read_error_or_eof(loop, handle, GetLastError(),
1330 assert(bytes == sizeof(ipc_frame.header));
1331 assert(ipc_frame.header.flags <= (UV_IPC_TCP_SERVER | UV_IPC_RAW_DATA |
1332 UV_IPC_TCP_CONNECTION));
1334 if (ipc_frame.header.flags & UV_IPC_TCP_SERVER) {
1335 assert(avail - sizeof(ipc_frame.header) >=
1336 sizeof(ipc_frame.socket_info));
1338 /* Read the TCP socket info. */
1339 if (!ReadFile(handle->handle,
1340 &ipc_frame.socket_info,
1341 sizeof(ipc_frame) - sizeof(ipc_frame.header),
1344 uv_pipe_read_error_or_eof(loop, handle, GetLastError(),
1349 assert(bytes == sizeof(ipc_frame) - sizeof(ipc_frame.header));
1351 /* Store the pending socket info. */
1352 assert(!handle->pending_ipc_info.socket_info);
1353 handle->pending_ipc_info.socket_info =
1354 (WSAPROTOCOL_INFOW*)malloc(sizeof(*(handle->pending_ipc_info.socket_info)));
1355 if (!handle->pending_ipc_info.socket_info) {
1356 uv_fatal_error(ERROR_OUTOFMEMORY, "malloc");
1359 *(handle->pending_ipc_info.socket_info) = ipc_frame.socket_info;
1360 handle->pending_ipc_info.tcp_connection =
1361 ipc_frame.header.flags & UV_IPC_TCP_CONNECTION;
1364 if (ipc_frame.header.flags & UV_IPC_RAW_DATA) {
1365 handle->remaining_ipc_rawdata_bytes =
1366 ipc_frame.header.raw_data_length;
1370 avail = min(avail, (DWORD)handle->remaining_ipc_rawdata_bytes);
1374 buf = handle->alloc_cb((uv_handle_t*) handle, avail);
1375 assert(buf.len > 0);
1377 if (ReadFile(handle->handle,
1382 /* Successful read */
1384 assert(handle->remaining_ipc_rawdata_bytes >= bytes);
1385 handle->remaining_ipc_rawdata_bytes =
1386 handle->remaining_ipc_rawdata_bytes - bytes;
1387 if (handle->read2_cb) {
1388 handle->read2_cb(handle, bytes, buf,
1389 handle->pending_ipc_info.socket_info ? UV_TCP : UV_UNKNOWN_HANDLE);
1390 } else if (handle->read_cb) {
1391 handle->read_cb((uv_stream_t*)handle, bytes, buf);
1394 if (handle->pending_ipc_info.socket_info) {
1395 free(handle->pending_ipc_info.socket_info);
1396 handle->pending_ipc_info.socket_info = NULL;
1399 handle->read_cb((uv_stream_t*)handle, bytes, buf);
1402 /* Read again only if bytes == buf.len */
1403 if (bytes <= buf.len) {
1407 uv_pipe_read_error_or_eof(loop, handle, GetLastError(), uv_null_buf_);
1412 /* Post another 0-read if still reading and not closing. */
1413 if ((handle->flags & UV_HANDLE_READING) &&
1414 !(handle->flags & UV_HANDLE_READ_PENDING)) {
1415 uv_pipe_queue_read(loop, handle);
1419 DECREASE_PENDING_REQ_COUNT(handle);
1423 void uv_process_pipe_write_req(uv_loop_t* loop, uv_pipe_t* handle,
1425 assert(handle->type == UV_NAMED_PIPE);
1427 assert(handle->write_queue_size >= req->queued_bytes);
1428 handle->write_queue_size -= req->queued_bytes;
1430 UNREGISTER_HANDLE_REQ(loop, handle, req);
1432 if (handle->flags & UV_HANDLE_EMULATE_IOCP) {
1433 if (req->wait_handle != INVALID_HANDLE_VALUE) {
1434 UnregisterWait(req->wait_handle);
1435 req->wait_handle = INVALID_HANDLE_VALUE;
1437 if (req->event_handle) {
1438 CloseHandle(req->event_handle);
1439 req->event_handle = NULL;
1443 if (req->ipc_header) {
1444 if (req == &handle->ipc_header_write_req) {
1445 req->type = UV_UNKNOWN_REQ;
1451 if (!REQ_SUCCESS(req)) {
1452 uv__set_sys_error(loop, GET_REQ_ERROR(req));
1453 ((uv_write_cb)req->cb)(req, -1);
1455 ((uv_write_cb)req->cb)(req, 0);
1460 handle->write_reqs_pending--;
1462 if (handle->flags & UV_HANDLE_NON_OVERLAPPED_PIPE &&
1463 handle->non_overlapped_writes_tail) {
1464 assert(handle->write_reqs_pending > 0);
1465 uv_queue_non_overlapped_write(handle);
1468 if (handle->shutdown_req != NULL &&
1469 handle->write_reqs_pending == 0) {
1470 uv_want_endgame(loop, (uv_handle_t*)handle);
1473 DECREASE_PENDING_REQ_COUNT(handle);
1477 void uv_process_pipe_accept_req(uv_loop_t* loop, uv_pipe_t* handle,
1478 uv_req_t* raw_req) {
1479 uv_pipe_accept_t* req = (uv_pipe_accept_t*) raw_req;
1481 assert(handle->type == UV_NAMED_PIPE);
1483 if (REQ_SUCCESS(req)) {
1484 assert(req->pipeHandle != INVALID_HANDLE_VALUE);
1485 req->next_pending = handle->pending_accepts;
1486 handle->pending_accepts = req;
1488 if (handle->connection_cb) {
1489 handle->connection_cb((uv_stream_t*)handle, 0);
1492 if (req->pipeHandle != INVALID_HANDLE_VALUE) {
1493 CloseHandle(req->pipeHandle);
1494 req->pipeHandle = INVALID_HANDLE_VALUE;
1496 if (!(handle->flags & UV__HANDLE_CLOSING)) {
1497 uv_pipe_queue_accept(loop, handle, req, FALSE);
1501 DECREASE_PENDING_REQ_COUNT(handle);
1505 void uv_process_pipe_connect_req(uv_loop_t* loop, uv_pipe_t* handle,
1506 uv_connect_t* req) {
1507 assert(handle->type == UV_NAMED_PIPE);
1509 UNREGISTER_HANDLE_REQ(loop, handle, req);
1512 if (REQ_SUCCESS(req)) {
1513 uv_pipe_connection_init(handle);
1514 ((uv_connect_cb)req->cb)(req, 0);
1516 uv__set_sys_error(loop, GET_REQ_ERROR(req));
1517 ((uv_connect_cb)req->cb)(req, -1);
1521 DECREASE_PENDING_REQ_COUNT(handle);
1525 void uv_process_pipe_shutdown_req(uv_loop_t* loop, uv_pipe_t* handle,
1526 uv_shutdown_t* req) {
1527 assert(handle->type == UV_NAMED_PIPE);
1529 UNREGISTER_HANDLE_REQ(loop, handle, req);
1531 /* Initialize and optionally start the eof timer. */
1532 /* This makes no sense if we've already seen EOF. */
1533 if (handle->flags & UV_HANDLE_READABLE) {
1534 eof_timer_init(handle);
1536 /* If reading start the timer right now. */
1537 /* Otherwise uv_pipe_queue_read will start it. */
1538 if (handle->flags & UV_HANDLE_READ_PENDING) {
1539 eof_timer_start(handle);
1547 DECREASE_PENDING_REQ_COUNT(handle);
1551 static void eof_timer_init(uv_pipe_t* pipe) {
1554 assert(pipe->eof_timer == NULL);
1555 assert(pipe->flags & UV_HANDLE_CONNECTION);
1557 pipe->eof_timer = (uv_timer_t*) malloc(sizeof *pipe->eof_timer);
1559 r = uv_timer_init(pipe->loop, pipe->eof_timer);
1560 assert(r == 0); /* timers can't fail */
1561 pipe->eof_timer->data = pipe;
1562 uv_unref((uv_handle_t*) pipe->eof_timer);
1566 static void eof_timer_start(uv_pipe_t* pipe) {
1567 assert(pipe->flags & UV_HANDLE_CONNECTION);
1569 if (pipe->eof_timer != NULL) {
1570 uv_timer_start(pipe->eof_timer, eof_timer_cb, eof_timeout, 0);
1575 static void eof_timer_stop(uv_pipe_t* pipe) {
1576 assert(pipe->flags & UV_HANDLE_CONNECTION);
1578 if (pipe->eof_timer != NULL) {
1579 uv_timer_stop(pipe->eof_timer);
1584 static void eof_timer_cb(uv_timer_t* timer, int status) {
1585 uv_pipe_t* pipe = (uv_pipe_t*) timer->data;
1586 uv_loop_t* loop = timer->loop;
1588 assert(status == 0); /* timers can't fail */
1589 assert(pipe->type == UV_NAMED_PIPE);
1591 /* This should always be true, since we start the timer only */
1592 /* in uv_pipe_queue_read after successfully calling ReadFile, */
1593 /* or in uv_process_pipe_shutdown_req if a read is pending, */
1594 /* and we always immediately stop the timer in */
1595 /* uv_process_pipe_read_req. */
1596 assert(pipe->flags & UV_HANDLE_READ_PENDING);
1598 /* If there are many packets coming off the iocp then the timer callback */
1599 /* may be called before the read request is coming off the queue. */
1600 /* Therefore we check here if the read request has completed but will */
1601 /* be processed later. */
1602 if ((pipe->flags & UV_HANDLE_READ_PENDING) &&
1603 HasOverlappedIoCompleted(&pipe->read_req.overlapped)) {
1607 /* Force both ends off the pipe. */
1608 CloseHandle(pipe->handle);
1609 pipe->handle = INVALID_HANDLE_VALUE;
1611 /* Stop reading, so the pending read that is going to fail will */
1612 /* not be reported to the user. */
1613 uv_read_stop((uv_stream_t*) pipe);
1615 /* Report the eof and update flags. This will get reported even if the */
1616 /* user stopped reading in the meantime. TODO: is that okay? */
1617 uv_pipe_read_eof(loop, pipe, uv_null_buf_);
1621 static void eof_timer_destroy(uv_pipe_t* pipe) {
1622 assert(pipe->flags && UV_HANDLE_CONNECTION);
1624 if (pipe->eof_timer) {
1625 uv_close((uv_handle_t*) pipe->eof_timer, eof_timer_close_cb);
1626 pipe->eof_timer = NULL;
1631 static void eof_timer_close_cb(uv_handle_t* handle) {
1632 assert(handle->type == UV_TIMER);
1637 int uv_pipe_open(uv_pipe_t* pipe, uv_file file) {
1638 HANDLE os_handle = (HANDLE)_get_osfhandle(file);
1640 if (os_handle == INVALID_HANDLE_VALUE ||
1641 uv_set_pipe_handle(pipe->loop, pipe, os_handle, 0) == -1) {
1642 uv__set_sys_error(pipe->loop, WSAEINVAL);
1646 uv_pipe_connection_init(pipe);
1647 pipe->handle = os_handle;
1648 pipe->flags |= UV_HANDLE_READABLE | UV_HANDLE_WRITABLE;
1651 assert(!(pipe->flags & UV_HANDLE_NON_OVERLAPPED_PIPE));
1652 pipe->ipc_pid = uv_parent_pid();
1653 assert(pipe->ipc_pid != -1);