added management function to purge stale sessions in the management server

ndb/include/util/Bitmask.hpp:
  added bitXORC
ndb/include/util/SocketServer.hpp:
  added method to apply function on each session
ndb/src/common/util/SocketServer.cpp:
  added method to apply function on each session

5 files changed, 135 insertions, 15 deletions

diff --git a/ndb/src/mgmsrv/MgmtSrvr.cpp b/ndb/src/mgmsrv/MgmtSrvr.cpp
index 2e30d73290b..a49b29af275 100644
--- a/ndb/src/mgmsrv/MgmtSrvr.cpp
+++ b/ndb/src/mgmsrv/MgmtSrvr.cpp
@@ -400,11 +400,13 @@ MgmtSrvr::getPort() const {
 
 /* Constructor */
 MgmtSrvr::MgmtSrvr(NodeId nodeId,
+		   SocketServer *socket_server,
 		   const BaseString &configFilename,
 		   LocalConfig &local_config,
 		   Config * config):
   _blockNumber(1), // Hard coded block number since it makes it easy to send
                    // signals to other management servers.
+  m_socket_server(socket_server),
   _ownReference(0),
   m_local_config(local_config),
   theSignalIdleList(NULL),
@@ -2094,6 +2096,25 @@ MgmtSrvr::getNodeType(NodeId nodeId) const
   return nodeTypes[nodeId];
 }
 
+void
+MgmtSrvr::get_connected_nodes(NodeBitmask &connected_nodes) const
+{
+  if (theFacade && theFacade->theClusterMgr) 
+  {
+    for(Uint32 i = 0; i < MAX_NODES; i++)
+    {
+      if (getNodeType(i) == NDB_MGM_NODE_TYPE_NDB)
+      {
+	const ClusterMgr::Node &node= theFacade->theClusterMgr->getNodeInfo(i);
+	if (node.connected)
+	{
+	  connected_nodes.bitOR(node.m_state.m_connected_nodes);
+	}
+      }
+    }
+  }
+}
+
 bool
 MgmtSrvr::alloc_node_id(NodeId * nodeId, 
 			enum ndb_mgm_node_type type,
@@ -2106,7 +2127,7 @@ MgmtSrvr::alloc_node_id(NodeId * nodeId,
 		       *nodeId, type, client_addr));
   if (g_no_nodeid_checks) {
     if (*nodeId == 0) {
-      error_string.appfmt("no-nodeid-ckecks set in manegment server.\n"
+      error_string.appfmt("no-nodeid-checks set in management server.\n"
 			  "node id must be set explicitly in connectstring");
       DBUG_RETURN(false);
     }
@@ -2115,16 +2136,11 @@ MgmtSrvr::alloc_node_id(NodeId * nodeId,
   Guard g(m_node_id_mutex);
   int no_mgm= 0;
   NodeBitmask connected_nodes(m_reserved_nodes);
-  for(Uint32 i = 0; i < MAX_NODES; i++)
+  get_connected_nodes(connected_nodes);
   {
-    if (getNodeType(i) == NDB_MGM_NODE_TYPE_NDB &&
-	theFacade && theFacade->theClusterMgr) {
-      const ClusterMgr::Node &node= theFacade->theClusterMgr->getNodeInfo(i);
-      if (node.connected) {
-	connected_nodes.bitOR(node.m_state.m_connected_nodes);
-      }
-    } else if (getNodeType(i) == NDB_MGM_NODE_TYPE_MGM)
-      no_mgm++;
+    for(Uint32 i = 0; i < MAX_NODES; i++)
+      if (getNodeType(i) == NDB_MGM_NODE_TYPE_MGM)
+	no_mgm++;
   }
   bool found_matching_id= false;
   bool found_matching_type= false;
@@ -2227,6 +2243,10 @@ MgmtSrvr::alloc_node_id(NodeId * nodeId,
 	m_connect_address[id_found].s_addr= 0;
     }
     m_reserved_nodes.set(id_found);
+    char tmp_str[128];
+    m_reserved_nodes.getText(tmp_str);
+    g_EventLogger.info("Mgmt server state: nodeid %d reserved for ip %s, m_reserved_nodes %s.",
+		       id_found, get_connect_address(id_found), tmp_str);
     DBUG_RETURN(true);
   }
 
@@ -2283,6 +2303,36 @@ MgmtSrvr::alloc_node_id(NodeId * nodeId,
       error_string.appfmt("No node defined with id=%d in config file.",
 			  *nodeId);
   }
+
+  g_EventLogger.warning("Allocate nodeid (%d) failed. Connection from ip %s. "
+			"Returned error string \"%s\"",
+			*nodeId,
+			client_addr != 0 ? inet_ntoa(((struct sockaddr_in *)(client_addr))->sin_addr) : "<none>",
+			error_string.c_str());
+
+  NodeBitmask connected_nodes2;
+  get_connected_nodes(connected_nodes2);
+  {
+    BaseString tmp_connected, tmp_not_connected;
+    for(Uint32 i = 0; i < MAX_NODES; i++)
+    {
+      if (connected_nodes2.get(i))
+      {
+	if (!m_reserved_nodes.get(i))
+	  tmp_connected.appfmt(" %d", i);
+      }
+      else if (m_reserved_nodes.get(i))
+      {
+	tmp_not_connected.appfmt(" %d", i);
+      }
+    }
+    if (tmp_connected.length() > 0)
+      g_EventLogger.info("Mgmt server state: node id's %s connected but not reserved", 
+			 tmp_connected.c_str());
+    if (tmp_not_connected.length() > 0)
+      g_EventLogger.info("Mgmt server state: node id's %s not connected but reserved",
+			 tmp_not_connected.c_str());
+  }
   DBUG_RETURN(false);
 }
 
@@ -2531,10 +2581,15 @@ MgmtSrvr::Allocated_resources::~Allocated_resources()
 {
   Guard g(m_mgmsrv.m_node_id_mutex);
   if (!m_reserved_nodes.isclear()) {
+    m_mgmsrv.m_reserved_nodes.bitANDC(m_reserved_nodes); 
     // node has been reserved, force update signal to ndb nodes
     global_flag_send_heartbeat_now= 1;
+
+    char tmp_str[128];
+    m_mgmsrv.m_reserved_nodes.getText(tmp_str);
+    g_EventLogger.info("Mgmt server state: nodeid %d freed, m_reserved_nodes %s.",
+		       get_nodeid(), tmp_str);
   }
-  m_mgmsrv.m_reserved_nodes.bitANDC(m_reserved_nodes); 
 }
 
 void
@@ -2543,6 +2598,17 @@ MgmtSrvr::Allocated_resources::reserve_node(NodeId id)
   m_reserved_nodes.set(id);
 }
 
