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

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

MySQL数据库

开发平台：

Visual C++

consumer_restore.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 <NDBT_ReturnCodes.h>
  18. #include "consumer_restore.hpp"
  19. #include <NdbSleep.h>
  21. extern my_bool opt_core;
  23. extern FilteredNdbOut err;
  24. extern FilteredNdbOut info;
  25. extern FilteredNdbOut debug;
  27. static void callback(int, NdbConnection*, void*);
  29. bool
  30. BackupRestore::init()
  31. {
  32.   release();
  34.   if (!m_restore && !m_restore_meta)
  35.     return true;
  37.   m_ndb = new Ndb();
  39.   if (m_ndb == NULL)
  40.     return false;
  41.   
  42.   m_ndb->init(1024);
  43.   if (m_ndb->waitUntilReady(30) != 0)
  44.   {
  45.     err << "Failed to connect to ndb!!" << endl;
  46.     return false;
  47.   }
  48.   info << "Connected to ndb!!" << endl;
  50.   m_callback = new restore_callback_t[m_parallelism];
  52.   if (m_callback == 0)
  53.   {
  54.     err << "Failed to allocate callback structs" << endl;
  55.     return false;
  56.   }
  58.   m_free_callback= m_callback;
  59.   for (Uint32 i= 0; i < m_parallelism; i++) {
  60.     m_callback[i].restore= this;
  61.     m_callback[i].connection= 0;
  62.     if (i > 0)
  63.       m_callback[i-1].next= &(m_callback[i]);
  64.   }
  65.   m_callback[m_parallelism-1].next = 0;
  67.   return true;
  68. }
  70. void BackupRestore::release()
  71. {
  72.   if (m_ndb)
  73.   {
  74.     delete m_ndb;
  75.     m_ndb= 0;
  76.   }
  78.   if (m_callback)
  79.   {
  80.     delete [] m_callback;
  81.     m_callback= 0;
  82.   }
  83. }
  85. BackupRestore::~BackupRestore()
  86. {
  87.   release();
  88. }
  90. static
  91. int 
  92. match_blob(const char * name){
  93.   int cnt, id1, id2;
  94.   char buf[256];
  95.   if((cnt = sscanf(name, "%[^/]/%[^/]/NDB$BLOB_%d_%d", buf, buf, &id1, &id2)) == 4){
  96.     return id1;
  97.   }
  98.   
  99.   return -1;
  100. }
  102. const NdbDictionary::Table*
  103. BackupRestore::get_table(const NdbDictionary::Table* tab){
  104.   if(m_cache.m_old_table == tab)
  105.     return m_cache.m_new_table;
  106.   m_cache.m_old_table = tab;
  108.   int cnt, id1, id2;
  109.   char db[256], schema[256];
  110.   if((cnt = sscanf(tab->getName(), "%[^/]/%[^/]/NDB$BLOB_%d_%d", 
  111.    db, schema, &id1, &id2)) == 4){
  112.     m_ndb->setDatabaseName(db);
  113.     m_ndb->setSchemaName(schema);
  114.     
  115.     BaseString::snprintf(db, sizeof(db), "NDB$BLOB_%d_%d", 
  116.  m_new_tables[id1]->getTableId(), id2);
  117.     
  118.     m_cache.m_new_table = m_ndb->getDictionary()->getTable(db);
  119.     
  120.   } else {
  121.     m_cache.m_new_table = m_new_tables[tab->getTableId()];
  122.   }
  123.   assert(m_cache.m_new_table);
  124.   return m_cache.m_new_table;
  125. }
  127. bool
  128. BackupRestore::finalize_table(const TableS & table){
  129.   bool ret= true;
  130.   if (!m_restore && !m_restore_meta)
  131.     return ret;
  132.   if (table.have_auto_inc())
  133.   {
  134.     Uint64 max_val= table.get_max_auto_val();
  135.     Uint64 auto_val= m_ndb->readAutoIncrementValue(get_table(table.m_dictTable));
  136.     if (max_val+1 > auto_val || auto_val == ~(Uint64)0)
  137.       ret= m_ndb->setAutoIncrementValue(get_table(table.m_dictTable), max_val+1, false);
  138.   }
  139.   return ret;
  140. }
  142. bool
  143. BackupRestore::table(const TableS & table){
  144.   if (!m_restore && !m_restore_meta)
  145.     return true;
  147.   const char * name = table.getTableName();
  148.   
  149.   /**
  150.    * Ignore blob tables
  151.    */
  152.   if(match_blob(name) >= 0)
  153.     return true;
  154.   
  155.   const NdbTableImpl & tmptab = NdbTableImpl::getImpl(* table.m_dictTable);
  156.   if(tmptab.m_indexType != NdbDictionary::Index::Undefined){
  157.     m_indexes.push_back(table.m_dictTable);
  158.     return true;
  159.   }
  160.   
  161.   BaseString tmp(name);
  162.   Vector<BaseString> split;
  163.   if(tmp.split(split, "/") != 3){
  164.     err << "Invalid table name format " << name << endl;
  165.     return false;
  166.   }
  168.   m_ndb->setDatabaseName(split[0].c_str());
  169.   m_ndb->setSchemaName(split[1].c_str());
  170.   
  171.   NdbDictionary::Dictionary* dict = m_ndb->getDictionary();
  172.   if(m_restore_meta){
  173.     NdbDictionary::Table copy(*table.m_dictTable);
  175.     copy.setName(split[2].c_str());
  177.     if (dict->createTable(copy) == -1) 
  178.     {
  179.       err << "Create table " << table.getTableName() << " failed: "
  180.   << dict->getNdbError() << endl;
  181.       return false;
  182.     }
  183.     info << "Successfully restored table " << table.getTableName()<< endl ;
  184.   }  
  185.   
  186.   const NdbDictionary::Table* tab = dict->getTable(split[2].c_str());
  187.   if(tab == 0){
  188.     err << "Unable to find table: " << split[2].c_str() << endl;
  189.     return false;
  190.   }
  191.   if(m_restore_meta){
  192.     m_ndb->setAutoIncrementValue(tab, ~(Uint64)0, false);
  193.   }
  194.   const NdbDictionary::Table* null = 0;
  195.   m_new_tables.fill(table.m_dictTable->getTableId(), null);
  196.   m_new_tables[table.m_dictTable->getTableId()] = tab;
  197.   return true;
  198. }
  200. bool
  201. BackupRestore::endOfTables(){
  202.   if(!m_restore_meta)
  203.     return true;
  205.   NdbDictionary::Dictionary* dict = m_ndb->getDictionary();
  206.   for(size_t i = 0; i<m_indexes.size(); i++){
  207.     const NdbTableImpl & indtab = NdbTableImpl::getImpl(* m_indexes[i]);
  209.     BaseString tmp(indtab.m_primaryTable.c_str());
  210.     Vector<BaseString> split;
  211.     if(tmp.split(split, "/") != 3){
  212.       err << "Invalid table name format " << indtab.m_primaryTable.c_str()
  213.   << endl;
  214.       return false;
  215.     }
  216.     
  217.     m_ndb->setDatabaseName(split[0].c_str());
  218.     m_ndb->setSchemaName(split[1].c_str());
  219.     
  220.     const NdbDictionary::Table * prim = dict->getTable(split[2].c_str());
  221.     if(prim == 0){
  222.       err << "Unable to find base table "" << split[2].c_str() 
  223.   << "" for index "
  224.   << indtab.getName() << endl;
  225.       return false;
  226.     }
  227.     NdbTableImpl& base = NdbTableImpl::getImpl(*prim);
  228.     NdbIndexImpl* idx;
  229.     int id;
  230.     char idxName[255], buf[255];
  231.     if(sscanf(indtab.getName(), "%[^/]/%[^/]/%d/%s",
  232.       buf, buf, &id, idxName) != 4){
  233.       err << "Invalid index name format " << indtab.getName() << endl;
  234.       return false;
  235.     }
  236.     if(NdbDictInterface::create_index_obj_from_table(&idx, &indtab, &base))
  237.     {
  238.       err << "Failed to create index " << idxName
  239.   << " on " << split[2].c_str() << endl;
  240. return false;
  241.     }
  242.     idx->setName(idxName);
  243.     if(dict->createIndex(* idx) != 0)
  244.     {
  245.       delete idx;
  246.       err << "Failed to create index " << idxName
  247.   << " on " << split[2].c_str() << endl
  248.   << dict->getNdbError() << endl;
  250.       return false;
  251.     }
  252.     delete idx;
  253.     info << "Successfully created index " << idxName
  254.  << " on " << split[2].c_str() << endl;
  255.   }
  256.   return true;
  257. }
  259. void BackupRestore::tuple(const TupleS & tup)
  260. {
  261.   if (!m_restore) 
  262.     return;
  264.   while (m_free_callback == 0)
  265.   {
  266.     assert(m_transactions == m_parallelism);
  267.     // send-poll all transactions
  268.     // close transaction is done in callback
  269.     m_ndb->sendPollNdb(3000, 1);
  270.   }
  271.   
  272.   restore_callback_t * cb = m_free_callback;
  273.   
  274.   if (cb == 0)
  275.     assert(false);
  276.   
  277.   m_free_callback = cb->next;
  278.   cb->retries = 0;
  279.   cb->tup = tup; // must do copy!
  280.   tuple_a(cb);
  282. }
  284. void BackupRestore::tuple_a(restore_callback_t *cb)
  285. {
  286.   while (cb->retries < 10) 
  287.   {
  288.     /**
  289.      * start transactions
  290.      */
  291.     cb->connection = m_ndb->startTransaction();
  292.     if (cb->connection == NULL) 
  293.     {
  294.       if (errorHandler(cb)) 
  295.       {
  296. m_ndb->sendPollNdb(3000, 1);
  297. continue;
  298.       }
  299.       exitHandler();
  300.     } // if
  301.     
  302.     const TupleS &tup = cb->tup;
  303.     const NdbDictionary::Table * table = get_table(tup.getTable()->m_dictTable);
  305.     NdbOperation * op = cb->connection->getNdbOperation(table);
  306.     
  307.     if (op == NULL) 
  308.     {
  309.       if (errorHandler(cb)) 
  310. continue;
  311.       exitHandler();
  312.     } // if
  313.     
  314.     if (op->writeTuple() == -1) 
  315.     {
  316.       if (errorHandler(cb))
  317. continue;
  318.       exitHandler();
  319.     } // if
  320.     
  321.     int ret = 0;
  322.     for (int j = 0; j < 2; j++)
  323.     {
  324.       for (int i = 0; i < tup.getNoOfAttributes(); i++) 
  325.       {
  326. const AttributeDesc * attr_desc = tup.getDesc(i);
  327. const AttributeData * attr_data = tup.getData(i);
  328. int size = attr_desc->size;
  329. int arraySize = attr_desc->arraySize;
  330. char * dataPtr = attr_data->string_value;
  331. Uint32 length = (size * arraySize) / 8;
  333. if (j == 0 && tup.getTable()->have_auto_inc(i))
  334.   tup.getTable()->update_max_auto_val(dataPtr,size);
  336. if (attr_desc->m_column->getPrimaryKey())
  337. {
  338.   if (j == 1) continue;
  339.   ret = op->equal(i, dataPtr, length);
  340. }
  341. else
  342. {
  343.   if (j == 0) continue;
  344.   if (attr_data->null) 
  345.     ret = op->setValue(i, NULL, 0);
  346.   else
  347.     ret = op->setValue(i, dataPtr, length);
  348. }
  349. if (ret < 0) {
  350.   ndbout_c("Column: %d type %d %d %d %d",i,
  351.    attr_desc->m_column->getType(),
  352.    size, arraySize, attr_data->size);
  353.   break;
  354. }
  355.       }
  356.       if (ret < 0)
  357. break;
  358.     }
  359.     if (ret < 0)
  360.     {
  361.       if (errorHandler(cb)) 
  362. continue;
  363.       exitHandler();
  364.     }
  366.     // Prepare transaction (the transaction is NOT yet sent to NDB)
  367.     cb->connection->executeAsynchPrepare(Commit, &callback, cb);
  368.     m_transactions++;
  369.     return;
  370.   }
  371.   err << "Retried transaction " << cb->retries << " times.nLast error"
  372.       << m_ndb->getNdbError(cb->error_code) << endl
  373.       << "...Unable to recover from errors. Exiting..." << endl;
  374.   exitHandler();
  375. }
  377. void BackupRestore::cback(int result, restore_callback_t *cb)
  378. {
  379.   m_transactions--;
  381.   if (result < 0)
  382.   {
  383.     /**
  384.      * Error. temporary or permanent?
  385.      */
  386.     if (errorHandler(cb))
  387.       tuple_a(cb); // retry
  388.     else
  389.     {
  390.       err << "Restore: Failed to restore data due to a unrecoverable error. Exiting..." << endl;
  391.       exitHandler();
  392.     }
  393.   }
  394.   else
  395.   {
  396.     /**
  397.      * OK! close transaction
  398.      */
  399.     m_ndb->closeTransaction(cb->connection);
  400.     cb->connection= 0;
  401.     cb->next= m_free_callback;
  402.     m_free_callback= cb;
  403.     m_dataCount++;
  404.   }
  405. }
  407. /**
  408.  * returns true if is recoverable,
  409.  * Error handling based on hugo
  410.  *  false if it is an  error that generates an abort.
  411.  */
  412. bool BackupRestore::errorHandler(restore_callback_t *cb) 
  413. {
  414.   NdbError error;
  415.   if(cb->connection)
  416.   {
  417.     error= cb->connection->getNdbError();
  418.     m_ndb->closeTransaction(cb->connection);
  419.     cb->connection= 0;
  420.   }
  421.   else
  422.   {
  423.     error= m_ndb->getNdbError();
  424.   } 
  426.   Uint32 sleepTime = 100 + cb->retries * 300;
  427.   
  428.   cb->retries++;
  429.   cb->error_code = error.code;
  431.   switch(error.status)
  432.   {
  433.   case NdbError::Success:
  434.     return false;
  435.     // ERROR!
  436.     break;
  437.     
  438.   case NdbError::TemporaryError:
  439.     err << "Temporary error: " << error << endl;
  440.     NdbSleep_MilliSleep(sleepTime);
  441.     return true;
  442.     // RETRY
  443.     break;
  444.     
  445.   case NdbError::UnknownResult:
  446.     err << error << endl;
  447.     return false;
  448.     // ERROR!
  449.     break;
  450.     
  451.   default:
  452.   case NdbError::PermanentError:
  453.     //ERROR
  454.     err << error << endl;
  455.     return false;
  456.     break;
  457.   }
  458.   return false;
  459. }
  461. void BackupRestore::exitHandler() 
  462. {
  463.   release();
  464.   NDBT_ProgramExit(NDBT_FAILED);
  465.   if (opt_core)
  466.     abort();
  467.   else
  468.     exit(NDBT_FAILED);
  469. }
  472. void
  473. BackupRestore::tuple_free()
  474. {
  475.   if (!m_restore)
  476.     return;
  478.   // Poll all transactions
  479.   while (m_transactions)
  480.   {
  481.     m_ndb->sendPollNdb(3000);
  482.   }
  483. }
  485. void
  486. BackupRestore::endOfTuples()
  487. {
  488.   tuple_free();
  489. }
  491. void
  492. BackupRestore::logEntry(const LogEntry & tup)
  493. {
  494.   if (!m_restore)
  495.     return;
  497.   NdbConnection * trans = m_ndb->startTransaction();
  498.   if (trans == NULL) 
  499.   {
  500.     // Deep shit, TODO: handle the error
  501.     err << "Cannot start transaction" << endl;
  502.     exitHandler();
  503.   } // if
  504.   
  505.   const NdbDictionary::Table * table = get_table(tup.m_table->m_dictTable);
  506.   NdbOperation * op = trans->getNdbOperation(table);
  507.   if (op == NULL) 
  508.   {
  509.     err << "Cannot get operation: " << trans->getNdbError() << endl;
  510.     exitHandler();
  511.   } // if
  512.   
  513.   int check = 0;
  514.   switch(tup.m_type)
  515.   {
  516.   case LogEntry::LE_INSERT:
  517.     check = op->insertTuple();
  518.     break;
  519.   case LogEntry::LE_UPDATE:
  520.     check = op->updateTuple();
  521.     break;
  522.   case LogEntry::LE_DELETE:
  523.     check = op->deleteTuple();
  524.     break;
  525.   default:
  526.     err << "Log entry has wrong operation type."
  527.    << " Exiting...";
  528.     exitHandler();
  529.   }
  531.   if (check != 0) 
  532.   {
  533.     err << "Error defining op: " << trans->getNdbError() << endl;
  534.     exitHandler();
  535.   } // if
  536.   
  537.   Bitmask<4096> keys;
  538.   for (Uint32 i= 0; i < tup.size(); i++) 
  539.   {
  540.     const AttributeS * attr = tup[i];
  541.     int size = attr->Desc->size;
  542.     int arraySize = attr->Desc->arraySize;
  543.     const char * dataPtr = attr->Data.string_value;
  544.     
  545.     if (tup.m_table->have_auto_inc(attr->Desc->attrId))
  546.       tup.m_table->update_max_auto_val(dataPtr,size);
  548.     const Uint32 length = (size / 8) * arraySize;
  549.     if (attr->Desc->m_column->getPrimaryKey())
  550.     {
  551.       if(!keys.get(attr->Desc->attrId))
  552.       {
  553. keys.set(attr->Desc->attrId);
  554. check= op->equal(attr->Desc->attrId, dataPtr, length);
  555.       }
  556.     }
  557.     else
  558.       check= op->setValue(attr->Desc->attrId, dataPtr, length);
  559.     
  560.     if (check != 0) 
  561.     {
  562.       err << "Error defining op: " << trans->getNdbError() << endl;
  563.       exitHandler();
  564.     } // if
  565.   }
  566.   
  567.   const int ret = trans->execute(Commit);
  568.   if (ret != 0)
  569.   {
  570.     // Both insert update and delete can fail during log running
  571.     // and it's ok
  572.     // TODO: check that the error is either tuple exists or tuple does not exist?
  573.     bool ok= false;
  574.     NdbError errobj= trans->getNdbError();
  575.     switch(tup.m_type)
  576.     {
  577.     case LogEntry::LE_INSERT:
  578.       if(errobj.status == NdbError::PermanentError &&
  579.  errobj.classification == NdbError::ConstraintViolation)
  580. ok= true;
  581.       break;
  582.     case LogEntry::LE_UPDATE:
  583.     case LogEntry::LE_DELETE:
  584.       if(errobj.status == NdbError::PermanentError &&
  585.  errobj.classification == NdbError::NoDataFound)
  586. ok= true;
  587.       break;
  588.     }
  589.     if (!ok)
  590.     {
  591.       err << "execute failed: " << errobj << endl;
  592.       exitHandler();
  593.     }
  594.   }
  595.   
  596.   m_ndb->closeTransaction(trans);
  597.   m_logCount++;
  598. }
  600. void
  601. BackupRestore::endOfLogEntrys()
  602. {
  603.   if (!m_restore)
  604.     return;
  606.   info << "Restored " << m_dataCount << " tuples and "
  607.        << m_logCount << " log entries" << endl;
  608. }
  610. /*
  611.  *   callback : This is called when the transaction is polled
  612.  *              
  613.  *   (This function must have three arguments: 
  614.  *   - The result of the transaction, 
  615.  *   - The NdbConnection object, and 
  616.  *   - A pointer to an arbitrary object.)
  617.  */
  619. static void
  620. callback(int result, NdbConnection* trans, void* aObject)
  621. {
  622.   restore_callback_t *cb = (restore_callback_t *)aObject;
  623.   (cb->restore)->cback(result, cb);
  624. }
  626. #if 0 // old tuple impl
  627. void
  628. BackupRestore::tuple(const TupleS & tup)
  629. {
  630.   if (!m_restore)
  631.     return;
  632.   while (1) 
  633.   {
  634.     NdbConnection * trans = m_ndb->startTransaction();
  635.     if (trans == NULL) 
  636.     {
  637.       // Deep shit, TODO: handle the error
  638.       ndbout << "Cannot start transaction" << endl;
  639.       exitHandler();
  640.     } // if
  641.     
  642.     const TableS * table = tup.getTable();
  643.     NdbOperation * op = trans->getNdbOperation(table->getTableName());
  644.     if (op == NULL) 
  645.     {
  646.       ndbout << "Cannot get operation: ";
  647.       ndbout << trans->getNdbError() << endl;
  648.       exitHandler();
  649.     } // if
  650.     
  651.     // TODO: check return value and handle error
  652.     if (op->writeTuple() == -1) 
  653.     {
  654.       ndbout << "writeTuple call failed: ";
  655.       ndbout << trans->getNdbError() << endl;
  656.       exitHandler();
  657.     } // if
  658.     
  659.     for (int i = 0; i < tup.getNoOfAttributes(); i++) 
  660.     {
  661.       const AttributeS * attr = tup[i];
  662.       int size = attr->Desc->size;
  663.       int arraySize = attr->Desc->arraySize;
  664.       const char * dataPtr = attr->Data.string_value;
  665.       
  666.       const Uint32 length = (size * arraySize) / 8;
  667.       if (attr->Desc->m_column->getPrimaryKey()) 
  668. op->equal(i, dataPtr, length);
  669.     }
  670.     
  671.     for (int i = 0; i < tup.getNoOfAttributes(); i++) 
  672.     {
  673.       const AttributeS * attr = tup[i];
  674.       int size = attr->Desc->size;
  675.       int arraySize = attr->Desc->arraySize;
  676.       const char * dataPtr = attr->Data.string_value;
  677.       
  678.       const Uint32 length = (size * arraySize) / 8;
  679.       if (!attr->Desc->m_column->getPrimaryKey())
  680. if (attr->Data.null)
  681.   op->setValue(i, NULL, 0);
  682. else
  683.   op->setValue(i, dataPtr, length);
  684.     }
  685.     int ret = trans->execute(Commit);
  686.     if (ret != 0)
  687.     {
  688.       ndbout << "execute failed: ";
  689.       ndbout << trans->getNdbError() << endl;
  690.       exitHandler();
  691.     }
  692.     m_ndb->closeTransaction(trans);
  693.     if (ret == 0)
  694.       break;
  695.   }
  696.   m_dataCount++;
  697. }
  698. #endif
  700. template class Vector<NdbDictionary::Table*>;
  701. template class Vector<const NdbDictionary::Table*>;