Backport of Bug#10374 to mysql-next-mr

------------------------------------------------------------
revno: 2597.37.3
revision-id: sp1r-davi@mysql.com/endora.local-20080328123626-16430
parent: sp1r-anozdrin/alik@quad.opbmk-20080327125300-11290
committer: davi@mysql.com/endora.local
timestamp: Fri 2008-03-28 09:36:26 -0300
message:
  Bug#10374 GET_LOCK does not let connection to close on the server side if it's aborted

  The problem is that the server doesn't detect aborted connections which
  are waiting on a lock or sleeping (user sleep), wasting system resources
  for a connection that is already dead.

  The solution is to peek at the connection every five seconds to verify if
  the connection is not aborted. A aborted connection is detect by polling
  the connection socket for available data to be read or end of file and in
  case of eof, the wait is aborted and the connection killed.

include/violite.h

@@ -88,6 +88,7 @@ my_bool	vio_peer_addr(Vio* vio, char *buf, uint16 *port);
 /* Remotes in_addr */
 void	vio_in_addr(Vio *vio, struct in_addr *in);
 my_bool	vio_poll_read(Vio *vio,uint timeout);
+my_bool vio_peek_read(Vio *vio, uint *bytes);
 
 #ifdef HAVE_OPENSSL
 #include <openssl/opensslv.h>

mysql-test/r/dirty_close.result

@@ -7,3 +7,13 @@ n
 2
 3
 DROP TABLE t1;
+SELECT GET_LOCK("dangling", 0);
+GET_LOCK("dangling", 0)
+1
+SELECT GET_LOCK('dangling', 3600);;
+SELECT GET_LOCK('dangling', 3600);;
+SELECT RELEASE_LOCK('dangling');
+RELEASE_LOCK('dangling')
+1
+GET_LOCK('dangling', 3600)
+1

mysql-test/t/dirty_close.test

@@ -22,6 +22,38 @@ disconnect con2;
 
 # End of 4.1 tests
 
+#
+# Bug#10374 GET_LOCK does not let connection to close on the server side if it's aborted
+#
+
+connection default;
+SELECT GET_LOCK("dangling", 0);
+connect(con1, localhost, root,,);
+connection con1;
+--send SELECT GET_LOCK('dangling', 3600);
+connection default;
+let $wait_condition=
+  SELECT COUNT(*) = 1 FROM INFORMATION_SCHEMA.PROCESSLIST WHERE STATE = "User lock"
+  AND INFO = "SELECT GET_LOCK('dangling', 3600)";
+--source include/wait_condition.inc
+dirty_close con1;
+let $wait_condition=
+  SELECT COUNT(*) = 0 FROM INFORMATION_SCHEMA.PROCESSLIST WHERE STATE = "User lock"
+  AND INFO = "SELECT GET_LOCK('dangling', 3600)";
+--source include/wait_condition.inc
+connect(con1, localhost, root,,);
+--send SELECT GET_LOCK('dangling', 3600);
+connection default;
+let $wait_condition=
+  SELECT COUNT(*) = 1 FROM INFORMATION_SCHEMA.PROCESSLIST WHERE STATE = "User lock"
+  AND INFO = "SELECT GET_LOCK('dangling', 3600)";
+--source include/wait_condition.inc
+SELECT RELEASE_LOCK('dangling');
+connection con1;
+--reap
+connection default;
+disconnect con1;
+
 # Wait till all disconnects are completed
 --source include/wait_until_count_sessions.inc
 

sql/item_func.cc

@@ -3437,6 +3437,48 @@ void debug_sync_point(const char* lock_name, uint lock_timeout)
 
 #endif
 
+
+/**
+  Wait for a given condition to be signaled within the specified timeout.
+
+  @param cond the condition variable to wait on
+  @param lock the associated mutex
+  @param abstime the amount of time in seconds to wait
+
+  @retval return value from pthread_cond_timedwait
+*/
+
+#define INTERRUPT_INTERVAL (5 * ULL(1000000000))
+
+static int interruptible_wait(THD *thd, pthread_cond_t *cond,
+                              pthread_mutex_t *lock, double time)
+{
+  int error;
+  struct timespec abstime;
+  ulonglong slice, timeout= (ulonglong) (time * 1000000000.0);
+
+  do
+  {
+    /* Wait for a fixed interval. */
+    if (timeout > INTERRUPT_INTERVAL)
+      slice= INTERRUPT_INTERVAL;
+    else
+      slice= timeout;
+
+    timeout-= slice;
+    set_timespec_nsec(abstime, slice);
+    error= pthread_cond_timedwait(cond, lock, &abstime);
+    if (error == ETIMEDOUT || error == ETIME)
+    {
+      /* Return error if timed out or connection is broken. */
+      if (!timeout || !thd->vio_is_connected())
+        break;
+    }
+  } while (error && timeout);
+
+  return error;
+}
+
 /**
   Get a user level lock.  If the thread has an old lock this is first released.
 
@@ -3452,8 +3494,7 @@ longlong Item_func_get_lock::val_int()
 {
   DBUG_ASSERT(fixed == 1);
   String *res=args[0]->val_str(&value);
-  longlong timeout=args[1]->val_int();
-  struct timespec abstime;
+  double timeout= args[1]->val_real();
   THD *thd=current_thd;
   User_level_lock *ull;
   int error;
@@ -3517,12 +3558,11 @@ longlong Item_func_get_lock::val_int()
   thd->mysys_var->current_mutex= &LOCK_user_locks;
   thd->mysys_var->current_cond=  &ull->cond;
 
-  set_timespec(abstime,timeout);
   error= 0;
   while (ull->locked && !thd->killed)
   {
     DBUG_PRINT("info", ("waiting on lock"));
-    error= pthread_cond_timedwait(&ull->cond,&LOCK_user_locks,&abstime);
+    error= interruptible_wait(thd, &ull->cond, &LOCK_user_locks, timeout);
     if (error == ETIMEDOUT || error == ETIME)
     {
       DBUG_PRINT("info", ("lock wait timeout"));
@@ -3717,13 +3757,13 @@ void Item_func_benchmark::print(String *str, enum_query_type query_type)
 longlong Item_func_sleep::val_int()
 {
   THD *thd= current_thd;
-  struct timespec abstime;
   pthread_cond_t cond;
+  double timeout;
   int error;
 
   DBUG_ASSERT(fixed == 1);
 
-  double time= args[0]->val_real();
+  timeout= args[0]->val_real();
   /*
     On 64-bit OSX pthread_cond_timedwait() waits forever
     if passed abstime time has already been exceeded by 
@@ -3733,10 +3773,8 @@ longlong Item_func_sleep::val_int()
     We assume that the lines between this test and the call 
     to pthread_cond_timedwait() will be executed in less than 0.00001 sec.
   */
