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