release/src/router/mysql/storage/ndb/test/tools/hugoPkUpdate.cpp

   1 /* Copyright (c) 2003-2008 MySQL AB
   2
   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; version 2 of the License.
   6
   7    This program is distributed in the hope that it will be useful,
   8    but WITHOUT ANY WARRANTY; without even the implied warranty of
   9    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  10    GNU General Public License for more details.
  11
  12    You should have received a copy of the GNU General Public License
  13    along with this program; if not, write to the Free Software
  14    Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA */
  15
  16 #include <ndb_global.h>
  17
  18 #include <NdbOut.hpp>
  19
  20 #include <NdbApi.hpp>
  21 #include <NdbMain.h>
  22 #include <NDBT.hpp>
  23 #include <NDBT_Thread.hpp>
  24 #include <NDBT_Stats.hpp>
  25 #include <NdbSleep.h>
  26 #include <getarg.h>
  27
  28 #include <HugoTransactions.hpp>
  29
  30 static NDBT_ThreadFunc hugoPkUpdate;
  31
  32 struct ThrInput {
  33   const NdbDictionary::Table* pTab;
  34   int records;
  35   int batch;
  36   int stats;
  37 };
  38
  39 struct ThrOutput {
  40   NDBT_Stats latency;
  41 };
  42
  43 int main(int argc, const char** argv){
  44   ndb_init();
  45
  46   int _records = 0;
  47   int _loops = 1;
  48   int _threads = 1;
  49   int _stats = 0;
  50   int _abort = 0;
  51   int _batch = 1;
  52   const char* _tabname = NULL, *db = 0;
  53   int _help = 0;
  54
  55   struct getargs args[] = {
  56     { "aborts", 'a', arg_integer, &_abort, "percent of transactions that are aborted", "abort%" },
  57     { "loops", 'l', arg_integer, &_loops, "number of times to run this program(0=infinite loop)", "loops" },
  58     { "threads", 't', arg_integer, &_threads, "number of threads (default 1)", "threads" },
  59     { "stats", 's', arg_flag, &_stats, "report latency per batch", "stats" },
  60     //    { "batch", 'b', arg_integer, &_batch, "batch value", "batch" },
  61     { "records", 'r', arg_integer, &_records, "Number of records", "records" },
  62     { "usage", '?', arg_flag, &_help, "Print help", "" },
  63     { "database", 'd', arg_string, &db, "Database", "" }
  64   };
  65   int num_args = sizeof(args) / sizeof(args[0]);
  66   int optind = 0;
  67   char desc[] =
  68     "tabname\n"\
  69     "This program will update all records in a table using PK\n";
  70
  71   if(getarg(args, num_args, argc, argv, &optind) ||
  72      argv[optind] == NULL || _records == 0 || _help) {
  73     arg_printusage(args, num_args, argv[0], desc);
  74     return NDBT_ProgramExit(NDBT_WRONGARGS);
  75   }
  76   _tabname = argv[optind];
  77
  78   // Connect to Ndb
  79   Ndb_cluster_connection con;
  80   if(con.connect(12, 5, 1) != 0)
  81   {
  82     return NDBT_ProgramExit(NDBT_FAILED);
  83   }
  84
  85   if (con.wait_until_ready(30,0) < 0)
  86   {
  87     ndbout << "Cluster nodes not ready in 30 seconds." << endl;
  88     return NDBT_ProgramExit(NDBT_FAILED);
  89   }
  90
  91   Ndb MyNdb( &con, db ? db : "TEST_DB" );
  92
  93   if(MyNdb.init() != 0){
  94     ERR(MyNdb.getNdbError());
  95     return NDBT_ProgramExit(NDBT_FAILED);
  96   }
  97
  98   // Check if table exists in db
  99   const NdbDictionary::Table * pTab = NDBT_Table::discoverTableFromDb(&MyNdb, _tabname);
 100   if(pTab == NULL){
 101     ndbout << " Table " << _tabname << " does not exist!" << endl;
 102     return NDBT_ProgramExit(NDBT_WRONGARGS);
 103   }
 104
 105   // threads
 106   NDBT_ThreadSet ths(_threads);
 107
 108   // create Ndb object for each thread
 109   if (ths.connect(&con, db ? db : "TEST_DB") == -1) {
 110     ndbout << "connect failed: err=" << ths.get_err() << endl;
 111     return NDBT_ProgramExit(NDBT_FAILED);
 112   }
 113
 114   // input is options
 115   ThrInput input;
 116   ths.set_input(&input);
 117   input.pTab = pTab;
 118   input.records = _records;
 119   input.batch = _batch;
 120   input.stats = _stats;
 121
 122   // output is stats
 123   ThrOutput output;
 124   ths.set_output<ThrOutput>();
 125
 126   int i = 0;
 127   while (i < _loops || _loops == 0) {
 128     ndbout << i << ": ";
 129
 130     ths.set_func(hugoPkUpdate);
 131     ths.start();
 132     ths.stop();
 133
 134     if (ths.get_err())
 135       NDBT_ProgramExit(NDBT_FAILED);
 136
 137     if (_stats) {
 138       NDBT_Stats latency;
 139
 140       // add stats from each thread
 141       int n;
 142       for (n = 0; n < ths.get_count(); n++) {
 143         NDBT_Thread& thr = ths.get_thread(n);
 144         ThrOutput* output = (ThrOutput*)thr.get_output();
 145         latency += output->latency;
 146       }
 147
 148       ndbout
 149         << "latency per batch (us): "
 150         << " samples=" << latency.getCount()
 151         << " min=" << (int)latency.getMin()
 152         << " max=" << (int)latency.getMax()
 153         << " mean=" << (int)latency.getMean()
 154         << " stddev=" << (int)latency.getStddev()
 155         << endl;
 156     }
 157     i++;
 158   }
 159
 160   return NDBT_ProgramExit(NDBT_OK);
 161 }
 162
 163 static void hugoPkUpdate(NDBT_Thread& thr)
 164 {
 165   const ThrInput* input = (const ThrInput*)thr.get_input();
 166   ThrOutput* output = (ThrOutput*)thr.get_output();
 167
 168   HugoTransactions hugoTrans(*input->pTab);
 169   output->latency.reset();
 170   if (input->stats)
 171     hugoTrans.setStatsLatency(&output->latency);
 172
 173   NDBT_ThreadSet& ths = thr.get_thread_set();
 174   hugoTrans.setThrInfo(ths.get_count(), thr.get_thread_no());
 175
 176   int ret;
 177   ret = hugoTrans.pkUpdateRecords(thr.get_ndb(),
 178                                   input->records,
 179                                   input->batch);
 180   if (ret != 0)
 181     thr.set_err(ret);
 182 }