bzr push into mysql-5.1-telco-7.0 branch (magnus.blaudd:3674 to 3676)

[Magnus Blåudd]

 3676 Magnus Blåudd	2010-08-11 [merge]
      Merge 6.3 -> 7.0

 3675 Magnus Blåudd	2010-08-11 [merge]
      Merg

    modified:
      storage/ndb/include/portlib/ndb_socket_poller.h
      storage/ndb/src/common/transporter/TCP_Transporter.cpp
      storage/ndb/src/common/transporter/TransporterRegistry.cpp
      storage/ndb/src/common/util/OutputStream.cpp
      storage/ndb/src/common/util/socket_io.cpp
      storage/ndb/test/tools/log_listner.cpp
      storage/ndb/tools/waiter.cpp
 3674 Jonas Oreland	2010-08-11
      ndb - fix compiler warning

    modified:
      storage/ndb/src/kernel/vm/Ndbinfo.cpp
=== modified file 'storage/ndb/include/portlib/ndb_socket_poller.h'
--- a/storage/ndb/include/portlib/ndb_socket_poller.h	2010-06-02 15:00:06 +0000
+++ b/storage/ndb/include/portlib/ndb_socket_poller.h	2010-08-11 10:21:17 +0000
@@ -18,6 +18,8 @@
 #ifndef NDB_SOCKET_POLLER_H
 #define NDB_SOCKET_POLLER_H
 
+#include <portlib/NdbTick.h>
+
 /*
   Portability layer used for waiting on socket events
 */
@@ -243,7 +245,11 @@ public:
 #endif
   }
 
-  int poll(int timeout) {
+  /*
+    Wait for event(s) on socket(s) without retry of interrupted wait
+  */
+  int poll_unsafe(int timeout)
+  {
 #ifdef HAVE_POLL
     return ::poll(m_pfds, m_count, timeout);
 #else
@@ -265,9 +271,56 @@ public:
                   timeout == -1 ? NULL : &tv);
 #endif
   }
+
+  /*
+    Wait for event(s) on socket(s), retry interrupted wait
+    if there is still time left
+  */
+  int poll(int timeout)
+  {
+    do
+    {
+      const NDB_TICKS start = NdbTick_CurrentMillisecond();
+
+      const int res = poll_unsafe(timeout);
+      if (likely(res >= 0))
+        return res; // Default return path
+
+      const int error = my_socket_errno();
+      if (res == -1 &&
+          (error == EINTR || error == EAGAIN))
+      {
+        // Retry if any time left of timeout
+
+        // Subtract function call time from remaining timeout
+        timeout -= (int)(NdbTick_CurrentMillisecond() - start);
+
+        if (timeout <= 0)
+          return 0; // Timeout occured
+
+        //fprintf(stderr, "Got interrupted, retrying... timeout left: %d\n",
+        //        timeout_millis);
+
+        continue; // Retry interrupted poll
+      }
+
+      // Unhandled error code, return it
+      return res;
+
+    } while (true);
+
+    abort(); // Never reached
+  }
+
 };
 
 
