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

   1 /* Copyright (c) 2003-2005, 2007 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 hugoPkRead;
  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;
  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(not 0)", "batch" },
  61     { "records", 'r', arg_integer, &_records, "Number of records", "records" },
  62     { "usage", '?', arg_flag, &_help, "Print help", "" }
  63   };
  64   int num_args = sizeof(args) / sizeof(args[0]);
  65   int optind = 0;
  66   char desc[] =
  67     "tabname\n"\
  68     "This program will read 'r' records from one table in Ndb. \n"\
  69     "It will verify every column read by calculating the expected value.\n";
  70
  71   if(getarg(args, num_args, argc, argv, &optind) ||
  72      argv[optind] == NULL || _records == 0 || _batch == 0 || _help) {
  73     arg_printusage(args, num_args, argv[0], desc);
  74     return NDBT_ProgramExit(NDBT_WRONGARGS);
  75   }
  76   _tabname = argv[optind];
  77
  78
  79   // Connect to Ndb
  80   Ndb_cluster_connection con;
  81   if(con.connect(12, 5, 1) != 0)
  82   {
  83     return NDBT_ProgramExit(NDBT_FAILED);
  84   }
  85
  86   Ndb MyNdb(&con, "TEST_DB" );
  87
  88   if(MyNdb.init() != 0){
  89     ERR(MyNdb.getNdbError());
  90     return NDBT_ProgramExit(NDBT_FAILED);
  91   }
  92
  93   while(MyNdb.waitUntilReady() != 0)
  94     ndbout << "Waiting for ndb to become ready..." << endl;
  95
  96   // Check if table exists in db
  97   const NdbDictionary::Table * pTab = NDBT_Table::discoverTableFromDb(&MyNdb, _tabname);
  98   if(pTab == NULL){
  99     ndbout << " Table " << _tabname << " does not exist!" << endl;
 100     return NDBT_ProgramExit(NDBT_WRONGARGS);
 101   }
 102
 103   // threads
 104   NDBT_ThreadSet ths(_threads);
 105
 106   // create Ndb object for each thread
 107   if (ths.connect(&con, "TEST_DB") == -1) {
 108     ndbout << "connect failed: err=" << ths.get_err() << endl;
 109     return NDBT_ProgramExit(NDBT_FAILED);
 110   }
 111
 112   // input is options
 113   ThrInput input;
 114   ths.set_input(&input);
 115   input.pTab = pTab;
 116   input.records = _records;
 117   input.batch = _batch;
 118   input.stats = _stats;
 119
 120   // output is stats
 121   ThrOutput output;
 122   ths.set_output<ThrOutput>();
 123
 124   int i = 0;
 125   while (i < _loops || _loops == 0) {
 126     ndbout << i << ": ";
 127
 128     ths.set_func(hugoPkRead);
 129     ths.start();
 130     ths.stop();
 131
 132     if (ths.get_err())
 133       NDBT_ProgramExit(NDBT_FAILED);
 134
 135     if (_stats) {
 136       NDBT_Stats latency;
 137
 138       // add stats from each thread
 139       int n;
 140       for (n = 0; n < ths.get_count(); n++) {
 141         NDBT_Thread& thr = ths.get_thread(n);
 142         ThrOutput* output = (ThrOutput*)thr.get_output();
 143         latency += output->latency;
 144       }
 145
 146       ndbout
 147         << "latency per batch (us): "
 148         << " samples=" << latency.getCount()
 149         << " min=" << (int)latency.getMin()
 150         << " max=" << (int)latency.getMax()
 151         << " mean=" << (int)latency.getMean()
 152         << " stddev=" << (int)latency.getStddev()
 153         << endl;
 154     }
 155     i++;
 156   }
 157
 158   return NDBT_ProgramExit(NDBT_OK);
 159 }
 160
 161 static void hugoPkRead(NDBT_Thread& thr)
 162 {
 163   const ThrInput* input = (const ThrInput*)thr.get_input();
 164   ThrOutput* output = (ThrOutput*)thr.get_output();
 165
 166   HugoTransactions hugoTrans(*input->pTab);
 167   output->latency.reset();
 168   if (input->stats)
 169     hugoTrans.setStatsLatency(&output->latency);
 170
 171   int ret;
 172   ret = hugoTrans.pkReadRecords(thr.get_ndb(),
 173                                 input->records,
 174                                 input->batch);
 175   if (ret != 0)
 176     thr.set_err(ret);
 177 }