Bug#47571: idle named pipe connection is unkillable

Bug#31621: Windows server hanging during shutdown using named pipes 
           and idle connection
            
Problem: when idle pipe connection is forcefully closed with KILL
statement or when the server goes down, thread that is closing connection
would hang infinitely in CloseHandle(). The reason for the hang is that 
named pipe operations are performed synchronously. In this mode all IOs
on pipe are serialized, that is CloseHandle() will not abort ReadFile() 
in another thread, but wait for ReadFile() to complete.
            
The fix implements asynchrnous mode for named pipes, where operation of file
are not synchronized. Read/Write operation would fire an async IO and wait for
either IO completion or timeout.
            
Note, that with this patch timeouts are properly handled for named pipes.
      
Post-review: Win32 timeout code has been fixed for named pipes and shared
memory. We do not store pointer to NET in vio structure, only the read and 
write timeouts.


include/violite.h:
  Add pipe_overlapped to Vio structure for async IO for named pipes.
sql-common/client.c:
  Use asynchronous pipe IO.
sql/mysqld.cc:
  Use asynchronous pipe IO.
vio/vio.c:
  -Refactor timeouts for win32 protocols: shared memory and named pipes.
  Store read/write timeout in VIO structure, instead of storing pointer
  to NET. New function vio_win32_timeout called indirectly via 
  vio_timeout changes these values.
vio/vio_priv.h:
  Remove vio_ignore_timeout.
  Add vio_win32_timeout to be used for named pipes and shared memory.
vio/viosocket.c:
  Use async IO for named pipes.
  After issuing IO, wait for either IO completion, pipe_close_event
  or timeout.
          
  Refactor timeouts for named pipe and shared memory.

include/violite.h

@@ -44,7 +44,7 @@ enum enum_vio_type
 Vio*	vio_new(my_socket sd, enum enum_vio_type type, uint flags);
 #ifdef __WIN__
 Vio* vio_new_win32pipe(HANDLE hPipe);
-Vio* vio_new_win32shared_memory(NET *net,HANDLE handle_file_map,
+Vio* vio_new_win32shared_memory(HANDLE handle_file_map,
                                 HANDLE handle_map,
                                 HANDLE event_server_wrote,
                                 HANDLE event_server_read,
@@ -221,7 +221,11 @@ struct st_vio
   HANDLE  event_conn_closed;
   size_t  shared_memory_remain;
   char    *shared_memory_pos;
-  NET     *net;
 #endif /* HAVE_SMEM */
+#ifdef _WIN32
+  OVERLAPPED pipe_overlapped;
+  DWORD read_timeout_millis;
+  DWORD write_timeout_millis;
+#endif
 };
 #endif /* vio_violite_h_ */

sql-common/client.c

@@ -389,7 +389,7 @@ HANDLE create_named_pipe(MYSQL *mysql, uint connect_timeout, char **arg_host,
 			    0,
 			    NULL,
 			    OPEN_EXISTING,
-			    0,
+			    FILE_FLAG_OVERLAPPED,
 			    NULL )) != INVALID_HANDLE_VALUE)
       break;
     if (GetLastError() != ERROR_PIPE_BUSY)
