bk commit into 5.1 tree (jonas:1.2615)

[jonas]

Below is the list of changes that have just been committed into a local
5.1 repository of jonas. When jonas does a push these changes will
be propagated to the main repository and, within 24 hours after the
push, to the public repository.
For information on how to access the public repository
see http://dev.mysql.com/doc/mysql/en/installing-source-tree.html

ChangeSet@stripped, 2007-10-13 11:06:18+02:00, jonas@stripped +7 -0
  Merge perch.ndb.mysql.com:/home/jonas/src/51-telco
  into  perch.ndb.mysql.com:/home/jonas/src/51-trp
  MERGE: 1.2482.8.63

  storage/ndb/include/transporter/TransporterRegistry.hpp@stripped, 2007-10-13 11:01:17+02:00,
jonas@stripped +0 -0
    Auto merged
    MERGE: 1.25.1.3

  storage/ndb/src/common/transporter/TCP_Transporter.cpp@stripped, 2007-10-13 11:06:15+02:00,
jonas@stripped +1 -4
    merge
    MERGE: 1.17.1.2

  storage/ndb/src/common/transporter/TCP_Transporter.hpp@stripped, 2007-10-13 11:06:15+02:00,
jonas@stripped +3 -1
    merge
    MERGE: 1.14.2.1

  storage/ndb/src/common/transporter/Transporter.hpp@stripped, 2007-10-13 11:01:17+02:00,
jonas@stripped +0 -0
    Auto merged
    MERGE: 1.17.1.2

  storage/ndb/src/common/transporter/TransporterRegistry.cpp@stripped, 2007-10-13
11:01:17+02:00, jonas@stripped +0 -0
    Auto merged
    MERGE: 1.76.1.1

  storage/ndb/src/kernel/vm/Configuration.cpp@stripped, 2007-10-13 11:01:17+02:00,
jonas@stripped +0 -0
    Auto merged
    MERGE: 1.59.1.1

  storage/ndb/src/ndbapi/TransporterFacade.cpp@stripped, 2007-10-13 11:01:17+02:00,
jonas@stripped +0 -0
    Auto merged
    MERGE: 1.68.1.3

# This is a BitKeeper patch.  What follows are the unified diffs for the
# set of deltas contained in the patch.  The rest of the patch, the part
# that BitKeeper cares about, is below these diffs.
# User:	jonas
# Host:	perch.ndb.mysql.com
# Root:	/home/jonas/src/51-trp/RESYNC

--- 1.26/storage/ndb/include/transporter/TransporterRegistry.hpp	2007-10-13 11:06:24
+02:00
+++ 1.27/storage/ndb/include/transporter/TransporterRegistry.hpp	2007-10-13 11:06:24
+02:00
@@ -35,6 +35,8 @@
 
 #include <mgmapi/mgmapi.h>
 
+#include <NodeBitmask.hpp>
+
 // A transporter is always in an IOState.
 // NoHalt is used initially and as long as it is no restrictions on
 // sending or receiving.
@@ -208,6 +210,13 @@
    *   Get #free bytes in send buffer for <em>node</node>
    */
   Uint32 get_free_buffer(Uint32 node) const ;
+
+  /**
+   * Set or clear overloaded bit.
+   * Query if any overloaded bit is set.
+   */
+  void set_status_overloaded(Uint32 nodeId, bool val);
+  const NodeBitmask& get_status_overloaded() const;
   
   /**
    * prepareSend
@@ -325,6 +334,11 @@
    */
   PerformState* performStates;
   IOState*      ioStates;
+
+  /**
+   * Overloaded bits, for fast check.
+   */
+  NodeBitmask m_status_overloaded;
  
   /**
    * Unpack signal data
@@ -360,5 +374,18 @@
   int m_shm_own_pid;
   int m_transp_count;
 };
+
+inline void
+TransporterRegistry::set_status_overloaded(Uint32 nodeId, bool val)
+{
+  assert(nodeId < MAX_NODES);
+  m_status_overloaded.set(nodeId, val);
+}
+
+inline const NodeBitmask&
+TransporterRegistry::get_status_overloaded() const
+{
+  return m_status_overloaded;
+}
 
 #endif // Define of TransporterRegistry_H

--- 1.18/storage/ndb/src/common/transporter/TCP_Transporter.cpp	2007-10-13 11:06:24 +02:00
+++ 1.19/storage/ndb/src/common/transporter/TCP_Transporter.cpp	2007-10-13 11:06:24 +02:00
@@ -101,6 +101,7 @@
   m_send_bytes = 0;
   m_send_buffers = 0;
   m_send_buffer_pos = 0;
+  overloadedPct = 80; // make configurable in next patch
 }
 
 TCP_Transporter::~TCP_Transporter() {

--- 1.17/storage/ndb/src/common/transporter/TCP_Transporter.hpp	2007-10-13 11:06:24 +02:00
+++ 1.18/storage/ndb/src/common/transporter/TCP_Transporter.hpp	2007-10-13 11:06:24 +02:00
@@ -113,10 +113,9 @@
   virtual void disconnectImpl();
   
 private:
-  /**
-   * Send buffers
-   */
-  SendBuffer m_sendBuffer;
+  Uint32 m_send_bytes;
+  Uint32 m_send_buffer_pos;
+  struct iovec * m_send_buffers;
   
   // Sending/Receiving socket used by both client and server
   NDB_SOCKET_TYPE theSocket;   