+NodeId
+MgmtSrvr::Allocated_resources::get_nodeid() const
+{
+  for(Uint32 i = 0; i < MAX_NODES; i++)
+  {
+    if (m_reserved_nodes.get(i))
+      return i;
+  }
+  return 0;
+}
+
 int
 MgmtSrvr::setDbParameter(int node, int param, const char * value,
 			 BaseString& msg){
diff --git a/ndb/src/mgmsrv/MgmtSrvr.hpp b/ndb/src/mgmsrv/MgmtSrvr.hpp
index c796e1e9219..b3257491123 100644
--- a/ndb/src/mgmsrv/MgmtSrvr.hpp
+++ b/ndb/src/mgmsrv/MgmtSrvr.hpp
@@ -96,7 +96,10 @@ public:
     // methods to reserve/allocate resources which
     // will be freed when running destructor
     void reserve_node(NodeId id);
-    bool is_reserved(NodeId nodeId) { return m_reserved_nodes.get(nodeId);}
+    bool is_reserved(NodeId nodeId) { return m_reserved_nodes.get(nodeId); }
+    bool is_reserved(NodeBitmask mask) { return !mask.bitAND(m_reserved_nodes).isclear(); }
+    bool isclear() { return m_reserved_nodes.isclear(); }
+    NodeId get_nodeid() const;
   private:
     MgmtSrvr &m_mgmsrv;
     NodeBitmask m_reserved_nodes;
@@ -173,6 +176,7 @@ public:
   /* Constructor */
 
   MgmtSrvr(NodeId nodeId,                    /* Local nodeid */
+	   SocketServer *socket_server,
 	   const BaseString &config_filename,      /* Where to save config */
 	   LocalConfig &local_config,  /* Ndb.cfg filename */
 	   Config * config); 
@@ -499,6 +503,9 @@ public:
   int setDbParameter(int node, int parameter, const char * value, BaseString&);
   
   const char *get_connect_address(Uint32 node_id) { return inet_ntoa(m_connect_address[node_id]); }
+  void get_connected_nodes(NodeBitmask &connected_nodes) const;
+  SocketServer *get_socket_server() { return m_socket_server; }
+
   //**************************************************************************
 private:
   //**************************************************************************
@@ -525,6 +532,8 @@ private:
   
   int _blockNumber;
   NodeId _ownNodeId;
+  SocketServer *m_socket_server;
+
   BlockReference _ownReference; 
   NdbMutex *m_configMutex;
   const Config * _config;
diff --git a/ndb/src/mgmsrv/Services.cpp b/ndb/src/mgmsrv/Services.cpp
index 2672d8c9d4b..0394c4e80bb 100644
--- a/ndb/src/mgmsrv/Services.cpp
+++ b/ndb/src/mgmsrv/Services.cpp
@@ -242,6 +242,8 @@ ParserRow<MgmApiSession> commands[] = {
     MGM_ARG("node", Int, Optional, "Node"),  
     MGM_ARG("filter", String, Mandatory, "Event category"),
 
+  MGM_CMD("purge stale sessions", &MgmApiSession::purge_stale_sessions, ""),
+
   MGM_END()
 };
 
@@ -1412,6 +1414,46 @@ done:
   m_output->println("");
 }
 
