src/ceph/src/test/bench/small_io_bench_fs.cc

   1 // -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
   2
   3 #include <boost/scoped_ptr.hpp>
   4 #include <boost/lexical_cast.hpp>
   5 #include <boost/program_options/option.hpp>
   6 #include <boost/program_options/options_description.hpp>
   7 #include <boost/program_options/variables_map.hpp>
   8 #include <boost/program_options/cmdline.hpp>
   9 #include <boost/program_options/parsers.hpp>
  10 #include <iostream>
  11 #include <set>
  12 #include <sstream>
  13 #include <stdlib.h>
  14 #include <fstream>
  15
  16 #include "common/Formatter.h"
  17
  18 #include "bencher.h"
  19 #include "rados_backend.h"
  20 #include "detailed_stat_collector.h"
  21 #include "distribution.h"
  22 #include "global/global_init.h"
  23 #include "os/filestore/FileStore.h"
  24 #include "testfilestore_backend.h"
  25 #include "common/perf_counters.h"
  26
  27 namespace po = boost::program_options;
  28 using namespace std;
  29
  30 struct MorePrinting : public DetailedStatCollector::AdditionalPrinting {
  31   CephContext *cct;
  32   explicit MorePrinting(CephContext *cct) : cct(cct) {}
  33   void operator()(std::ostream *out) override {
  34     bufferlist bl;
  35     Formatter *f = Formatter::create("json-pretty");
  36     cct->get_perfcounters_collection()->dump_formatted(f, 0);
  37     f->flush(bl);
  38     delete f;
  39     bl.append('\0');
  40     *out << bl.c_str() << std::endl;
  41   }
  42 };
  43
  44 int main(int argc, char **argv)
  45 {
  46   po::options_description desc("Allowed options");
  47   desc.add_options()
  48     ("help", "produce help message")
  49     ("num-concurrent-ops", po::value<unsigned>()->default_value(10),
  50      "set number of concurrent ops")
  51     ("num-objects", po::value<unsigned>()->default_value(500),
  52      "set number of objects to use")
  53     ("object-size", po::value<unsigned>()->default_value(4<<20),
  54      "set object size")
  55     ("io-size", po::value<unsigned>()->default_value(4<<10),
  56      "set io size")
  57     ("write-ratio", po::value<double>()->default_value(0.75),
  58      "set ratio of read to write")
  59     ("duration", po::value<unsigned>()->default_value(0),
  60      "set max duration, 0 for unlimited")
  61     ("max-ops", po::value<unsigned>()->default_value(0),
  62      "set max ops, 0 for unlimited")
  63     ("seed", po::value<unsigned>(),
  64      "seed")
  65     ("num-colls", po::value<unsigned>()->default_value(20),
  66      "number of collections")
  67     ("op-dump-file", po::value<string>()->default_value(""),
  68      "set file for dumping op details, omit for stderr")
  69     ("filestore-path", po::value<string>(),
  70      "path to filestore directory, mandatory")
  71     ("journal-path", po::value<string>(),
  72      "path to journal, mandatory")
  73     ("offset-align", po::value<unsigned>()->default_value(4096),
  74      "align offset by")
  75     ("write-infos", po::value<bool>()->default_value(false),
  76       "write info objects with main writes")
  77     ("sequential", po::value<bool>()->default_value(false),
  78      "do sequential writes like rbd")
  79     ("disable-detailed-ops", po::value<bool>()->default_value(false),
  80      "don't dump per op stats")
  81     ("num-writers", po::value<unsigned>()->default_value(1),
  82      "num write threads")
  83     ;
  84
  85   vector<string> ceph_option_strings;
  86   po::variables_map vm;
  87   try {
  88     po::parsed_options parsed =
  89       po::command_line_parser(argc, argv).options(desc).allow_unregistered().run();
  90     po::store(
  91               parsed,
  92               vm);
  93     po::notify(vm);
  94
  95     ceph_option_strings = po::collect_unrecognized(parsed.options,
  96                                                    po::include_positional);
  97   } catch(po::error &e) {
  98     std::cerr << e.what() << std::endl;
  99     return 1;
 100   }
 101   vector<const char *> ceph_options, def_args;
 102   ceph_options.reserve(ceph_option_strings.size());
 103   for (vector<string>::iterator i = ceph_option_strings.begin();
 104        i != ceph_option_strings.end();
 105        ++i) {
 106     ceph_options.push_back(i->c_str());
 107   }
 108
 109   auto cct = global_init(
 110     &def_args, ceph_options, CEPH_ENTITY_TYPE_CLIENT,
 111     CODE_ENVIRONMENT_UTILITY,
 112     CINIT_FLAG_NO_DEFAULT_CONFIG_FILE);
 113   common_init_finish(g_ceph_context);
 114   g_ceph_context->_conf->apply_changes(NULL);
 115
 116   if (!vm.count("filestore-path") || !vm.count("journal-path")) {
 117     cout << "Must provide filestore-path and journal-path" << std::endl
 118          << desc << std::endl;
 119     return 1;
 120   }
 121
 122   if (vm.count("help")) {
 123     cout << desc << std::endl;
 124     return 1;
 125   }
 126
 127   rngen_t rng;
 128   if (vm.count("seed"))
 129     rng = rngen_t(vm["seed"].as<unsigned>());
 130
 131   set<pair<double, Bencher::OpType> > ops;
 132   ops.insert(make_pair(vm["write-ratio"].as<double>(), Bencher::WRITE));
 133   ops.insert(make_pair(1-vm["write-ratio"].as<double>(), Bencher::READ));
 134
 135   FileStore fs(g_ceph_context, vm["filestore-path"].as<string>(),
 136                vm["journal-path"].as<string>());
 137   ObjectStore::Sequencer osr(__func__);
 138
 139   if (fs.mkfs() < 0) {
 140     cout << "mkfs failed" << std::endl;
 141     return 1;
 142   }
 143
 144   if (fs.mount() < 0) {
 145     cout << "mount failed" << std::endl;
 146     return 1;
 147   }
 148
 149   ostream *detailed_ops = 0;
 150   ofstream myfile;
 151   if (vm["disable-detailed-ops"].as<bool>()) {
 152     detailed_ops = 0;
 153   } else if (vm["op-dump-file"].as<string>().size()) {
 154     myfile.open(vm["op-dump-file"].as<string>().c_str());
 155     detailed_ops = &myfile;
 156   } else {
 157     detailed_ops = &cerr;
 158   }
 159
 160   ceph::shared_ptr<StatCollector> col(
 161     new DetailedStatCollector(
 162       1, new JSONFormatter, detailed_ops, &cout,
 163       new MorePrinting(g_ceph_context)));
 164
 165   cout << "Creating objects.." << std::endl;
 166   bufferlist bl;
 167   for (uint64_t i = 0; i < vm["object-size"].as<unsigned>(); ++i) {
 168     bl.append(0);
 169   }
 170
 171   for (uint64_t num = 0; num < vm["num-colls"].as<unsigned>(); ++num) {
 172     spg_t pgid(pg_t(num, 0), shard_id_t::NO_SHARD);
 173     std::cout << "collection " << pgid << std::endl;
 174     ObjectStore::Transaction t;
 175     t.create_collection(coll_t(pgid), 0);
 176     fs.apply_transaction(&osr, std::move(t));
 177   }
 178   {
 179     ObjectStore::Transaction t;
 180     t.create_collection(coll_t(), 0);
 181     fs.apply_transaction(&osr, std::move(t));
 182   }
 183
 184   vector<ceph::shared_ptr<Bencher> > benchers(
 185     vm["num-writers"].as<unsigned>());
 186   for (vector<ceph::shared_ptr<Bencher> >::iterator i = benchers.begin();
 187        i != benchers.end();
 188        ++i) {
 189     set<string> objects;
 190     for (uint64_t num = 0; num < vm["num-objects"].as<unsigned>(); ++num) {
 191       unsigned col_num = num % vm["num-colls"].as<unsigned>();
 192       spg_t pgid(pg_t(col_num, 0), shard_id_t::NO_SHARD);
 193       stringstream obj;
 194       obj << "obj_" << num << "_bencher_" << (i - benchers.begin());
 195       objects.insert(coll_t(pgid).to_str() + string("/") + obj.str());
 196     }
 197     Distribution<
 198       boost::tuple<string, uint64_t, uint64_t, Bencher::OpType> > *gen = 0;
 199     if (vm["sequential"].as<bool>()) {
 200       std::cout << "Using Sequential generator" << std::endl;
 201       gen = new SequentialLoad(
 202         objects,
 203         vm["object-size"].as<unsigned>(),
 204         vm["io-size"].as<unsigned>(),
 205         new WeightedDist<Bencher::OpType>(rng, ops)
 206         );
 207     } else {
 208       std::cout << "Using random generator" << std::endl;
 209       gen = new FourTupleDist<string, uint64_t, uint64_t, Bencher::OpType>(
 210         new RandomDist<string>(rng, objects),
 211         new Align(
 212           new UniformRandom(
 213             rng,
 214             0,
 215             vm["object-size"].as<unsigned>() - vm["io-size"].as<unsigned>()),
 216           vm["offset-align"].as<unsigned>()
 217           ),
 218         new Uniform(vm["io-size"].as<unsigned>()),
 219         new WeightedDist<Bencher::OpType>(rng, ops)
 220         );
 221     }
 222
 223     Bencher *bencher = new Bencher(
 224       gen,
 225       col,
 226       new TestFileStoreBackend(&fs, vm["write-infos"].as<bool>()),
 227       vm["num-concurrent-ops"].as<unsigned>(),
 228       vm["duration"].as<unsigned>(),
 229       vm["max-ops"].as<unsigned>());
 230
 231     bencher->init(objects, vm["object-size"].as<unsigned>(), &std::cout);
 232     cout << "Created objects..." << std::endl;
 233     (*i).reset(bencher);
 234   }
 235
 236   for (vector<ceph::shared_ptr<Bencher> >::iterator i = benchers.begin();
 237        i != benchers.end();
 238        ++i) {
 239     (*i)->create("bencher");
 240   }
 241   for (vector<ceph::shared_ptr<Bencher> >::iterator i = benchers.begin();
 242        i != benchers.end();
 243        ++i) {
 244     (*i)->join();
 245   }
 246
 247   fs.umount();
 248   if (vm["op-dump-file"].as<string>().size()) {
 249     myfile.close();
 250   }
 251   return 0;
 252 }