@@ -182,16 +181,18 @@
 inline
 bool
 TCP_Transporter::hasDataToSend() const {
-  return m_sendBuffer.dataSize > 0;
+  return m_send_bytes > 0;
 }
 
 inline
 void
 TCP_Transporter::update_status_overloaded() {
+#if 0
   const Uint32 used = m_sendBuffer.dataSize;
   const Uint32 total = m_sendBuffer.sizeOfBuffer;
   const bool val = (100 * used > overloadedPct * total);
   set_status_overloaded(val);
+#endif
 }
 
 inline

--- 1.18/storage/ndb/src/common/transporter/Transporter.hpp	2007-10-13 11:06:24 +02:00
+++ 1.19/storage/ndb/src/common/transporter/Transporter.hpp	2007-10-13 11:06:24 +02:00
@@ -86,6 +86,10 @@
   };
 
   virtual Uint32 get_free_buffer() const = 0;
+
+  void set_status_overloaded(bool val) {
+    m_transporter_registry.set_status_overloaded(remoteNodeId, val);
+  }
   
 protected:
   Transporter(TransporterRegistry &,

--- 1.79/storage/ndb/src/common/transporter/TransporterRegistry.cpp	2007-10-13 11:06:24
+02:00
+++ 1.80/storage/ndb/src/common/transporter/TransporterRegistry.cpp	2007-10-13 11:06:24
+02:00
@@ -99,7 +99,8 @@
   nTCPTransporters = 0;
   nSCITransporters = 0;
   nSHMTransporters = 0;
-  
+  m_send_buffer_size = 0;
+
   // Initialize the transporter arrays
   for (unsigned i=0; i<maxTransporters; i++) {
     theTCPTransporters[i] = NULL;
@@ -270,7 +271,7 @@
 bool
 TransporterRegistry::createTCPTransporter(TransporterConfiguration *config) {
 #ifdef NDB_TCP_TRANSPORTER
-
+  
   if(!nodeIdSpecified){
     init(config->localNodeId);
   }
@@ -280,23 +281,9 @@
   
   if(theTransporters[config->remoteNodeId] != NULL)
     return false;
-   
+  
   TCP_Transporter * t = new TCP_Transporter(*this, config);
-
-#if 0
-					    config->tcp.sendBufferSize,
-					    config->tcp.maxReceiveSize,
-					    config->localHostName,
-					    config->remoteHostName,
-					    config->s_port,
-					    config->isMgmConnection,
-					    localNodeId,
-					    config->remoteNodeId,
-					    config->serverNodeId,
-					    config->checksum,
-					    config->signalId);
-#endif
-
+  
   if (t == NULL) 
     return false;
   else if (!t->initTransporter()) {
@@ -311,7 +298,8 @@
   performStates[t->getRemoteNodeId()]       = DISCONNECTED;
   nTransporters++;
   nTCPTransporters++;
-
+  
+  m_send_buffer_size += config->tcp.sendBufferSize;
   return true;
 #else
   return false;
@@ -550,7 +538,7 @@
 	  reportError(callbackObj, nodeId, TE_SEND_BUFFER_FULL);
 	  return SEND_OK;
 	}
-	
+
 	WARNING("Signal to " << nodeId << " lost(buffer)");
 	reportError(callbackObj, nodeId, TE_SIGNAL_LOST_SEND_BUFFER_FULL);
 	return SEND_BUFFER_FULL;
@@ -643,6 +631,85 @@
     
     return SEND_BLOCKED;
   }
+}
+
+
+/**
+ * Transporter(s) that does not support (nativly)
+ *   the thr_client_r interface
+ *   uses this method...
+ */
+SendStatus
+TransporterRegistry::prepareSend(Transporter* t, 
+				 struct iovec* buf, 
+				 Uint32 lenBytes)
+{
+  Uint32 i;
+  Uint32 cnt = (lenBytes + 32767) / 32768;
+  SendStatus ss = SEND_OK;
+  
+  if(t->isConnected())
+  {
+    Uint32 * insertPtr = t->getWritePtr(lenBytes, 0);
+    if(insertPtr != 0)
+    {
+      for (i = 0; i<cnt; i++)
+      {
+	Uint32 sz = buf[i].iov_len;
+	assert((sz & 3) == 0);
+	sz /= 4;
+	memcpy(insertPtr, buf[i].iov_base, sz);
+	insertPtr += sz;
+      }
+      t->updateWritePtr(lenBytes, 0);
+      goto done;
+    }
+	
+    /**
+     * @note: on linux/i386 the granularity is 10ms
+     *        so sleepTime = 2 generates a 10 ms sleep.
+     */
+    int sleepTime = 2;
+    for(Uint32 loop = 0; loop<50; loop++){
+      if((nSHMTransporters+nSCITransporters) == 0)
+	NdbSleep_MilliSleep(sleepTime); 
+      insertPtr = t->getWritePtr(lenBytes, 0);
+      if(insertPtr != 0)
+      {
+	for (i = 0; i<cnt; i++)
+	{
+	  Uint32 sz = buf[i].iov_len;
+	  assert((sz & 3) == 0);
+	  sz /= 4;
+	  memcpy(insertPtr, buf[i].iov_base, sz);
+	  insertPtr += sz;
+	}
+	t->updateWritePtr(lenBytes, 0);
+	break;
+      }
+    }
+
+    Uint32 nodeId = t->remoteNodeId;
+    if(insertPtr != 0)
+    {
+      /**
+       * Send buffer full, but resend works
+       */
+      reportError(callbackObj, nodeId, TE_SEND_BUFFER_FULL);
+      goto done;
+    }
+    
+    WARNING("Signal to " << nodeId << " lost(buffer)");
+    reportError(callbackObj, nodeId, TE_SIGNAL_LOST_SEND_BUFFER_FULL);
+    ss = SEND_BUFFER_FULL;
+    goto done;
+  } 
+
+  ss = SEND_DISCONNECTED;
+
+done:
+  release_buffers(buf, cnt);
+  return ss;
 }
 
 void