+struct PurgeStruct
+{
+  NodeBitmask free_nodes;/* free nodes as reported
+			  * by ndbd in apiRegReqConf
+			  */
+  BaseString *str;
+};
+
+void
+MgmApiSession::stop_session_if_not_connected(SocketServer::Session *_s, void *data)
+{
+  MgmApiSession *s= (MgmApiSession *)_s;
+  struct PurgeStruct &ps= *(struct PurgeStruct *)data;
+  if (s->m_allocated_resources->is_reserved(ps.free_nodes))
+  {
+    ps.str->appfmt(" %d", s->m_allocated_resources->get_nodeid());
+    s->stopSession();
+  }
+}
+
+void
+MgmApiSession::purge_stale_sessions(Parser_t::Context &ctx,
+				    const class Properties &args)
+{
+  struct PurgeStruct ps;
+  BaseString str;
+  ps.str = &str;
+
+  m_mgmsrv.get_connected_nodes(ps.free_nodes);
+  ps.free_nodes.bitXORC(NodeBitmask()); // invert connected_nodes to get free nodes
+
+  m_mgmsrv.get_socket_server()->foreachSession(stop_session_if_not_connected,&ps);
+
+  m_output->println("purge stale sessions reply");
+  if (str.length() > 0)
+    m_output->println("purged:%s",str.c_str());
+  m_output->println("result: Ok");
+  m_output->println("");
+}
+
 template class MutexVector<int>;
 template class Vector<ParserRow<MgmApiSession> const*>;
 template class Vector<unsigned short>;
diff --git a/ndb/src/mgmsrv/Services.hpp b/ndb/src/mgmsrv/Services.hpp
index e47820826b6..bfc915f18f1 100644
--- a/ndb/src/mgmsrv/Services.hpp
+++ b/ndb/src/mgmsrv/Services.hpp
@@ -28,7 +28,9 @@
 /** Undefine this to remove backwards compatibility for "GET CONFIG". */
 #define MGM_GET_CONFIG_BACKWARDS_COMPAT
 
-class MgmApiSession : public SocketServer::Session {
+class MgmApiSession : public SocketServer::Session
+{
+  static void stop_session_if_not_connected(SocketServer::Session *_s, void *data);
 private:
   typedef Parser<MgmApiSession> Parser_t;
 
@@ -84,6 +86,8 @@ public:
 
   void setParameter(Parser_t::Context &ctx, const class Properties &args);
   void listen_event(Parser_t::Context &ctx, const class Properties &args);
+
+  void purge_stale_sessions(Parser_t::Context &ctx, const class Properties &args);
   
   void repCommand(Parser_t::Context &ctx, const class Properties &args);
 };
diff --git a/ndb/src/mgmsrv/main.cpp b/ndb/src/mgmsrv/main.cpp
index 15767e4766d..7a57fdeb77a 100644
--- a/ndb/src/mgmsrv/main.cpp
+++ b/ndb/src/mgmsrv/main.cpp
@@ -83,7 +83,6 @@ struct MgmGlobals {
 int g_no_nodeid_checks= 0;
 static MgmGlobals glob;
 
-
 /******************************************************************************
  * Function prototypes
  ******************************************************************************/
@@ -226,7 +225,7 @@ int main(int argc, char** argv)
   if (!readGlobalConfig())
     goto error_end;
 
-  glob.mgmObject = new MgmtSrvr(glob.localNodeId,
+  glob.mgmObject = new MgmtSrvr(glob.localNodeId, glob.socketServer,
 				BaseString(glob.config_filename),
 				local_config,
 				glob.cluster_config);