2 files changed, 260 insertions, 0 deletions

diff --git a/src/crimson/os/seastore/cached_extent.cc b/src/crimson/os/seastore/cached_extent.cc
index ffe89df44bd..085a519cb68 100644
--- a/src/crimson/os/seastore/cached_extent.cc
+++ b/src/crimson/os/seastore/cached_extent.cc
@@ -182,4 +182,183 @@ std::ostream &operator<<(std::ostream &out, const lba_pin_list_t &rhs)
   return out << ']';
 }
 
+bool BufferSpace::is_range_loaded(extent_len_t offset, extent_len_t length) const
+{
+  assert(length > 0);
+  auto i = buffer_map.upper_bound(offset);
+  if (i == buffer_map.begin()) {
+    return false;
+  }
+  --i;
+  auto& [i_offset, i_bl] = *i;
+  assert(offset >= i_offset);
+  assert(i_bl.length() > 0);
+  if (offset + length > i_offset + i_bl.length()) {
+    return false;
+  } else {
+    return true;
+  }
+}
+
+ceph::bufferlist BufferSpace::get_buffer(extent_len_t offset, extent_len_t length) const
+{
+  assert(length > 0);
+  auto i = buffer_map.upper_bound(offset);
+  assert(i != buffer_map.begin());
+  --i;
+  auto& [i_offset, i_bl] = *i;
+  assert(offset >= i_offset);
+  assert(i_bl.length() > 0);
+  assert(offset + length <= i_offset + i_bl.length());
+  ceph::bufferlist res;
+  res.substr_of(i_bl, offset - i_offset, length);
+  return res;
+}
+
+load_ranges_t BufferSpace::load_ranges(extent_len_t offset, extent_len_t length)
+{
+  assert(length > 0);
+  load_ranges_t ret;
+  auto next = buffer_map.upper_bound(offset);
+
+  // must be assigned for the main-loop
+  map_t::iterator previous;
+  extent_len_t range_offset;
+  extent_len_t range_length;
+
+  // returns whether to proceed main-loop or not
+  auto f_merge_next_check_hole = [this, &next, &range_offset, &range_length](
+      ceph::bufferlist& previous_bl,
+      extent_len_t hole_length,
+      extent_len_t next_offset,
+      const ceph::bufferlist& next_bl) {
+    range_length -= hole_length;
+    previous_bl.append(next_bl);
+    if (range_length <= next_bl.length()) {
+      // "next" end includes or beyonds the range
+      buffer_map.erase(next);
+      return false;
+    } else {
+      range_offset = next_offset + next_bl.length();
+      range_length -= next_bl.length();
+      // erase next should destruct next_bl
+      next = buffer_map.erase(next);
+      return true;
+    }
+  };
+
+  // returns whether to proceed main-loop or not
+  auto f_prepare_without_merge_previous = [
+      this, offset, length,
+      &ret, &previous, &next, &range_length,
+      &f_merge_next_check_hole]() {
+    if (next == buffer_map.end()) {
+      // "next" reaches end,
+      // range has no "next" to merge
+      create_hole_insert_map(ret, offset, length, next);
+      return false;
+    }
+    // "next" is valid
+    auto& [n_offset, n_bl] = *next;
+    // next is from upper_bound()
+    assert(offset < n_offset);
+    extent_len_t hole_length = n_offset - offset;
+    if (length < hole_length) {
+      // "next" is beyond the range end,
+      // range has no "next" to merge
+      create_hole_insert_map(ret, offset, length, next);
+      return false;
+    }
+    // length >= hole_length
+    // insert hole as "previous"
+    previous = create_hole_insert_map(ret, offset, hole_length, next);
+    auto& p_bl = previous->second;
+    range_length = length;
+    return f_merge_next_check_hole(p_bl, hole_length, n_offset, n_bl);
+  };
+
+  /*
+   * prepare main-loop
+   */
+  if (next == buffer_map.begin()) {
+    // "previous" is invalid
+    if (!f_prepare_without_merge_previous()) {
+      return ret;
+    }
+  } else {
+    // "previous" is valid
+    previous = std::prev(next);
+    auto& [p_offset, p_bl] = *previous;
+    assert(offset >= p_offset);
+    extent_len_t p_end = p_offset + p_bl.length();
+    if (offset <= p_end) {
+      // "previous" is adjacent or overlaps the range
+      range_offset = p_end;
+      assert(offset + length > p_end);
+      range_length = offset + length - p_end;
+      // start the main-loop (merge "previous")
+    } else {
+      // "previous" is not adjacent to the range
+      // range and buffer_map should not overlap
+      assert(offset > p_end);
+      if (!f_prepare_without_merge_previous()) {
+        return ret;
+      }
+    }
+  }
+
+  /*
+   * main-loop: merge the range with "previous" and look at "next"
+   *
+   * "previous": the previous buffer_map entry, must be valid, must be mergable
+   * "next": the next buffer_map entry, maybe end, maybe mergable
+   * range_offset/length: the current range right after "previous"
+   */
+  assert(std::next(previous) == next);
+  auto& [p_offset, p_bl] = *previous;
+  assert(range_offset == p_offset + p_bl.length());
+  assert(range_length > 0);
+  while (next != buffer_map.end()) {
+    auto& [n_offset, n_bl] = *next;
+    assert(range_offset < n_offset);
+    extent_len_t hole_length = n_offset - range_offset;
+    if (range_length < hole_length) {
+      // "next" offset is beyond the range end
+      break;
+    }
+    // range_length >= hole_length
+    create_hole_append_bl(ret, p_bl, range_offset, hole_length);
+    if (!f_merge_next_check_hole(p_bl, hole_length, n_offset, n_bl)) {
+      return ret;
+    }
+    assert(std::next(previous) == next);
+    assert(range_offset == p_offset + p_bl.length());
+    assert(range_length > 0);
+  }
+  // range has no "next" to merge:
+  // 1. "next" reaches end
+  // 2. "next" offset is beyond the range end
+  create_hole_append_bl(ret, p_bl, range_offset, range_length);
+  return ret;
+}
+
+ceph::bufferptr BufferSpace::to_full_ptr(extent_len_t length)
+{
+  assert(length > 0);
+  assert(buffer_map.size() == 1);
+  auto it = buffer_map.begin();
+  auto& [i_off, i_buf] = *it;
+  assert(i_off == 0);
+  if (!i_buf.is_contiguous()) {
+    // Allocate page aligned ptr, also see create_extent_ptr_*()
+    i_buf.rebuild();
+  }
+  assert(i_buf.get_num_buffers() == 1);
+  ceph::bufferptr ptr(i_buf.front());
+  assert(ptr.is_page_aligned());
+  assert(ptr.length() == length);
+  buffer_map.clear();
+  return ptr;
+}
+
 }
