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

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

MySQL数据库

开发平台：

Visual C++

Channel.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 "Channel.hpp"
  19. Channel::Channel() 
  20. {
  21.   reset();
  22. }
  24. Channel::~Channel() 
  25. {
  26.   /**
  27.    * Destroy list of selected tables
  28.    */
  29.   for(Uint32 i=0; i < m_selectedTables.size(); i++) {
  30.     delete m_selectedTables[i];
  31.     m_selectedTables[i] = 0;
  32.   }
  33.   m_selectedTables=0;
  34. }
  36. void
  37. Channel::reset() 
  38. {
  39.   for (Uint32 i=0; i<MAX_NO_OF_NODE_GROUPS; i++) {
  40.     for (Uint32 j=0; j<NO_OF_POSITIONS; j++) {
  41.       state[i][j].set(1,0);
  42.     }
  43.   }
  44.   m_noOfNodeGroups = 0;
  45.   m_requestorEnabled = true;
  46.   m_transferEnabled = true;
  47.   m_applyEnabled = true;
  48.   m_deleteEnabled = true;
  49.   m_autoStartEnabled = false;
  50.   m_stopEpochId = intervalMax;
  51.   setSubKey(0);
  52.   setSubId(0);
  53.   m_stateSub = NO_SUBSCRIPTION_EXISTS;
  54.   m_stateRep = CONSISTENT;
  55.   m_metaScanEpochs = emptyInterval;
  56.   m_dataScanEpochs = emptyInterval;
  57. }
  59. bool 
  60. Channel::requestTransfer(Uint32 nodeGrp, Interval * i)
  61. {
  62.   invariant();
  63.   Interval tmp1, tmp2;
  65.   // i = PS - SSReq - SS - App
  66.   intervalLeftMinus(state[nodeGrp][PS], state[nodeGrp][SSReq], &tmp1);
  67.   intervalLeftMinus(tmp1, state[nodeGrp][SS], &tmp2);
  68.   intervalLeftMinus(tmp2, state[nodeGrp][App], i);
  69.   
  70.   i->onlyLeft(GREP_SYSTEM_TABLE_MAX_RANGE);
  71.   i->onlyUpToValue(m_stopEpochId);
  72.   if (i->isEmpty()) return false;
  74.   add(SSReq, nodeGrp, *i);
  75.   invariant();
  76.   return true;
  77. }
  79. bool 
  80. Channel::requestApply(Uint32 nodeGrp, Uint32 * epoch)
  81. {
  82.   invariant();
  83.   Interval tmp1, tmp2;
  85.   // tmp2 = SS - AppReq - App
  86.   intervalLeftMinus(state[nodeGrp][SS], state[nodeGrp][AppReq], &tmp1);
  87.   intervalLeftMinus(tmp1, state[nodeGrp][App], &tmp2);
  89.   tmp2.onlyUpToValue(m_stopEpochId);
  90.   if (tmp2.isEmpty()) return false;
  91.   tmp2.onlyLeft(1);
  93.   // Check that all GCI Buffers for epoch exists in SS
  94.   for (Uint32 i=0; i<m_noOfNodeGroups; i++) {
  95.     if (!state[nodeGrp][SS].inInterval(tmp2.first()))
  96.       return false;
  97.   }
  99.   invariant();
  100.   add(AppReq, nodeGrp, tmp2);
  101.   invariant();
  102.   *epoch = tmp2.first();
  103.   return true;
  104. }
  106. bool 
  107. Channel::requestDelete(Uint32 nodeGrp, Interval * i)
  108. {
  109.   invariant();
  110.   Interval tmp1;
  112.   // i = (App cut PS) - DelReq
  113.   intervalCut(state[nodeGrp][App], state[nodeGrp][PS], &tmp1);
  114.   intervalLeftMinus(tmp1, state[nodeGrp][DelReq], i);
  115.   
  116.   if (i->isEmpty()) return false;
  117.   i->onlyLeft(GREP_SYSTEM_TABLE_MAX_RANGE);
  119.   invariant();
  120.   add(DelReq, nodeGrp, *i);
  121.   invariant();
  122.   return true;
  123. }
  125. void 
  126. Channel::add(Position pos, Uint32 nodeGrp, const Interval i) 
  127. {
  128.   Interval r;
  129.   intervalAdd(state[nodeGrp][pos], i, &r);
  130.   state[nodeGrp][pos].set(r);
  131. }
  133. void
  134. Channel::clear(Position p, Uint32 nodeGrp, const Interval i)
  135. {
  136.   Interval r;
  137.   intervalLeftMinus(state[nodeGrp][p], i, &r);
  138.   state[nodeGrp][p].set(r);
  139. }
  141. bool
  142. Channel::isSynchable(Uint32 nodeGrp)
  143. {
  144.   return true;
  145.   /*
  146.     @todo This should be implemented...
  148.     Interval tmp1, tmp2;
  149.     intervalAdd(state[nodeGrp][PS], state[nodeGrp][SSReq], &tmp1);
  150.     intervalAdd(tmp1, state[nodeGrp][SSReq], &tmp2);
  151.     intervalAdd(tmp2, state[nodeGrp][SS], &tmp1);
  152.     intervalAdd(tmp1, state[nodeGrp][AppReq], &tmp2);
  153.     intervalAdd(tmp2, state[nodeGrp][App], &tmp1);
  154.     if (intervalInclude(state[nodeGrp][PS], tmp1.right()))
  155.     return true;
  156.     else 
  157.     return false;
  158.   */
  159. }
  161. /**
  162.  * Return the cut of all App:s.
  163.  */
  164. void
  165. Channel::getFullyAppliedEpochs(Interval * interval)
  166. {
  167.   if (m_noOfNodeGroups < 1) {
  168.     *interval = emptyInterval;
  169.     return;
  170.   }
  172.   *interval = universeInterval;
  173.   for (Uint32 i=0; i<m_noOfNodeGroups; i++) {
  174.      if (state[i][App].isEmpty()) {
  175.        *interval = emptyInterval;
  176.       return;
  177.     }
  178.     
  179.      if (interval->first() < state[i][App].first()) {
  180.        interval->setFirst(state[i][App].first());
  181.      }
  182.      if (state[i][App].last() < interval->last()) {
  183.        interval->setLast(state[i][App].last());
  184.      }
  185.   }
  186.   interval->normalize();
  187.   return;
  188. }
  190. /**
  191.  * Return true if it is ok to remove the subscription and then stop channel
  192.  */
  193. bool 
  194. Channel::isStoppable()
  195. {
  196.   /**
  197.    * Check that AppReq are empty for all nodegrps
  198.    */
  199.   for (Uint32 i=0; i<m_noOfNodeGroups; i++) {
  200.     if (!state[i][AppReq].isEmpty()) {
  201.       RLOG(("Stop disallowed. AppReq is non-empty"));
  202.       return false;
  203.     }
  204.   }
  206.   /**
  207.    * If stop immediately, then it is ok to stop now
  208.    */
  209.   if (m_stopEpochId == 0) {
  210.     RLOG(("Stop allowed. AppReq empty and immediate stop requested"));
  211.     return true;
  212.   }
  214.   /**
  215.    * If stop on a certain epoch, then 
  216.    * check that stopEpochId is equal to the last applied epoch
  217.    */
  218.   Interval interval;
  219.   getFullyAppliedEpochs(&interval);
  220.   if (m_stopEpochId > interval.last()) {
  221.     RLOG(("Stop disallowed. AppReq empty. Stop %d, LastApplied %d",
  222.   m_stopEpochId, interval.last()));
  223.     return false;
  224.   }
  226.   return true;
  227. }
  229. GrepError::Code
  230. Channel::setStopEpochId(Uint32 n) 
  231. {
  232.   /**
  233.    * If n equal to zero, use next possible epoch (max(App, AppReq))
  234.    */
  235.   if (n == 0) {
  236.     for (Uint32 i=0; i<m_noOfNodeGroups; i++) {
  237.       n = (state[i][App].last() > n) ? state[i][App].last() : n;
  238.       n = (state[i][AppReq].last() > n) ? state[i][AppReq].last() : n;
  239.     }
  240.   }
  242.   /**
  243.    *  If n >= max(App, AppReq) then set value, else return error code
  244.    */
  245.   for (Uint32 i=0; i<m_noOfNodeGroups; i++) {
  246.     if (n < state[i][App].last()) return GrepError::ILLEGAL_STOP_EPOCH_ID;
  247.     if (n < state[i][AppReq].last()) return GrepError::ILLEGAL_STOP_EPOCH_ID;
  248.   }
  250.   m_stopEpochId = n;
  251.   return GrepError::NO_ERROR;
  252. };
  254. bool 
  255. Channel::shouldStop() 
  256. {
  257.   /**
  258.    * If (m_stopEpochId == App) then channel should stop
  259.    */
  260.   for (Uint32 i=0; i<m_noOfNodeGroups; i++) {
  261.     if(m_stopEpochId != state[i][App].last()) return false;
  262.   }
  263.   return true;
  264. }
  266. /*****************************************************************************
  267.  * SELECTIVE TABLE INTERFACE
  268.  *****************************************************************************/
  270. GrepError::Code  
  271. Channel::addTable(const char * tableName)
  272. {
  273.   if(strlen(tableName)>MAX_TAB_NAME_SIZE)
  274.     return GrepError::REP_NOT_PROPER_TABLE;
  275.   /**
  276.    * No of separators are the number of table_name_separator found in tableName
  277.    * since a table is defined as <db>/<schema>/tablename.
  278.    * if noOfSeparators is not equal to 2, then it is not a valid
  279.    * table name.
  280.    */
  281.   Uint32 noOfSeps = 0;
  282.   if(strlen(tableName) < 5)
  283.     return GrepError::REP_NOT_PROPER_TABLE;
  284.   for(Uint32 i =0; i < strlen(tableName); i++)
  285.     if(tableName[i]==table_name_separator)
  286.       noOfSeps++;
  287.   if(noOfSeps!=2)
  288.     return GrepError::REP_NOT_PROPER_TABLE;
  289.   table * t= new table(tableName);
  290.   for(Uint32 i=0; i<m_selectedTables.size(); i++) {
  291.     if(strcmp(tableName, m_selectedTables[i]->tableName)==0)
  292.       return GrepError::REP_TABLE_ALREADY_SELECTED;
  293.   }
  294.   m_selectedTables.push_back(t);
  295.   return GrepError::NO_ERROR;
  296. }
  298. GrepError::Code  
  299. Channel::removeTable(const char * tableName)
  300. {
  301.   if(strlen(tableName)>MAX_TAB_NAME_SIZE)
  302.     return GrepError::REP_NOT_PROPER_TABLE;
  303.   /**
  304.    * No of separators are the number of table_name_separator found in tableName
  305.    * since a table is defined as <db>/<schema>/tablename.
  306.    * If noOfSeparators is not equal to 2, 
  307.    * then it is not a valid table name.
  308.    */
  309.   Uint32 noOfSeps = 0;
  310.   if(strlen(tableName) < 5)
  311.     return GrepError::REP_NOT_PROPER_TABLE;
  312.   for(Uint32 i =0; i < strlen(tableName); i++)
  313.     if(tableName[i]==table_name_separator)
  314.       noOfSeps++;
  315.   if(noOfSeps!=2)
  316.     return GrepError::REP_NOT_PROPER_TABLE;
  317.   for(Uint32 i=0; i<m_selectedTables.size(); i++) {
  318.     if(strcmp(tableName, m_selectedTables[i]->tableName)==0) {
  319.       delete m_selectedTables[i];
  320.       m_selectedTables.erase(i);
  321.       return GrepError::NO_ERROR;
  322.     }
  323.   }
  324.   return GrepError::REP_TABLE_NOT_FOUND;
  325. }
  327. void 
  328. Channel::printTables() 
  329. {
  330.   if(m_selectedTables.size() == 0)
  331.     ndbout_c("|   ALL TABLES                                        "
  332.      "                    |");
  333.   else {
  334.     for(Uint32 i=0; i<m_selectedTables.size(); i++)
  335.       ndbout_c("|   %-69s |", m_selectedTables[i]->tableName);
  336.   }
  337. }
  339. Vector<struct table *> *  
  340. Channel::getSelectedTables() 
  341. {
  342.   if(m_selectedTables.size() == 0) return 0;
  343.   return &m_selectedTables;
  344. }
  346. /*****************************************************************************
  347.  * PRINT
  348.  *****************************************************************************/
  350. void 
  351. Channel::print(Position pos) 
  352. {
  353.   switch(pos){
  354.   case PS:      ndbout << "PS Rep"; break;
  355.   case SSReq:   ndbout << "Tra-Req"; break;
  356.   case SS:      ndbout << "SS Rep"; break;
  357.   case AppReq:  ndbout << "App-Req"; break;
  358.   case App:     ndbout << "Applied"; break;
  359.   case DelReq:  ndbout << "Del-Req"; break;
  360.   default:      REPABORT("Unknown replication position");
  361.   }
  362. }
  364. void 
  365. Channel::print() 
  366. {
  367.   for (Uint32 i=0; i<m_noOfNodeGroups; i++) {
  368.     print(i); 
  369.   }
  370. }
  372. void 
  373. Channel::print(Position pos, Uint32 nodeGrp) 
  374. {
  375.   print(pos); 
  376.   if (state[nodeGrp][pos].first() == 1 && state[nodeGrp][pos].last() == 0) {
  377.     ndbout << " EMPTY";
  378.   } else {
  379.     ndbout << " [" << state[nodeGrp][pos].first() << "-" 
  380.    << state[nodeGrp][pos].last() << "]";
  381.   }
  382. }
  384. static const char* 
  385. channelline =
  386. "+-------------------------------------------------------------------------+n"
  387. ;
  389. void
  390. Channel::getEpochState(Position p, 
  391.        Uint32 nodeGrp, 
  392.        Uint32 * first,
  393.        Uint32 * last) {
  394.   if(state[nodeGrp][p].isEmpty()) {
  395.     *first = 1;
  396.     *last  = 0;
  397.     return;
  398.   }  
  399.   *first = state[nodeGrp][p].first();
  400.   *last  = state[nodeGrp][p].last();
  401. }
  404. void 
  405. Channel::print(Uint32 nodeGrp) 
  406. {
  407.   ndbout << channelline;
  408.   ndbout_c("|                        |       Meta scan       |"
  409.    "        Data scan       |");
  410.   ndbout.print("|                        ");
  411.   if (m_metaScanEpochs.isEmpty()) {
  412.     ndbout.print("|                       ");
  413.   } else {
  414.     ndbout.print("| %10u-%-10u ",
  415.  m_metaScanEpochs.first(), m_metaScanEpochs.last());
  416.   }
  417.   if (m_dataScanEpochs.isEmpty()) {
  418.     ndbout_c("|                        |");
  419.   } else {
  420.     ndbout_c("|  %10u-%-10u |",
  421.      m_dataScanEpochs.first(), m_dataScanEpochs.last());
  422.   }
  424.   /* --- */
  426.   ndbout << channelline;
  427.   ndbout_c("|    Source Rep Server   |    Being Transfered   |"
  428.    " Destination Rep Server |");
  429.   if (state[nodeGrp][PS].isEmpty()) {
  430.     ndbout.print("|                        ");
  431.   } else {
  432.     ndbout.print("|  %10u-%-10u ",
  433.  state[nodeGrp][PS].first(), state[nodeGrp][PS].last());
  434.   }
  435.   if (state[nodeGrp][SSReq].isEmpty()) {
  436.     ndbout.print("|                       ");
  437.   } else {
  438.     ndbout.print("| %10u-%-10u ",
  439.  state[nodeGrp][SSReq].first(), state[nodeGrp][SSReq].last());
  440.   }
  441.   if (state[nodeGrp][SS].isEmpty()) {
  442.     ndbout_c("|                        |");
  443.   } else {
  444.     ndbout_c("|  %10u-%-10u |",
  445.      state[nodeGrp][SS].first(), state[nodeGrp][SS].last());
  446.   }
  448.   /* --- */
  450.   ndbout << channelline;
  451.   ndbout_c("|      Being Applied     |        Applied        |"
  452.    "      Being Deleted     |");
  453.   if (state[nodeGrp][AppReq].isEmpty()) {
  454.     ndbout.print("|                        ");
  455.   } else {
  456.     ndbout.print("|  %10u-%-10u ", state[nodeGrp][AppReq].first(), 
  457.  state[nodeGrp][AppReq].last());
  458.   }
  459.   if (state[nodeGrp][App].isEmpty()) {
  460.     ndbout.print("|                       ");
  461.   } else {
  462.     ndbout.print("| %10u-%-10u ",
  463.  state[nodeGrp][App].first(), state[nodeGrp][App].last());
  464.   }
  465.   if (state[nodeGrp][DelReq].isEmpty()) {
  466.     ndbout_c("|                        |");
  467.   } else {
  468.     ndbout_c("|  %10u-%-10u |",
  469.      state[nodeGrp][DelReq].first(), state[nodeGrp][DelReq].last());
  470.   }
  471. }
  473. /*****************************************************************************
  474.  * Private Methods
  475.  *****************************************************************************/
  477. void
  478. Channel::invariant() 
  479. {
  480.   for (Uint32 j=0; j<MAX_NO_OF_NODE_GROUPS; j++) 
  481.   {
  482.     if (!intervalDisjoint(state[j][SSReq], state[j][SS]))
  483.       REPABORT("Invariant 1 violated");
  484.     if (!intervalDisjoint(state[j][AppReq], state[j][App]))
  485.       REPABORT("Invariant 2 violated");
  486.   }
  487. }