-  if (time < 0.00001)
+  if (timeout < 0.00001)
     return 0;
-    
-  set_timespec_nsec(abstime, (ulonglong)(time * ULL(1000000000)));
 
   pthread_cond_init(&cond, NULL);
   pthread_mutex_lock(&LOCK_user_locks);
@@ -3748,7 +3786,7 @@ longlong Item_func_sleep::val_int()
   error= 0;
   while (!thd->killed)
   {
-    error= pthread_cond_timedwait(&cond, &LOCK_user_locks, &abstime);
+    error= interruptible_wait(thd, &cond, &LOCK_user_locks, timeout);
     if (error == ETIMEDOUT || error == ETIME)
       break;
     error= 0;

sql/sql_class.cc

@@ -1618,6 +1618,30 @@ void THD::rollback_item_tree_changes()
 }
 
 
+#ifndef EMBEDDED_LIBRARY
+
+/**
+  Check that the endpoint is still available.
+*/
+
+bool THD::vio_is_connected()
+{
+  uint bytes= 0;
+
+  /* End of input is signaled by poll if the socket is aborted. */
+  if (vio_poll_read(net.vio, 0))
+    return TRUE;
+
+  /* Socket is aborted if signaled but no data is available. */
+  if (vio_peek_read(net.vio, &bytes))
+    return TRUE;
+
+  return bytes ? TRUE : FALSE;
+}
+
+#endif
+
+
 /*****************************************************************************
 ** Functions to provide a interface to select results
 *****************************************************************************/

sql/sql_class.h

@@ -1961,9 +1961,12 @@ class THD :public Statement,
     DBUG_VOID_RETURN;
   }
   inline bool vio_ok() const { return net.vio != 0; }
+  /** Return FALSE if connection to client is broken. */
+  bool vio_is_connected();
 #else
   void clear_error();
-  inline bool vio_ok() const { return true; }
+  inline bool vio_ok() const { return TRUE; }
+  inline bool vio_is_connected() { return TRUE; }
 #endif
   /**
     Mark the current error as fatal. Warning: this does not

vio/viosocket.c

@@ -360,9 +360,24 @@ void vio_in_addr(Vio *vio, struct in_addr *in)
 
 my_bool vio_poll_read(Vio *vio,uint timeout)
 {
-#ifndef HAVE_POLL
-  return 0;
-#else
+#ifdef __WIN__
+  int res, fd= vio->sd;
+  fd_set readfds, errorfds;
+  struct timeval tm;
+  DBUG_ENTER("vio_poll");
+  tm.tv_sec= timeout;
+  tm.tv_usec= 0;
+  FD_ZERO(&readfds);
+  FD_ZERO(&errorfds);
+  FD_SET(fd, &readfds);
+  FD_SET(fd, &errorfds);
+  if ((res= select(fd, &readfds, NULL, &errorfds, &tm) <= 0))
+  {
+    DBUG_RETURN(res < 0 ? 0 : 1);
+  }
+  res= FD_ISSET(fd, &readfds) || FD_ISSET(fd, &errorfds);
+  DBUG_RETURN(!res);
+#elif defined(HAVE_POLL)
   struct pollfd fds;
   int res;
   DBUG_ENTER("vio_poll");
@@ -373,11 +388,37 @@ my_bool vio_poll_read(Vio *vio,uint timeout)
   {
     DBUG_RETURN(res < 0 ? 0 : 1);		/* Don't return 1 on errors */
   }
-  DBUG_RETURN(fds.revents & POLLIN ? 0 : 1);
+  DBUG_RETURN(fds.revents & (POLLIN | POLLERR | POLLHUP) ? 0 : 1);
+#else
+  return 0;
 #endif
 }
 
 
+my_bool vio_peek_read(Vio *vio, uint *bytes)
+{
+#ifdef __WIN__
+  int len;
+  if (ioctlsocket(vio->sd, FIONREAD, &len))
+    return TRUE;
+  *bytes= len;
+  return FALSE;
+#elif FIONREAD
+  int len;
+  if (ioctl(vio->sd, FIONREAD, &len) < 0)
+    return TRUE;
+  *bytes= len;
+  return FALSE;
+#else
+  char buf[1024];
+  ssize_t res= recv(vio->sd, &buf, sizeof(buf), MSG_PEEK);
+  if (res < 0)
+    return TRUE;
+  *bytes= res;
+  return FALSE;
+#endif
+}
+
 void vio_timeout(Vio *vio, uint which, uint timeout)
 {
 #if defined(SO_SNDTIMEO) && defined(SO_RCVTIMEO)