src/ceph/src/os/filestore/JournalThrottle.h

   1 // -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
   2 // vim: ts=8 sw=2 smarttab
   3
   4 #ifndef CEPH_JOURNAL_THROTTLE_H
   5 #define CEPH_JOURNAL_THROTTLE_H
   6
   7 #include "common/Throttle.h"
   8
   9 #include <list>
  10 #include <deque>
  11 #include <condition_variable>
  12 #include <thread>
  13 #include <vector>
  14 #include <chrono>
  15 #include <iostream>
  16
  17 /**
  18  * JournalThrottle
  19  *
  20  * Throttle designed to implement dynamic throttling as the journal fills
  21  * up.  The goal is to not delay ops at all when the journal is relatively
  22  * empty, delay ops somewhat as the journal begins to fill (with the delay
  23  * getting linearly longer as the journal fills up to a high water mark),
  24  * and to delay much more aggressively (though still linearly with usage)
  25  * until we hit the max value.
  26  *
  27  * The implementation simply wraps BackoffThrottle with a queue of
  28  * journaled but not synced ops.
  29  *
  30  * The usage pattern is as follows:
  31  * 1) Call get(seq, bytes) before taking the op_queue_throttle
  32  * 2) Once the journal is flushed, flush(max_op_id_flushed)
  33  */
  34 class JournalThrottle {
  35   BackoffThrottle throttle;
  36
  37   std::mutex lock;
  38   /// deque<id, count>
  39   std::deque<std::pair<uint64_t, uint64_t> > journaled_ops;
  40   using locker = std::unique_lock<std::mutex>;
  41
  42 public:
  43   /**
  44    * set_params
  45    *
  46    * Sets params.  If the params are invalid, returns false
  47    * and populates errstream (if non-null) with a user compreshensible
  48    * explanation.
  49    */
  50   bool set_params(
  51     double low_threshhold,
  52     double high_threshhold,
  53     double expected_throughput,
  54     double high_multiple,
  55     double max_multiple,
  56     uint64_t throttle_max,
  57     std::ostream *errstream);
  58
  59   /**
  60    * gets specified throttle for id mono_id, waiting as necessary
  61    *
  62    * @param c [in] amount to take
  63    * @return duration waited
  64    */
  65   std::chrono::duration<double> get(uint64_t c);
  66
  67   /**
  68    * take
  69    *
  70    * Takes specified throttle without waiting
  71    */
  72   uint64_t take(uint64_t c);
  73
  74   /**
  75    * register_throttle_seq
  76    *
  77    * Registers a sequence number with an amount of throttle to
  78    * release upon flush()
  79    *
  80    * @param seq [in] seq
  81    */
  82   void register_throttle_seq(uint64_t seq, uint64_t c);
  83
  84
  85   /**
  86    * Releases throttle held by ids <= mono_id
  87    *
  88    * @param mono_id [in] id up to which to flush
  89    * @returns pair<ops_flushed, bytes_flushed>
  90    */
  91   std::pair<uint64_t, uint64_t> flush(uint64_t mono_id);
  92
  93   uint64_t get_current();
  94   uint64_t get_max();
  95
  96   JournalThrottle(
  97     unsigned expected_concurrency ///< [in] determines size of conds
  98     ) : throttle(g_ceph_context, "filestore_journal", expected_concurrency) {}
  99 };
 100
 101 #endif