SignalSender.cpp
上传用户:romrleung
上传日期:2022-05-23
资源大小:18897k
文件大小:7k
源码类别:

MySQL数据库

开发平台:

Visual C++

  1. /* Copyright (C) 2003 MySQL AB
  2.    This program is free software; you can redistribute it and/or modify
  3.    it under the terms of the GNU General Public License as published by
  4.    the Free Software Foundation; either version 2 of the License, or
  5.    (at your option) any later version.
  6.    This program is distributed in the hope that it will be useful,
  7.    but WITHOUT ANY WARRANTY; without even the implied warranty of
  8.    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  9.    GNU General Public License for more details.
  10.    You should have received a copy of the GNU General Public License
  11.    along with this program; if not, write to the Free Software
  12.    Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA */
  13. #include "SignalSender.hpp"
  14. #include <NdbSleep.h>
  15. #include <SignalLoggerManager.hpp>
  16. #include <signaldata/NFCompleteRep.hpp>
  17. #include <signaldata/NodeFailRep.hpp>
  18. SimpleSignal::SimpleSignal(bool dealloc){
  19.   memset(this, 0, sizeof(* this));
  20.   deallocSections = dealloc;
  21. }
  22. SimpleSignal::~SimpleSignal(){
  23.   if(!deallocSections)
  24.     return;
  25.   if(ptr[0].p != 0) delete []ptr[0].p;
  26.   if(ptr[1].p != 0) delete []ptr[1].p;
  27.   if(ptr[2].p != 0) delete []ptr[2].p;
  28. }
  29. void 
  30. SimpleSignal::set(class SignalSender& ss,
  31.   Uint8  trace, Uint16 recBlock, Uint16 gsn, Uint32 len){
  32.   
  33.   header.theTrace                = trace;
  34.   header.theReceiversBlockNumber = recBlock;
  35.   header.theVerId_signalNumber   = gsn;
  36.   header.theLength               = len;
  37.   header.theSendersBlockRef      = refToBlock(ss.getOwnRef());
  38. }
  39. void
  40. SimpleSignal::print(FILE * out){
  41.   fprintf(out, "---- Signal ----------------n");
  42.   SignalLoggerManager::printSignalHeader(out, header, 0, 0, false);
  43.   SignalLoggerManager::printSignalData(out, header, theData);
  44.   for(Uint32 i = 0; i<header.m_noOfSections; i++){
  45.     Uint32 len = ptr[i].sz;
  46.     fprintf(out, " --- Section %d size=%d ---n", i, len);
  47.     Uint32 * signalData = ptr[i].p;
  48.     while(len >= 7){
  49.       fprintf(out, 
  50.               " H'%.8x H'%.8x H'%.8x H'%.8x H'%.8x H'%.8x H'%.8xn",
  51.               signalData[0], signalData[1], signalData[2], signalData[3], 
  52.               signalData[4], signalData[5], signalData[6]);
  53.       len -= 7;
  54.       signalData += 7;
  55.     }
  56.     if(len > 0){
  57.       fprintf(out, " H'%.8x", signalData[0]);
  58.       for(Uint32 i = 1; i<len; i++)
  59.         fprintf(out, " H'%.8x", signalData[i]);
  60.       fprintf(out, "n");
  61.     }
  62.   }
  63. }
  64. SignalSender::SignalSender(TransporterFacade *facade)
  65.   : m_lock(0)
  66. {
  67.   m_cond = NdbCondition_Create();
  68.   theFacade = facade;
  69.   m_blockNo = theFacade->open(this, execSignal, execNodeStatus);
  70.   assert(m_blockNo > 0);
  71. }
  72. SignalSender::~SignalSender(){
  73.   int i;
  74.   if (m_lock)
  75.     unlock();
  76.   theFacade->close(m_blockNo,0);
  77.   // free these _after_ closing theFacade to ensure that
  78.   // we delete all signals
  79.   for (i= m_jobBuffer.size()-1; i>= 0; i--)
  80.     delete m_jobBuffer[i];
  81.   for (i= m_usedBuffer.size()-1; i>= 0; i--)
  82.     delete m_usedBuffer[i];
  83.   NdbCondition_Destroy(m_cond);
  84. }
  85. int SignalSender::lock()
  86. {
  87.   if (NdbMutex_Lock(theFacade->theMutexPtr))
  88.     return -1;
  89.   m_lock= 1;
  90.   return 0;
  91. }
  92. int SignalSender::unlock()
  93. {
  94.   if (NdbMutex_Unlock(theFacade->theMutexPtr))
  95.     return -1;
  96.   m_lock= 0;
  97.   return 0;
  98. }
  99. Uint32
  100. SignalSender::getOwnRef() const {
  101.   return numberToRef(m_blockNo, theFacade->ownId());
  102. }
  103. Uint32
  104. SignalSender::getAliveNode() const{
  105.   return theFacade->get_an_alive_node();
  106. }
  107. const ClusterMgr::Node & 
  108. SignalSender::getNodeInfo(Uint16 nodeId) const {
  109.   return theFacade->theClusterMgr->getNodeInfo(nodeId);
  110. }
  111. Uint32
  112. SignalSender::getNoOfConnectedNodes() const {
  113.   return theFacade->theClusterMgr->getNoOfConnectedNodes();
  114. }
  115. SendStatus
  116. SignalSender::sendSignal(Uint16 nodeId, const SimpleSignal * s){
  117.   return theFacade->theTransporterRegistry->prepareSend(&s->header,
  118. 1, // JBB
  119. &s->theData[0],
  120. nodeId, 
  121. &s->ptr[0]);
  122. }
  123. template<class T>
  124. SimpleSignal *
  125. SignalSender::waitFor(Uint32 timeOutMillis, T & t)
  126. {
  127.   SimpleSignal * s = t.check(m_jobBuffer);
  128.   if(s != 0){
  129.     return s;
  130.   }
  131.   
  132.   NDB_TICKS now = NdbTick_CurrentMillisecond();
  133.   NDB_TICKS stop = now + timeOutMillis;
  134.   Uint32 wait = (timeOutMillis == 0 ? 10 : timeOutMillis);
  135.   do {
  136.     NdbCondition_WaitTimeout(m_cond,
  137.      theFacade->theMutexPtr, 
  138.      wait);
  139.     
  140.     
  141.     SimpleSignal * s = t.check(m_jobBuffer);
  142.     if(s != 0){
  143.       m_usedBuffer.push_back(s);
  144.       return s;
  145.     }
  146.     
  147.     now = NdbTick_CurrentMillisecond();
  148.     wait = (timeOutMillis == 0 ? 10 : stop - now);
  149.   } while(stop > now || timeOutMillis == 0);
  150.   
  151.   return 0;
  152. class WaitForAny {
  153. public:
  154.   SimpleSignal * check(Vector<SimpleSignal*> & m_jobBuffer){
  155.     if(m_jobBuffer.size() > 0){
  156.       SimpleSignal * s = m_jobBuffer[0];
  157.       m_jobBuffer.erase(0);
  158.       return s;
  159.     }
  160.     return 0;
  161.   }
  162. };
  163.   
  164. SimpleSignal *
  165. SignalSender::waitFor(Uint32 timeOutMillis){
  166.   
  167.   WaitForAny w;
  168.   return waitFor(timeOutMillis, w);
  169. }
  170. class WaitForNode {
  171. public:
  172.   Uint32 m_nodeId;
  173.   SimpleSignal * check(Vector<SimpleSignal*> & m_jobBuffer){
  174.     Uint32 len = m_jobBuffer.size();
  175.     for(Uint32 i = 0; i<len; i++){
  176.       if(refToNode(m_jobBuffer[i]->header.theSendersBlockRef) == m_nodeId){
  177. SimpleSignal * s = m_jobBuffer[i];
  178. m_jobBuffer.erase(i);
  179. return s;
  180.       }
  181.     }
  182.     return 0;
  183.   }
  184. };
  185. SimpleSignal *
  186. SignalSender::waitFor(Uint16 nodeId, Uint32 timeOutMillis){
  187.   
  188.   WaitForNode w;
  189.   w.m_nodeId = nodeId;
  190.   return waitFor(timeOutMillis, w);
  191. }
  192. #include <NdbApiSignal.hpp>
  193. void
  194. SignalSender::execSignal(void* signalSender, 
  195.  NdbApiSignal* signal, 
  196.  class LinearSectionPtr ptr[3]){
  197.   SimpleSignal * s = new SimpleSignal(true);
  198.   s->header = * signal;
  199.   memcpy(&s->theData[0], signal->getDataPtr(), 4 * s->header.theLength);
  200.   for(Uint32 i = 0; i<s->header.m_noOfSections; i++){
  201.     s->ptr[i].p = new Uint32[ptr[i].sz];
  202.     s->ptr[i].sz = ptr[i].sz;
  203.     memcpy(s->ptr[i].p, ptr[i].p, 4 * ptr[i].sz);
  204.   }
  205.   SignalSender * ss = (SignalSender*)signalSender;
  206.   ss->m_jobBuffer.push_back(s);
  207.   NdbCondition_Signal(ss->m_cond);
  208. }
  209.   
  210. void 
  211. SignalSender::execNodeStatus(void* signalSender, 
  212.      Uint32 nodeId, 
  213.      bool alive, 
  214.      bool nfCompleted){
  215.   if (alive) {
  216.     // node connected
  217.     return;
  218.   }
  219.   SimpleSignal * s = new SimpleSignal(true);
  220.   SignalSender * ss = (SignalSender*)signalSender;
  221.   // node disconnected
  222.   if(nfCompleted)
  223.   {
  224.     // node shutdown complete
  225.     s->header.theVerId_signalNumber = GSN_NF_COMPLETEREP;
  226.     NFCompleteRep *rep = (NFCompleteRep *)s->getDataPtrSend();
  227.     rep->blockNo = 0;
  228.     rep->nodeId = 0;
  229.     rep->failedNodeId = nodeId;
  230.     rep->unused = 0;
  231.     rep->from = 0;
  232.   }
  233.   else
  234.   {
  235.     // node failure
  236.     s->header.theVerId_signalNumber = GSN_NODE_FAILREP;
  237.     NodeFailRep *rep = (NodeFailRep *)s->getDataPtrSend();
  238.     rep->failNo = 0;
  239.     rep->masterNodeId = 0;
  240.     rep->noOfNodes = 1;
  241.     NodeBitmask::clear(rep->theNodes);
  242.     NodeBitmask::set(rep->theNodes,nodeId);
  243.   }
  244.   ss->m_jobBuffer.push_back(s);
  245.   NdbCondition_Signal(ss->m_cond);
  246. }
  247. #if __SUNPRO_CC != 0x560
  248. template SimpleSignal* SignalSender::waitFor<WaitForNode>(unsigned, WaitForNode&);
  249. template SimpleSignal* SignalSender::waitFor<WaitForAny>(unsigned, WaitForAny&);
  250. #endif
  251. template class Vector<SimpleSignal*>;
  252.