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

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

MySQL数据库

开发平台：

Visual C++

RepState.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 "RepState.hpp"
  19. #include <signaldata/SumaImpl.hpp>
  20. #include <NdbApiSignal.hpp>
  21. #include <Properties.hpp>
  22. //#define DBUG_REQUESTOR
  24. #ifdef DBUG_REQUESTOR
  25. #define DBUG_REQUESTOR_PRINT(X) ndbout_c(X);
  26. #else
  27. #define DBUG_REQUESTOR_PRINT(X)
  28. #endif
  30. /****************************************************************************
  31.  * Constructor / Destructor / Init
  32.  ****************************************************************************/
  33. RepState::RepState() 
  34. {
  35.   m_connected = UNKNOWN;
  36.   m_repConnected = UNKNOWN;
  37.   m_mutex = NdbMutex_Create();
  38.   m_stopEpoch = 0;
  39.   m_subIdToRemove = 0;
  40.   m_subKeyToRemove = 0;
  41. }
  43. RepState::~RepState() 
  44. {
  45.   NdbMutex_Destroy(m_mutex);
  46. }
  48. void
  49. RepState::setSubscriptionRequests(FuncRequestCreateSubscriptionId f1,
  50.   FuncRequestCreateSubscription f2,
  51.   FuncRequestRemoveSubscription f3)
  52. {
  53.   m_funcRequestCreateSubscriptionId = f1;
  54.   m_funcRequestCreateSubscription = f2;
  55.   m_funcRequestRemoveSubscription = f3;
  56. }
  58. void
  59. RepState::setIntervalRequests(FuncRequestTransfer f1, 
  60.       FuncRequestApply f2,
  61.       FuncRequestDeleteSS f3, 
  62.       FuncRequestDeletePS f4)
  63. {
  64.   m_funcRequestTransfer = f1;
  65.   m_funcRequestApply = f2;
  66.   m_funcRequestDeleteSS = f3;
  67.   m_funcRequestDeletePS = f4;
  68. }
  70. void
  71. RepState::setStartRequests(FuncRequestStartMetaLog * f5,
  72.    FuncRequestStartDataLog * f6,
  73.    FuncRequestStartMetaScan * f7,
  74.    FuncRequestStartDataScan * f8, 
  75.    FuncRequestEpochInfo * f9) 
  76. {
  77.   m_funcRequestStartMetaLog = f5;
  78.   m_funcRequestStartDataLog = f6;
  79.   m_funcRequestStartMetaScan = f7;
  80.   m_funcRequestStartDataScan = f8;
  81.   m_funcRequestEpochInfo = f9;
  82. }
  85. /****************************************************************************
  86.  * Private Helper functions
  87.  ****************************************************************************/
  89. void 
  90. RepState::requestTransfer(NdbApiSignal * signal) 
  91. {
  92.   DBUG_REQUESTOR_PRINT("RepState: Transfer calculations started");
  93.   for(Uint32 nodeGrp=0; nodeGrp<m_channel.getNoOfNodeGroups(); nodeGrp++) {
  94.     DBUG_REQUESTOR_PRINT("RepState: Transfer calc for node grp");
  95.     Interval i;
  96.     if (m_channel.requestTransfer(nodeGrp, &i)) {
  97.       m_funcRequestTransfer(m_extSender, signal, nodeGrp, i.first(), i.last());
  98.     }
  99.   }
  100. }
  102. void
  103. RepState::requestApply(NdbApiSignal * signal) 
  104. {
  105.   DBUG_REQUESTOR_PRINT("RepState: Apply calculations started");
  106.   for(Uint32 nodeGrp=0; nodeGrp<m_channel.getNoOfNodeGroups(); nodeGrp++) {
  107.     DBUG_REQUESTOR_PRINT("RepState: Apply calc for node grp");
  108.     Uint32 gci;
  109.     if (m_channel.requestApply(nodeGrp, &gci)) {
  110.       Uint32 force = (m_channel.getState() == Channel::LOG) ? 0 : 1;
  111.       m_funcRequestApply(m_applier, signal, nodeGrp, gci, gci, force);
  112.     }
  113.   }
  114. }
  116. void
  117. RepState::requestDelete(NdbApiSignal * signal) 
  118. {
  119.   DBUG_REQUESTOR_PRINT("RepState: Delete calculations started");
  120.   for(Uint32 nodeGrp=0; nodeGrp<m_channel.getNoOfNodeGroups(); nodeGrp++) {
  121.     DBUG_REQUESTOR_PRINT("RepState: Delete calc for node grp");
  122.     Interval i;
  123.     if (m_channel.requestDelete(nodeGrp, &i)){
  124.       m_funcRequestDeleteSS(m_gciContainer, signal, nodeGrp, 
  125.     i.first(), i.last());
  126.       m_funcRequestDeletePS(m_extSender, signal, nodeGrp, i.first(), i.last());
  127.     }
  128.   }
  129. }
  131. void
  132. RepState::requestEpochInfo(NdbApiSignal * signal) 
  133. {
  134.   DBUG_REQUESTOR_PRINT("RepState: Epoch Info calculations");
  135.   for(Uint32 nodeGrp=0; nodeGrp<m_channel.getNoOfNodeGroups(); nodeGrp++) {
  136.     m_funcRequestEpochInfo(m_extSender, signal, nodeGrp);
  137.   }
  138. }
  140. /****************************************************************************
  141.  * Public 
  142.  ****************************************************************************/
  144. GrepError::Code
  145. RepState::add(Channel::Position s, Uint32 nodeGrp, const Interval i) 
  146. {
  147.   m_channel.add(s, nodeGrp, i);
  149.   if(s == Channel::PS) 
  150.   {
  151.     m_connected = CONNECTED;
  152.     m_connected_counter = 0;
  153.   }
  155.   Interval fullEpochs;
  156.   m_channel.getFullyAppliedEpochs(&fullEpochs);
  157.   if(s == Channel::App &&
  158.      m_channel.getState() == Channel::DATASCAN_COMPLETED && 
  159.      fullEpochs.last() >= m_channel.getDataScanEpochs().last() &&
  160.      fullEpochs.last() >= m_channel.getMetaScanEpochs().last())
  161.   {
  162.     RLOG(("[%d-%d] fully applied. Channel state changed to LOG",
  163.   fullEpochs.first(), fullEpochs.last()));
  164.     m_channel.setState(Channel::LOG);
  165.     disableAutoStart();
  166.   }
  168.   return GrepError::NO_ERROR;
  169. }
  171. GrepError::Code 
  172. RepState::clear(Channel::Position s, Uint32 nodeGrp, const Interval i) 
  173. {
  174.   m_channel.clear(s, nodeGrp, i);
  175.   return GrepError::NO_ERROR;
  176. }
  178. /****************************************************************************
  179.  * Execute 
  180.  * 
  181.  * This method should only be called from Requestor!
  182.  ****************************************************************************/
  184. GrepError::Code 
  185. RepState::protectedExecute()
  186. {
  187.   GrepError::Code err;
  188.   
  189.   NdbMutex_Lock(m_mutex);
  190.   
  191.   NdbApiSignal* signal = m_extSender->getSignal();
  192.   if (signal == NULL) {
  193.     err = GrepError::COULD_NOT_ALLOCATE_MEM_FOR_SIGNAL;
  194.   } else {
  195.     err = execute(signal);
  196.   }
  197.   NdbMutex_Unlock(m_mutex);
  198.   return err;
  199. }
  201. GrepError::Code 
  202. RepState::execute(NdbApiSignal* signal)
  203. {
  204.   Uint32 subId = m_channel.getSubId();
  205.   Uint32 subKey = m_channel.getSubKey();
  207.   if (!m_channel.m_requestorEnabled) 
  208.     return GrepError::NO_ERROR;
  210.   /**
  211.    * @todo Should have subscriptions in here
  212.    */
  213.   requestEpochInfo(signal);
  215.   /**
  216.    * Update connected counter (Silence time)
  217.    */
  218.   m_connected_counter++;
  219.   if (m_connected_counter > REQUESTOR_EXECUTES_NEEDED_FOR_UNKNOWN_CONNECTION) {
  220.     m_connected = UNKNOWN;
  221.   }
  223.   switch (m_channel.getState()) 
  224.   {
  225.   case Channel::CONSISTENT:
  226.     if (isAutoStartEnabled()) {
  227.       switch (m_channel.getStateSub()) 
  228.       {
  229.       case Channel::NO_SUBSCRIPTION_EXISTS:
  230. m_funcRequestCreateSubscriptionId(m_extSender, signal);  
  231. m_channel.setStateSub(Channel::CREATING_SUBSCRIPTION_ID);
  232. break;
  234.       case Channel::CREATING_SUBSCRIPTION_ID:
  235. break;
  237.       case Channel::SUBSCRIPTION_ID_CREATED:
  238. if(m_channel.isSelective())
  239.           m_funcRequestCreateSubscription(m_extSender, signal,
  240.                                           m_channel.getSubId(),
  241.                                           m_channel.getSubKey(),
  242.                                           m_channel.getSelectedTables());
  243.         else
  244.           m_funcRequestCreateSubscription(m_extSender, signal,
  245.                                           m_channel.getSubId(),
  246.                                           m_channel.getSubKey(),
  247.                                           0);
  248. m_channel.setStateSub(Channel::STARTING_SUBSCRIPTION);
  249. break;
  251.       case Channel::STARTING_SUBSCRIPTION:
  252. break;
  254.       case Channel::SUBSCRIPTION_STARTED:
  255. m_funcRequestStartMetaLog(m_extSender, signal, 
  256.   m_channel.getSubId(),
  257.   m_channel.getSubKey());
  258. m_channel.setState(Channel::METALOG_STARTING);
  259. break;
  260.       } 
  261.     }
  262.     break;
  264.   case Channel::METALOG_STARTING:
  265.     break;
  267.   case Channel::METALOG_STARTED:
  268.     if (isAutoStartEnabled()) {
  269.       m_funcRequestStartMetaScan(m_extSender, signal, subId, subKey);
  270.       m_channel.setState(Channel::METASCAN_STARTING);
  271.     }
  272.     break;
  274.   case Channel::METASCAN_STARTING:
  275.     break;
  277.   case Channel::METASCAN_COMPLETED:
  278.     if (isAutoStartEnabled()) {
  279.       m_funcRequestStartDataLog(m_extSender, signal, subId, subKey);
  280.       m_channel.setState(Channel::DATALOG_STARTING);
  281.     }
  282.     break;
  284.   case Channel::DATALOG_STARTING:
  285.     break;
  287.   case Channel::DATALOG_STARTED:
  288.     if (isAutoStartEnabled()) {
  289.       m_funcRequestStartDataScan(m_extSender, signal, subId, subKey);
  290.       m_channel.setState(Channel::DATASCAN_STARTING);
  291.     }
  292.     break;
  294.   case Channel::DATASCAN_STARTING:
  295.     break;
  297.   case Channel::DATASCAN_COMPLETED:
  298.     break;
  300.   case Channel::LOG:
  301.     if (m_channel.shouldStop()) {
  302.       disableTransfer();
  303.       m_channel.setState(Channel::STOPPING);
  304.     }
  305.     break;
  307.   case Channel::STOPPING:
  308.     if (m_channel.m_transferEnabled) 
  309.     {
  310.       REPABORT("Illegal stopping state while transfer is still enabled");
  311.     }
  312.     /**
  313.      * check if channel has a subscription, if not,
  314.      * check if we have marked a subscription that we want to remove
  315.      * and remove it. This is used to clean up "dangling subscriptions"
  316.      * after various crashes
  317.      */
  318.     if(!m_channel.subscriptionExists())
  319.     { 
  320.       if(m_subIdToRemove && m_subKeyToRemove) 
  321.       {
  322. m_funcRequestRemoveSubscription(m_extSender, signal, 
  323. m_subIdToRemove, 
  324. m_subKeyToRemove);
  325. eventSubscriptionDeleted( m_subIdToRemove, 
  326.   m_subKeyToRemove);
  327. return GrepError::NO_ERROR;
  328.       }  
  329.       else {
  330. return GrepError::SUBSCRIPTION_ID_NOT_FOUND;
  331.       }
  332.     } else {
  333.       if (m_channel.isStoppable())
  334. {
  335.   
  336.   m_funcRequestRemoveSubscription(m_extSender, signal, 
  337.   m_channel.getSubId(), 
  338.   m_channel.getSubKey());
  339.   eventSubscriptionDeleted(m_channel.getSubId(), 
  340.    m_channel.getSubKey());  
  341. }
  342.       else 
  343. return GrepError::CHANNEL_NOT_STOPPABLE;
  344.       
  345.     }
  346.     break;
  348.   default:
  349.     REPABORT("Illegal replication state");
  350.   }
  351.   if (m_channel.m_transferEnabled)  requestTransfer(signal);
  352.   if (m_channel.m_applyEnabled)     requestApply(signal);
  353.   if (m_channel.m_deleteEnabled)    requestDelete(signal); 
  354.   return GrepError::NO_ERROR;
  355. }
  357. /****************************************************************************
  358.  * Request
  359.  * 
  360.  * This method should only be called from Main Thread!
  361.  ****************************************************************************/
  363. GrepError::Code 
  364. RepState::protectedRequest(GrepReq::Request req, Uint32 arg)
  365. {
  366.   return protectedRequest(req, arg, 0);
  367. }
  369. GrepError::Code 
  370. RepState::protectedRequest(GrepReq::Request req, Uint32 arg1, Uint32 arg2)
  371. {
  372.   GrepError::Code code;
  373.   NdbMutex_Lock(m_mutex);
  375.   NdbApiSignal* signal = m_extSender->getSignal();
  376.   if (signal == NULL) {
  377.     code = GrepError::COULD_NOT_ALLOCATE_MEM_FOR_SIGNAL;
  378.   } else {
  379.     code = request(req, arg1, arg2, signal);
  380.   }
  382.   NdbMutex_Unlock(m_mutex);
  383.   return code;
  384. }
  386. GrepError::Code
  387. RepState::protectedAddTable(const char * fullTableName)
  388. {
  389.   GrepError::Code code;
  390.   NdbMutex_Lock(m_mutex);
  391.   code  =  m_channel.addTable(fullTableName);
  392.   NdbMutex_Unlock(m_mutex);
  393.   return code;
  394. }
  396. GrepError::Code
  397. RepState::protectedRemoveTable(const char * fullTableName)
  398. {
  399.   GrepError::Code code;
  400.   if(m_channel.getStateSub() !=  Channel::NO_SUBSCRIPTION_EXISTS)
  401.     return GrepError::START_ALREADY_IN_PROGRESS;
  402.   NdbMutex_Lock(m_mutex);
  403.   code  =  m_channel.removeTable(fullTableName);
  404.   NdbMutex_Unlock(m_mutex);
  405.   return code;
  406. }
  408. GrepError::Code 
  409. RepState::request(GrepReq::Request request, Uint32 arg1, Uint32 arg2, 
  410.   NdbApiSignal* signal) 
  411. {
  412.   switch (request) 
  413.   {
  414.     /*************************************************************************
  415.      * STATUS etc
  416.      *************************************************************************/
  418.   case GrepReq::STATUS:
  419.     printStatus();
  420.     break;
  422.   case GrepReq::REMOVE_BUFFERS:
  423.     return GrepError::NOT_YET_IMPLEMENTED;
  424.     
  425.     /*************************************************************************
  426.      * START
  427.      *************************************************************************/
  429.   case GrepReq::CREATE_SUBSCR:
  430.     if (m_channel.getStateSub() != Channel::NO_SUBSCRIPTION_EXISTS) 
  431.       return GrepError::SUBSCRIPTION_ID_ALREADY_EXIST;
  433.     m_funcRequestCreateSubscriptionId(m_extSender, signal);  
  434.     m_channel.setStateSub(Channel::CREATING_SUBSCRIPTION_ID);
  435.     return GrepError::NO_ERROR;
  437.   case GrepReq::START_SUBSCR:
  438.     if (m_channel.getState() == Channel::STOPPING)
  439.       return GrepError::ILLEGAL_ACTION_WHEN_STOPPING;
  440.     if (m_channel.getStateSub() != Channel::SUBSCRIPTION_ID_CREATED)
  441.       return GrepError::SUBSCRIPTION_ID_NOT_FOUND;
  442.     if(m_channel.isSelective())
  443.       m_funcRequestCreateSubscription(m_extSender, signal,
  444.                                       m_channel.getSubId(),
  445.                                       m_channel.getSubKey(),
  446.                                       m_channel.getSelectedTables());
  447.     else
  448.       m_funcRequestCreateSubscription(m_extSender, signal,
  449.                                       m_channel.getSubId(),
  450.                                       m_channel.getSubKey(),
  451.                                       0);
  452.     m_channel.setStateSub(Channel::STARTING_SUBSCRIPTION);
  453.     return GrepError::NO_ERROR;
  455.   case GrepReq::START_METALOG:
  456.     if (m_channel.getState() == Channel::STOPPING)
  457.       return GrepError::ILLEGAL_ACTION_WHEN_STOPPING;
  458.     if (m_channel.getStateSub() != Channel::SUBSCRIPTION_STARTED)
  459.       return GrepError::SUBSCRIPTION_NOT_STARTED;
  460.     if (m_channel.getState() != Channel::CONSISTENT)
  461.       return GrepError::START_OF_COMPONENT_IN_WRONG_STATE;
  462.     
  463.     m_funcRequestStartMetaLog(m_extSender, signal, 
  464.       m_channel.getSubId(), 
  465.       m_channel.getSubKey());  
  466.     m_channel.setState(Channel::METALOG_STARTING);
  467.     return GrepError::NO_ERROR;
  469.   case GrepReq::START_METASCAN:
  470.     if (m_channel.getState() == Channel::STOPPING)
  471.       return GrepError::ILLEGAL_ACTION_WHEN_STOPPING;
  472.     if (m_channel.getStateSub() != Channel::SUBSCRIPTION_STARTED)
  473.       return GrepError::SUBSCRIPTION_NOT_STARTED;
  474.     if (m_channel.getState() != Channel::METALOG_STARTED)
  475.       return GrepError::START_OF_COMPONENT_IN_WRONG_STATE;
  476.  
  477.     m_funcRequestStartMetaScan(m_extSender, signal,
  478.        m_channel.getSubId(), 
  479.        m_channel.getSubKey());  
  480.     m_channel.setState(Channel::METASCAN_STARTING);
  481.     return GrepError::NO_ERROR;
  483.   case GrepReq::START_DATALOG:
  484.     if (m_channel.getState() == Channel::STOPPING)
  485.       return GrepError::ILLEGAL_ACTION_WHEN_STOPPING;
  486.     if (m_channel.getStateSub() != Channel::SUBSCRIPTION_STARTED)
  487.       return GrepError::SUBSCRIPTION_NOT_STARTED;
  488.     if (m_channel.getState() != Channel::METASCAN_COMPLETED)
  489.       return GrepError::START_OF_COMPONENT_IN_WRONG_STATE;
  491.     m_funcRequestStartDataLog(m_extSender, signal,
  492.       m_channel.getSubId(), 
  493.       m_channel.getSubKey());  
  494.     m_channel.setState(Channel::DATALOG_STARTING);
  495.     return GrepError::NO_ERROR;
  497.   case GrepReq::START_DATASCAN:
  498.     if (m_channel.getState() == Channel::STOPPING)
  499.       return GrepError::ILLEGAL_ACTION_WHEN_STOPPING;
  500.     if (m_channel.getStateSub() != Channel::SUBSCRIPTION_STARTED)
  501.       return GrepError::SUBSCRIPTION_NOT_STARTED;
  502.     if (m_channel.getState() != Channel::DATALOG_STARTED)
  503.       return GrepError::START_OF_COMPONENT_IN_WRONG_STATE;
  505.     m_funcRequestStartDataScan(m_extSender, signal,
  506.       m_channel.getSubId(), 
  507.       m_channel.getSubKey());  
  508.     m_channel.setState(Channel::DATASCAN_STARTING);
  509.     return GrepError::NO_ERROR;
  511.   case GrepReq::START_REQUESTOR:
  512.     enable();
  513.     return GrepError::NO_ERROR;
  514.     
  515.   case GrepReq::START_TRANSFER:
  516.     if (m_channel.getState() == Channel::STOPPING)
  517.       return GrepError::ILLEGAL_ACTION_WHEN_STOPPING;
  518.     enableTransfer();
  519.     return GrepError::NO_ERROR;
  521.   case GrepReq::START_APPLY:
  522.     enableApply();
  523.     return GrepError::NO_ERROR;
  525.   case GrepReq::START_DELETE:
  526.     enableDelete();
  527.     return GrepError::NO_ERROR;
  529.   case GrepReq::START:
  530.     if (isAutoStartEnabled())
  531.       return GrepError::START_ALREADY_IN_PROGRESS;
  533.     enableAutoStart();
  534.     return GrepError::NO_ERROR;
  535.     
  536.     /*************************************************************************
  537.      * STOP
  538.      *************************************************************************/
  540.   case GrepReq::STOP:
  541.     if (m_channel.getStateSub() == Channel::NO_SUBSCRIPTION_EXISTS)
  542.       return GrepError::SUBSCRIPTION_NOT_STARTED;
  543.     if (m_channel.getState() == Channel::STOPPING)
  544.       return GrepError::ILLEGAL_ACTION_WHEN_STOPPING;
  546.     if (arg1 == 0) { 
  547.       /**
  548.        * Stop immediately
  549.        */
  550.       disableTransfer();
  551.       m_channel.setState(Channel::STOPPING);
  552.       m_channel.setStopEpochId(0);
  553.       return GrepError::NO_ERROR;
  554.     } else {
  555.       /**
  556.        * Set future stop epoch
  557.        */
  558.       return m_channel.setStopEpochId(arg1);
  559.     }
  561.   case GrepReq::STOP_SUBSCR:
  562.     {
  563.       if(m_subIdToRemove == 0 && m_subKeyToRemove == 0) {
  564.   m_subIdToRemove   = arg1;
  565.   m_subKeyToRemove  = arg2;
  566.       } else {
  567. return GrepError::ILLEGAL_ACTION_WHEN_STOPPING;
  568.       }
  569.       
  570.       if(m_channel.getSubId() != 0 && m_channel.getSubKey() != 0)
  571. return GrepError::ILLEGAL_USE_OF_COMMAND;
  572.       if (m_channel.getState() == Channel::STOPPING)
  573. return GrepError::ILLEGAL_ACTION_WHEN_STOPPING;
  574.       disableTransfer();
  575.       m_channel.setState(Channel::STOPPING);
  576.       return GrepError::NO_ERROR;
  577.     }
  578.   case GrepReq::STOP_METALOG:
  579.   case GrepReq::STOP_METASCAN:
  580.   case GrepReq::STOP_DATALOG:
  581.   case GrepReq::STOP_DATASCAN:
  582.     return GrepError::NOT_YET_IMPLEMENTED;
  584.   case GrepReq::STOP_REQUESTOR:
  585.     disable();
  586.     return GrepError::NO_ERROR;
  587.     
  588.   case GrepReq::STOP_TRANSFER:
  589.     disableTransfer();
  590.     return GrepError::NO_ERROR;
  592.   case GrepReq::STOP_APPLY:
  593.     disableApply();
  594.     return GrepError::NO_ERROR;
  596.   case GrepReq::STOP_DELETE:
  597.     disableDelete();
  598.     return GrepError::NO_ERROR;
  600.   default:
  601.     ndbout_c("RepCommandInterpreter: Illegal request received");
  602.     return GrepError::NOT_YET_IMPLEMENTED;
  603.   }
  604.   return GrepError::NOT_YET_IMPLEMENTED;
  605. }
  607. /****************************************************************************
  608.  * 
  609.  ****************************************************************************/
  611. /*
  612. GrepError::Code
  613. RepState::slowStop() 
  614. {
  615.   switch(m_channel.getState()) 
  616.   {
  617.   case Channel::LOG:
  618.     m_channel.setState(Channel::LOG_SLOW_STOP);
  619.     return GrepError::NO_ERROR;
  620.   default:
  621.     return GrepError::REQUESTOR_ILLEGAL_STATE_FOR_SLOWSTOP;
  622.   }
  623. }
  625. GrepError::Code
  626. RepState::fastStop() 
  627. {
  628.   switch(m_channel.getState()) 
  629.   {
  630.   case Channel::LOG:
  631.     m_channel.setState(Channel::LOG_FAST_STOP);
  632.     return GrepError::NO_ERROR;
  633.   default:
  634.     return GrepError::REQUESTOR_ILLEGAL_STATE_FOR_FASTSTOP;
  635.   }
  636. }
  637. */
  639. /****************************************************************************
  640.  * Print Status
  641.  ****************************************************************************/
  643. static const char* 
  644. headerText =
  645. "+-------------------------------------------------------------------------+n"
  646. "|                         MySQL Replication Server                        |n"
  647. "+-------------------------------------------------------------------------+n"
  648. ;
  650. static const char* 
  651. channelHeaderText =
  652. "+-------------------------------------------------------------------------+n"
  653. "|                   Applier Channel 1 Replication Status                  |n"
  654. "+-------------------------------------------------------------------------+n"
  655. ;
  657. static const char* 
  658. line =
  659. "+-------------------------------------------------------------------------+n"
  660. ;
  663. Properties *
  664. RepState::getStatus() 
  665. {
  666.   Properties * prop = new Properties();
  667.   if(prop == NULL)
  668.     return NULL;
  669.   NdbMutex_Lock(m_mutex);
  671.   prop->put("nodegroups", (int)m_channel.getNoOfNodeGroups());
  672. //  prop->put("epoch_state", m_channel.getEpochState());
  673.   NdbMutex_Unlock(m_mutex);
  674.   return prop;
  675. }
  678. Properties * RepState::query(QueryCounter counter, Uint32 replicationId) 
  679. {
  680.   Properties * prop = new Properties();
  681.   if(prop == NULL)
  682.     return NULL;
  683.   NdbMutex_Lock(m_mutex);
  684.   if(counter != ~(Uint32)0)
  685.     getEpochState((Channel::Position)counter, prop );
  686.   prop->put("no_of_nodegroups", m_channel.getNoOfNodeGroups());
  687.   prop->put("subid", m_channel.getNoOfNodeGroups());
  688.   prop->put("subkey", m_channel.getSubKey());
  689.   prop->put("connected_db", m_connected);
  690.   prop->put("connected_rep", m_repConnected);
  691.   prop->put("state_sub", (int)m_channel.getStateSub());
  692.   prop->put("state", (int)m_channel.getState());    
  693.   
  694.   NdbMutex_Unlock(m_mutex);
  695.   return prop;
  696.   
  697. }
  699. void
  700. RepState::getEpochState(Channel::Position pos, Properties * p)
  701. {
  702.   char first_buf[20];
  703.   char last_buf[20];
  704.   int pos_first = 0, pos_last = 0;
  705.   Uint32 first = 0, last = 0;
  706.   for(Uint32 i = 0; i < m_channel.getNoOfNodeGroups() ; i++) 
  707.   {    
  708.     m_channel.getEpochState(pos, i, &first, &last);
  709.     pos_first += sprintf(first_buf+pos_first,"%d%s",first,",");
  710.     pos_last  += sprintf(last_buf+pos_last,"%d%s",last,",");    
  711.   }
  712. /**
  713.  * remove trailing comma
  714.  */
  715.   pos_first--;
  716.   pos_last--;
  717.   first_buf[pos_first]= '';
  718.   last_buf[pos_last]= '';
  719. #if 0
  720.   sprintf(first_buf+pos_first,"","");
  721.   sprintf(last_buf + pos_last,"","");    
  722. #endif
  724.   p->put("first", first_buf);
  725.   p->put("last", last_buf);
  726.   
  727. }
  730. void
  731. RepState::printStatus() 
  732. {
  733.   /***************************************************************************
  734.    * Global Status
  735.    ***************************************************************************/
  736.   ndbout << headerText;
  737.   switch (m_connected)
  738.   {
  739.   case CONNECTED: 
  740.     ndbout << "| Source:        Connected    "; break;
  741.   case DISCONNECTED:
  742.     ndbout << "| Source:        Disconnected "; break;
  743.   case CONNECTABLE:
  744.     ndbout << "| Source:        Disconnected "; break;
  745.   default:
  746.     ndbout << "| Source:        Unknown      "; break;
  747.   }
  748.   switch (m_repConnected)
  749.   {
  750.   case CONNECTED:
  751.     ndbout << "(Rep: Connected)       "; break;
  752.   case DISCONNECTED:
  753.     ndbout << "(Rep: Disconnected)    "; break;
  754.   case CONNECTABLE:
  755.     ndbout << "(Rep: Disconnected)    "; break;
  756.   default:
  757.     ndbout << "(Rep: Unknown)         "; break;
  758.   }
  759.   ndbout << "                     |" << endl;
  760.   ndbout << "| Autostart:     " << (isAutoStartEnabled() ? "On " : "Off")
  761.  << "                          ";
  762.   ndbout_c("  Silence time:  %10u |", m_connected_counter);
  764.   /***************************************************************************
  765.    * Channel Status
  766.    ***************************************************************************/
  767.   ndbout << channelHeaderText;
  768.   switch(m_channel.getStateSub()) {
  769.   case Channel::NO_SUBSCRIPTION_EXISTS:
  770.     ndbout_c("| Subscription:  Non-existing                      "
  771.      "                       |");
  772.     break;
  773.   case Channel::CREATING_SUBSCRIPTION_ID:
  774.     ndbout_c("| Subscription:  Non-existing (Id is being created)"
  775.      "                       |");
  776.     break;
  777.   case Channel::SUBSCRIPTION_ID_CREATED:
  778.     ndbout_c("| Subscription:  %-3d-%-6d in state: Not yet started     "
  779.      "                |", 
  780.      m_channel.getSubId(), m_channel.getSubKey());
  781.     break;
  782.   case Channel::STARTING_SUBSCRIPTION:
  783.     ndbout_c("| Subscription:  %-3d-%-6d in state: Being started       "
  784.      "                |", 
  785.      m_channel.getSubId(), m_channel.getSubKey());
  786.     break;
  787.   case Channel::SUBSCRIPTION_STARTED:
  788.     ndbout_c("| Subscription:  %-3d-%-6d in state: Started             "
  789.      "                |", 
  790.      m_channel.getSubId(), m_channel.getSubKey());
  791.     break;
  792.   default:
  793.     REPABORT("Illegal subscription state");
  794.   }
  795.   ndbout << "| Stop epoch:    ";
  796.   if (m_channel.getStopEpochId() == intervalMax) {
  797.     ndbout << "No stop defined                             "; 
  798.   } else {
  799.     ndbout.print("%-10d                                  ",
  800.  m_channel.getStopEpochId());
  801.   }
  802.   ndbout << "             |" << endl;
  803.   
  804.   ndbout << "| State:         ";
  805.   switch(m_channel.getState()) 
  806.   {
  807.   case Channel::CONSISTENT:     
  808.     ndbout << "Local database is subscription consistent   "; 
  809.     break;
  810.   case Channel::METALOG_STARTING:
  811.     ndbout << "Starting (Phase 1: Metalog starting)        "; 
  812.     break;
  813.   case Channel::METALOG_STARTED:
  814.     ndbout << "Starting (Phase 2: Metalog started)         "; 
  815.     break;
  816.   case Channel::METASCAN_STARTING:
  817.     ndbout << "Starting (Phase 3: Metascan starting)       "; 
  818.     break;
  819.   case Channel::METASCAN_COMPLETED:
  820.     ndbout << "Starting (Phase 4: Metascan completed)      "; 
  821.     break;
  822.   case Channel::DATALOG_STARTING:
  823.     ndbout << "Starting (Phase 5: Datalog starting)        "; 
  824.     break;
  825.   case Channel::DATALOG_STARTED:
  826.     ndbout << "Starting (Phase 6: Datalog started)         "; 
  827.     break;
  828.   case Channel::DATASCAN_STARTING:
  829.     ndbout << "Starting (Phase 7: Datascan completed)      "; 
  830.     break;
  831.   case Channel::DATASCAN_COMPLETED:
  832.     ndbout << "Starting (Phase 8: Datascan completed)      "; 
  833.     break;
  834.   case Channel::LOG:            
  835.     ndbout << "Logging                                     "; 
  836.     break;
  837.   case Channel::STOPPING:
  838.     ndbout << "Stopping (Stopped when all epochs applied)  ";
  839.     break;
  840.   }
  841.   ndbout << "             |" << endl;
  843. /* @todo
  844.   ndbout_c("| Syncable:      Yes/Scan/No/Unknown (Not implemented)"
  845.    "                    |");
  846. */
  847.   ndbout << "| Requestor:     " << (isEnabled() ? "On " : "Off") 
  848.  << " (Transfer: " << (isTransferEnabled() ? "On,  " : "Off, ")
  849.  << "Apply: " << (isApplyEnabled() ? "On,  " : "Off, ")
  850.  << "Delete: " << (isDeleteEnabled() ? "On) " : "Off)")
  851.  << "             |" << endl;
  852.   ndbout_c("| Tables being replicated using this channel:         "
  853.    "                    |");
  854.   m_channel.printTables();
  856.   /**
  857.    * Print node groups
  858.    */
  859.   if (getNoOfNodeGroups() == 0) 
  860.   {
  861.     ndbout_c("| No node groups are known.           "
  862.      "                                    |");
  863.   } 
  864.   else 
  865.   {
  866.     m_channel.print();
  867.   }
  868.   ndbout << line;
  869. }