+/*
+  ndb_poll
+  - Utility function for waiting on events on one socket
+    with retry of interrupted wait
+*/
+
 static inline
 int
 ndb_poll(ndb_socket_t sock,

=== modified file 'storage/ndb/src/common/transporter/TCP_Transporter.cpp'
--- a/storage/ndb/src/common/transporter/TCP_Transporter.cpp	2010-06-01 12:19:50 +0000
+++ b/storage/ndb/src/common/transporter/TCP_Transporter.cpp	2010-08-11 10:21:17 +0000
@@ -255,10 +255,15 @@ bool TCP_Transporter::setSocketNonBlocki
 bool
 TCP_Transporter::send_is_possible(int timeout_millisec) const
 {
+  ndb_socket_poller poller;
+
   if (!my_socket_valid(theSocket))
     return false;
 
-  if (ndb_poll(theSocket, false, true, false, timeout_millisec) <= 0)
+  poller.clear();
+  poller.add(theSocket, false, true, false);
+
+  if (poller.poll_unsafe(timeout_millisec) <= 0)
     return false; // Timeout or error occured
 
   return true;

=== modified file 'storage/ndb/src/common/transporter/TransporterRegistry.cpp'
--- a/storage/ndb/src/common/transporter/TransporterRegistry.cpp	2010-06-01 12:23:20 +0000
+++ b/storage/ndb/src/common/transporter/TransporterRegistry.cpp	2010-08-11 10:21:17 +0000
@@ -1052,7 +1052,7 @@ TransporterRegistry::poll_TCP(Uint32 tim
     hasdata |= t->hasReceiveData();
   }
 
-  tcpReadSelectReply = m_socket_poller.poll(hasdata ? 0 : timeOutMillis);
+  tcpReadSelectReply = m_socket_poller.poll_unsafe(hasdata ? 0 : timeOutMillis);
 
   return tcpReadSelectReply || hasdata;
 }

=== modified file 'storage/ndb/src/common/util/OutputStream.cpp'
--- a/storage/ndb/src/common/util/OutputStream.cpp	2010-03-04 12:59:58 +0000
+++ b/storage/ndb/src/common/util/OutputStream.cpp	2010-08-11 10:21:17 +0000
@@ -174,7 +174,7 @@ BufferedSockOutputStream::println(const 
 }
 
 void BufferedSockOutputStream::flush(){
-  int elapsed;
+  int elapsed = 0;
   if (write_socket(m_socket, m_timeout_ms, &elapsed,
                    (const char*)m_buffer.get_data(), m_buffer.length()) != 0)
   {

=== modified file 'storage/ndb/src/common/util/socket_io.cpp'
--- a/storage/ndb/src/common/util/socket_io.cpp	2010-06-01 12:19:50 +0000
+++ b/storage/ndb/src/common/util/socket_io.cpp	2010-08-11 10:21:17 +0000
@@ -18,10 +18,36 @@
 
 #include <ndb_global.h>
 
-#include <NdbTCP.h>
-#include <socket_io.h>
-#include <NdbOut.hpp>
-#include <NdbTick.h>
+#include <portlib/NdbTCP.h>
+#include <portlib/NdbTick.h>
+#include <util/socket_io.h>
+#include <util/BaseString.hpp>
+
+static inline
+int
+poll_socket(ndb_socket_t socket, bool read, bool write,
+            int timeout_millis, int* total_elapsed_millis)
+{
+  const NDB_TICKS start = NdbTick_CurrentMillisecond();
+
+  timeout_millis -= *total_elapsed_millis;
+
+  if (timeout_millis <= 0)
+    return 0; // Timeout occured
+
+  const int res =
+    ndb_poll(socket, read, write, false, timeout_millis);
+
+  // Calculate elapsed time in this function
+  const int elapsed_millis = (int)(NdbTick_CurrentMillisecond() - start);
+  assert(elapsed_millis >= 0);
+
+  // Update the total elapsed time
+  *total_elapsed_millis += elapsed_millis;
+
+  return res;
+}
+
 
 extern "C"
 int
@@ -30,13 +56,11 @@ read_socket(NDB_SOCKET_TYPE socket, int 
   if(buflen < 1)
     return 0;
 
-  const int selectRes = ndb_poll(socket, true, false, false, timeout_millis);
-  if(selectRes == 0)
-    return 0;
-
-  if(selectRes == -1){
-    return -1;
-  }
+  int elapsed_millis = 0;
+  const int res = poll_socket(socket, true, false,
+                              timeout_millis, &elapsed_millis);
+  if (res <= 0)
+    return res;
 
   return my_recv(socket, &buf[0], buflen, 0);
 }
@@ -50,19 +74,15 @@ readln_socket(NDB_SOCKET_TYPE socket, in
 
   if(mutex)
     NdbMutex_Unlock(mutex);
-  Uint64 tick= NdbTick_CurrentMillisecond();
-  int selectRes = ndb_poll(socket, true, false, false, timeout_millis);
-  *time= (int)(NdbTick_CurrentMillisecond() - tick);
+
+  const int res = poll_socket(socket, true, false,
+                              timeout_millis, time);
+
   if(mutex)
     NdbMutex_Lock(mutex);
 
-  if(selectRes == 0){
-    return 0;
-  }
-
-  if(selectRes == -1){
-    return -1;
-  }
+  if (res <= 0)
+    return res;
 
   char* ptr = buf;
   int len = buflen;
@@ -125,14 +145,13 @@ readln_socket(NDB_SOCKET_TYPE socket, in
       }
     }
 
-    tick= NdbTick_CurrentMillisecond();
-    int selectRes = ndb_poll(socket, true, false, false,
-                             timeout_millis - *time);
-    *time= (int)(NdbTick_CurrentMillisecond() - tick);
-
-    if(selectRes != 1){
+    if (poll_socket(socket, true, false, timeout_millis, time) != 1)
+    {
+      // Read some bytes but didn't find newline before all time was
+      // used up => return error
       return -1;
     }
+
   } while (len > 0);
   
   return -1;
@@ -143,13 +162,8 @@ int
 write_socket(NDB_SOCKET_TYPE socket, int timeout_millis, int *time,
 	     const char buf[], int len){
 
-  Uint64 tick= NdbTick_CurrentMillisecond();
-  const int selectRes = ndb_poll(socket, false, true, false, timeout_millis);
-  *time= (int)(NdbTick_CurrentMillisecond() - tick);
-
-  if(selectRes != 1){
+  if (poll_socket(socket, false, true, timeout_millis, time) != 1)
     return -1;
-  }
 
   const char * tmp = &buf[0];
   while(len > 0){
@@ -163,14 +177,8 @@ write_socket(NDB_SOCKET_TYPE socket, int
     if(len == 0)
       break;
 
-    Uint64 tick= NdbTick_CurrentMillisecond();
-    const int selectRes2 = ndb_poll(socket, false, true, false,
-                                    timeout_millis - *time);
-    *time= (int)(NdbTick_CurrentMillisecond() - tick);
-
-    if(selectRes2 != 1){
+    if (poll_socket(socket, false, true, timeout_millis, time) != 1)
       return -1;
-    }
   }
   
   return 0;

=== modified file 'storage/ndb/test/tools/log_listner.cpp'
--- a/storage/ndb/test/tools/log_listner.cpp	2010-05-04 14:34:54 +0000
+++ b/storage/ndb/test/tools/log_listner.cpp	2010-08-11 10:21:17 +0000
@@ -20,6 +20,11 @@ int filter[] = { 15, NDB_MGM_EVENT_CATEG
 		 15, NDB_MGM_EVENT_CATEGORY_CONGESTION,
 		 0 };
 
+extern "C"
+void catch_signal(int signum)
+{
+}
+
 int 
 main(int argc, char** argv)
 {
@@ -30,6 +35,13 @@ main(int argc, char** argv)
 #ifndef DBUG_OFF
   opt_debug= "d:t:O,/tmp/eventlog.trace";
 #endif
+
+#ifndef _WIN32
+  // Catching signal to allow testing of EINTR safeness
+  // with "while killall -USR1 eventlog; do true; done"
+  signal(SIGUSR1, catch_signal);
+#endif
+
   if ((ho_error=handle_options(&argc, &argv, my_long_options, 
 			       ndb_std_get_one_option)))
     return NDBT_ProgramExit(NDBT_WRONGARGS);

=== modified file 'storage/ndb/tools/waiter.cpp'
--- a/storage/ndb/tools/waiter.cpp	2010-05-04 14:34:54 +0000
+++ b/storage/ndb/tools/waiter.cpp	2010-08-11 10:21:17 +0000
@@ -79,6 +79,11 @@ static void usage()
   ndb_usage(short_usage_sub, load_default_groups, my_long_options);
 }
 
+extern "C"
+void catch_signal(int signum)
+{
+}
+
 int main(int argc, char** argv){
   NDB_INIT(argv[0]);
   ndb_opt_set_usage_funcs(NULL, short_usage_sub, usage);
@@ -88,6 +93,12 @@ int main(int argc, char** argv){
   opt_debug= "d:t:O,/tmp/ndb_waiter.trace";
 #endif
 
+#ifndef _WIN32
+  // Catching signal to allow testing of EINTR safeness
+  // with "while killall -USR1 ndbwaiter; do true; done"
+  signal(SIGUSR1, catch_signal);
+#endif
+
   if (handle_options(&argc, &argv, my_long_options,
                      ndb_std_get_one_option))
     return NDBT_ProgramExit(NDBT_WRONGARGS);

Attachment: [text/bzr-bundle] bzr/[e-mail]