--- 1.62/storage/ndb/src/kernel/vm/Configuration.cpp	2007-10-13 11:06:24 +02:00
+++ 1.63/storage/ndb/src/kernel/vm/Configuration.cpp	2007-10-13 11:06:24 +02:00
@@ -422,6 +422,13 @@
     }
   }
 
+  if (globalTransporterRegistry.alloc_buffers() != true)
+  {
+    ERROR_SET(fatal, NDBD_EXIT_MEMALLOC, 
+	      "Failed to alloc send buffers", 
+	      "Failed to alloc send buffers");
+  }
+
   /**
    * Setup cluster configuration data
    */

--- 1.69/storage/ndb/src/ndbapi/TransporterFacade.cpp	2007-10-13 11:06:24 +02:00
+++ 1.70/storage/ndb/src/ndbapi/TransporterFacade.cpp	2007-10-13 11:06:24 +02:00
@@ -60,6 +60,21 @@
  * Call back functions
  *****************************************************************************/
 
+void* ndb_thread_add_thread_id(void *param)
+{
+  return NULL;
+}
+
+void *ndb_thread_remove_thread_id(void *param)
+{
+  return NULL;
+}
+
+void ndb_thread_fill_thread_object(void *param, uint *len, bool server)
+{
+  *len = 0;
+}
+
 void
 reportError(void * callbackObj, NodeId nodeId,
 	    TransporterError errorCode, const char *info)
@@ -175,11 +190,25 @@
   }
   return false;
 }
-#ifdef TRACE_APIREGREQ
-#define TRACE_GSN(gsn) true
-#else
-#define TRACE_GSN(gsn) (gsn != GSN_API_REGREQ && gsn != GSN_API_REGCONF)
+inline
+bool
+TRACE_GSN(Uint32 gsn)
+{
+  switch(gsn){
+#ifndef TRACE_APIREGREQ
+  case GSN_API_REGREQ:
+  case GSN_API_REGCONF:
+    return false;
 #endif
+#if 0
+  case GSN_SUB_GCP_COMPLETE_REP:
+  case GSN_SUB_GCP_COMPLETE_ACK:
+    return true;
+#endif
+  default:
+    return true;
+  }
+}
 #endif
 
 /**