@@ -623,7 +623,7 @@ HANDLE create_shared_memory(MYSQL *mysql,NET *net, uint connect_timeout)
 err2:
   if (error_allow == 0)
   {
-    net->vio= vio_new_win32shared_memory(net,handle_file_map,handle_map,
+    net->vio= vio_new_win32shared_memory(handle_file_map,handle_map,
                                          event_server_wrote,
                                          event_server_read,event_client_wrote,
                                          event_client_read,event_conn_closed);
@@ -2028,7 +2028,7 @@ CLI_MYSQL_REAL_CONNECT(MYSQL *mysql,const char *host, const char *user,
     }
     else
     {
-      net->vio=vio_new_win32pipe(hPipe);
+      net->vio= vio_new_win32pipe(hPipe);
       my_snprintf(host_info=buff, sizeof(buff)-1,
                   ER(CR_NAMEDPIPE_CONNECTION), unix_socket);
     }

sql/mysqld.cc

@@ -1717,7 +1717,7 @@ static void network_init(void)
     saPipeSecurity.lpSecurityDescriptor = &sdPipeDescriptor;
     saPipeSecurity.bInheritHandle = FALSE;
     if ((hPipe= CreateNamedPipe(pipe_name,
-				PIPE_ACCESS_DUPLEX,
+				PIPE_ACCESS_DUPLEX|FILE_FLAG_OVERLAPPED,
 				PIPE_TYPE_BYTE |
 				PIPE_READMODE_BYTE |
 				PIPE_WAIT,
@@ -5203,17 +5203,26 @@ pthread_handler_t handle_connections_sockets(void *arg __attribute__((unused)))
 pthread_handler_t handle_connections_namedpipes(void *arg)
 {
   HANDLE hConnectedPipe;
-  BOOL fConnected;
+  OVERLAPPED connectOverlapped = {0};
   THD *thd;
   my_thread_init();
   DBUG_ENTER("handle_connections_namedpipes");
-  (void) my_pthread_getprio(pthread_self());		// For debugging
+  connectOverlapped.hEvent = CreateEvent(NULL, TRUE, FALSE, NULL);
 
   DBUG_PRINT("general",("Waiting for named pipe connections."));
   while (!abort_loop)
   {
     /* wait for named pipe connection */
-    fConnected = ConnectNamedPipe(hPipe, NULL);
+    BOOL fConnected= ConnectNamedPipe(hPipe, &connectOverlapped);
+    if (!fConnected && (GetLastError() == ERROR_IO_PENDING))
+    {
+        /*
+          ERROR_IO_PENDING says async IO has started but not yet finished.
+          GetOverlappedResult will wait for completion.
+        */
+        DWORD bytes;
+        fConnected= GetOverlappedResult(hPipe, &connectOverlapped,&bytes, TRUE);
+    }
     if (abort_loop)
       break;
     if (!fConnected)
@@ -5222,7 +5231,7 @@ pthread_handler_t handle_connections_namedpipes(void *arg)
     {
       CloseHandle(hPipe);
       if ((hPipe= CreateNamedPipe(pipe_name,
-                                  PIPE_ACCESS_DUPLEX,
+                                  PIPE_ACCESS_DUPLEX|FILE_FLAG_OVERLAPPED,
                                   PIPE_TYPE_BYTE |
                                   PIPE_READMODE_BYTE |
                                   PIPE_WAIT,
@@ -5242,7 +5251,7 @@ pthread_handler_t handle_connections_namedpipes(void *arg)
     hConnectedPipe = hPipe;
     /* create new pipe for new connection */
     if ((hPipe = CreateNamedPipe(pipe_name,
-				 PIPE_ACCESS_DUPLEX,
+				 PIPE_ACCESS_DUPLEX|FILE_FLAG_OVERLAPPED,
 				 PIPE_TYPE_BYTE |
 				 PIPE_READMODE_BYTE |
 				 PIPE_WAIT,
@@ -5264,7 +5273,7 @@ pthread_handler_t handle_connections_namedpipes(void *arg)
       CloseHandle(hConnectedPipe);
       continue;
     }
-    if (!(thd->net.vio = vio_new_win32pipe(hConnectedPipe)) ||
+    if (!(thd->net.vio= vio_new_win32pipe(hConnectedPipe)) ||
 	my_net_init(&thd->net, thd->net.vio))
     {
       close_connection(thd, ER_OUT_OF_RESOURCES, 1);
@@ -5275,7 +5284,7 @@ pthread_handler_t handle_connections_namedpipes(void *arg)
     thd->security_ctx->host= my_strdup(my_localhost, MYF(0));
     create_new_thread(thd);
   }
-
+  CloseHandle(connectOverlapped.hEvent);
   decrement_handler_count();
   DBUG_RETURN(0);
 }
@@ -5452,8 +5461,7 @@ pthread_handler_t handle_connections_shared_memory(void *arg)
       errmsg= "Could not set client to read mode";
       goto errorconn;
     }
-    if (!(thd->net.vio= vio_new_win32shared_memory(&thd->net,
-                                                   handle_client_file_map,
+    if (!(thd->net.vio= vio_new_win32shared_memory(handle_client_file_map,
                                                    handle_client_map,
                                                    event_client_wrote,
                                                    event_client_read,

vio/vio.c

@@ -43,7 +43,7 @@ static void vio_init(Vio* vio, enum enum_vio_type type,
   if ((flags & VIO_BUFFERED_READ) &&
       !(vio->read_buffer= (char*)my_malloc(VIO_READ_BUFFER_SIZE, MYF(MY_WME))))
     flags&= ~VIO_BUFFERED_READ;
-#ifdef __WIN__ 
+#ifdef _WIN32
   if (type == VIO_TYPE_NAMEDPIPE)
   {
     vio->viodelete	=vio_delete;
@@ -59,9 +59,16 @@ static void vio_init(Vio* vio, enum enum_vio_type type,
     vio->in_addr	=vio_in_addr;
     vio->vioblocking	=vio_blocking;
     vio->is_blocking	=vio_is_blocking;
-    vio->timeout	=vio_ignore_timeout;
+
+    vio->timeout=vio_win32_timeout;
+    /* Set default timeout */
+    vio->read_timeout_millis = INFINITE;
+    vio->write_timeout_millis = INFINITE;
+
+    memset(&(vio->pipe_overlapped), 0, sizeof(OVERLAPPED));
+    vio->pipe_overlapped.hEvent= CreateEvent(NULL, TRUE, FALSE, NULL);
+    DBUG_VOID_RETURN;
   }
-  else					/* default is VIO_TYPE_TCPIP */
 #endif
 #ifdef HAVE_SMEM 
   if (type == VIO_TYPE_SHARED_MEMORY)
@@ -79,9 +86,14 @@ static void vio_init(Vio* vio, enum enum_vio_type type,
     vio->in_addr	=vio_in_addr;
     vio->vioblocking	=vio_blocking;
     vio->is_blocking	=vio_is_blocking;
-    vio->timeout	=vio_ignore_timeout;
+
+    /* Currently, shared memory is on Windows only, hence the below is ok*/
+    vio->timeout= vio_win32_timeout; 
+    /* Set default timeout */
+    vio->read_timeout_millis= INFINITE;
+    vio->write_timeout_millis= INFINITE;
+    DBUG_VOID_RETURN;
   }
-  else
 #endif   
 #ifdef HAVE_OPENSSL 
   if (type == VIO_TYPE_SSL)
@@ -100,8 +112,8 @@ static void vio_init(Vio* vio, enum enum_vio_type type,
     vio->vioblocking	=vio_ssl_blocking;
     vio->is_blocking	=vio_is_blocking;
     vio->timeout	=vio_timeout;
+    DBUG_VOID_RETURN;
   }
-  else					/* default is VIO_TYPE_TCPIP */
 #endif /* HAVE_OPENSSL */
   {
     vio->viodelete	=vio_delete;
@@ -193,7 +205,7 @@ Vio *vio_new_win32pipe(HANDLE hPipe)
 }
 
 #ifdef HAVE_SMEM
-Vio *vio_new_win32shared_memory(NET *net,HANDLE handle_file_map, HANDLE handle_map,
+Vio *vio_new_win32shared_memory(HANDLE handle_file_map, HANDLE handle_map,
                                 HANDLE event_server_wrote, HANDLE event_server_read,
                                 HANDLE event_client_wrote, HANDLE event_client_read,
                                 HANDLE event_conn_closed)
@@ -212,7 +224,6 @@ Vio *vio_new_win32shared_memory(NET *net,HANDLE handle_file_map, HANDLE handle_m
     vio->event_conn_closed= event_conn_closed;
     vio->shared_memory_remain= 0;
     vio->shared_memory_pos= handle_map;
-    vio->net= net;
     strmov(vio->desc, "shared memory");
   }
   DBUG_RETURN(vio);

vio/vio_priv.h

@@ -22,7 +22,10 @@
 #include <m_string.h>
 #include <violite.h>
 
-void	vio_ignore_timeout(Vio *vio, uint which, uint timeout);
+#ifdef _WIN32
+void	vio_win32_timeout(Vio *vio, uint which, uint timeout);
+#endif
+
 void	vio_timeout(Vio *vio,uint which, uint timeout);
 
 #ifdef HAVE_OPENSSL

vio/viosocket.c

@@ -261,19 +261,13 @@ int vio_close(Vio * vio)
 {
   int r=0;
   DBUG_ENTER("vio_close");
-#ifdef __WIN__
-  if (vio->type == VIO_TYPE_NAMEDPIPE)
-  {
-#if defined(__NT__) && defined(MYSQL_SERVER)
-    CancelIo(vio->hPipe);
-    DisconnectNamedPipe(vio->hPipe);
-#endif
-    r=CloseHandle(vio->hPipe);
-  }
-  else
-#endif /* __WIN__ */
+
  if (vio->type != VIO_CLOSED)
   {
+    DBUG_ASSERT(vio->type ==  VIO_TYPE_TCPIP ||
+      vio->type == VIO_TYPE_SOCKET ||
+      vio->type == VIO_TYPE_SSL);
+
     DBUG_ASSERT(vio->sd >= 0);
     if (shutdown(vio->sd, SHUT_RDWR))
       r= -1;
@@ -417,44 +411,97 @@ void vio_timeout(Vio *vio, uint which, uint timeout)
 
 
 #ifdef __WIN__
-size_t vio_read_pipe(Vio * vio, uchar* buf, size_t size)
+
+/*
+  Finish pending IO on pipe. Honor wait timeout
+*/
+static int pipe_complete_io(Vio* vio, char* buf, size_t size, DWORD timeout_millis)
 {
   DWORD length;
+  DWORD ret;
+
+  DBUG_ENTER("pipe_complete_io");
+
+  ret= WaitForSingleObject(vio->pipe_overlapped.hEvent, timeout_millis);
+  /*
+    WaitForSingleObjects will normally return WAIT_OBJECT_O (success, IO completed)
+    or WAIT_TIMEOUT.
+  */
+  if(ret != WAIT_OBJECT_0)
+  {
+    CancelIo(vio->hPipe);
+    DBUG_PRINT("error",("WaitForSingleObject() returned  %d", ret));
+    DBUG_RETURN(-1);
+  }
+
+  if (!GetOverlappedResult(vio->hPipe,&(vio->pipe_overlapped),&length, FALSE))
+  {
+    DBUG_PRINT("error",("GetOverlappedResult() returned last error  %d", 
+      GetLastError()));
+    DBUG_RETURN(-1);
+  }
+
+  DBUG_RETURN(length);
+}
+
+
+size_t vio_read_pipe(Vio * vio, uchar *buf, size_t size)
+{
+  DWORD bytes_read;
   DBUG_ENTER("vio_read_pipe");
   DBUG_PRINT("enter", ("sd: %d  buf: 0x%lx  size: %u", vio->sd, (long) buf,
                        (uint) size));
 
-  if (!ReadFile(vio->hPipe, buf, size, &length, NULL))
-    DBUG_RETURN(-1);
+  if (!ReadFile(vio->hPipe, buf, (DWORD)size, &bytes_read,
+      &(vio->pipe_overlapped)))
+  {
+    if (GetLastError() != ERROR_IO_PENDING)
+    {
+      DBUG_PRINT("error",("ReadFile() returned last error %d",
+        GetLastError()));
+      DBUG_RETURN((size_t)-1);
+    }
+    bytes_read= pipe_complete_io(vio, buf, size,vio->read_timeout_millis);
+  }
 
-  DBUG_PRINT("exit", ("%d", length));
-  DBUG_RETURN((size_t) length);
+  DBUG_PRINT("exit", ("%d", bytes_read));
+  DBUG_RETURN(bytes_read);
 }
 
 
 size_t vio_write_pipe(Vio * vio, const uchar* buf, size_t size)
 {
-  DWORD length;
+  DWORD bytes_written;
   DBUG_ENTER("vio_write_pipe");
   DBUG_PRINT("enter", ("sd: %d  buf: 0x%lx  size: %u", vio->sd, (long) buf,
                        (uint) size));
 
-  if (!WriteFile(vio->hPipe, (char*) buf, size, &length, NULL))
-    DBUG_RETURN(-1);
+  if (!WriteFile(vio->hPipe, buf, (DWORD)size, &bytes_written,
+      &(vio->pipe_overlapped)))
+  {
+    if (GetLastError() != ERROR_IO_PENDING)
+    {
+      DBUG_PRINT("vio_error",("WriteFile() returned last error %d",
+        GetLastError()));
+      DBUG_RETURN((size_t)-1);
+    }
+    bytes_written = pipe_complete_io(vio, (char *)buf, size, 
+        vio->write_timeout_millis);
+  }
 
-  DBUG_PRINT("exit", ("%d", length));
-  DBUG_RETURN((size_t) length);
+  DBUG_PRINT("exit", ("%d", bytes_written));
+  DBUG_RETURN(bytes_written);
 }
 
+
 int vio_close_pipe(Vio * vio)
 {
   int r;
   DBUG_ENTER("vio_close_pipe");
-#if defined(__NT__) && defined(MYSQL_SERVER)
-  CancelIo(vio->hPipe);
+
+  CloseHandle(vio->pipe_overlapped.hEvent);
   DisconnectNamedPipe(vio->hPipe);
-#endif
-  r=CloseHandle(vio->hPipe);
+  r= CloseHandle(vio->hPipe);
   if (r)
   {
     DBUG_PRINT("vio_error", ("close() failed, error: %d",GetLastError()));
@@ -466,10 +513,23 @@ int vio_close_pipe(Vio * vio)
 }
 
 
-void vio_ignore_timeout(Vio *vio __attribute__((unused)),
-			uint which __attribute__((unused)),
-			uint timeout __attribute__((unused)))
+void vio_win32_timeout(Vio *vio, uint which , uint timeout_sec)
 {
+    DWORD timeout_millis;
+    /*
+      Windows is measuring timeouts in milliseconds. Check for possible int 
+      overflow.
+    */
+    if (timeout_sec > UINT_MAX/1000)
+      timeout_millis= INFINITE;
+    else
+      timeout_millis= timeout_sec * 1000;
+
+    /* which == 1 means "write", which == 0 means "read".*/
+    if(which)
+      vio->write_timeout_millis= timeout_millis;
+    else
+      vio->read_timeout_millis= timeout_millis;
 }
 
 
@@ -504,7 +564,7 @@ size_t vio_read_shared_memory(Vio * vio, uchar* buf, size_t size)
          WAIT_ABANDONED_0 and WAIT_TIMEOUT - fail.  We can't read anything
       */
       if (WaitForMultipleObjects(array_elements(events), events, FALSE,
-                                 vio->net->read_timeout*1000) != WAIT_OBJECT_0)
+                                 vio->read_timeout_millis) != WAIT_OBJECT_0)
       {
         DBUG_RETURN(-1);
       };
@@ -561,7 +621,7 @@ size_t vio_write_shared_memory(Vio * vio, const uchar* buf, size_t size)
   while (remain != 0)
   {
     if (WaitForMultipleObjects(array_elements(events), events, FALSE,
-                               vio->net->write_timeout*1000) != WAIT_OBJECT_0)
+                               vio->write_timeout_millis) != WAIT_OBJECT_0)
     {
       DBUG_RETURN((size_t) -1);
     }