release/src/router/mysql/storage/ndb/src/ndbapi/ClusterMgr.hpp

   1 /* Copyright (c) 2003-2007 MySQL AB
   2
   3    This program is free software; you can redistribute it and/or modify
   4    it under the terms of the GNU General Public License as published by
   5    the Free Software Foundation; version 2 of the License.
   6
   7    This program is distributed in the hope that it will be useful,
   8    but WITHOUT ANY WARRANTY; without even the implied warranty of
   9    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  10    GNU General Public License for more details.
  11
  12    You should have received a copy of the GNU General Public License
  13    along with this program; if not, write to the Free Software
  14    Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA */
  15
  16 #ifndef ClusterMgr_H
  17 #define ClusterMgr_H
  18
  19 #include "API.hpp"
  20 #include <ndb_limits.h>
  21 #include <NdbThread.h>
  22 #include <NdbMutex.h>
  23 #include <NdbCondition.h>
  24 #include <signaldata/ArbitSignalData.hpp>
  25 #include <signaldata/NodeStateSignalData.hpp>
  26 #include <NodeInfo.hpp>
  27 #include <NodeState.hpp>
  28
  29 extern "C" void* runClusterMgr_C(void * me);
  30
  31
  32 /**
  33  * @class ClusterMgr
  34  */
  35 class ClusterMgr {
  36   friend void* runClusterMgr_C(void * me);
  37   friend void  execute(void *, struct SignalHeader * const,
  38                        Uint8, Uint32 * const, LinearSectionPtr ptr[3]);
  39 public:
  40   ClusterMgr(class TransporterFacade &);
  41   ~ClusterMgr();
  42   void init(struct ndb_mgm_configuration_iterator & config);
  43
  44   void reportConnected(NodeId nodeId);
  45   void reportDisconnected(NodeId nodeId);
  46
  47   bool checkUpgradeCompatability(Uint32 nodeVersion);
  48
  49   void doStop();
  50   void startThread();
  51
  52   void forceHB();
  53   void set_max_api_reg_req_interval(unsigned int millisec) { m_max_api_reg_req_interval = millisec; }
  54
  55 private:
  56   void threadMain();
  57
  58   int  theStop;
  59   class TransporterFacade & theFacade;
  60
  61 public:
  62   enum Cluster_state {
  63     CS_waiting_for_clean_cache = 0,
  64     CS_waiting_for_first_connect,
  65     CS_connected
  66   };
  67   struct Node {
  68     Node();
  69     bool defined;
  70     bool connected;     // Transporter connected
  71     bool compatible;    // Version is compatible
  72     bool nfCompleteRep; // NF Complete Rep has arrived
  73     bool m_alive;       // Node is alive
  74     bool m_api_reg_conf;// API_REGCONF has arrived
  75
  76     NodeInfo  m_info;
  77     NodeState m_state;
  78
  79     /**
  80      * Heartbeat stuff
  81      */
  82     Uint32 hbFrequency; // Heartbeat frequence
  83     Uint32 hbCounter;   // # milliseconds passed since last hb sent
  84   };
  85
  86   const Node &  getNodeInfo(NodeId) const;
  87   Uint32        getNoOfConnectedNodes() const;
  88   bool          isClusterAlive() const;
  89   void          hb_received(NodeId);
  90
  91   Uint32        m_connect_count;
  92 private:
  93   Uint32        m_max_api_reg_req_interval;
  94   Uint32        noOfAliveNodes;
  95   Uint32        noOfConnectedNodes;
  96   Node          theNodes[MAX_NODES];
  97   NdbThread*    theClusterMgrThread;
  98
  99   NodeBitmask   waitForHBFromNodes; // used in forcing HBs
 100   NdbCondition* waitForHBCond;
 101   bool          waitingForHB;
 102
 103   enum Cluster_state m_cluster_state;
 104   /**
 105    * Used for controlling start/stop of the thread
 106    */
 107   NdbMutex*     clusterMgrThreadMutex;
 108
 109   void showState(NodeId nodeId);
 110   void reportNodeFailed(NodeId nodeId, bool disconnect = false);
 111
 112   /**
 113    * Signals received
 114    */
 115   void execAPI_REGREQ    (const Uint32 * theData);
 116   void execAPI_REGCONF   (const Uint32 * theData);
 117   void execAPI_REGREF    (const Uint32 * theData);
 118   void execNODE_FAILREP  (const Uint32 * theData);
 119   void execNF_COMPLETEREP(const Uint32 * theData);
 120
 121   inline void set_node_alive(Node& node, bool alive){
 122     if(node.m_alive && !alive)
 123     {
 124       assert(noOfAliveNodes);
 125       noOfAliveNodes--;
 126     }
 127     else if(!node.m_alive && alive)
 128     {
 129       noOfAliveNodes++;
 130     }
 131     node.m_alive = alive;
 132   }
 133 };
 134
 135 inline
 136 const ClusterMgr::Node &
 137 ClusterMgr::getNodeInfo(NodeId nodeId) const {
 138   return theNodes[nodeId];
 139 }
 140
 141 inline
 142 Uint32
 143 ClusterMgr::getNoOfConnectedNodes() const {
 144   return noOfConnectedNodes;
 145 }
 146
 147 inline
 148 bool
 149 ClusterMgr::isClusterAlive() const {
 150   return noOfAliveNodes != 0;
 151 }
 152 inline
 153 void
 154 ClusterMgr::hb_received(NodeId nodeId) {
 155   theNodes[nodeId].m_info.m_heartbeat_cnt= 0;
 156 }
 157
 158 /*****************************************************************************/
 159
 160 /**
 161  * @class ArbitMgr
 162  * Arbitration manager.  Runs in separate thread.
 163  * Started only by a request from the kernel.
 164  */
 165
 166 extern "C" void* runArbitMgr_C(void* me);
 167
 168 class ArbitMgr
 169 {
 170 public:
 171   ArbitMgr(class TransporterFacade &);
 172   ~ArbitMgr();
 173
 174   inline void setRank(unsigned n) { theRank = n; }
 175   inline void setDelay(unsigned n) { theDelay = n; }
 176
 177   void doStart(const Uint32* theData);
 178   void doChoose(const Uint32* theData);
 179   void doStop(const Uint32* theData);
 180
 181   friend void* runArbitMgr_C(void* me);
 182
 183 private:
 184   class TransporterFacade & theFacade;
 185   unsigned theRank;
 186   unsigned theDelay;
 187
 188   void threadMain();
 189   NdbThread* theThread;
 190   NdbMutex* theThreadMutex;     // not really needed
 191
 192   struct ArbitSignal {
 193     GlobalSignalNumber gsn;
 194     ArbitSignalData data;
 195     NDB_TICKS timestamp;
 196
 197     ArbitSignal() {}
 198
 199     inline void init(GlobalSignalNumber aGsn, const Uint32* aData) {
 200       gsn = aGsn;
 201       if (aData != NULL)
 202         memcpy(&data, aData, sizeof(data));
 203       else
 204         memset(&data, 0, sizeof(data));
 205     }
 206
 207     inline void setTimestamp() {
 208       timestamp = NdbTick_CurrentMillisecond();
 209     }
 210
 211     inline NDB_TICKS getTimediff() {
 212       NDB_TICKS now = NdbTick_CurrentMillisecond();
 213       return now < timestamp ? 0 : now - timestamp;
 214     }
 215   };
 216
 217   NdbMutex* theInputMutex;
 218   NdbCondition* theInputCond;
 219   int theInputTimeout;
 220   bool theInputFull;            // the predicate
 221   ArbitSignal theInputBuffer;   // shared buffer
 222
 223   void sendSignalToThread(ArbitSignal& aSignal);
 224
 225   enum State {                  // thread states
 226     StateInit,
 227     StateStarted,               // thread started
 228     StateChoose1,               // received one valid REQ
 229     StateChoose2,               // received two valid REQs
 230     StateFinished               // finished one way or other
 231   };
 232   State theState;
 233
 234   enum Stop {                   // stop code in ArbitSignal.data.code
 235     StopExit = 1,               // at API exit
 236     StopRequest = 2,            // request from kernel
 237     StopRestart = 3             // stop before restart
 238   };
 239
 240   void threadStart(ArbitSignal& aSignal);       // handle thread events
 241   void threadChoose(ArbitSignal& aSignal);
 242   void threadTimeout();
 243   void threadStop(ArbitSignal& aSignal);
 244
 245   ArbitSignal theStartReq;
 246   ArbitSignal theChooseReq1;
 247   ArbitSignal theChooseReq2;
 248   ArbitSignal theStopOrd;
 249
 250   void sendStartConf(ArbitSignal& aSignal, Uint32);
 251   void sendChooseRef(ArbitSignal& aSignal, Uint32);
 252   void sendChooseConf(ArbitSignal& aSignal, Uint32);
 253   void sendStopRep(ArbitSignal& aSignal, Uint32);
 254
 255   void sendSignalToQmgr(ArbitSignal& aSignal);
 256 };
 257
 258 #endif