Upgrade libuv to 4eff34da4

deps/uv/include/uv-unix.h

@@ -67,25 +67,27 @@ typedef struct {
 
 #define UV_STREAM_PRIVATE_FIELDS \
   uv_read_cb read_cb; \
-  uv_alloc_cb alloc_cb;
+  uv_alloc_cb alloc_cb; \
-
-
-/* UV_TCP */
-#define UV_TCP_PRIVATE_FIELDS \
-  int delayed_error; \
-  uv_connection_cb connection_cb; \
-  int accepted_fd; \
   uv_connect_t *connect_req; \
   uv_shutdown_t *shutdown_req; \
   ev_io read_watcher; \
   ev_io write_watcher; \
   ngx_queue_t write_queue; \
-  ngx_queue_t write_completed_queue;
+  ngx_queue_t write_completed_queue; \
+  int delayed_error; \
+  uv_connection_cb connection_cb; \
+  int accepted_fd;
+
+
+/* UV_TCP */
+#define UV_TCP_PRIVATE_FIELDS
 
 
 /* UV_NAMED_PIPE */
 #define UV_PIPE_PRIVATE_TYPEDEF
-#define UV_PIPE_PRIVATE_FIELDS
+#define UV_PIPE_PRIVATE_FIELDS \
+  UV_TCP_PRIVATE_FIELDS \
+  const char* pipe_fname; /* strdup'ed */ \
 
 
 /* UV_PREPARE */ \

deps/uv/include/uv-win.h

@@ -96,7 +96,6 @@ typedef struct uv_buf_t {
   struct uv_req_s accept_req;             \
 
 #define uv_pipe_server_fields             \
-    char* name;                           \
     uv_pipe_accept_t accept_reqs[4];      \
     uv_pipe_accept_t* pending_accepts;
 
@@ -104,6 +103,7 @@ typedef struct uv_buf_t {
   HANDLE handle;
 
 #define UV_PIPE_PRIVATE_FIELDS            \
+  char* name;                             \
   union {                                 \
     struct { uv_pipe_server_fields };     \
     struct { uv_pipe_connection_fields }; \

deps/uv/src/uv-win.c

@@ -222,6 +222,7 @@ static char uv_zero_[] = "";
 /* mark if IPv6 sockets are supported */
 static BOOL uv_allow_ipv6 = FALSE;
 
+
 /*
  * Subclass of uv_handle_t. Used for integration of c-ares.
  */
@@ -374,6 +375,7 @@ static uv_err_code uv_translate_sys_error(int sys_errno) {
     case ERROR_NO_UNICODE_TRANSLATION:      return UV_ECHARSET;
     case ERROR_BROKEN_PIPE:                 return UV_EOF;
     case ERROR_PIPE_BUSY:                   return UV_EBUSY;
+    case ERROR_SEM_TIMEOUT:                 return UV_ETIMEDOUT;
     default:                                return UV_UNKNOWN;
   }
 }
@@ -517,6 +519,7 @@ void uv_init() {
 static void uv_req_init(uv_req_t* req) {
   uv_counters()->req_init++;
   req->type = UV_UNKNOWN_REQ;
+  req->error = uv_ok_;
 }
 
 
@@ -1028,10 +1031,14 @@ static void uv_pipe_queue_accept(uv_pipe_t* handle, uv_pipe_accept_t* req) {
   /* Prepare the overlapped structure. */
   memset(&(req->overlapped), 0, sizeof(req->overlapped));
 
-  if (!ConnectNamedPipe(pipeHandle, &req->overlapped) &&
+  if (!ConnectNamedPipe(pipeHandle, &req->overlapped) && GetLastError() != ERROR_IO_PENDING) {
-      GetLastError() != ERROR_IO_PENDING && GetLastError() != ERROR_PIPE_CONNECTED) {
+    if (GetLastError() == ERROR_PIPE_CONNECTED) {
-    /* Make this req pending reporting an error. */
+      req->pipeHandle = pipeHandle;
-    req->error = uv_new_sys_error(GetLastError());
+      req->error = uv_ok_;
+    } else {
+      /* Make this req pending reporting an error. */
+      req->error = uv_new_sys_error(GetLastError());
+    }
     uv_insert_pending_req((uv_req_t*) req);
     handle->reqs_pending++;
     return;
@@ -2314,9 +2321,7 @@ static void uv_poll() {
     /* Package was dequeued */
     req = uv_overlapped_to_req(overlapped);
 
-    if (success) {
+    if (!success) {
-      req->error = uv_ok_;
-    } else {
       req->error = uv_new_sys_error(GetLastError());
     }
 
@@ -2970,6 +2975,7 @@ int uv_pipe_init(uv_pipe_t* handle) {
   handle->type = UV_NAMED_PIPE;
   handle->reqs_pending = 0;
   handle->pending_accepts = NULL;
+  handle->name = NULL;
 
   uv_counters()->pipe_init++;
 
@@ -3033,48 +3039,120 @@ int uv_pipe_listen(uv_pipe_t* handle, uv_connection_cb cb) {
   return 0;
 }
 
+
+static int uv_set_pipe_handle(uv_pipe_t* handle, HANDLE pipeHandle) {
+  DWORD mode = PIPE_TYPE_BYTE | PIPE_READMODE_BYTE | PIPE_WAIT;
+
+  if (!SetNamedPipeHandleState(pipeHandle, &mode, NULL, NULL)) {
+    return -1;
+  }
+
+  if (CreateIoCompletionPort(pipeHandle,
+                             uv_iocp_,
+                             (ULONG_PTR)handle,
+                             0) == NULL) {
+    return -1;
+  }
+  
+  return 0;
+}
+
+
+static DWORD WINAPI pipe_connect_thread_proc(void* parameter) {
+  HANDLE pipeHandle = INVALID_HANDLE_VALUE;
+  int errno;
+  uv_pipe_t* handle;
+  uv_connect_t* req;
+
+  req = (uv_connect_t*)parameter;
+  assert(req);
+  handle = (uv_pipe_t*)req->handle;
+  assert(handle);
+  
+  /* We're here because CreateFile on a pipe returned ERROR_PIPE_BUSY.  We wait for the pipe to become available with WaitNamedPipe. */
+  while (WaitNamedPipe(handle->name, 30000)) {
+    /* The pipe is now available, try to connect. */
+    pipeHandle = CreateFile(handle->name,
+                            GENERIC_READ | GENERIC_WRITE,
+                            0,
+                            NULL,
+                            OPEN_EXISTING,
+                            FILE_FLAG_OVERLAPPED,
+                            NULL);
+
+    if (pipeHandle != INVALID_HANDLE_VALUE) {
+      break;
+    }
+  } 
+
+  if (pipeHandle != INVALID_HANDLE_VALUE && !uv_set_pipe_handle(handle, pipeHandle)) {
+    handle->handle = pipeHandle;
+    req->error = uv_ok_;
+  } else {
+    req->error = uv_new_sys_error(GetLastError());
+  }
+
+  memset(&req->overlapped, 0, sizeof(req->overlapped));
+ 
+  /* Post completed */
+  if (!PostQueuedCompletionStatus(uv_iocp_,
+                                0,
+                                0,
+                                &req->overlapped)) {
+    uv_fatal_error(GetLastError(), "PostQueuedCompletionStatus");
+  }   
+
+  return 0;
+}
+
+
 /* TODO: make this work with UTF8 name */
-/* TODO: run this in the thread pool */
 int uv_pipe_connect(uv_connect_t* req, uv_pipe_t* handle,
     const char* name, uv_connect_cb cb) {
   int errno;
-  DWORD mode;
+  HANDLE pipeHandle;
+
+  handle->handle = INVALID_HANDLE_VALUE;
 
   uv_req_init((uv_req_t*) req);
   req->type = UV_CONNECT;
   req->handle = (uv_stream_t*) handle;
   req->cb = cb;
 
-  memset(&req->overlapped, 0, sizeof(req->overlapped));
+  pipeHandle = CreateFile(name,
+                          GENERIC_READ | GENERIC_WRITE,
+                          0,
+                          NULL,
+                          OPEN_EXISTING,
+                          FILE_FLAG_OVERLAPPED,
+                          NULL);
 
-  handle->handle = CreateFile(name,
+  if (pipeHandle == INVALID_HANDLE_VALUE) {
-                              GENERIC_READ | GENERIC_WRITE,
+    if (GetLastError() == ERROR_PIPE_BUSY) {
-                              0,
+      /* Wait for the server to make a pipe instance available. */
-                              NULL,
+      handle->name = strdup(name);
-                              OPEN_EXISTING,
+      if (!handle->name) {
-                              FILE_FLAG_OVERLAPPED,
+        uv_fatal_error(ERROR_OUTOFMEMORY, "malloc");
-                              NULL);
+      }
 
-  if (handle->handle == INVALID_HANDLE_VALUE &&
+      if (!QueueUserWorkItem(&pipe_connect_thread_proc, req, WT_EXECUTELONGFUNCTION)) {
-      GetLastError() != ERROR_IO_PENDING) {
+        errno = GetLastError();
+        goto error;
+      }
+
+      return 0;
+    }
+  
+    errno = GetLastError();
+    goto error;
+  }
+  
+  if (uv_set_pipe_handle((uv_pipe_t*)req->handle, pipeHandle)) {
     errno = GetLastError();
     goto error;
   }
 
-  mode = PIPE_TYPE_BYTE | PIPE_READMODE_BYTE | PIPE_WAIT;
+  handle->handle = pipeHandle;
-
-  if (!SetNamedPipeHandleState(handle->handle, &mode, NULL, NULL)) {
-    errno = GetLastError();
-    goto error;
-  }
-
-  if (CreateIoCompletionPort(handle->handle,
-                             uv_iocp_,
-                             (ULONG_PTR)handle,
-                             0) == NULL) {
-    errno = GetLastError();
-    goto error;
-  }
 
   req->error = uv_ok_;
   uv_insert_pending_req((uv_req_t*) req);
@@ -3082,12 +3160,10 @@ int uv_pipe_connect(uv_connect_t* req, uv_pipe_t* handle,
   return 0;
 
 error:
-  if (handle->handle != INVALID_HANDLE_VALUE) {
+  if (pipeHandle != INVALID_HANDLE_VALUE) {
-    CloseHandle(handle->handle);
+    CloseHandle(pipeHandle);
   }
-  req->error = uv_new_sys_error(errno);
+  uv_set_sys_error(errno);
-  uv_insert_pending_req((uv_req_t*) req);
-  handle->reqs_pending++;
   return -1;
 }
 
@@ -3097,6 +3173,11 @@ static void close_pipe(uv_pipe_t* handle, int* status, uv_err_t* err) {
   int i;
   HANDLE pipeHandle;
 
+  if (handle->name) {
+    free(handle->name);
+    handle->name;
+  }
+
   if (handle->flags & UV_HANDLE_PIPESERVER) {
     for (i = 0; i < COUNTOF(handle->accept_reqs); i++) {
       pipeHandle = handle->accept_reqs[i].pipeHandle;
@@ -3105,7 +3186,7 @@ static void close_pipe(uv_pipe_t* handle, int* status, uv_err_t* err) {
       }
     }
 
-  } else {
+  } else if (handle->handle != INVALID_HANDLE_VALUE) {
     CloseHandle(handle->handle);
   }
 

deps/uv/test/benchmark-pump.c

@@ -261,13 +261,6 @@ static void maybe_connect_some() {
       req = (uv_connect_t*) req_alloc();
       r = uv_pipe_connect(req, pipe, TEST_PIPENAME, connect_cb);
       ASSERT(r == 0);
-
-#ifdef _WIN32
-      /* HACK: This is temporary to give the pipes server enough time to create new handles.
-       * This will go away once uv_pipe_connect can deal with UV_EBUSY.
-       */
-      Sleep(1);
-#endif
     }
   }
 }

deps/uv/test/echo-server.c

@@ -124,7 +124,7 @@ static uv_buf_t echo_alloc(uv_stream_t* handle, size_t suggested_size) {
 
 
 static void on_connection(uv_handle_t* server, int status) {
-  uv_handle_t* handle;
+  uv_stream_t* stream;
   int r;
 
   if (status != 0) {
@@ -132,25 +132,31 @@ static void on_connection(uv_handle_t* server, int status) {
   }
   ASSERT(status == 0);
 
-  if (serverType == TCP) {
+  switch (serverType) {
-    handle = (uv_handle_t*) malloc(sizeof(uv_tcp_t));
+  case TCP:
-    ASSERT(handle != NULL);
+    stream = malloc(sizeof(uv_tcp_t));
+    ASSERT(stream != NULL);
+    uv_tcp_init((uv_tcp_t*)stream);
+    break;
 
-    uv_tcp_init((uv_tcp_t*)handle);
+  case PIPE:
-  } else {
+    stream = malloc(sizeof(uv_pipe_t));
-    handle = (uv_handle_t*) malloc(sizeof(uv_pipe_t));
+    ASSERT(stream != NULL);
-    ASSERT(handle != NULL);
+    uv_pipe_init((uv_pipe_t*)stream);
+    break;
 
-    uv_pipe_init((uv_pipe_t*)handle);
+  default:
+    ASSERT(0 && "Bad serverType");
+    abort();
   }
 
   /* associate server with stream */
-  handle->data = server;
+  stream->data = server;
 
-  r = uv_accept(server, (uv_stream_t*)handle);
+  r = uv_accept(server, stream);
   ASSERT(r == 0);
 
-  r = uv_read_start((uv_stream_t*)handle, echo_alloc, after_read);
+  r = uv_read_start(stream, echo_alloc, after_read);
   ASSERT(r == 0);
 }
 
@@ -233,22 +239,19 @@ static int pipe_echo_start(char* pipeName) {
 
   r = uv_pipe_init(&pipeServer);
   if (r) {
-    /* TODO: Error codes */
+    fprintf(stderr, "uv_pipe_init: %s\n", uv_strerror(uv_last_error()));
-    fprintf(stderr, "Pipe creation error\n");
     return 1;
   }
 
   r = uv_pipe_bind(&pipeServer, pipeName);
   if (r) {
-    /* TODO: Error codes */
+    fprintf(stderr, "uv_pipe_bind: %s\n", uv_strerror(uv_last_error()));
-    fprintf(stderr, "create error\n");
     return 1;
   }
 
   r = uv_pipe_listen(&pipeServer, on_connection);
   if (r) {
-    /* TODO: Error codes */
+    fprintf(stderr, "uv_pipe_listen: %s\n", uv_strerror(uv_last_error()));
-    fprintf(stderr, "Listen error on IPv6\n");
     return 1;
   }
 

deps/uv/test/task.h

@@ -33,8 +33,7 @@
 #ifdef _WIN32
 # define TEST_PIPENAME "\\\\.\\pipe\\uv-test"
 #else
-# /* TODO: define unix pipe name */
+# define TEST_PIPENAME "/tmp/uv-test-sock"
-# define TEST_PIPENAME ""
 #endif
 
 typedef enum {