fixed compile of shm transporter

ndb/src/mgmsrv/Services.cpp:
  changed getsockname to getpeername

7 files changed, 130 insertions, 106 deletions

diff --git a/ndb/src/common/mgmcommon/ConfigInfo.cpp b/ndb/src/common/mgmcommon/ConfigInfo.cpp
index 60b77a4cd8c..f78fc086cac 100644
--- a/ndb/src/common/mgmcommon/ConfigInfo.cpp
+++ b/ndb/src/common/mgmcommon/ConfigInfo.cpp
@@ -104,6 +104,7 @@ ConfigInfo::m_SectionRules[] = {
   { "OSE",  fixHostname, "HostName2" },
 
   { "TCP",  fixPortNumber, 0 }, // has to come after fixHostName
+  { "SHM",  fixPortNumber, 0 }, // has to come after fixHostName
   //{ "SHM",  fixShmKey, 0 },
 
   /**
@@ -1559,14 +1560,14 @@ const ConfigInfo::ParamInfo ConfigInfo::m_ParamInfo[] = {
     0x7FFFFFFF },
 
   {
-    CFG_TCP_SERVER_PORT,
+    CFG_CONNECTION_SERVER_PORT,
     "PortNumber",
     "TCP",
     "Port used for this transporter",
     ConfigInfo::USED,
     false,
     ConfigInfo::INT,
-    NDB_BASE_PORT+2,
+    MANDATORY,
     0,
     0x7FFFFFFF },
 
@@ -1696,6 +1697,18 @@ const ConfigInfo::ParamInfo ConfigInfo::m_ParamInfo[] = {
     0x7FFFFFFF },
   
   {
+    CFG_CONNECTION_SERVER_PORT,
+    "PortNumber",
+    "SHM",
+    "Port used for this transporter",
+    ConfigInfo::USED,
+    false,
+    ConfigInfo::INT,
+    MANDATORY,
+    0,
+    0x7FFFFFFF },
+
+  {
     KEY_INTERNAL,
     "ProcessId1",
     "SHM",
diff --git a/ndb/src/common/mgmcommon/IPCConfig.cpp b/ndb/src/common/mgmcommon/IPCConfig.cpp
index 6dd8e7c1589..a8536bf4fa7 100644
--- a/ndb/src/common/mgmcommon/IPCConfig.cpp
+++ b/ndb/src/common/mgmcommon/IPCConfig.cpp
@@ -365,6 +365,16 @@ IPCConfig::configureTransporters(Uint32 nodeId,
     Uint32 type = ~0;
     if(iter.get(CFG_TYPE_OF_SECTION, &type)) continue;
 
+    Uint32 tmp_server_port= 0;
+    if(iter.get(CFG_CONNECTION_SERVER_PORT, &tmp_server_port)) break;
+    if (nodeId <= nodeId1 && nodeId <= nodeId2) {
+      if (server_port && server_port != tmp_server_port) {
+	ndbout << "internal error in config setup of server ports line= " << __LINE__ << endl;
+	exit(-1);
+      }
+      server_port= tmp_server_port;
+    }
+
     switch(type){
     case CONNECTION_TYPE_SHM:{
       SHM_TransporterConfiguration conf;
@@ -378,6 +388,8 @@ IPCConfig::configureTransporters(Uint32 nodeId,
       if(iter.get(CFG_SHM_KEY, &conf.shmKey)) break;
       if(iter.get(CFG_SHM_BUFFER_MEM, &conf.shmSize)) break;
       
+      conf.port= tmp_server_port;
+
       if(!tr.createTransporter(&conf)){
 	ndbout << "Failed to create SHM Transporter from: " 
 	       << conf.localNodeId << " to: " << conf.remoteNodeId << endl;
@@ -429,10 +441,11 @@ IPCConfig::configureTransporters(Uint32 nodeId,
       if(iter.get(CFG_TCP_HOSTNAME_1, &host1)) break;
       if(iter.get(CFG_TCP_HOSTNAME_2, &host2)) break;
       
-      if(iter.get(CFG_TCP_SERVER_PORT, &conf.port)) break;
       if(iter.get(CFG_TCP_SEND_BUFFER_SIZE, &conf.sendBufferSize)) break;
       if(iter.get(CFG_TCP_RECEIVE_BUFFER_SIZE, &conf.maxReceiveSize)) break;
       
+      conf.port= tmp_server_port;
+
       const char * proxy;
       if (!iter.get(CFG_TCP_PROXY, &proxy)) {
 	if (strlen(proxy) > 0 && nodeId2 == nodeId) {
@@ -441,14 +454,6 @@ IPCConfig::configureTransporters(Uint32 nodeId,
 	}
       }
       
-      if (nodeId <= nodeId1 && nodeId <= nodeId2) {
-	if (server_port && server_port != conf.port) {
-	  ndbout << "internal error in config setup of server ports line= " << __LINE__ << endl;
-	  exit(-1);
-	}
-	server_port= conf.port;
-      }
-
       conf.localNodeId    = nodeId;
       conf.remoteNodeId   = remoteNodeId;
       conf.localHostName  = (nodeId == nodeId1 ? host1 : host2);
diff --git a/ndb/src/common/mgmcommon/NdbConfig.c b/ndb/src/common/mgmcommon/NdbConfig.c
index 68ddb5fbec2..26eb6b5af34 100644
--- a/ndb/src/common/mgmcommon/NdbConfig.c
+++ b/ndb/src/common/mgmcommon/NdbConfig.c
@@ -24,12 +24,13 @@ NdbConfig_AllocHomePath(int _len)
   const char *path= NdbEnv_GetEnv("NDB_HOME", 0, 0);
   int len= _len;
   int path_len= 0;
+  char *buf;
 
   if (path)
     path_len= strlen(path);
 
   len+= path_len;
-  char *buf= malloc(len);
+  buf= malloc(len);
   if (path_len > 0)
     snprintf(buf, len, "%s%c", path, DIR_SEPARATOR);
   else
diff --git a/ndb/src/common/transporter/SHM_Transporter.cpp b/ndb/src/common/transporter/SHM_Transporter.cpp
index 7c673f93c22..96ca42f1829 100644
--- a/ndb/src/common/transporter/SHM_Transporter.cpp
+++ b/ndb/src/common/transporter/SHM_Transporter.cpp
@@ -29,20 +29,19 @@
 #endif
 
 
-SHM_Transporter::SHM_Transporter(NodeId lNodeId,
+SHM_Transporter::SHM_Transporter(TransporterRegistry &t_reg,
+				 const char *lHostName,
+				 const char *rHostName, 
+				 int r_port,
+				 NodeId lNodeId,
 				 NodeId rNodeId, 
-				 key_t _shmKey,
-				 Uint32 _shmSize,
 				 bool compression, 
 				 bool checksum, 
-				 bool signalId) :
-  Transporter(lNodeId,
-	      rNodeId,
-	      0,
-	      compression,
-	      checksum,
-	      signalId),
-  isServer(lNodeId < rNodeId),
+				 bool signalId,
+				 key_t _shmKey,
+				 Uint32 _shmSize) :
+  Transporter(t_reg, lHostName, rHostName, r_port, lNodeId, rNodeId,
+	      0, compression, checksum, signalId),
   shmKey(_shmKey),
   shmSize(_shmSize)
 {
@@ -68,16 +67,6 @@ SHM_Transporter::initTransporter(){
   return true;
 }
     
-bool
-SHM_Transporter::connectImpl(Uint32 timeOutMillis){
-  bool res;
-  if(isServer)
-    res = connectServer(timeOutMillis);
-  else
-    res = connectClient(timeOutMillis);
-  return res;
-}
-
 void
 SHM_Transporter::setupBuffers(){
   Uint32 sharedSize = 0;
diff --git a/ndb/src/common/transporter/SHM_Transporter.hpp b/ndb/src/common/transporter/SHM_Transporter.hpp
index da4566515e3..1fd91562028 100644
--- a/ndb/src/common/transporter/SHM_Transporter.hpp
+++ b/ndb/src/common/transporter/SHM_Transporter.hpp
@@ -32,13 +32,17 @@ typedef Uint32 key_t;
 class SHM_Transporter : public Transporter {
   friend class TransporterRegistry;
 public:
-  SHM_Transporter(NodeId lNodeId,
+  SHM_Transporter(TransporterRegistry &,
+		  const char *lHostName,
+		  const char *rHostName, 
+		  int r_port,
+		  NodeId lNodeId,
 		  NodeId rNodeId, 
-		  key_t shmKey,
-		  Uint32 shmSize,
 		  bool compression, 
 		  bool checksum, 
-		  bool signalId);
+		  bool signalId,
+		  key_t shmKey,
+		  Uint32 shmSize);
   
   /**
    * SHM destructor
@@ -75,14 +79,6 @@ protected:
   void disconnectImpl();
 
   /**
-   * Invokes the connectServer or connectClient.
-   * @param timeOutMillis - the timeout the connect thread waits before 
-   * retrying.
-   * @return True if connectImpl successful, otherwise false.
-   */
-  bool connectImpl(Uint32 timeOutMillis);
-  
-  /**
    * Blocking
    *
    * -# Create shm segment
@@ -94,7 +90,7 @@ protected:
    *            i.e., both agrees that the other one has setup the segment.
    *            Otherwise false.
    */
-  bool connectServer(Uint32 timeOutMillis);
+  virtual bool connect_server_impl(NDB_SOCKET_TYPE sockfd);
 
   /**
    * Blocking
@@ -108,7 +104,9 @@ protected:
    *            i.e., both agrees that the other one has setup the segment.
    *            Otherwise false.
    */
-  bool connectClient(Uint32 timeOutMillis);
+  virtual bool connect_client_impl(NDB_SOCKET_TYPE sockfd);
+
+  bool connect_common(NDB_SOCKET_TYPE sockfd);
 
 
   /**
@@ -127,7 +125,6 @@ private:
   bool _shmSegCreated;
   bool _attached;
     
-  const bool  isServer;
   key_t shmKey;
   volatile Uint32 * serverStatusFlag;
   volatile Uint32 * clientStatusFlag;  
diff --git a/ndb/src/common/transporter/SHM_Transporter.unix.cpp b/ndb/src/common/transporter/SHM_Transporter.unix.cpp
index afbf124432e..edf314fbc35 100644
--- a/ndb/src/common/transporter/SHM_Transporter.unix.cpp
+++ b/ndb/src/common/transporter/SHM_Transporter.unix.cpp
@@ -23,83 +23,98 @@
 #include <NdbSleep.h>
 #include <NdbOut.hpp>
 
+#include <InputStream.hpp>
+#include <OutputStream.hpp>
+
 #include <sys/ipc.h>
 #include <sys/shm.h>
 
-
-
 bool
-SHM_Transporter::connectServer(Uint32 timeOutMillis){
+SHM_Transporter::connect_server_impl(NDB_SOCKET_TYPE sockfd)
+{
+  SocketOutputStream s_output(sockfd);
+  SocketInputStream s_input(sockfd);
+
   if(!_shmSegCreated){
     shmId = shmget(shmKey, shmSize, IPC_CREAT | 960);
     if(shmId == -1){
       perror("shmget: ");
-      reportThreadError(remoteNodeId, TE_SHM_UNABLE_TO_CREATE_SEGMENT);
-      NdbSleep_MilliSleep(timeOutMillis);
+      report_error(TE_SHM_UNABLE_TO_CREATE_SEGMENT);
+      NdbSleep_MilliSleep(m_timeOutMillis);
+      NDB_CLOSE_SOCKET(sockfd);
       return false;
     }
     _shmSegCreated = true;
   }
 
-  if(!_attached){
-    shmBuf = (char *)shmat(shmId, 0, 0);
-    if(shmBuf == 0){
-      perror("shmat: ");
-      reportThreadError(remoteNodeId, TE_SHM_UNABLE_TO_ATTACH_SEGMENT);
-      NdbSleep_MilliSleep(timeOutMillis);
-      return false;
-    }
-    _attached = true;
-  }
-  
-  struct shmid_ds info;
-  const int res = shmctl(shmId, IPC_STAT, &info);
-  if(res == -1){
-    perror("shmctl: ");
-    reportThreadError(remoteNodeId, TE_SHM_IPC_STAT);
-    NdbSleep_MilliSleep(timeOutMillis);
+  s_output.println("shm server 1 ok");
+
+  char buf[256];
+  if (s_input.gets(buf, 256) == 0) {
+    NDB_CLOSE_SOCKET(sockfd);
     return false;
   }
-  
-  if(info.shm_nattch == 2 && !setupBuffersDone) {
-    setupBuffers();
-    setupBuffersDone=true;
-  }
 
-  if(setupBuffersDone) {
-    NdbSleep_MilliSleep(timeOutMillis);
-    if(*serverStatusFlag==1 && *clientStatusFlag==1)
-      return true;
-  }
-  
+  int r= connect_common(sockfd);
 
-  if(info.shm_nattch > 2){
-    reportThreadError(remoteNodeId, TE_SHM_DISCONNECT);
-    NdbSleep_MilliSleep(timeOutMillis);
-    return false;
+  if (r) {
+    s_output.println("shm server 2 ok");
+    if (s_input.gets(buf, 256) == 0) {
+      NDB_CLOSE_SOCKET(sockfd);
+      return false;
+    }
   }
-  
-  NdbSleep_MilliSleep(timeOutMillis);
-  return false;
+
+  NDB_CLOSE_SOCKET(sockfd);
+  return r;
 }
 
 bool
-SHM_Transporter::connectClient(Uint32 timeOutMillis){
-  if(!_shmSegCreated){
+SHM_Transporter::connect_client_impl(NDB_SOCKET_TYPE sockfd)
+{
+  SocketInputStream s_input(sockfd);
+  SocketOutputStream s_output(sockfd);
+
+  char buf[256];
+  if (s_input.gets(buf, 256) == 0) {
+    NDB_CLOSE_SOCKET(sockfd);
+    return false;
+  }
 
+  if(!_shmSegCreated){
     shmId = shmget(shmKey, shmSize, 0);
     if(shmId == -1){
-      NdbSleep_MilliSleep(timeOutMillis);
+      NdbSleep_MilliSleep(m_timeOutMillis);
+      NDB_CLOSE_SOCKET(sockfd);
       return false;
     }
     _shmSegCreated = true;
   }
 
+  s_output.println("shm client 1 ok");
+
+  int r= connect_common(sockfd);
+
+  if (r) {
+    if (s_input.gets(buf, 256) == 0) {
+      NDB_CLOSE_SOCKET(sockfd);
+      return false;
+    }
+    s_output.println("shm client 2 ok");
+  }
+
+  NDB_CLOSE_SOCKET(sockfd);
+  return r;
+}
+
+bool
+SHM_Transporter::connect_common(NDB_SOCKET_TYPE sockfd)
+{
   if(!_attached){
     shmBuf = (char *)shmat(shmId, 0, 0);
     if(shmBuf == 0){
-      reportThreadError(remoteNodeId, TE_SHM_UNABLE_TO_ATTACH_SEGMENT);
-      NdbSleep_MilliSleep(timeOutMillis);
+      report_error(TE_SHM_UNABLE_TO_ATTACH_SEGMENT);
+      NdbSleep_MilliSleep(m_timeOutMillis);
       return false;
     }
     _attached = true;
@@ -109,8 +124,8 @@ SHM_Transporter::connectClient(Uint32 timeOutMillis){
 
   const int res = shmctl(shmId, IPC_STAT, &info);
   if(res == -1){
-    reportThreadError(remoteNodeId, TE_SHM_IPC_STAT);
-    NdbSleep_MilliSleep(timeOutMillis);
+    report_error(TE_SHM_IPC_STAT);
+    NdbSleep_MilliSleep(m_timeOutMillis);
     return false;
   }
   
@@ -121,18 +136,18 @@ SHM_Transporter::connectClient(Uint32 timeOutMillis){
   }
 
   if(setupBuffersDone) {
-    NdbSleep_MilliSleep(timeOutMillis);
+    NdbSleep_MilliSleep(m_timeOutMillis);
     if(*serverStatusFlag==1 && *clientStatusFlag==1)
       return true;
   }
 
   if(info.shm_nattch > 2){
-    reportThreadError(remoteNodeId, TE_SHM_DISCONNECT);
-    NdbSleep_MilliSleep(timeOutMillis);
+    report_error(TE_SHM_DISCONNECT);
+    NdbSleep_MilliSleep(m_timeOutMillis);
     return false;
   }
 
-  NdbSleep_MilliSleep(timeOutMillis);
+  NdbSleep_MilliSleep(m_timeOutMillis);
   return false;
 }
 
@@ -141,12 +156,12 @@ SHM_Transporter::checkConnected(){
   struct shmid_ds info;
   const int res = shmctl(shmId, IPC_STAT, &info);
   if(res == -1){
-    reportError(callbackObj, remoteNodeId, TE_SHM_IPC_STAT);
+    report_error(TE_SHM_IPC_STAT);
     return false;
   }
  
   if(info.shm_nattch != 2){
-    reportError(callbackObj, remoteNodeId, TE_SHM_DISCONNECT);
+    report_error(TE_SHM_DISCONNECT);
     return false;
   }
   return true;
@@ -168,7 +183,7 @@ SHM_Transporter::disconnectImpl(){
   if(isServer && _shmSegCreated){
     const int res = shmctl(shmId, IPC_RMID, 0);
     if(res == -1){
-      reportError(callbackObj, remoteNodeId, TE_SHM_UNABLE_TO_REMOVE_SEGMENT);
+      report_error(TE_SHM_UNABLE_TO_REMOVE_SEGMENT);
       return;
     }
     _shmSegCreated = false;
diff --git a/ndb/src/common/transporter/TransporterRegistry.cpp b/ndb/src/common/transporter/TransporterRegistry.cpp
index c046e4f4236..3d42c40f720 100644
--- a/ndb/src/common/transporter/TransporterRegistry.cpp
+++ b/ndb/src/common/transporter/TransporterRegistry.cpp
@@ -352,13 +352,17 @@ TransporterRegistry::createTransporter(SHM_TransporterConfiguration *config) {
   if(theTransporters[config->remoteNodeId] != NULL)
     return false;
 
-  SHM_Transporter * t = new SHM_Transporter(config->localNodeId,
+  SHM_Transporter * t = new SHM_Transporter(*this,
+					    "localhost",
+					    "localhost",
+					    config->port,
+					    localNodeId,
 					    config->remoteNodeId,
-					    config->shmKey,
-					    config->shmSize,
 					    config->compression,
 					    config->checksum,
-					    config->signalId
+					    config->signalId,
+					    config->shmKey,
+					    config->shmSize
 					    );
   if (t == NULL)
     return false;