src/ceph/src/test/omap_bench.cc

   1 /*
   2  * Generate latency statistics for a configurable number of write
   3  * operations of configurable size.
   4  *
   5  *  Created on: May 21, 2012
   6  *      Author: Eleanor Cawthon
   7  *
   8  * This is free software; you can redistribute it and/or
   9  * modify it under the terms of the GNU Lesser General Public
  10  * License version 2.1, as published by the Free Software
  11  * Foundation.  See file COPYING.
  12  */
  13
  14 #include "include/rados/librados.hpp"
  15 #include "include/Context.h"
  16 #include "common/ceph_context.h"
  17 #include "common/Mutex.h"
  18 #include "common/Cond.h"
  19 #include "include/utime.h"
  20 #include "global/global_context.h"
  21 #include "common/ceph_argparse.h"
  22 #include "test/omap_bench.h"
  23
  24 #include <string>
  25 #include <iostream>
  26 #include <cassert>
  27 #include <climits>
  28 #include <cmath>
  29
  30 using namespace std;
  31 using ceph::bufferlist;
  32
  33 int OmapBench::setup(int argc, const char** argv) {
  34   //parse key_value_store_bench args
  35   vector<const char*> args;
  36   argv_to_vec(argc,argv,args);
  37   for (unsigned i = 0; i < args.size(); i++) {
  38     if(i < args.size() - 1) {
  39       if (strcmp(args[i], "-t") == 0) {
  40         threads = atoi(args[i+1]);
  41       } else if (strcmp(args[i], "-o") == 0) {
  42         objects = atoi(args[i+1]);
  43       } else if (strcmp(args[i], "--entries") == 0) {
  44         entries_per_omap = atoi(args[i+1]);
  45       } else if (strcmp(args[i], "--keysize") == 0) {
  46         key_size = atoi(args[i+1]);
  47       } else if (strcmp(args[i], "--valsize") == 0) {
  48         value_size = atoi(args[i+1]);
  49       } else if (strcmp(args[i], "--inc") == 0) {
  50         increment = atoi(args[i+1]);
  51       } else if (strcmp(args[i], "--omaptype") == 0) {
  52         if(strcmp("rand",args[i+1]) == 0) {
  53           omap_generator = OmapBench::generate_non_uniform_omap;
  54         }
  55         else if (strcmp("uniform", args[i+1]) == 0) {
  56           omap_generator = OmapBench::generate_uniform_omap;
  57         }
  58       } else if (strcmp(args[i], "--name") == 0) {
  59         rados_id = args[i+1];
  60       }
  61     } else if (strcmp(args[i], "--help") == 0) {
  62       cout << "\nUsage: ostorebench [options]\n"
  63            << "Generate latency statistics for a configurable number of "
  64            << "key value pair operations of\n"
  65            << "configurable size.\n\n"
  66            << "OPTIONS\n"
  67            << " -t              number of threads to use (default "<<threads;
  68       cout << ")\n"
  69            << " -o              number of objects to write (default "<<objects;
  70       cout << ")\n"
  71            << " --entries       number of entries per (default "
  72            << entries_per_omap;
  73       cout <<")\n"
  74            << " --keysize       number of characters per key "
  75            << "(default "<<key_size;
  76       cout << ")\n"
  77            << " --valsize       number of characters per value "
  78            << "(default "<<value_size;
  79       cout << ")\n"
  80            << " --inc           specify the increment to use in the displayed "
  81            << "histogram (default "<<increment;
  82       cout << ")\n"
  83            << " --omaptype      specify how omaps should be generated - "
  84            << "rand for random sizes between\n"
  85            << "                        0 and max size, uniform for all sizes"
  86            << " to be specified size.\n"
  87            << "                        (default uniform)\n";
  88       cout << " --name          the rados id to use (default "<< rados_id
  89            << ")\n";
  90       exit(1);
  91     }
  92   }
  93   int r = rados.init(rados_id.c_str());
  94   if (r < 0) {
  95     cout << "error during init" << std::endl;
  96     return r;
  97   }
  98   r = rados.conf_parse_argv(argc, argv);
  99   if (r < 0) {
 100     cout << "error during parsing args" << std::endl;
 101     return r;
 102   }
 103   r = rados.conf_parse_env(NULL);
 104   if (r < 0) {
 105     cout << "error during parsing env" << std::endl;
 106     return r;
 107   }
 108   r = rados.conf_read_file(NULL);
 109   if (r < 0) {
 110     cout << "error during read file" << std::endl;
 111     return r;
 112   }
 113   r = rados.connect();
 114   if (r < 0) {
 115     cout << "error during connect" << std::endl;
 116     return r;
 117   }
 118   r = rados.ioctx_create(pool_name.c_str(), io_ctx);
 119   if (r < 0) {
 120     cout << "error creating io ctx" << std::endl;
 121     rados.shutdown();
 122     return r;
 123   }
 124   return 0;
 125 }
 126
 127 //Writer functions
 128 Writer::Writer(OmapBench *omap_bench) : ob(omap_bench) {
 129   stringstream name;
 130   ob->data_lock.Lock();
 131   name << omap_bench->prefix << ++(ob->data.started_ops);
 132   ob->data_lock.Unlock();
 133   oid = name.str();
 134 }
 135 void Writer::start_time() {
 136   begin_time = ceph_clock_now();
 137 }
 138 void Writer::stop_time() {
 139   end_time = ceph_clock_now();
 140 }
 141 double Writer::get_time() {
 142   return (end_time - begin_time) * 1000;
 143 }
 144 string Writer::get_oid() {
 145   return oid;
 146 }
 147 std::map<std::string, bufferlist> & Writer::get_omap() {
 148   return omap;
 149 }
 150
 151 //AioWriter functions
 152 AioWriter::AioWriter(OmapBench *ob) : Writer(ob) {
 153   aioc = NULL;
 154 }
 155 AioWriter::~AioWriter() {
 156   if(aioc) aioc->release();
 157 }
 158 librados::AioCompletion * AioWriter::get_aioc() {
 159   return aioc;
 160 }
 161 void AioWriter::set_aioc(librados::callback_t complete,
 162     librados::callback_t safe) {
 163   aioc = ob->rados.aio_create_completion(this, complete, safe);
 164 }
 165
 166
 167 //Helper methods
 168 void OmapBench::aio_is_safe(rados_completion_t c, void *arg) {
 169   AioWriter *aiow = reinterpret_cast<AioWriter *>(arg);
 170   aiow->stop_time();
 171   Mutex * data_lock = &aiow->ob->data_lock;
 172   Mutex * thread_is_free_lock = &aiow->ob->thread_is_free_lock;
 173   Cond * thread_is_free = &aiow->ob->thread_is_free;
 174   int &busythreads_count = aiow->ob->busythreads_count;
 175   o_bench_data &data = aiow->ob->data;
 176   int INCREMENT = aiow->ob->increment;
 177   int err = aiow->get_aioc()->get_return_value();
 178   if (err < 0) {
 179     cout << "error writing AioCompletion";
 180     return;
 181   }
 182   double time = aiow->get_time();
 183   delete aiow;
 184   data_lock->Lock();
 185   data.avg_latency = (data.avg_latency * data.completed_ops + time)
 186       / (data.completed_ops + 1);
 187   data.completed_ops++;
 188   if (time < data.min_latency) {
 189     data.min_latency = time;
 190   }
 191   if (time > data.max_latency) {
 192     data.max_latency = time;
 193   }
 194   data.total_latency += time;
 195   ++(data.freq_map[time / INCREMENT]);
 196   if(data.freq_map[time/INCREMENT] > data.mode.second) {
 197     data.mode.first = time/INCREMENT;
 198     data.mode.second = data.freq_map[time/INCREMENT];
 199   }
 200   data_lock->Unlock();
 201
 202   thread_is_free_lock->Lock();
 203   busythreads_count--;
 204   thread_is_free->Signal();
 205   thread_is_free_lock->Unlock();
 206 }
 207
 208 string OmapBench::random_string(int len) {
 209   string ret;
 210   string alphanum = "0123456789"
 211       "ABCDEFGHIJKLMNOPQRSTUVWXYZ"
 212       "abcdefghijklmnopqrstuvwxyz";
 213
 214   for (int i = 0; i < len; ++i) {
 215     ret.push_back(alphanum[rand() % (alphanum.size() - 1)]);
 216   }
 217
 218   return ret;
 219 }
 220
 221 int OmapBench::run() {
 222   return (((OmapBench *)this)->*OmapBench::test)(omap_generator);
 223 }
 224
 225 int OmapBench::print_written_omap() {
 226   for (int i = 1; i <= objects; i++) {
 227     int err = 0;
 228     librados::ObjectReadOperation key_read;
 229     set<string> out_keys;
 230     map<string, bufferlist> out_vals;
 231     std::stringstream objstrm;
 232     objstrm << prefix;
 233     objstrm << i;
 234     cout << "\nPrinting omap for "<<objstrm.str() << std::endl;
 235     // FIXME: we ignore pmore here.  this shouldn't happen for benchmark
 236     // keys, though, unless the OSD limit is *really* low.
 237     key_read.omap_get_keys2("", LONG_MAX, &out_keys, nullptr, &err);
 238     io_ctx.operate(objstrm.str(), &key_read, NULL);
 239     if (err < 0) {
 240       cout << "error " << err;
 241       cout << " getting omap key set " << std::endl;
 242       return err;
 243     }
 244
 245     librados::ObjectReadOperation val_read;
 246     val_read.omap_get_vals_by_keys(out_keys, &out_vals, &err);
 247     if (err < 0) {
 248       cout << "error " << err;
 249       cout << " getting omap value set " << std::endl;
 250       return err;
 251     }
 252     io_ctx.operate(objstrm.str(), &val_read, NULL);
 253
 254     for (set<string>::iterator iter = out_keys.begin();
 255         iter != out_keys.end(); ++iter) {
 256       cout << *iter << "\t" << (out_vals)[*iter] << std::endl;
 257     }
 258   }
 259   return 0;
 260 }
 261
 262 void OmapBench::print_results() {
 263   cout << "========================================================";
 264   cout << "\nNumber of kvmaps written:\t" << objects;
 265   cout << "\nNumber of ops at once:\t" << threads;
 266   cout << "\nEntries per kvmap:\t\t" << entries_per_omap;
 267   cout << "\nCharacters per key:\t" << key_size;
 268   cout << "\nCharacters per val:\t" << value_size;
 269   cout << std::endl;
 270   cout << std::endl;
 271   cout << "Average latency:\t" << data.avg_latency;
 272   cout << "ms\nMinimum latency:\t" << data.min_latency;
 273   cout << "ms\nMaximum latency:\t" << data.max_latency;
 274   cout << "ms\nMode latency:\t\t"<<"between "<<data.mode.first * increment;
 275   cout << " and " <<data.mode.first * increment + increment;
 276   cout << "ms\nTotal latency:\t\t" << data.total_latency;
 277   cout << "ms"<<std::endl;
 278   cout << std::endl;
 279   cout << "Histogram:" << std::endl;
 280   for(int i = floor(data.min_latency / increment); i <
 281       ceil(data.max_latency / increment); i++) {
 282     cout << ">= "<< i * increment;
 283     cout << "ms";
 284     int spaces;
 285     if (i == 0) spaces = 4;
 286     else spaces = 3 - floor(log10(i));
 287     for (int j = 0; j < spaces; j++) {
 288       cout << " ";
 289     }
 290     cout << "[";
 291     for(int j = 0; j < ((data.freq_map)[i])*45/(data.mode.second); j++) {
 292       cout << "*";
 293     }
 294     cout << std::endl;
 295   }
 296   cout << "\n========================================================"
 297        << std::endl;
 298 }
 299
 300 int OmapBench::write_omap_asynchronously(AioWriter *aiow,
 301     const std::map<std::string,bufferlist> &omap) {
 302   librados::ObjectWriteOperation owo;
 303   owo.create(false);
 304   owo.omap_clear();
 305   owo.omap_set(omap);
 306   aiow->start_time();
 307   int err = io_ctx.aio_operate(aiow->get_oid(), aiow->get_aioc(), &owo);
 308   if (err < 0) {
 309     cout << "writing omap failed with code "<<err;
 310     cout << std::endl;
 311     return err;
 312   }
 313   return 0;
 314 }
 315
 316 //Omap Generators
 317 int OmapBench::generate_uniform_omap(const int omap_entries, const int key_size,
 318     const int value_size, std::map<std::string,bufferlist> * out_omap) {
 319   bufferlist bl;
 320
 321   //setup omap
 322   for (int i = 0; i < omap_entries; i++) {
 323     bufferlist omap_val;
 324     omap_val.append(random_string(value_size));
 325     string key = random_string(key_size);
 326     (*out_omap)[key]= omap_val;
 327   }
 328   return 0;
 329 }
 330
 331 int OmapBench::generate_non_uniform_omap(const int omap_entries,
 332     const int key_size, const int value_size,
 333     std::map<std::string,bufferlist> * out_omap) {
 334   bufferlist bl;
 335
 336   int num_entries = rand() % omap_entries + 1;
 337   int key_len = rand() % key_size +1;
 338   int val_len = rand() % value_size +1;
 339
 340   //setup omap
 341   for (int i = 0; i < num_entries; i++) {
 342     bufferlist omap_val;
 343     omap_val.append(random_string(val_len));
 344     string key = random_string(key_len);
 345     (*out_omap)[key] = omap_val;
 346   }
 347   return 0;
 348 }
 349
 350 int OmapBench::generate_small_non_random_omap(const int omap_entries,
 351     const int key_size, const int value_size,
 352     std::map<std::string,bufferlist> * out_omap) {
 353   bufferlist bl;
 354   stringstream key;
 355
 356   //setup omap
 357   for (int i = 0; i < omap_entries; i++) {
 358     bufferlist omap_val;
 359     omap_val.append("Value ");
 360     omap_val.append(i);
 361     key << "Key " << i;
 362     (*out_omap)[key.str()]= omap_val;
 363   }
 364   return 0;
 365 }
 366
 367 //tests
 368 int OmapBench::test_write_objects_in_parallel(omap_generator_t omap_gen) {
 369   comp = NULL;
 370   AioWriter *this_aio_writer;
 371
 372   Mutex::Locker l(thread_is_free_lock);
 373   for (int i = 0; i < objects; i++) {
 374     assert(busythreads_count <= threads);
 375     //wait for a writer to be free
 376     if (busythreads_count == threads) {
 377       int err = thread_is_free.Wait(thread_is_free_lock);
 378       assert(busythreads_count < threads);
 379       if (err < 0) {
 380         return err;
 381       }
 382     }
 383
 384     //set up the write
 385     this_aio_writer = new AioWriter(this);
 386     this_aio_writer->set_aioc(NULL,safe);
 387
 388     //perform the write
 389     busythreads_count++;
 390     int err = omap_gen(entries_per_omap, key_size, value_size,
 391         & this_aio_writer->get_omap());
 392     if (err < 0) {
 393       return err;
 394     }
 395     err = OmapBench::write_omap_asynchronously(this_aio_writer,
 396         (this_aio_writer->get_omap()));
 397
 398
 399     if (err < 0) {
 400       return err;
 401     }
 402   }
 403   while(busythreads_count > 0) {
 404     thread_is_free.Wait(thread_is_free_lock);
 405   }
 406
 407   return 0;
 408 }
 409
 410 /**
 411  * runs the specified test with the specified parameters and generates
 412  * a histogram of latencies
 413  */
 414 int main(int argc, const char** argv) {
 415   OmapBench ob;
 416   int err = ob.setup(argc, argv);
 417   if (err<0) {
 418     cout << "error during setup: "<<err;
 419     cout << std::endl;
 420     exit(1);
 421   }
 422   err = ob.run();
 423   if (err < 0) {
 424     cout << "writing objects failed with code " << err;
 425     cout << std::endl;
 426     return err;
 427   }
 428
 429   ob.print_results();
 430
 431   //uncomment to show omaps
 432   /*err = ob.return print_written_omap();
 433   if (err < 0) {
 434     cout << "printing omaps failed with code " << err;
 435     cout << std::endl;
 436     return err;
 437   }
 438   */
 439   return 0;
 440
 441 }