ndb - Embryo of overload protection

  Add method to query free send buffer size


ndb/include/transporter/TransporterRegistry.hpp:
  Allow accessing free send buffer size
ndb/src/common/transporter/SCI_Transporter.cpp:
  Allow accessing free send buffer size
ndb/src/common/transporter/SCI_Transporter.hpp:
  Allow accessing free send buffer size
ndb/src/common/transporter/SHM_Buffer.hpp:
  Allow accessing free send buffer size
ndb/src/common/transporter/SHM_Transporter.cpp:
  Allow accessing free send buffer size
ndb/src/common/transporter/SHM_Transporter.hpp:
  Allow accessing free send buffer size
ndb/src/common/transporter/SendBuffer.cpp:
  Allow accessing free send buffer size
ndb/src/common/transporter/SendBuffer.hpp:
  Allow accessing free send buffer size
ndb/src/common/transporter/TCP_Transporter.cpp:
  Allow accessing free send buffer size
ndb/src/common/transporter/TCP_Transporter.hpp:
  Allow accessing free send buffer size
ndb/src/common/transporter/Transporter.hpp:
  Allow accessing free send buffer size
ndb/src/common/transporter/TransporterRegistry.cpp:
  Allow accessing free send buffer size
ndb/src/mgmsrv/ConfigInfo.cpp:
  Increse min values for SHM and TCP transport send buffer size

ndb/include/transporter/TransporterRegistry.hpp

@@ -179,6 +179,13 @@ public:
   bool createTransporter(struct SHM_TransporterConfiguration * config);
   bool createTransporter(struct OSE_TransporterConfiguration * config);
 
+  /**
+   * Get free buffer space
+   *
+   *   Get #free bytes in send buffer for <em>node</node>
+   */
+  Uint32 get_free_buffer(Uint32 node) const ;
+  
   /**
    * prepareSend
    *

ndb/src/common/transporter/SCI_Transporter.cpp

@@ -1023,7 +1023,8 @@ SCI_Transporter::initSCI() {
   DBUG_RETURN(true);
 } 
  
- 
- 
- 
- 
+Uint32
+SCI_Transporter::get_free_buffer() const
+{
+  return (m_TargetSegm[m_ActiveAdapterId].writer)->get_free_buffer();
+}

ndb/src/common/transporter/SCI_Transporter.hpp

@@ -134,6 +134,7 @@ public:
    */ 
   bool getConnectionStatus(); 
 
+  virtual Uint32 get_free_buffer() const;   
 private: 
   SCI_Transporter(TransporterRegistry &t_reg,
                   const char *local_host,

ndb/src/common/transporter/SHM_Buffer.hpp

@@ -157,6 +157,7 @@ public:
 
   inline Uint32 getWriteIndex() const { return m_writeIndex;}
   inline Uint32 getBufferSize() const { return m_bufferSize;}
+  inline Uint32 get_free_buffer() const;
   
   inline void copyIndexes(SHM_Writer * standbyWriter);
 
@@ -213,4 +214,20 @@ SHM_Writer::updateWritePtr(Uint32 sz){
   * m_sharedWriteIndex = tWriteIndex;
 }
 
+inline
+Uint32
+SHM_Writer::get_free_buffer() const
+{
+  Uint32 tReadIndex  = * m_sharedReadIndex;
+  Uint32 tWriteIndex = m_writeIndex;
+  
+  Uint32 free;
+  if(tReadIndex <= tWriteIndex){
+    free = m_bufferSize + tReadIndex - tWriteIndex;
+  } else {
+    free = tReadIndex - tWriteIndex;
+  }
+  return free;
+}
+ 
 #endif

ndb/src/common/transporter/SHM_Transporter.cpp

@@ -362,3 +362,9 @@ SHM_Transporter::doSend()
     kill(m_remote_pid, g_ndb_shm_signum);
   }
 }
+
+Uint32
+SHM_Transporter::get_free_buffer() const 
+{
+  return writer->get_free_buffer();
+}

ndb/src/common/transporter/SHM_Transporter.hpp

@@ -138,6 +138,8 @@ protected:
   Uint32 m_last_signal;
   Uint32 m_signal_threshold;
 
+  virtual Uint32 get_free_buffer() const;
+  
 private:
   bool _shmSegCreated;
   bool _attached;

ndb/src/common/transporter/SendBuffer.cpp

@@ -60,7 +60,7 @@ SendBuffer::bufferSize() {
 }
 
 Uint32
-SendBuffer::bufferSizeRemaining() {
+SendBuffer::bufferSizeRemaining() const {
   return (sizeOfBuffer - dataSize);
 }
 

ndb/src/common/transporter/SendBuffer.hpp

@@ -51,7 +51,7 @@ public:
   bool initBuffer(Uint32 aRemoteNodeId);  
 
   // Number of bytes remaining in the buffer
-  Uint32 bufferSizeRemaining();
+  Uint32 bufferSizeRemaining() const;
 
   // Number of bytes of data in the buffer 
   int bufferSize(); 

ndb/src/common/transporter/TCP_Transporter.cpp

@@ -250,6 +250,11 @@ TCP_Transporter::sendIsPossible(struct timeval * timeout) {
 #endif
 }
 
+Uint32
+TCP_Transporter::get_free_buffer() const 
+{
+  return m_sendBuffer.bufferSizeRemaining();
+}
 
 Uint32 *
 TCP_Transporter::getWritePtr(Uint32 lenBytes, Uint32 prio){

ndb/src/common/transporter/TCP_Transporter.hpp

@@ -99,6 +99,7 @@ private:
    */
   virtual void updateReceiveDataPtr(Uint32 bytesRead);
 
+  virtual Uint32 get_free_buffer() const;
 protected:
   /**
    * Setup client/server and perform connect/accept

ndb/src/common/transporter/Transporter.hpp

@@ -69,6 +69,8 @@ public:
    */
   NodeId getLocalNodeId() const;
 
+  virtual Uint32 get_free_buffer() const = 0;
+  
 protected:
   Transporter(TransporterRegistry &,
 	      TransporterType,

ndb/src/common/transporter/TransporterRegistry.cpp

@@ -523,6 +523,18 @@ TransporterRegistry::removeTransporter(NodeId nodeId) {
   theTransporters[nodeId] = NULL;        
 }
 
+Uint32
+TransporterRegistry::get_free_buffer(Uint32 node) const
+{
+  Transporter *t;
+  if(likely((t = theTransporters[node]) != 0))
+  {
+    return t->get_free_buffer();
+  }
+  return 0;
+}
+
+
 SendStatus
 TransporterRegistry::prepareSend(const SignalHeader * const signalHeader, 
 				 Uint8 prio,

ndb/src/mgmsrv/ConfigInfo.cpp

@@ -1656,7 +1656,7 @@ const ConfigInfo::ParamInfo ConfigInfo::m_ParamInfo[] = {
     false,
     ConfigInfo::CI_INT,
     "256K",
-    "16K",
+    "64K",
     STR_VALUE(MAX_INT_RNIL) },
 
   {
@@ -1844,7 +1844,7 @@ const ConfigInfo::ParamInfo ConfigInfo::m_ParamInfo[] = {
     false,
     ConfigInfo::CI_INT,
     "1M",
-    "4K",
+    "64K",
     STR_VALUE(MAX_INT_RNIL) },
 
   {