diff --git a/src/crimson/os/seastore/cached_extent.h b/src/crimson/os/seastore/cached_extent.h
index fe6a4895c40..026e677bbbc 100644
--- a/src/crimson/os/seastore/cached_extent.h
+++ b/src/crimson/os/seastore/cached_extent.h
@@ -41,6 +41,8 @@ void intrusive_ptr_release(CachedExtent *);
 
 #endif
 
+// Note: BufferSpace::to_full_ptr() also creates extent ptr.
+
 inline ceph::bufferptr create_extent_ptr_rand(extent_len_t len) {
   assert(is_aligned(len, CEPH_PAGE_SIZE));
   assert(len > 0);
@@ -167,6 +169,85 @@ struct trans_spec_view_t {
     boost::intrusive::compare<cmp_t>>;
 };
 
+struct load_range_t {
+  extent_len_t offset;
+  ceph::bufferptr ptr;
+
+  extent_len_t get_length() const {
+    return ptr.length();
+  }
+
+  extent_len_t get_end() const {
+    extent_len_t end = offset + ptr.length();
+    assert(end > offset);
+    return end;
+  }
+};
+struct load_ranges_t {
+  extent_len_t length = 0;
+  std::list<load_range_t> ranges;
+
+  void push_back(extent_len_t offset, ceph::bufferptr ptr) {
+    assert(ranges.empty() ||
+           (ranges.back().get_end() < offset));
+    assert(ptr.length());
+    length += ptr.length();
+    ranges.push_back({offset, std::move(ptr)});
+  }
+};
+
+/// manage small chunks of extent
+class BufferSpace {
+  using map_t = std::map<extent_len_t, ceph::bufferlist>;
+public:
+  BufferSpace() = default;
+
+  /// Returns true if offset~length is fully loaded
+  bool is_range_loaded(extent_len_t offset, extent_len_t length) const;
+
+  /// Returns the bufferlist of offset~length
+  ceph::bufferlist get_buffer(extent_len_t offset, extent_len_t length) const;
+
+  /// Returns the ranges to load, merge the buffer_map if possible
+  load_ranges_t load_ranges(extent_len_t offset, extent_len_t length);
+
+  /// Converts to ptr when fully loaded
+  ceph::bufferptr to_full_ptr(extent_len_t length);
+
+private:
+  // create and append the read-hole to
+  // load_ranges_t and bl
+  static void create_hole_append_bl(
+      load_ranges_t& ret,
+      ceph::bufferlist& bl,
+      extent_len_t hole_offset,
+      extent_len_t hole_length) {
+    ceph::bufferptr hole_ptr = create_extent_ptr_rand(hole_length);
+    bl.append(hole_ptr);
+    ret.push_back(hole_offset, std::move(hole_ptr));
+  }
+
+  // create and insert the read-hole to buffer_map,
+  // and append to load_ranges_t
+  // returns the iterator containing the inserted read-hole
+  auto create_hole_insert_map(
+      load_ranges_t& ret,
+      extent_len_t hole_offset,
+      extent_len_t hole_length,
+      const map_t::const_iterator& next_it) {
+    assert(!buffer_map.contains(hole_offset));
+    ceph::bufferlist bl;
+    create_hole_append_bl(ret, bl, hole_offset, hole_length);
+    auto it = buffer_map.insert(
+        next_it, std::pair{hole_offset, std::move(bl)});
+    assert(next_it == std::next(it));
+    return it;
+  }
+
+  /// extent offset -> buffer, won't overlap nor contiguous
+  map_t buffer_map;
+};
+
 class ExtentIndex;
 class CachedExtent
   : public boost::intrusive_ref_counter<