Merge joreland@bk-internal.mysql.com:/home/bk/mysql-5.0-ndb

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

ndb/include/mgmapi/mgmapi.h

@@ -994,8 +994,12 @@ extern "C" {
    *
    * @note the socket is now able to be used as a transporter connection
    */
-  NDB_SOCKET_TYPE ndb_mgm_convert_to_transporter(NdbMgmHandle handle);
+  NDB_SOCKET_TYPE ndb_mgm_convert_to_transporter(NdbMgmHandle *handle);
 
+  /**
+   * Get the node id of the mgm server we're connected to
+   */
+  Uint32 ndb_mgm_get_mgmd_nodeid(NdbMgmHandle handle);
 
   /**
    * Config iterator

ndb/include/mgmcommon/ConfigRetriever.hpp

@@ -76,6 +76,7 @@ public:
   const char *get_mgmd_host() const;
   const char *get_connectstring(char *buf, int buf_sz) const;
   NdbMgmHandle get_mgmHandle() { return m_handle; };
+  NdbMgmHandle* get_mgmHandlePtr() { return &m_handle; };
 
   Uint32 get_configuration_nodeid() const;
 private:

ndb/include/transporter/TransporterRegistry.hpp

@@ -116,11 +116,20 @@ public:
    */
   bool connect_server(NDB_SOCKET_TYPE sockfd);
 
+  bool connect_client(NdbMgmHandle *h);
+
   /**
-   * use a mgmd connection to connect as a transporter
+   * Given a SocketClient, creates a NdbMgmHandle, turns it into a transporter
+   * and returns the socket.
    */
   NDB_SOCKET_TYPE connect_ndb_mgmd(SocketClient *sc);
 
+  /**
+   * Given a connected NdbMgmHandle, turns it into a transporter
+   * and returns the socket.
+   */
+  NDB_SOCKET_TYPE connect_ndb_mgmd(NdbMgmHandle *h);
+
   /**
    * Remove all transporters
    */

ndb/src/common/transporter/Transporter.cpp

@@ -124,6 +124,15 @@ Transporter::connect_client() {
   else
     sockfd= m_socket_client->connect();
 
+  connect_client(sockfd);
+}  
+
+bool
+Transporter::connect_client(NDB_SOCKET_TYPE sockfd) {
+
+  if(m_connected)
+    return true;
+
   if (sockfd == NDB_INVALID_SOCKET)
     return false;
 

ndb/src/common/transporter/Transporter.hpp

@@ -44,6 +44,7 @@ public:
    *    Use isConnected() to check status
    */
   bool connect_client();
+  bool connect_client(NDB_SOCKET_TYPE sockfd);
   bool connect_server(NDB_SOCKET_TYPE socket);
 
   /**

ndb/src/common/transporter/TransporterRegistry.cpp

@@ -1365,6 +1365,8 @@ TransporterRegistry::add_transporter_interface(NodeId remoteNodeId,
 bool
 TransporterRegistry::start_service(SocketServer& socket_server)
 {
+  struct ndb_mgm_reply mgm_reply;
+
   DBUG_ENTER("TransporterRegistry::start_service");
   if (m_transporter_interface.size() > 0 && !nodeIdSpecified)
   {
@@ -1521,63 +1523,87 @@ TransporterRegistry::get_transporter(NodeId nodeId) {
   return theTransporters[nodeId];
 }
 
-NDB_SOCKET_TYPE TransporterRegistry::connect_ndb_mgmd(SocketClient *sc)
+bool TransporterRegistry::connect_client(NdbMgmHandle *h)
 {
-  NdbMgmHandle h= ndb_mgm_create_handle();
-  struct ndb_mgm_reply mgm_reply;
+  DBUG_ENTER("TransporterRegistry::connect_client(NdbMgmHandle)");
 
-  if ( h == NULL )
-  {
-    return NDB_INVALID_SOCKET;
-  }
+  Uint32 mgm_nodeid= ndb_mgm_get_mgmd_nodeid(*h);
 
-  /**
-   * Set connectstring
-   */
-  {
-    char c[100];
-    char *cs= &c[0];
-    unsigned len= strlen(sc->get_server_name())+20;
-    if( len > sizeof(c) )
-    {
-      /*
-       * server name is long. malloc enough for it and the port number
-       */
-      cs= (char*)malloc(len*sizeof(char));
-      if(!cs)
-      {
-	ndb_mgm_destroy_handle(&h);
-	return NDB_INVALID_SOCKET;
-      }
-    }
-    snprintf(cs,len,"%s:%u",sc->get_server_name(),sc->get_port());
-    ndb_mgm_set_connectstring(h, cs);
-    if(cs != &c[0])
-      free(cs);
-  }
+  if(!mgm_nodeid)
+    return false;
 
-  if(ndb_mgm_connect(h, 0, 0, 0)<0)
+  Transporter * t = theTransporters[mgm_nodeid];
+  if (!t)
+    return false;
+
+  DBUG_RETURN(t->connect_client(connect_ndb_mgmd(h)));
+}
+
+/**
+ * Given a connected NdbMgmHandle, turns it into a transporter
+ * and returns the socket.
+ */
+NDB_SOCKET_TYPE TransporterRegistry::connect_ndb_mgmd(NdbMgmHandle *h)
+{
+  struct ndb_mgm_reply mgm_reply;
+
+  if ( h==NULL || *h == NULL )
   {
-    ndb_mgm_destroy_handle(&h);
     return NDB_INVALID_SOCKET;
   }
 
   for(unsigned int i=0;i < m_transporter_interface.size();i++)
-    if (ndb_mgm_set_connection_int_parameter(h,
+    if (m_transporter_interface[i].m_s_service_port < 0
+	&& ndb_mgm_set_connection_int_parameter(*h,
 				   get_localNodeId(),
 				   m_transporter_interface[i].m_remote_nodeId,
 				   CFG_CONNECTION_SERVER_PORT,
 				   m_transporter_interface[i].m_s_service_port,
 				   &mgm_reply) < 0)
     {
-      ndb_mgm_destroy_handle(&h);
+      ndb_mgm_destroy_handle(h);
       return NDB_INVALID_SOCKET;
     }
 
+  /**
+   * convert_to_transporter also disposes of the handle (i.e. we don't leak
+   * memory here.
+   */
   NDB_SOCKET_TYPE sockfd= ndb_mgm_convert_to_transporter(h);
   if ( sockfd == NDB_INVALID_SOCKET)
-    ndb_mgm_destroy_handle(&h);
+    ndb_mgm_destroy_handle(h);
   return sockfd;
 }
 
+/**
+ * Given a SocketClient, creates a NdbMgmHandle, turns it into a transporter
+ * and returns the socket.
+ */
+NDB_SOCKET_TYPE TransporterRegistry::connect_ndb_mgmd(SocketClient *sc)
+{
+  NdbMgmHandle h= ndb_mgm_create_handle();
+
+  if ( h == NULL )
+  {
+    return NDB_INVALID_SOCKET;
+  }
+
+  /**
+   * Set connectstring
+   */
+  {
+    BaseString cs;
+    cs.assfmt("%s:%u",sc->get_server_name(),sc->get_port());
+    ndb_mgm_set_connectstring(h, cs.c_str());
+  }
+
+  if(ndb_mgm_connect(h, 0, 0, 0)<0)
+  {
+    ndb_mgm_destroy_handle(&h);
+    return NDB_INVALID_SOCKET;
+  }
+
+  return connect_ndb_mgmd(&h);
+}
+
 template class Vector<TransporterRegistry::Transporter_interface>;

ndb/src/kernel/main.cpp

@@ -190,6 +190,13 @@ int main(int argc, char** argv)
     exit(-1);
   }
 
+  // Re-use the mgm handle as a transporter
+  if(!globalTransporterRegistry.connect_client(
+		 theConfig->get_config_retriever()->get_mgmHandlePtr()))
+      ERROR_SET(fatal, ERR_INVALID_CONFIG,
+		"Connection to mgmd terminated before setup was complete", 
+		"StopOnError missing");
+
   if (!globalTransporterRegistry.start_clients()){
     ndbout_c("globalTransporterRegistry.start_clients() failed");
     exit(-1);

ndb/src/kernel/vm/Configuration.cpp

@@ -189,7 +189,6 @@ Configuration::fetch_configuration(){
   }
 
   m_mgmd_port= 0;
-  m_mgmd_host= 0;
   m_config_retriever= new ConfigRetriever(getConnectString(),
 					  NDB_VERSION, NODE_TYPE_DB);
 
@@ -211,7 +210,7 @@ Configuration::fetch_configuration(){
   }
   
   m_mgmd_port= m_config_retriever->get_mgmd_port();
-  m_mgmd_host= m_config_retriever->get_mgmd_host();
+  m_mgmd_host.assign(m_config_retriever->get_mgmd_host());
 
   ConfigRetriever &cr= *m_config_retriever;
   

ndb/src/kernel/vm/Configuration.hpp

@@ -17,6 +17,7 @@
 #ifndef Configuration_H
 #define Configuration_H
 
+#include <util/BaseString.hpp>
 #include <mgmapi.h>
 #include <ndb_types.h>
 
@@ -67,7 +68,7 @@ public:
   const ndb_mgm_configuration_iterator * getOwnConfigIterator() const;
 
   Uint32 get_mgmd_port() const {return m_mgmd_port;};
-  const char *get_mgmd_host() const {return m_mgmd_host;};
+  const char *get_mgmd_host() const {return m_mgmd_host.c_str();};
   ConfigRetriever* get_config_retriever() { return m_config_retriever; };
 
   class LogLevel * m_logLevel;
@@ -99,7 +100,7 @@ private:
   bool _initialStart;
   char * _connectString;
   Uint32 m_mgmd_port;
-  const char *m_mgmd_host;
+  BaseString m_mgmd_host;
   bool _daemonMode;
 
   void calcSizeAlt(class ConfigValues * );

ndb/src/mgmapi/mgmapi.cpp

@@ -2189,23 +2189,54 @@ ndb_mgm_get_connection_int_parameter(NdbMgmHandle handle,
 
 extern "C"
 NDB_SOCKET_TYPE
-ndb_mgm_convert_to_transporter(NdbMgmHandle handle)
+ndb_mgm_convert_to_transporter(NdbMgmHandle *handle)
 {
   NDB_SOCKET_TYPE s;
 
-  CHECK_HANDLE(handle, NDB_INVALID_SOCKET);
-  CHECK_CONNECTED(handle, NDB_INVALID_SOCKET);
+  CHECK_HANDLE((*handle), NDB_INVALID_SOCKET);
+  CHECK_CONNECTED((*handle), NDB_INVALID_SOCKET);
 
-  handle->connected= 0;   // we pretend we're disconnected
-  s= handle->socket;
+  (*handle)->connected= 0;   // we pretend we're disconnected
+  s= (*handle)->socket;
 
   SocketOutputStream s_output(s);
   s_output.println("transporter connect");
   s_output.println("");
 
-  ndb_mgm_destroy_handle(&handle); // set connected=0, so won't disconnect
+  ndb_mgm_destroy_handle(handle); // set connected=0, so won't disconnect
 
   return s;
 }
 
+extern "C"
+Uint32
+ndb_mgm_get_mgmd_nodeid(NdbMgmHandle handle)
+{
+  Uint32 nodeid=0;
+
+  DBUG_ENTER("ndb_mgm_get_mgmd_nodeid");
+  CHECK_HANDLE(handle, 0);
+  CHECK_CONNECTED(handle, 0);
+  
+  Properties args;
+
+  const ParserRow<ParserDummy> reply[]= {
+    MGM_CMD("get mgmd nodeid reply", NULL, ""),
+    MGM_ARG("nodeid", Int, Mandatory, "Node ID"),
+    MGM_END()
+  };
+  
+  const Properties *prop;
+  prop = ndb_mgm_call(handle, reply, "get mgmd nodeid", &args);
+  CHECK_REPLY(prop, 0);
+
+  if(!prop->get("nodeid",&nodeid)){
+    ndbout_c("Unable to get value");
+    return 0;
+  }
+
+  delete prop;
+  DBUG_RETURN(nodeid);
+}
+
 template class Vector<const ParserRow<ParserDummy>*>;

ndb/src/mgmsrv/Services.cpp

@@ -266,6 +266,8 @@ ParserRow<MgmApiSession> commands[] = {
 
   MGM_CMD("transporter connect", &MgmApiSession::transporter_connect, ""),
 
+  MGM_CMD("get mgmd nodeid", &MgmApiSession::get_mgmd_nodeid, ""),
+
   MGM_END()
 };
 
@@ -1554,5 +1556,13 @@ MgmApiSession::transporter_connect(Parser_t::Context &ctx,
   m_mgmsrv.transporter_connect(s);
 }
 
+void
+MgmApiSession::get_mgmd_nodeid(Parser_t::Context &ctx,
+			       Properties const &args) {
+  m_output->println("get mgmd nodeid reply");
+  m_output->println("nodeid:%u",m_mgmsrv.getOwnNodeId());  
+  m_output->println("");
+}
+
 template class MutexVector<int>;
 template class Vector<ParserRow<MgmApiSession> const*>;

ndb/src/mgmsrv/Services.hpp

@@ -99,6 +99,8 @@ public:
   void check_connection(Parser_t::Context &ctx, const class Properties &args);
 
   void transporter_connect(Parser_t::Context &ctx, Properties const &args);
+
+  void get_mgmd_nodeid(Parser_t::Context &ctx, Properties const &args);
   
   void repCommand(Parser_t::Context &ctx, const class Properties &args);
 };