开通博客赚积分
发布资源赚积分
分类

源码开发语言/平台
当前位置：首页 > 源码/资料 > 数据库系统 > MySQL数据库 > 查看源码
ClusterMgr.cpp
资源名称：mysql-4.1.16-win-src.zip　[点击查看]
上传用户：romrleung
上传日期：2022-05-23
资源大小：18897k
文件大小：21k
源码类别：

MySQL数据库

开发平台：

Visual C++

ClusterMgr.cpp：源码内容
  1. /* Copyright (C) 2003 MySQL AB
  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; either version 2 of the License, or
  6.    (at your option) any later version.
  8.    This program is distributed in the hope that it will be useful,
  9.    but WITHOUT ANY WARRANTY; without even the implied warranty of
  10.    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  11.    GNU General Public License for more details.
  13.    You should have received a copy of the GNU General Public License
  14.    along with this program; if not, write to the Free Software
  15.    Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA */
  17. #include <ndb_global.h>
  18. #include <my_pthread.h>
  19. #include <ndb_limits.h>
  20. #include <ndb_version.h>
  22. #include "TransporterFacade.hpp"
  23. #include "ClusterMgr.hpp"
  24. #include <IPCConfig.hpp>
  25. #include "NdbApiSignal.hpp"
  26. #include "API.hpp"
  27. #include <NdbSleep.h>
  28. #include <NdbOut.hpp>
  29. #include <NdbTick.h>
  32. #include <signaldata/NodeFailRep.hpp>
  33. #include <signaldata/NFCompleteRep.hpp>
  34. #include <signaldata/ApiRegSignalData.hpp>
  36. #include <mgmapi.h>
  37. #include <mgmapi_configuration.hpp>
  38. #include <mgmapi_config_parameters.h>
  40. int global_flag_send_heartbeat_now= 0;
  42. // Just a C wrapper for threadMain
  43. extern "C" 
  44. void*
  45. runClusterMgr_C(void * me)
  46. {
  47.   ((ClusterMgr*) me)->threadMain();
  48.   /** 
  49.    * Sleep to allow another thread that is not exiting to take control 
  50.    * of signals allocated by this thread
  51.    *
  52.    * see Ndb::~Ndb() in Ndbinit.cpp
  53.    */  
  54. #ifdef NDB_OSE
  55.   NdbSleep_MilliSleep(50);
  56. #endif
  57.   return NULL;
  58. }
  60. extern "C" {
  61.   void ndbSetOwnVersion();
  62. }
  63. ClusterMgr::ClusterMgr(TransporterFacade & _facade):
  64.   theStop(0),
  65.   theFacade(_facade)
  66. {
  67.   DBUG_ENTER("ClusterMgr::ClusterMgr");
  68.   ndbSetOwnVersion();
  69.   clusterMgrThreadMutex = NdbMutex_Create();
  70.   noOfAliveNodes= 0;
  71.   noOfConnectedNodes= 0;
  72.   theClusterMgrThread= 0;
  73.   DBUG_VOID_RETURN;
  74. }
  76. ClusterMgr::~ClusterMgr()
  77. {
  78.   DBUG_ENTER("ClusterMgr::~ClusterMgr");
  79.   doStop();  
  80.   NdbMutex_Destroy(clusterMgrThreadMutex);
  81.   DBUG_VOID_RETURN;
  82. }
  84. void
  85. ClusterMgr::init(ndb_mgm_configuration_iterator & iter){
  86.   for(iter.first(); iter.valid(); iter.next()){
  87.     Uint32 tmp = 0;
  88.     if(iter.get(CFG_NODE_ID, &tmp))
  89.       continue;
  91.     theNodes[tmp].defined = true;
  92. #if 0
  93.     ndbout << "--------------------------------------" << endl;
  94.     ndbout << "--------------------------------------" << endl;
  95.     ndbout_c("ClusterMgr: Node %d defined as %s", tmp, config.getNodeType(tmp));
  96. #endif
  98.     unsigned type;
  99.     if(iter.get(CFG_TYPE_OF_SECTION, &type))
  100.       continue;
  102.     switch(type){
  103.     case NODE_TYPE_DB:
  104.       theNodes[tmp].m_info.m_type = NodeInfo::DB;
  105.       break;
  106.     case NODE_TYPE_API:
  107.       theNodes[tmp].m_info.m_type = NodeInfo::API;
  108.       break;
  109.     case NODE_TYPE_MGM:
  110.       theNodes[tmp].m_info.m_type = NodeInfo::MGM;
  111.       break;
  112.     case NODE_TYPE_REP:
  113.       theNodes[tmp].m_info.m_type = NodeInfo::REP;
  114.       break;
  115.     case NODE_TYPE_EXT_REP:
  116.       theNodes[tmp].m_info.m_type = NodeInfo::REP;
  117.       {
  118. Uint32 hbFreq = 10000;
  119. //ndb_mgm_get_int_parameter(iter, CFG_, &hbFreq);
  120. theNodes[tmp].hbFrequency = hbFreq;
  121. assert(100 <= hbFreq && hbFreq < 60 * 60 * 1000);
  122.       }
  123.       break;
  124.     default:
  125.       type = type;
  126. #if 0
  127.       ndbout_c("ClusterMgr: Unknown node type: %d", type);
  128. #endif
  129.     }
  130.   }
  131. }
  133. void
  134. ClusterMgr::startThread() {
  135.   NdbMutex_Lock(clusterMgrThreadMutex);
  136.   
  137.   theStop = 0;
  138.   
  139.   theClusterMgrThread = NdbThread_Create(runClusterMgr_C,
  140.                                          (void**)this,
  141.                                          32768,
  142.                                          "ndb_clustermgr",
  143.                                          NDB_THREAD_PRIO_LOW);
  144.   NdbMutex_Unlock(clusterMgrThreadMutex);
  145. }
  147. void
  148. ClusterMgr::doStop( ){
  149.   DBUG_ENTER("ClusterMgr::doStop");
  150.   NdbMutex_Lock(clusterMgrThreadMutex);
  151.   if(theStop){
  152.     NdbMutex_Unlock(clusterMgrThreadMutex);
  153.     DBUG_VOID_RETURN;
  154.   }
  155.   void *status;
  156.   theStop = 1;
  157.   if (theClusterMgrThread) {
  158.     NdbThread_WaitFor(theClusterMgrThread, &status);  
  159.     NdbThread_Destroy(&theClusterMgrThread);
  160.   }
  161.   NdbMutex_Unlock(clusterMgrThreadMutex);
  162.   DBUG_VOID_RETURN;
  163. }
  165. void
  166. ClusterMgr::threadMain( ){
  167.   NdbApiSignal signal(numberToRef(API_CLUSTERMGR, theFacade.ownId()));
  168.   
  169.   signal.theVerId_signalNumber   = GSN_API_REGREQ;
  170.   signal.theReceiversBlockNumber = QMGR;
  171.   signal.theTrace                = 0;
  172.   signal.theLength               = ApiRegReq::SignalLength;
  174.   ApiRegReq * req = CAST_PTR(ApiRegReq, signal.getDataPtrSend());
  175.   req->ref = numberToRef(API_CLUSTERMGR, theFacade.ownId());
  176.   req->version = NDB_VERSION;
  178.   
  179.   Uint32 timeSlept = 100;
  180.   Uint64 now = NdbTick_CurrentMillisecond();
  182.   while(!theStop){
  183.     /**
  184.      * Start of Secure area for use of Transporter
  185.      */
  186.     int send_heartbeat_now= global_flag_send_heartbeat_now;
  187.     global_flag_send_heartbeat_now= 0;
  189.     theFacade.lock_mutex();
  190.     for (int i = 1; i < MAX_NODES; i++){
  191.       /**
  192.        * Send register request (heartbeat) to all available nodes 
  193.        * at specified timing intervals
  194.        */
  195.       const NodeId nodeId = i;
  196.       Node & theNode = theNodes[nodeId];
  197.       
  198.       if (!theNode.defined)
  199. continue;
  201.       if (theNode.connected == false){
  202. theFacade.doConnect(nodeId);
  203. continue;
  204.       }
  205.       
  206.       if (!theNode.compatible){
  207. continue;
  208.       }
  209.       
  210.       theNode.hbCounter += timeSlept;
  211.       if (theNode.hbCounter >= theNode.hbFrequency ||
  212.   send_heartbeat_now) {
  213. /**
  214.  * It is now time to send a new Heartbeat
  215.  */
  216. if (theNode.hbCounter >= theNode.hbFrequency) {
  217.   theNode.hbSent++;
  218.   theNode.hbCounter = 0;
  219. }
  221. /**
  222.  * If the node is of type REP, 
  223.  * then the receiver of the signal should be API_CLUSTERMGR
  224.  */
  225. if (theNode.m_info.m_type == NodeInfo::REP) {
  226.   signal.theReceiversBlockNumber = API_CLUSTERMGR;
  227. }
  228. #if 0 
  229. ndbout_c("ClusterMgr: Sending API_REGREQ to node %d", (int)nodeId);
  230. #endif
  231. theFacade.sendSignalUnCond(&signal, nodeId);
  232.       }//if
  233.       
  234.       if (theNode.hbSent == 4 && theNode.hbFrequency > 0){
  235. reportNodeFailed(i);
  236.       }//if
  237.     }
  238.     
  239.     /**
  240.      * End of secure area. Let other threads in
  241.      */
  242.     theFacade.unlock_mutex();
  243.     
  244.     // Sleep for 100 ms between each Registration Heartbeat
  245.     Uint64 before = now;
  246.     NdbSleep_MilliSleep(100); 
  247.     now = NdbTick_CurrentMillisecond();
  248.     timeSlept = (now - before);
  249.   }
  250. }
  252. #if 0
  253. void
  254. ClusterMgr::showState(NodeId nodeId){
  255.   ndbout << "-- ClusterMgr - NodeId = " << nodeId << endl;
  256.   ndbout << "theNodeList      = " << theNodeList[nodeId] << endl;
  257.   ndbout << "theNodeState     = " << theNodeState[nodeId] << endl;
  258.   ndbout << "theNodeCount     = " << theNodeCount[nodeId] << endl;
  259.   ndbout << "theNodeStopDelay = " << theNodeStopDelay[nodeId] << endl;
  260.   ndbout << "theNodeSendDelay = " << theNodeSendDelay[nodeId] << endl;
  261. }
  262. #endif
  264. ClusterMgr::Node::Node()
  265.   : m_state(NodeState::SL_NOTHING) { 
  266.   compatible = nfCompleteRep = true;
  267.   connected = defined = m_alive = false; 
  268. }
  270. /******************************************************************************
  271.  * API_REGREQ and friends
  272.  ******************************************************************************/
  274. void
  275. ClusterMgr::execAPI_REGREQ(const Uint32 * theData){
  276.   const ApiRegReq * const apiRegReq = (ApiRegReq *)&theData[0];
  277.   const NodeId nodeId = refToNode(apiRegReq->ref);
  279. #if 0
  280.   ndbout_c("ClusterMgr: Recd API_REGREQ from node %d", nodeId);
  281. #endif
  283.   assert(nodeId > 0 && nodeId < MAX_NODES);
  285.   Node & node = theNodes[nodeId];
  286.   assert(node.defined == true);
  287.   assert(node.connected == true);
  289.   if(node.m_info.m_version != apiRegReq->version){
  290.     node.m_info.m_version = apiRegReq->version;
  292.     if (getMajor(node.m_info.m_version) < getMajor(NDB_VERSION) ||
  293. getMinor(node.m_info.m_version) < getMinor(NDB_VERSION)) {
  294.       node.compatible = false;
  295.     } else {
  296.       node.compatible = true;
  297.     }
  298.   }
  300.   NdbApiSignal signal(numberToRef(API_CLUSTERMGR, theFacade.ownId()));
  301.   signal.theVerId_signalNumber   = GSN_API_REGCONF;
  302.   signal.theReceiversBlockNumber = API_CLUSTERMGR;
  303.   signal.theTrace                = 0;
  304.   signal.theLength               = ApiRegConf::SignalLength;
  305.   
  306.   ApiRegConf * const conf = CAST_PTR(ApiRegConf, signal.getDataPtrSend());
  307.   conf->qmgrRef = numberToRef(API_CLUSTERMGR, theFacade.ownId());
  308.   conf->version = NDB_VERSION;
  309.   conf->apiHeartbeatFrequency = node.hbFrequency;
  310.   theFacade.sendSignalUnCond(&signal, nodeId);
  311. }
  313. int global_mgmt_server_check = 0; // set to one in mgmtsrvr main;
  315. void
  316. ClusterMgr::execAPI_REGCONF(const Uint32 * theData){
  317.   const ApiRegConf * const apiRegConf = (ApiRegConf *)&theData[0];
  318.   const NodeId nodeId = refToNode(apiRegConf->qmgrRef);
  319.   
  320. #if 0 
  321.   ndbout_c("ClusterMgr: Recd API_REGCONF from node %d", nodeId);
  322. #endif
  324.   assert(nodeId > 0 && nodeId < MAX_NODES);
  325.   
  326.   Node & node = theNodes[nodeId];
  327.   assert(node.defined == true);
  328.   assert(node.connected == true);
  330.   if(node.m_info.m_version != apiRegConf->version){
  331.     node.m_info.m_version = apiRegConf->version;
  332.     if (global_mgmt_server_check == 1)
  333.       node.compatible = ndbCompatible_mgmt_ndb(NDB_VERSION,
  334.        node.m_info.m_version);
  335.     else
  336.       node.compatible = ndbCompatible_api_ndb(NDB_VERSION,
  337.       node.m_info.m_version);
  338.   }
  339.   
  340.   node.m_state = apiRegConf->nodeState;
  341.   if (node.compatible && (node.m_state.startLevel == NodeState::SL_STARTED  ||
  342.   node.m_state.startLevel == NodeState::SL_SINGLEUSER)){
  343.     set_node_alive(node, true);
  344.   } else {
  345.     set_node_alive(node, false);
  346.   }//if
  347.   node.hbSent = 0;
  348.   node.hbCounter = 0;
  349.   if (node.m_info.m_type != NodeInfo::REP) {
  350.     node.hbFrequency = (apiRegConf->apiHeartbeatFrequency * 10) - 50;
  351.   }
  352. }
  354. void
  355. ClusterMgr::execAPI_REGREF(const Uint32 * theData){
  356.   
  357.   ApiRegRef * ref = (ApiRegRef*)theData;
  358.   
  359.   const NodeId nodeId = refToNode(ref->ref);
  360.   
  361.   assert(nodeId > 0 && nodeId < MAX_NODES);
  362.   
  363.   Node & node = theNodes[nodeId];
  364.   assert(node.connected == true);
  365.   assert(node.defined == true);
  367.   node.compatible = false;
  368.   set_node_alive(node, false);
  369.   node.m_state = NodeState::SL_NOTHING;
  370.   node.m_info.m_version = ref->version;
  372.   switch(ref->errorCode){
  373.   case ApiRegRef::WrongType:
  374.     ndbout_c("Node %d reports that this node should be a NDB node", nodeId);
  375.     abort();
  376.   case ApiRegRef::UnsupportedVersion:
  377.   default:
  378.     break;
  379.   }
  380. }
  382. void
  383. ClusterMgr::execNODE_FAILREP(const Uint32 * theData){
  384.   NodeFailRep * const nodeFail = (NodeFailRep *)&theData[0];
  385.   for(int i = 1; i<MAX_NODES; i++){
  386.     if(NodeBitmask::get(nodeFail->theNodes, i)){
  387.       reportNodeFailed(i);
  388.     }
  389.   }
  390. }
  392. void
  393. ClusterMgr::execNF_COMPLETEREP(const Uint32 * theData){
  394.   NFCompleteRep * const nfComp = (NFCompleteRep *)theData;
  396.   const NodeId nodeId = nfComp->failedNodeId;
  397.   assert(nodeId > 0 && nodeId < MAX_NODES);
  398.   
  399.   theFacade.ReportNodeFailureComplete(nodeId);
  400.   theNodes[nodeId].nfCompleteRep = true;
  401. }
  403. void
  404. ClusterMgr::reportConnected(NodeId nodeId){
  405.   /**
  406.    * Ensure that we are sending heartbeat every 100 ms
  407.    * until we have got the first reply from NDB providing
  408.    * us with the real time-out period to use.
  409.    */
  410.   assert(nodeId > 0 && nodeId < MAX_NODES);
  412.   noOfConnectedNodes++;
  414.   Node & theNode = theNodes[nodeId];
  415.   theNode.connected = true;
  416.   theNode.hbSent = 0;
  417.   theNode.hbCounter = 0;
  418.   
  419.   if (theNode.m_info.m_type != NodeInfo::REP) {
  420.     theNode.hbFrequency = 0;
  421.   }
  422.   theNode.m_info.m_version = 0;
  423.   theNode.compatible = true;
  424.   theNode.nfCompleteRep = true;
  425.   
  426.   theFacade.ReportNodeAlive(nodeId);
  427. }
  429. void
  430. ClusterMgr::reportDisconnected(NodeId nodeId){
  431.   assert(nodeId > 0 && nodeId < MAX_NODES);
  432.   assert(noOfConnectedNodes > 0);
  434.   noOfConnectedNodes--;
  435.   theNodes[nodeId].connected = false;
  436.   theNodes[nodeId].m_info.m_connectCount ++;
  437.   reportNodeFailed(nodeId);
  438. }
  440. void
  441. ClusterMgr::reportNodeFailed(NodeId nodeId){
  443.   Node & theNode = theNodes[nodeId];
  444.  
  445.   set_node_alive(theNode, false);
  446.   if(theNode.connected)
  447.     theFacade.doDisconnect(nodeId);
  448.   
  449.   const bool report = (theNode.m_state.startLevel != NodeState::SL_NOTHING);  
  450.   theNode.m_state.startLevel = NodeState::SL_NOTHING;
  451.   
  452.   if(report){
  453.     theFacade.ReportNodeDead(nodeId);
  454.   }  
  456.   theNode.nfCompleteRep = false;
  457.   
  458.   if(noOfAliveNodes == 0){
  459.     NFCompleteRep rep;
  460.     for(Uint32 i = 1; i<MAX_NODES; i++){
  461.       if(theNodes[i].defined && theNodes[i].nfCompleteRep == false){
  462. rep.failedNodeId = i;
  463. execNF_COMPLETEREP((Uint32*)&rep);
  464.       }
  465.     }
  466.   }
  467. }
  469. /******************************************************************************
  470.  * Arbitrator
  471.  ******************************************************************************/
  472. ArbitMgr::ArbitMgr(TransporterFacade & _fac)
  473.   : theFacade(_fac)
  474. {
  475.   DBUG_ENTER("ArbitMgr::ArbitMgr");
  477.   theThreadMutex = NdbMutex_Create();
  478.   theInputCond = NdbCondition_Create();
  479.   theInputMutex = NdbMutex_Create();
  480.   
  481.   theRank = 0;
  482.   theDelay = 0;
  483.   theThread = 0;
  485.   theInputTimeout = 0;
  486.   theInputFull = false;
  487.   memset(&theInputFull, 0, sizeof(theInputFull));
  488.   theState = StateInit;
  490.   memset(&theStartReq, 0, sizeof(theStartReq));
  491.   memset(&theChooseReq1, 0, sizeof(theChooseReq1));
  492.   memset(&theChooseReq2, 0, sizeof(theChooseReq2));
  493.   memset(&theStopOrd, 0, sizeof(theStopOrd));
  495.   DBUG_VOID_RETURN;
  496. }
  498. ArbitMgr::~ArbitMgr()
  499. {
  500.   DBUG_ENTER("ArbitMgr::~ArbitMgr");
  501.   NdbMutex_Destroy(theThreadMutex);
  502.   NdbCondition_Destroy(theInputCond);
  503.   NdbMutex_Destroy(theInputMutex);
  504.   DBUG_VOID_RETURN;
  505. }
  507. // Start arbitrator thread.  This is kernel request.
  508. // First stop any previous thread since it is a left-over
  509. // which was never used and which now has wrong ticket.
  510. void
  511. ArbitMgr::doStart(const Uint32* theData)
  512. {
  513.   ArbitSignal aSignal;
  514.   NdbMutex_Lock(theThreadMutex);
  515.   if (theThread != NULL) {
  516.     aSignal.init(GSN_ARBIT_STOPORD, NULL);
  517.     aSignal.data.code = StopRestart;
  518.     sendSignalToThread(aSignal);
  519.     void* value;
  520.     NdbThread_WaitFor(theThread, &value);
  521.     NdbThread_Destroy(&theThread);
  522.     theState = StateInit;
  523.     theInputFull = false;
  524.   }
  525.   aSignal.init(GSN_ARBIT_STARTREQ, theData);
  526.   sendSignalToThread(aSignal);
  527.   theThread = NdbThread_Create(
  528.     runArbitMgr_C, (void**)this, 32768, "ndb_arbitmgr",
  529.     NDB_THREAD_PRIO_HIGH);
  530.   NdbMutex_Unlock(theThreadMutex);
  531. }
  533. // The "choose me" signal from a candidate.
  534. void
  535. ArbitMgr::doChoose(const Uint32* theData)
  536. {
  537.   ArbitSignal aSignal;
  538.   aSignal.init(GSN_ARBIT_CHOOSEREQ, theData);
  539.   sendSignalToThread(aSignal);
  540. }
  542. // Stop arbitrator thread via stop signal from the kernel
  543. // or when exiting API program.
  544. void
  545. ArbitMgr::doStop(const Uint32* theData)
  546. {
  547.   DBUG_ENTER("ArbitMgr::doStop");
  548.   ArbitSignal aSignal;
  549.   NdbMutex_Lock(theThreadMutex);
  550.   if (theThread != NULL) {
  551.     aSignal.init(GSN_ARBIT_STOPORD, theData);
  552.     if (theData == 0) {
  553.       aSignal.data.code = StopExit;
  554.     } else {
  555.       aSignal.data.code = StopRequest;
  556.     }
  557.     sendSignalToThread(aSignal);
  558.     void* value;
  559.     NdbThread_WaitFor(theThread, &value);
  560.     NdbThread_Destroy(&theThread);
  561.     theState = StateInit;
  562.   }
  563.   NdbMutex_Unlock(theThreadMutex);
  564.   DBUG_VOID_RETURN;
  565. }
  567. // private methods
  569. extern "C" 
  570. void*
  571. runArbitMgr_C(void* me)
  572. {
  573.   ((ArbitMgr*) me)->threadMain();
  574.   return NULL;
  575. }
  577. void
  578. ArbitMgr::sendSignalToThread(ArbitSignal& aSignal)
  579. {
  580. #ifdef DEBUG_ARBIT
  581.   char buf[17] = "";
  582.   ndbout << "arbit recv: ";
  583.   ndbout << " gsn=" << aSignal.gsn;
  584.   ndbout << " send=" << aSignal.data.sender;
  585.   ndbout << " code=" << aSignal.data.code;
  586.   ndbout << " node=" << aSignal.data.node;
  587.   ndbout << " ticket=" << aSignal.data.ticket.getText(buf, sizeof(buf));
  588.   ndbout << " mask=" << aSignal.data.mask.getText(buf, sizeof(buf));
  589.   ndbout << endl;
  590. #endif
  591.   aSignal.setTimestamp();       // signal arrival time
  592.   NdbMutex_Lock(theInputMutex);
  593.   while (theInputFull) {
  594.     NdbCondition_WaitTimeout(theInputCond, theInputMutex, 1000);
  595.   }
  596.   theInputBuffer = aSignal;
  597.   theInputFull = true;
  598.   NdbCondition_Signal(theInputCond);
  599.   NdbMutex_Unlock(theInputMutex);
  600. }
  602. void
  603. ArbitMgr::threadMain()
  604. {
  605.   ArbitSignal aSignal;
  606.   aSignal = theInputBuffer;
  607.   threadStart(aSignal);
  608.   bool stop = false;
  609.   while (! stop) {
  610.     NdbMutex_Lock(theInputMutex);
  611.     while (! theInputFull) {
  612.       NdbCondition_WaitTimeout(theInputCond, theInputMutex, theInputTimeout);
  613.       threadTimeout();
  614.     }
  615.     aSignal = theInputBuffer;
  616.     theInputFull = false;
  617.     NdbCondition_Signal(theInputCond);
  618.     NdbMutex_Unlock(theInputMutex);
  619.     switch (aSignal.gsn) {
  620.     case GSN_ARBIT_CHOOSEREQ:
  621.       threadChoose(aSignal);
  622.       break;
  623.     case GSN_ARBIT_STOPORD:
  624.       stop = true;
  625.       break;
  626.     }
  627.   }
  628.   threadStop(aSignal);
  629. }
  631. // handle events in the thread
  633. void
  634. ArbitMgr::threadStart(ArbitSignal& aSignal)
  635. {
  636.   theStartReq = aSignal;
  637.   sendStartConf(theStartReq, ArbitCode::ApiStart);
  638.   theState = StateStarted;
  639.   theInputTimeout = 1000;
  640. }
  642. void
  643. ArbitMgr::threadChoose(ArbitSignal& aSignal)
  644. {
  645.   switch (theState) {
  646.   case StateStarted:            // first REQ
  647.     if (! theStartReq.data.match(aSignal.data)) {
  648.       sendChooseRef(aSignal, ArbitCode::ErrTicket);
  649.       break;
  650.     }
  651.     theChooseReq1 = aSignal;
  652.     if (theDelay == 0) {
  653.       sendChooseConf(aSignal, ArbitCode::WinChoose);
  654.       theState = StateFinished;
  655.       theInputTimeout = 1000;
  656.       break;
  657.     }
  658.     theState = StateChoose1;
  659.     theInputTimeout = 1;
  660.     return;
  661.   case StateChoose1:            // second REQ within Delay
  662.     if (! theStartReq.data.match(aSignal.data)) {
  663.       sendChooseRef(aSignal, ArbitCode::ErrTicket);
  664.       break;
  665.     }
  666.     theChooseReq2 = aSignal;
  667.     theState = StateChoose2;
  668.     theInputTimeout = 1;
  669.     return;
  670.   case StateChoose2:            // too many REQs - refuse all
  671.     if (! theStartReq.data.match(aSignal.data)) {
  672.       sendChooseRef(aSignal, ArbitCode::ErrTicket);
  673.       break;
  674.     }
  675.     sendChooseRef(theChooseReq1, ArbitCode::ErrToomany);
  676.     sendChooseRef(theChooseReq2, ArbitCode::ErrToomany);
  677.     sendChooseRef(aSignal, ArbitCode::ErrToomany);
  678.     theState = StateFinished;
  679.     theInputTimeout = 1000;
  680.     return;
  681.   default:
  682.     sendChooseRef(aSignal, ArbitCode::ErrState);
  683.     break;
  684.   }
  685. }
  687. void
  688. ArbitMgr::threadTimeout()
  689. {
  690.   switch (theState) {
  691.   case StateStarted:
  692.     break;
  693.   case StateChoose1:
  694.     if (theChooseReq1.getTimediff() < theDelay)
  695.       break;
  696.     sendChooseConf(theChooseReq1, ArbitCode::WinChoose);
  697.     theState = StateFinished;
  698.     theInputTimeout = 1000;
  699.     break;
  700.   case StateChoose2:
  701.     sendChooseConf(theChooseReq1, ArbitCode::WinChoose);
  702.     sendChooseConf(theChooseReq2, ArbitCode::LoseChoose);
  703.     theState = StateFinished;
  704.     theInputTimeout = 1000;
  705.     break;
  706.   default:
  707.     break;
  708.   }
  709. }
  711. void
  712. ArbitMgr::threadStop(ArbitSignal& aSignal)
  713. {
  714.   switch (aSignal.data.code) {
  715.   case StopExit:
  716.     switch (theState) {
  717.     case StateStarted:
  718.       sendStopRep(theStartReq, 0);
  719.       break;
  720.     case StateChoose1:                  // just in time
  721.       sendChooseConf(theChooseReq1, ArbitCode::WinChoose);
  722.       break;
  723.     case StateChoose2:
  724.       sendChooseConf(theChooseReq1, ArbitCode::WinChoose);
  725.       sendChooseConf(theChooseReq2, ArbitCode::LoseChoose);
  726.       break;
  727.     case StateInit:
  728.     case StateFinished:
  729.       //??
  730.       break;
  731.     }
  732.     break;
  733.   case StopRequest:
  734.     break;
  735.   case StopRestart:
  736.     break;
  737.   }
  738. }
  740. // output routines
  742. void
  743. ArbitMgr::sendStartConf(ArbitSignal& aSignal, Uint32 code)
  744. {
  745.   ArbitSignal copySignal = aSignal;
  746.   copySignal.gsn = GSN_ARBIT_STARTCONF;
  747.   copySignal.data.code = code;
  748.   sendSignalToQmgr(copySignal);
  749. }
  751. void
  752. ArbitMgr::sendChooseConf(ArbitSignal& aSignal, Uint32 code)
  753. {
  754.   ArbitSignal copySignal = aSignal;
  755.   copySignal.gsn = GSN_ARBIT_CHOOSECONF;
  756.   copySignal.data.code = code;
  757.   sendSignalToQmgr(copySignal);
  758. }
  760. void
  761. ArbitMgr::sendChooseRef(ArbitSignal& aSignal, Uint32 code)
  762. {
  763.   ArbitSignal copySignal = aSignal;
  764.   copySignal.gsn = GSN_ARBIT_CHOOSEREF;
  765.   copySignal.data.code = code;
  766.   sendSignalToQmgr(copySignal);
  767. }
  769. void
  770. ArbitMgr::sendStopRep(ArbitSignal& aSignal, Uint32 code)
  771. {
  772.   ArbitSignal copySignal = aSignal;
  773.   copySignal.gsn = GSN_ARBIT_STOPREP;
  774.   copySignal.data.code = code;
  775.   sendSignalToQmgr(copySignal);
  776. }
  778. /**
  779.  * Send signal to QMGR.  The input includes signal number and
  780.  * signal data.  The signal data is normally a copy of a received
  781.  * signal so it contains expected arbitrator node id and ticket.
  782.  * The sender in signal data is the QMGR node id.
  783.  */
  784. void
  785. ArbitMgr::sendSignalToQmgr(ArbitSignal& aSignal)
  786. {
  787.   NdbApiSignal signal(numberToRef(API_CLUSTERMGR, theFacade.ownId()));
  789.   signal.theVerId_signalNumber = aSignal.gsn;
  790.   signal.theReceiversBlockNumber = QMGR;
  791.   signal.theTrace  = 0;
  792.   signal.theLength = ArbitSignalData::SignalLength;
  794.   ArbitSignalData* sd = CAST_PTR(ArbitSignalData, signal.getDataPtrSend());
  796.   sd->sender = numberToRef(API_CLUSTERMGR, theFacade.ownId());
  797.   sd->code = aSignal.data.code;
  798.   sd->node = aSignal.data.node;
  799.   sd->ticket = aSignal.data.ticket;
  800.   sd->mask = aSignal.data.mask;
  802. #ifdef DEBUG_ARBIT
  803.   char buf[17] = "";
  804.   ndbout << "arbit send: ";
  805.   ndbout << " gsn=" << aSignal.gsn;
  806.   ndbout << " recv=" << aSignal.data.sender;
  807.   ndbout << " code=" << aSignal.data.code;
  808.   ndbout << " node=" << aSignal.data.node;
  809.   ndbout << " ticket=" << aSignal.data.ticket.getText(buf, sizeof(buf));
  810.   ndbout << " mask=" << aSignal.data.mask.getText(buf, sizeof(buf));
  811.   ndbout << endl;
  812. #endif
  814.   theFacade.lock_mutex();
  815.   theFacade.sendSignalUnCond(&signal, aSignal.data.sender);
  816.   theFacade.unlock_mutex();
  817. }