src/ceph/src/test/bench/detailed_stat_collector.cc

   1 // -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
   2
   3 #include "detailed_stat_collector.h"
   4 #include <sys/time.h>
   5 #include <utility>
   6 #include <boost/tuple/tuple.hpp>
   7
   8 void DetailedStatCollector::Op::dump(
   9   ostream *out,
  10   Formatter *f)
  11 {
  12   if (!out)
  13     return;
  14   f->open_object_section(type.c_str());
  15   f->dump_string("type", type);
  16   f->dump_float("start", start);
  17   f->dump_float("latency", latency);
  18   f->dump_int("size", size);
  19   f->dump_int("seq", seq);
  20   f->close_section();
  21   f->flush(*out);
  22   *out << std::endl;
  23 }
  24
  25 static utime_t cur_time()
  26 {
  27   struct timeval tv;
  28   gettimeofday(&tv, 0);
  29   return utime_t(&tv);
  30 }
  31
  32 DetailedStatCollector::Aggregator::Aggregator()
  33   : recent_size(0), total_size(0), recent_latency(0),
  34     total_latency(0), recent_ops(0), total_ops(0), started(false)
  35 {}
  36
  37 void DetailedStatCollector::Aggregator::add(const Op &op)
  38 {
  39   if (!started) {
  40     last = first = op.start;
  41     started = true;
  42   }
  43   ++recent_ops;
  44   ++total_ops;
  45   recent_size += op.size;
  46   total_size += op.size;
  47   recent_latency += op.latency;
  48   total_latency += op.latency;
  49 }
  50
  51 void DetailedStatCollector::Aggregator::dump(Formatter *f)
  52 {
  53   utime_t now = cur_time();
  54   f->dump_stream("time") << now;
  55   f->dump_float("avg_recent_latency", recent_latency / recent_ops);
  56   f->dump_float("avg_total_latency", total_latency / total_ops);
  57   f->dump_float("avg_recent_iops", recent_ops / (now - last));
  58   f->dump_float("avg_total_iops", total_ops / (now - first));
  59   f->dump_float("avg_recent_throughput", recent_size / (now - last));
  60   f->dump_float("avg_total_throughput", total_size / (now - first));
  61   f->dump_float("avg_recent_throughput_mb",
  62                 (recent_size / (now - last)) / (1024*1024));
  63   f->dump_float("avg_total_throughput_mb",
  64                 (total_size / (now - first)) / (1024*1024));
  65   f->dump_float("duration", now - last);
  66   last = now;
  67   recent_latency = 0;
  68   recent_size = 0;
  69   recent_ops = 0;
  70 }
  71
  72 DetailedStatCollector::DetailedStatCollector(
  73   double bin_size,
  74   Formatter *formatter,
  75   ostream *out,
  76   ostream *summary_out,
  77   AdditionalPrinting *details
  78   ) : bin_size(bin_size), f(formatter), out(out),
  79       summary_out(summary_out), details(details),
  80       lock("Stat::lock"), cur_seq(0) {
  81   last_dump = cur_time();
  82 }
  83
  84 uint64_t DetailedStatCollector::next_seq()
  85 {
  86   Mutex::Locker l(lock);
  87   if (summary_out && ((cur_time() - last_dump) > bin_size)) {
  88     f->open_object_section("stats");
  89     for (map<string, Aggregator>::iterator i = aggregators.begin();
  90          i != aggregators.end();
  91          ++i) {
  92       f->open_object_section(i->first.c_str());
  93       i->second.dump(f.get());
  94       f->close_section();
  95     }
  96     f->close_section();
  97     f->flush(*summary_out);
  98     *summary_out << std::endl;
  99     if (details) {
 100       (*details)(summary_out);
 101       *summary_out << std::endl;
 102     }
 103     last_dump = cur_time();
 104   }
 105   return cur_seq++;
 106 }
 107
 108 void DetailedStatCollector::start_write(uint64_t seq, uint64_t length)
 109 {
 110   Mutex::Locker l(lock);
 111   utime_t now(cur_time());
 112   not_committed.insert(make_pair(seq, make_pair(length, now)));
 113   not_applied.insert(make_pair(seq, make_pair(length, now)));
 114 }
 115
 116 void DetailedStatCollector::start_read(uint64_t seq, uint64_t length)
 117 {
 118   Mutex::Locker l(lock);
 119   utime_t now(cur_time());
 120   not_read.insert(make_pair(seq, make_pair(length, now)));
 121 }
 122
 123 void DetailedStatCollector::write_applied(uint64_t seq)
 124 {
 125   Mutex::Locker l(lock);
 126   Op op(
 127     "write_applied",
 128     not_applied[seq].second,
 129     cur_time() - not_applied[seq].second,
 130     not_applied[seq].first,
 131     seq);
 132   op.dump(out, f.get());
 133   aggregators["write_applied"].add(op);
 134   not_applied.erase(seq);
 135 }
 136
 137 void DetailedStatCollector::write_committed(uint64_t seq)
 138 {
 139   Mutex::Locker l(lock);
 140   Op op(
 141     "write_committed",
 142     not_committed[seq].second,
 143     cur_time() - not_committed[seq].second,
 144     not_committed[seq].first,
 145     seq);
 146   op.dump(out, f.get());
 147   aggregators["write_committed"].add(op);
 148   not_committed.erase(seq);
 149 }
 150
 151 void DetailedStatCollector::read_complete(uint64_t seq)
 152 {
 153   Mutex::Locker l(lock);
 154   Op op(
 155     "read",
 156     not_read[seq].second,
 157     cur_time() - not_read[seq].second,
 158     not_read[seq].first,
 159     seq);
 160   op.dump(out, f.get());
 161   aggregators["read"].add(op);
 162   not_read.erase(seq);
 163 }