ceph/src/common/Readahead.cc

   1 // -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
   2 // vim: ts=8 sw=2 smarttab
   3
   4 #include "Readahead.h"
   5
   6 using namespace std;
   7
   8 Readahead::Readahead()
   9   : m_trigger_requests(10),
  10     m_readahead_min_bytes(0),
  11     m_readahead_max_bytes(NO_LIMIT),
  12     m_alignments(),
  13     m_lock("Readahead::m_lock"),
  14     m_nr_consec_read(0),
  15     m_consec_read_bytes(0),
  16     m_last_pos(0),
  17     m_readahead_pos(0),
  18     m_readahead_trigger_pos(0),
  19     m_readahead_size(0),
  20     m_pending(0),
  21     m_pending_lock("Readahead::m_pending_lock") {
  22 }
  23
  24 Readahead::~Readahead() {
  25 }
  26
  27 Readahead::extent_t Readahead::update(const vector<extent_t>& extents, uint64_t limit) {
  28   m_lock.Lock();
  29   for (vector<extent_t>::const_iterator p = extents.begin(); p != extents.end(); ++p) {
  30     _observe_read(p->first, p->second);
  31   }
  32   if (m_readahead_pos >= limit|| m_last_pos >= limit) {
  33     m_lock.Unlock();
  34     return extent_t(0, 0);
  35   }
  36   pair<uint64_t, uint64_t> extent = _compute_readahead(limit);
  37   m_lock.Unlock();
  38   return extent;
  39 }
  40
  41 Readahead::extent_t Readahead::update(uint64_t offset, uint64_t length, uint64_t limit) {
  42   m_lock.Lock();
  43   _observe_read(offset, length);
  44   if (m_readahead_pos >= limit || m_last_pos >= limit) {
  45     m_lock.Unlock();
  46     return extent_t(0, 0);
  47   }
  48   extent_t extent = _compute_readahead(limit);
  49   m_lock.Unlock();
  50   return extent;
  51 }
  52
  53 void Readahead::_observe_read(uint64_t offset, uint64_t length) {
  54   if (offset == m_last_pos) {
  55     m_nr_consec_read++;
  56     m_consec_read_bytes += length;
  57   } else {
  58     m_nr_consec_read = 0;
  59     m_consec_read_bytes = 0;
  60     m_readahead_trigger_pos = 0;
  61     m_readahead_size = 0;
  62     m_readahead_pos = 0;
  63   }
  64   m_last_pos = offset + length;
  65 }
  66
  67 Readahead::extent_t Readahead::_compute_readahead(uint64_t limit) {
  68   uint64_t readahead_offset = 0;
  69   uint64_t readahead_length = 0;
  70   if (m_nr_consec_read >= m_trigger_requests) {
  71     // currently reading sequentially
  72     if (m_last_pos >= m_readahead_trigger_pos) {
  73       // need to read ahead
  74       if (m_readahead_size == 0) {
  75         // initial readahead trigger
  76         m_readahead_size = m_consec_read_bytes;
  77         m_readahead_pos = m_last_pos;
  78       } else {
  79         // continuing readahead trigger
  80         m_readahead_size *= 2;
  81         if (m_last_pos > m_readahead_pos) {
  82           m_readahead_pos = m_last_pos;
  83         }
  84       }
  85       m_readahead_size = MAX(m_readahead_size, m_readahead_min_bytes);
  86       m_readahead_size = MIN(m_readahead_size, m_readahead_max_bytes);
  87       readahead_offset = m_readahead_pos;
  88       readahead_length = m_readahead_size;
  89
  90       // Snap to the first alignment possible
  91       uint64_t readahead_end = readahead_offset + readahead_length;
  92       for (vector<uint64_t>::iterator p = m_alignments.begin(); p != m_alignments.end(); ++p) {
  93         // Align the readahead, if possible.
  94         uint64_t alignment = *p;
  95         uint64_t align_prev = readahead_end / alignment * alignment;
  96         uint64_t align_next = align_prev + alignment;
  97         uint64_t dist_prev = readahead_end - align_prev;
  98         uint64_t dist_next = align_next - readahead_end;
  99         if (dist_prev < readahead_length / 2 && dist_prev < dist_next) {
 100           // we can snap to the previous alignment point by a less than 50% reduction in size
 101           assert(align_prev > readahead_offset);
 102           readahead_length = align_prev - readahead_offset;
 103           break;
 104         } else if(dist_next < readahead_length / 2) {
 105           // we can snap to the next alignment point by a less than 50% increase in size
 106           assert(align_next > readahead_offset);
 107           readahead_length = align_next - readahead_offset;
 108           break;
 109         }
 110         // Note that m_readahead_size should remain unadjusted.
 111       }
 112
 113       if (m_readahead_pos + readahead_length > limit) {
 114         readahead_length = limit - m_readahead_pos;
 115       }
 116
 117       m_readahead_trigger_pos = m_readahead_pos + readahead_length / 2;
 118       m_readahead_pos += readahead_length;
 119     }
 120   }
 121   return extent_t(readahead_offset, readahead_length);
 122 }
 123
 124 void Readahead::inc_pending(int count) {
 125   assert(count > 0);
 126   m_pending_lock.Lock();
 127   m_pending += count;
 128   m_pending_lock.Unlock();
 129 }
 130
 131 void Readahead::dec_pending(int count) {
 132   assert(count > 0);
 133   m_pending_lock.Lock();
 134   assert(m_pending >= count);
 135   m_pending -= count;
 136   if (m_pending == 0) {
 137     std::list<Context *> pending_waiting(std::move(m_pending_waiting));
 138     m_pending_lock.Unlock();
 139
 140     for (auto ctx : pending_waiting) {
 141       ctx->complete(0);
 142     }
 143   } else {
 144     m_pending_lock.Unlock();
 145   }
 146 }
 147
 148 void Readahead::wait_for_pending() {
 149   C_SaferCond ctx;
 150   wait_for_pending(&ctx);
 151   ctx.wait();
 152 }
 153
 154 void Readahead::wait_for_pending(Context *ctx) {
 155   m_pending_lock.Lock();
 156   if (m_pending > 0) {
 157     m_pending_lock.Unlock();
 158     m_pending_waiting.push_back(ctx);
 159     return;
 160   }
 161   m_pending_lock.Unlock();
 162
 163   ctx->complete(0);
 164 }
 165 void Readahead::set_trigger_requests(int trigger_requests) {
 166   m_lock.Lock();
 167   m_trigger_requests = trigger_requests;
 168   m_lock.Unlock();
 169 }
 170
 171 uint64_t Readahead::get_min_readahead_size(void) {
 172   Mutex::Locker lock(m_lock);
 173   return m_readahead_min_bytes;
 174 }
 175
 176 uint64_t Readahead::get_max_readahead_size(void) {
 177   Mutex::Locker lock(m_lock);
 178   return m_readahead_max_bytes;
 179 }
 180
 181 void Readahead::set_min_readahead_size(uint64_t min_readahead_size) {
 182   m_lock.Lock();
 183   m_readahead_min_bytes = min_readahead_size;
 184   m_lock.Unlock();
 185 }
 186
 187 void Readahead::set_max_readahead_size(uint64_t max_readahead_size) {
 188   m_lock.Lock();
 189   m_readahead_max_bytes = max_readahead_size;
 190   m_lock.Unlock();
 191 }
 192
 193 void Readahead::set_alignments(const vector<uint64_t> &alignments) {
 194   m_lock.Lock();
 195   m_alignments = alignments;
 196   m_lock.Unlock();
 197 }