Merge perch.ndb.mysql.com:/home/jonas/src/50-work

into  perch.ndb.mysql.com:/home/jonas/src/mysql-5.0-ndb

ndb/src/common/transporter/Packer.cpp

@@ -20,7 +20,12 @@
 #include <TransporterCallback.hpp>
 #include <RefConvert.hpp>
 
+#ifdef ERROR_INSERT
+Uint32 MAX_RECEIVED_SIGNALS = 1024;
+#else
 #define MAX_RECEIVED_SIGNALS 1024
+#endif
+
 Uint32
 TransporterRegistry::unpack(Uint32 * readPtr,
 			    Uint32 sizeOfData,

ndb/src/common/transporter/TCP_Transporter.hpp

@@ -102,6 +102,10 @@ private:
   virtual void updateReceiveDataPtr(Uint32 bytesRead);
 
   virtual Uint32 get_free_buffer() const;
+
+  inline bool hasReceiveData () const {
+    return receiveBuffer.sizeOfData > 0;
+  }
 protected:
   /**
    * Setup client/server and perform connect/accept

ndb/src/common/transporter/TransporterRegistry.cpp

@@ -841,6 +841,7 @@ TransporterRegistry::poll_OSE(Uint32 timeOutMillis)
 Uint32 
 TransporterRegistry::poll_TCP(Uint32 timeOutMillis)
 {
+  bool hasdata = false;
   if (false && nTCPTransporters == 0)
   {
     tcpReadSelectReply = 0;
@@ -885,6 +886,7 @@ TransporterRegistry::poll_TCP(Uint32 timeOutMillis)
       // Put the connected transporters in the socket read-set 
       FD_SET(socket, &tcpReadset);
     }
+    hasdata |= t->hasReceiveData();
   }
   
   // The highest socket value plus one
@@ -901,7 +903,7 @@ TransporterRegistry::poll_TCP(Uint32 timeOutMillis)
   }
 #endif
   
-  return tcpReadSelectReply;
+  return tcpReadSelectReply || hasdata;
 }
 #endif
 
@@ -937,8 +939,6 @@ TransporterRegistry::performReceive()
 #endif
 
 #ifdef NDB_TCP_TRANSPORTER
-  if(tcpReadSelectReply > 0)
-  {
   for (int i=0; i<nTCPTransporters; i++) 
   {
     checkJobBuffer();
@@ -946,21 +946,23 @@ TransporterRegistry::performReceive()
     const NodeId nodeId = t->getRemoteNodeId();
     const NDB_SOCKET_TYPE socket    = t->getSocket();
     if(is_connected(nodeId)){
-	if(t->isConnected() && FD_ISSET(socket, &tcpReadset)) 
+      if(t->isConnected())
       {
-	  const int receiveSize = t->doReceive();
-	  if(receiveSize > 0)
+        if (FD_ISSET(socket, &tcpReadset))
+	{
+	  t->doReceive();
+        }
+
+        if (t->hasReceiveData())
         {
           Uint32 * ptr;
           Uint32 sz = t->getReceiveData(&ptr);
-	    transporter_recv_from(callbackObj, nodeId);
           Uint32 szUsed = unpack(ptr, sz, nodeId, ioStates[nodeId]);
           t->updateReceiveDataPtr(szUsed);
 	}
       } 
     }
   }
-  }
 #endif
   
 #ifdef NDB_SCI_TRANSPORTER

ndb/src/kernel/blocks/cmvmi/Cmvmi.cpp

@@ -132,6 +132,7 @@ Cmvmi::~Cmvmi()
 
 #ifdef ERROR_INSERT
 NodeBitmask c_error_9000_nodes_mask;
+extern Uint32 MAX_RECEIVED_SIGNALS;
 #endif
 
 void Cmvmi::execNDB_TAMPER(Signal* signal) 
@@ -161,6 +162,22 @@ void Cmvmi::execNDB_TAMPER(Signal* signal)
     kill(getpid(), SIGABRT);
   }
 #endif
+
+#ifdef ERROR_INSERT
+  if (signal->theData[0] == 9003)
+  {
+    if (MAX_RECEIVED_SIGNALS < 1024)
+    {
+      MAX_RECEIVED_SIGNALS = 1024;
+    }
+    else
+    {
+      MAX_RECEIVED_SIGNALS = rand() % 128;
+    }
+    ndbout_c("MAX_RECEIVED_SIGNALS: %d", MAX_RECEIVED_SIGNALS);
+    CLEAR_ERROR_INSERT_VALUE;
+  }
+#endif
 }//execNDB_TAMPER()
 
 void Cmvmi::execSET_LOGLEVELORD(Signal* signal) 

ndb/test/ndbapi/testNdbApi.cpp

@@ -1306,6 +1306,36 @@ int runTestExecuteAsynch(NDBT_Context* ctx, NDBT_Step* step){
 
 template class Vector<NdbScanOperation*>;
 
+int 
+runBug28443(NDBT_Context* ctx, NDBT_Step* step)
+{
+  int result = NDBT_OK;
+  int records = ctx->getNumRecords();
+  
+  NdbRestarter restarter;
+
+  restarter.insertErrorInAllNodes(9003);
+
+  for (Uint32 i = 0; i<ctx->getNumLoops(); i++)
+  {
+    HugoTransactions hugoTrans(*ctx->getTab());
+    if (hugoTrans.loadTable(GETNDB(step), records, 2048) != 0)
+    {
+      result = NDBT_FAILED;
+      goto done;
+    }
+    if (runClearTable(ctx, step) != 0)
+    {
+      result = NDBT_FAILED;
+      goto done;
+    }
+  }
+  
+done:
+  restarter.insertErrorInAllNodes(9003);
+
+  return result;
+}
 
 NDBT_TESTSUITE(testNdbApi);
 TESTCASE("MaxNdb", 
@@ -1396,6 +1426,10 @@ TESTCASE("ExecuteAsynch",
 	 "Check that executeAsync() works (BUG#27495)\n"){ 
   INITIALIZER(runTestExecuteAsynch);
 }
+TESTCASE("Bug28443", 
+	 ""){ 
+  INITIALIZER(runBug28443);
+}
 NDBT_TESTSUITE_END(testNdbApi);
 
 int main(int argc, const char** argv){

ndb/test/run-test/daily-basic-tests.txt

@@ -629,6 +629,10 @@ max-time: 500
 cmd: testNdbApi
 args: -n ExecuteAsynch T1
 
+max-time: 1000
+cmd: testNdbApi
+args: -n BugBug28443
+
 #max-time: 500
 #cmd: testInterpreter
 #args: T1