silo/benchmarks/encstress.cc

   1 #include <iostream>
   2 #include <sstream>
   3 #include <vector>
   4 #include <utility>
   5 #include <string>
   6
   7 #include <stdlib.h>
   8 #include <unistd.h>
   9
  10 #include "../macros.h"
  11 #include "../varkey.h"
  12 #include "../thread.h"
  13 #include "../util.h"
  14 #include "../spinbarrier.h"
  15
  16 #include "../record/encoder.h"
  17 #include "bench.h"
  18
  19 using namespace std;
  20 using namespace util;
  21
  22 static size_t nkeys;
  23
  24 #define ENCSTRESS_REC_KEY_FIELDS(x, y) \
  25   x(int32_t,k0)
  26 #define ENCSTRESS_REC_VALUE_FIELDS(x, y) \
  27   x(int32_t,f0) \
  28   y(int32_t,f1) \
  29   y(int32_t,f2) \
  30   y(int32_t,f3) \
  31   y(int32_t,f4) \
  32   y(int32_t,f5) \
  33   y(int32_t,f6) \
  34   y(int32_t,f7)
  35 DO_STRUCT(encstress_rec, ENCSTRESS_REC_KEY_FIELDS, ENCSTRESS_REC_VALUE_FIELDS)
  36
  37 class encstress_worker : public bench_worker {
  38 public:
  39   encstress_worker(
  40       unsigned int worker_id,
  41       unsigned long seed, abstract_db *db,
  42       const map<string, abstract_ordered_index *> &open_tables,
  43       spin_barrier *barrier_a, spin_barrier *barrier_b)
  44     : bench_worker(worker_id, false, seed, db,
  45                    open_tables, barrier_a, barrier_b),
  46       tbl(open_tables.at("table"))
  47   {
  48   }
  49
  50   txn_result
  51   txn_read()
  52   {
  53     void *txn = db->new_txn(txn_flags, arena, txn_buf());
  54     const string k = u64_varkey(r.next() % nkeys).str();
  55     try {
  56       string v;
  57       ALWAYS_ASSERT(tbl->get(txn, k, v));
  58       if (likely(db->commit_txn(txn)))
  59         return txn_result(true, 0);
  60     } catch (abstract_db::abstract_abort_exception &ex) {
  61       db->abort_txn(txn);
  62     }
  63     return txn_result(false, 0);
  64   }
  65
  66   static txn_result
  67   TxnRead(bench_worker *w)
  68   {
  69     return static_cast<encstress_worker *>(w)->txn_read();
  70   }
  71
  72   virtual workload_desc_vec
  73   get_workload() const
  74   {
  75     workload_desc_vec w;
  76     w.push_back(workload_desc("Read", 1.0, TxnRead));
  77     return w;
  78   }
  79
  80 private:
  81   abstract_ordered_index *tbl;
  82 };
  83
  84 class encstress_loader : public bench_loader {
  85 public:
  86   encstress_loader(unsigned long seed,
  87                    abstract_db *db,
  88                    const map<string, abstract_ordered_index *> &open_tables)
  89     : bench_loader(seed, db, open_tables)
  90   {}
  91
  92 protected:
  93   virtual void
  94   load()
  95   {
  96     abstract_ordered_index *tbl = open_tables.at("table");
  97     try {
  98       // load
  99       const size_t batchsize = (db->txn_max_batch_size() == -1) ?
 100         10000 : db->txn_max_batch_size();
 101       ALWAYS_ASSERT(batchsize > 0);
 102       const size_t nbatches = nkeys / batchsize;
 103       if (nbatches == 0) {
 104         void *txn = db->new_txn(txn_flags, arena, txn_buf());
 105         for (size_t j = 0; j < nkeys; j++) {
 106           const encstress_rec::key key(j);
 107           encstress_rec::value rec;
 108           rec.f0 = 1; rec.f1 = 1; rec.f2 = 1; rec.f3 = 1;
 109           rec.f4 = 1; rec.f5 = 1; rec.f6 = 1; rec.f7 = 1;
 110           string buf;
 111           tbl->insert(txn, Encode(key), Encode(buf, rec));
 112         }
 113         if (verbose)
 114           cerr << "batch 1/1 done" << endl;
 115         ALWAYS_ASSERT(db->commit_txn(txn));
 116       } else {
 117         for (size_t i = 0; i < nbatches; i++) {
 118           size_t keyend = (i == nbatches - 1) ? nkeys : (i + 1) * batchsize;
 119           void *txn = db->new_txn(txn_flags, arena, txn_buf());
 120           for (size_t j = i * batchsize; j < keyend; j++) {
 121             const encstress_rec::key key(j);
 122             encstress_rec::value rec;
 123             rec.f0 = 1; rec.f1 = 1; rec.f2 = 1; rec.f3 = 1;
 124             rec.f4 = 1; rec.f5 = 1; rec.f6 = 1; rec.f7 = 1;
 125             string buf;
 126             tbl->insert(txn, Encode(key), Encode(buf, rec));
 127           }
 128           if (verbose)
 129             cerr << "batch " << (i + 1) << "/" << nbatches << " done" << endl;
 130           ALWAYS_ASSERT(db->commit_txn(txn));
 131         }
 132       }
 133     } catch (abstract_db::abstract_abort_exception &ex) {
 134       // shouldn't abort on loading!
 135       ALWAYS_ASSERT(false);
 136     }
 137     if (verbose)
 138       cerr << "[INFO] finished loading USERTABLE" << endl;
 139   }
 140 };
 141
 142 class encstress_bench_runner : public bench_runner {
 143 public:
 144   encstress_bench_runner(abstract_db *db)
 145     : bench_runner(db)
 146   {
 147     open_tables["table"] = db->open_index("table", sizeof(encstress_rec));
 148   }
 149
 150 protected:
 151   virtual vector<bench_loader *>
 152   make_loaders()
 153   {
 154     vector<bench_loader *> ret;
 155     ret.push_back(new encstress_loader(0, db, open_tables));
 156     return ret;
 157   }
 158
 159   virtual vector<bench_worker *>
 160   make_workers()
 161   {
 162     fast_random r(8544290);
 163     vector<bench_worker *> ret;
 164     for (size_t i = 0; i < nthreads; i++)
 165       ret.push_back(
 166         new encstress_worker(
 167           i, r.next(), db, open_tables,
 168           &barrier_a, &barrier_b));
 169     return ret;
 170   }
 171 };
 172
 173 void
 174 encstress_do_test(abstract_db *db, int argc, char **argv)
 175 {
 176   nkeys = size_t(scale_factor * 1000.0);
 177   ALWAYS_ASSERT(nkeys > 0);
 178   encstress_bench_runner r(db);
 179   r.run();
 180 }