crimson/os/seastore: add segment_cleaner

Adds SegmentCleaner component to manage rewriting
dirty segments.

Signed-off-by: Samuel Just <sjust@redhat.com>

src/crimson/os/seastore/CMakeLists.txt

@@ -7,6 +7,7 @@ add_library(crimson-seastore STATIC
   journal.cc
   cache.cc
   lba_manager.cc
+  segment_cleaner.cc
   lba_manager/btree/btree_lba_manager.cc
   lba_manager/btree/lba_btree_node_impl.cc
   lba_manager/btree/btree_range_pin.cc

src/crimson/os/seastore/cache.cc

@@ -395,6 +395,36 @@ Cache::replay_delta(
   }
 }
 
+Cache::get_next_dirty_extents_ret Cache::get_next_dirty_extents(
+  journal_seq_t seq)
+{
+  std::vector<CachedExtentRef> ret;
+  for (auto i = dirty.begin(); i != dirty.end(); ++i) {
+    CachedExtentRef cand;
+    if (i->dirty_from < seq) {
+      assert(ret.empty() || ret.back()->dirty_from <= i->dirty_from);
+      ret.push_back(&*i);
+    } else {
+      break;
+    }
+  }
+  return seastar::do_with(
+    std::move(ret),
+    [](auto &ret) {
+      return seastar::do_for_each(
+	ret,
+	[](auto &ext) {
+	  logger().debug(
+	    "get_next_dirty_extents: waiting on {}",
+	    *ext);
+	  return ext->wait_io();
+	}).then([&ret]() mutable {
+	  return seastar::make_ready_future<std::vector<CachedExtentRef>>(
+	    std::move(ret));
+	});
+    });
+}
+
 Cache::get_root_ret Cache::get_root(Transaction &t)
 {
   if (t.root) {

src/crimson/os/seastore/cache.h

@@ -14,6 +14,7 @@
 #include "crimson/common/errorator.h"
 #include "crimson/os/seastore/cached_extent.h"
 #include "crimson/os/seastore/root_block.h"
+#include "crimson/os/seastore/segment_cleaner.h"
 
 namespace crimson::os::seastore {
 
@@ -400,6 +401,13 @@ public:
     return out;
   }
 
+  /// returns extents with dirty_from < seq
+  using get_next_dirty_extents_ertr = crimson::errorator<>;
+  using get_next_dirty_extents_ret = get_next_dirty_extents_ertr::future<
+    std::vector<CachedExtentRef>>;
+  get_next_dirty_extents_ret get_next_dirty_extents(
+    journal_seq_t seq);
+
 private:
   SegmentManager &segment_manager; ///< ref to segment_manager
   RootBlockRef root;               ///< ref to current root

src/crimson/os/seastore/seastore.cc

@@ -39,6 +39,10 @@ struct SeastoreCollection final : public FuturizedCollection {
 SeaStore::SeaStore(const std::string& path)
   : segment_manager(segment_manager::create_ephemeral(
 		      segment_manager::DEFAULT_TEST_EPHEMERAL)),
+    segment_cleaner(
+      std::make_unique<SegmentCleaner>(
+	SegmentCleaner::config_t::default_from_segment_manager(
+	  *segment_manager))),
     cache(std::make_unique<Cache>(*segment_manager)),
     journal(new Journal(*segment_manager)),
     lba_manager(
@@ -46,11 +50,15 @@ SeaStore::SeaStore(const std::string& path)
     transaction_manager(
       new TransactionManager(
 	*segment_manager,
+	*segment_cleaner,
 	*journal,
 	*cache,
 	*lba_manager)),
     onode_manager(onode_manager::create_ephemeral())
-{}
+{
+  journal->set_segment_provider(&*segment_cleaner);
+  segment_cleaner->set_extent_callback(&*transaction_manager);
+}
 
 SeaStore::~SeaStore() = default;
 

src/crimson/os/seastore/seastore.h

@@ -16,6 +16,7 @@
 #include "include/uuid.h"
 
 #include "os/Transaction.h"
+#include "crimson/os/seastore/segment_cleaner.h"
 #include "crimson/os/futurized_store.h"
 #include "transaction.h"
 
@@ -119,6 +120,7 @@ public:
 
 private:
   std::unique_ptr<SegmentManager> segment_manager;
+  std::unique_ptr<SegmentCleaner> segment_cleaner;
   std::unique_ptr<Cache> cache;
   std::unique_ptr<Journal> journal;
   std::unique_ptr<LBAManager> lba_manager;

src/crimson/os/seastore/segment_cleaner.cc

@@ -0,0 +1,97 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+
+#include "crimson/common/log.h"
+
+#include "crimson/os/seastore/segment_cleaner.h"
+
+namespace {
+  seastar::logger& logger() {
+    return crimson::get_logger(ceph_subsys_filestore);
+  }
+}
+
+namespace crimson::os::seastore {
+
+SegmentCleaner::get_segment_ret SegmentCleaner::get_segment()
+{
+  // TODO
+  return get_segment_ret(
+    get_segment_ertr::ready_future_marker{},
+    next++);
+}
+
+void SegmentCleaner::update_journal_tail_target(journal_seq_t target)
+{
+  logger().debug(
+    "{}: {}",
+    __func__,
+    target);
+  assert(journal_tail_target == journal_seq_t() || target >= journal_tail_target);
+  if (journal_tail_target == journal_seq_t() || target > journal_tail_target) {
+    journal_tail_target = target;
+  }
+}
+
+void SegmentCleaner::update_journal_tail_committed(journal_seq_t committed)
+{
+  if (journal_tail_committed == journal_seq_t() ||
+      committed > journal_tail_committed) {
+    logger().debug(
+      "{}: update journal_tail_committed {}",
+      __func__,
+      committed);
+    journal_tail_committed = committed;
+  }
+  if (journal_tail_target == journal_seq_t() ||
+      committed > journal_tail_target) {
+    logger().debug(
+      "{}: update journal_tail_target {}",
+      __func__,
+      committed);
+    journal_tail_target = committed;
+  }
+}
+
+void SegmentCleaner::put_segment(segment_id_t segment)
+{
+  return;
+}
+
+SegmentCleaner::do_immediate_work_ret SegmentCleaner::do_immediate_work(
+  Transaction &t)
+{
+  auto next_target = get_dirty_tail_limit();
+  logger().debug(
+    "{}: journal_tail_target={} get_dirty_tail_limit()={}",
+    __func__,
+    journal_tail_target,
+    next_target);
+  if (journal_tail_target > next_target) {
+    return do_immediate_work_ertr::now();
+  }
+
+  return ecb->get_next_dirty_extents(
+    get_dirty_tail_limit()
+  ).then([=, &t](auto dirty_list) {
+    if (dirty_list.empty()) {
+      return do_immediate_work_ertr::now();
+    } else {
+      update_journal_tail_target(dirty_list.front()->get_dirty_from());
+    }
+    return seastar::do_with(
+      std::move(dirty_list),
+      [this, &t](auto &dirty_list) {
+	return crimson::do_for_each(
+	  dirty_list,
+	  [this, &t](auto &e) {
+	    logger().debug(
+	      "SegmentCleaner::do_immediate_work cleaning {}",
+	      *e);
+	    return ecb->rewrite_extent(t, e);
+	  });
+      });
+  });
+}
+
+}

src/crimson/os/seastore/segment_cleaner.h

@@ -0,0 +1,166 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+
+#pragma once
+
+#include <boost/intrusive/set.hpp>
+
+#include "common/ceph_time.h"
+
+#include "crimson/os/seastore/cached_extent.h"
+#include "crimson/os/seastore/journal.h"
+#include "crimson/os/seastore/seastore_types.h"
+#include "crimson/os/seastore/segment_manager.h"
+
+namespace crimson::os::seastore {
+class Transaction;
+
+class SegmentCleaner : public JournalSegmentProvider {
+public:
+  /// Config
+  struct config_t {
+    size_t num_segments = 0;
+    size_t segment_size = 0;
+    size_t target_journal_segments = 0;
+    size_t max_journal_segments = 0;
+
+    static config_t default_from_segment_manager(
+      SegmentManager &manager) {
+      return config_t{
+	manager.get_num_segments(),
+	static_cast<size_t>(manager.get_segment_size()),
+	2,
+	4};
+    }
+  };
+
+  /// Callback interface for querying and operating on segments
+  class ExtentCallbackInterface {
+  public:
+    /**
+     * get_next_dirty_extent
+     *
+     * returns all extents with dirty_from < bound
+     */
+    using get_next_dirty_extents_ertr = crimson::errorator<>;
+    using get_next_dirty_extents_ret = get_next_dirty_extents_ertr::future<
+      std::vector<CachedExtentRef>>;
+    virtual get_next_dirty_extents_ret get_next_dirty_extents(
+      journal_seq_t bound ///< [in] return extents with dirty_from < bound
+    ) = 0;
+
+    /**
+     * rewrite_extent
+     *
+     * Updates t with operations moving the passed extents to a new
+     * segment.  extent may be invalid, implementation must correctly
+     * handle finding the current instance if it is still alive and
+     * otherwise ignore it.
+     */
+    using rewrite_extent_ertr = crimson::errorator<
+      crimson::ct_error::input_output_error>;
+    using rewrite_extent_ret = rewrite_extent_ertr::future<>;
+    virtual rewrite_extent_ret rewrite_extent(
+      Transaction &t,
+      CachedExtentRef extent) = 0;
+  };
+
+private:
+  segment_id_t next = 0;
+  const config_t config;
+
+  journal_seq_t journal_tail_target;
+  journal_seq_t journal_tail_committed;
+  journal_seq_t journal_head;
+
+  ExtentCallbackInterface *ecb = nullptr;
+
+public:
+  SegmentCleaner(config_t config)
+    : config(config) {}
+
+  get_segment_ret get_segment() final;
+
+  // hack for testing until we get real space handling
+  void set_next(segment_id_t _next) {
+    next = _next;
+  }
+  segment_id_t get_next() const {
+    return next;
+  }
+
+
+  void put_segment(segment_id_t segment) final;
+
+  journal_seq_t get_journal_tail_target() const final {
+    return journal_tail_target;
+  }
+
+  void update_journal_tail_committed(journal_seq_t committed) final;
+
+  void update_journal_tail_target(journal_seq_t target);
+
+  void init_journal_tail(journal_seq_t tail) {
+    journal_tail_target = journal_tail_committed = tail;
+  }
+
+  void set_journal_head(journal_seq_t head) {
+    assert(journal_head == journal_seq_t() || head >= journal_head);
+    journal_head = head;
+  }
+
+  void set_extent_callback(ExtentCallbackInterface *cb) {
+    ecb = cb;
+  }
+
+  /**
+   * do_immediate_work
+   *
+   * Should be invoked prior to submission of any transaction,
+   * will piggy-back work required to maintain deferred work
+   * constraints.
+   */
+  using do_immediate_work_ertr = crimson::errorator<
+    crimson::ct_error::input_output_error>;
+  using do_immediate_work_ret = do_immediate_work_ertr::future<>;
+  do_immediate_work_ret do_immediate_work(
+    Transaction &t);
+
+
+  /**
+   * do_deferred_work
+   *
+   * Should be called at idle times -- will perform background
+   * operations based on deferred work constraints.
+   *
+   * If returned timespan is non-zero, caller should pause calling
+   * back into do_deferred_work before returned timespan has elapsed,
+   * or a foreground operation occurs.
+   */
+  using do_deferred_work_ertr = crimson::errorator<
+    crimson::ct_error::input_output_error>;
+  using do_deferred_work_ret = do_deferred_work_ertr::future<
+    ceph::timespan
+    >;
+  do_deferred_work_ret do_deferred_work(
+    Transaction &t);
+
+private:
+  journal_seq_t get_dirty_tail() const {
+    auto ret = journal_head;
+    ret.segment_seq -= std::min(
+      static_cast<size_t>(ret.segment_seq),
+      config.target_journal_segments);
+    return ret;
+  }
+
+  journal_seq_t get_dirty_tail_limit() const {
+    auto ret = journal_head;
+    ret.segment_seq -= std::min(
+      static_cast<size_t>(ret.segment_seq),
+      config.max_journal_segments);
+    return ret;
+  }
+};
+
+}

src/crimson/os/seastore/transaction_manager.cc

@@ -20,21 +20,22 @@ namespace crimson::os::seastore {
 
 TransactionManager::TransactionManager(
   SegmentManager &segment_manager,
+  SegmentCleaner &segment_cleaner,
   Journal &journal,
   Cache &cache,
   LBAManager &lba_manager)
   : segment_manager(segment_manager),
+    segment_cleaner(segment_cleaner),
     cache(cache),
     lba_manager(lba_manager),
     journal(journal)
-{
-  journal.set_segment_provider(this);
-}
+{}
 
 TransactionManager::mkfs_ertr::future<> TransactionManager::mkfs()
 {
   return journal.open_for_write().safe_then([this](auto addr) {
     logger().debug("TransactionManager::mkfs: about to do_with");
+    segment_cleaner.set_journal_head(addr);
     return seastar::do_with(
       create_transaction(),
       [this](auto &transaction) {
@@ -67,6 +68,7 @@ TransactionManager::mount_ertr::future<> TransactionManager::mount()
   }).safe_then([this] {
     return journal.open_for_write();
   }).safe_then([this](auto addr) {
+    segment_cleaner.set_journal_head(addr);
     return seastar::do_with(
       create_transaction(),
       [this](auto &t) {
@@ -156,23 +158,59 @@ TransactionManager::submit_transaction_ertr::future<>
 TransactionManager::submit_transaction(
   TransactionRef t)
 {
-  auto record = cache.try_construct_record(*t);
-  if (!record) {
-    return crimson::ct_error::eagain::make();
+  logger().debug("TransactionManager::submit_transaction");
+  return segment_cleaner.do_immediate_work(*t
+  ).safe_then([this, t=std::move(t)]() mutable -> submit_transaction_ertr::future<> {
+    auto record = cache.try_construct_record(*t);
+    if (!record) {
+      return crimson::ct_error::eagain::make();
+    }
+
+    return journal.submit_record(std::move(*record)).safe_then(
+      [this, t=std::move(t)](auto p) mutable {
+	auto [addr, journal_seq] = p;
+	segment_cleaner.set_journal_head(journal_seq);
+	cache.complete_commit(*t, addr, journal_seq);
+	lba_manager.complete_transaction(*t);
+      },
+      submit_transaction_ertr::pass_further{},
+      crimson::ct_error::all_same_way([](auto e) {
+	ceph_assert(0 == "Hit error submitting to journal");
+      }));
+  });
+}
+
+TransactionManager::get_next_dirty_extents_ret
+TransactionManager::get_next_dirty_extents(journal_seq_t seq)
+{
+  return cache.get_next_dirty_extents(seq);
+}
+
+TransactionManager::rewrite_extent_ret TransactionManager::rewrite_extent(
+  Transaction &t,
+  CachedExtentRef extent)
+{
+  {
+    auto updated = cache.update_extent_from_transaction(t, extent);
+    if (!updated) {
+      logger().debug(
+	"{}: {} is already retired, skipping",
+	__func__,
+	*extent);
+      return rewrite_extent_ertr::now();
+    }
+    extent = updated;
   }
 
-  logger().debug("TransactionManager::submit_transaction");
-
-  return journal.submit_record(std::move(*record)).safe_then(
-    [this, t=std::move(t)](auto p) mutable {
-      auto [addr, journal_seq] = p;
-      cache.complete_commit(*t, addr, journal_seq);
-      lba_manager.complete_transaction(*t);
-    },
-    submit_transaction_ertr::pass_further{},
-    crimson::ct_error::all_same_way([](auto e) {
-      ceph_assert(0 == "Hit error submitting to journal");
-    }));
+  if (extent->get_type() == extent_types_t::ROOT) {
+    logger().debug(
+      "{}: marking root {} for rewrite",
+      __func__,
+      *extent);
+    cache.duplicate_for_write(t, extent);
+    return rewrite_extent_ertr::now();
+  }
+  return lba_manager.rewrite_extent(t, extent);
 }
 
 TransactionManager::~TransactionManager() {}

src/crimson/os/seastore/transaction_manager.h

@@ -19,6 +19,7 @@
 
 #include "crimson/osd/exceptions.h"
 
+#include "crimson/os/seastore/segment_cleaner.h"
 #include "crimson/os/seastore/seastore_types.h"
 #include "crimson/os/seastore/cache.h"
 #include "crimson/os/seastore/segment_manager.h"
@@ -34,35 +35,15 @@ class Journal;
  * Abstraction hiding reading and writing to persistence.
  * Exposes transaction based interface with read isolation.
  */
-class TransactionManager : public JournalSegmentProvider {
+class TransactionManager : public SegmentCleaner::ExtentCallbackInterface {
 public:
   TransactionManager(
     SegmentManager &segment_manager,
+    SegmentCleaner &segment_cleaner,
     Journal &journal,
     Cache &cache,
     LBAManager &lba_manager);
 
-  segment_id_t next = 0;
-  get_segment_ret get_segment() final {
-    // TODO -- part of gc
-    return get_segment_ret(
-      get_segment_ertr::ready_future_marker{},
-      next++);
-  }
-
-  void put_segment(segment_id_t segment) final {
-    // TODO -- part of gc
-    return;
-  }
-
-  journal_seq_t get_journal_tail_target() const final {
-    // TODO -- part of gc
-    return journal_seq_t{};
-  }
-  void update_journal_tail_committed(journal_seq_t committed) final {
-    // TODO -- part of gc
-  }
-
   /// Writes initial metadata to disk
   using mkfs_ertr = crimson::errorator<
     crimson::ct_error::input_output_error
@@ -230,12 +211,24 @@ public:
     >;
   submit_transaction_ertr::future<> submit_transaction(TransactionRef);
 
+  /// SegmentCleaner::ExtentCallbackInterface
+
+  using SegmentCleaner::ExtentCallbackInterface::get_next_dirty_extents_ret;
+  get_next_dirty_extents_ret get_next_dirty_extents(
+    journal_seq_t seq) final;
+
+  using SegmentCleaner::ExtentCallbackInterface::rewrite_extent_ret;
+  rewrite_extent_ret rewrite_extent(
+    Transaction &t,
+    CachedExtentRef extent) final;
+
   ~TransactionManager();
 
 private:
   friend class Transaction;
 
   SegmentManager &segment_manager;
+  SegmentCleaner &segment_cleaner;
   Cache &cache;
   LBAManager &lba_manager;
   Journal &journal;

src/test/crimson/seastore/test_extmap_manager.cc

@@ -23,6 +23,7 @@ namespace {
 
 struct extentmap_manager_test_t : public seastar_test_suite_t {
   std::unique_ptr<SegmentManager> segment_manager;
+  SegmentCleaner segment_cleaner;
   Journal journal;
   Cache cache;
   LBAManagerRef lba_manager;
@@ -31,13 +32,18 @@ struct extentmap_manager_test_t : public seastar_test_suite_t {
 
   extentmap_manager_test_t()
     : segment_manager(create_ephemeral(segment_manager::DEFAULT_TEST_EPHEMERAL)),
+      segment_cleaner(SegmentCleaner::config_t::default_from_segment_manager(
+			*segment_manager)),
       journal(*segment_manager),
       cache(*segment_manager),
       lba_manager(
 	lba_manager::create_lba_manager(*segment_manager, cache)),
-      tm(*segment_manager, journal, cache, *lba_manager),
+      tm(*segment_manager, segment_cleaner, journal, cache, *lba_manager),
       extmap_manager(
-        extentmap_manager::create_extentmap_manager(tm)) {}
+        extentmap_manager::create_extentmap_manager(tm)) {
+    journal.set_segment_provider(&segment_cleaner);
+    segment_cleaner.set_extent_callback(&tm);
+  }
 
   seastar::future<> set_up_fut() final {
     return segment_manager->init().safe_then([this] {

src/test/crimson/seastore/test_transaction_manager.cc

@@ -3,6 +3,7 @@
 
 #include "test/crimson/gtest_seastar.h"
 
+#include "crimson/os/seastore/segment_cleaner.h"
 #include "crimson/os/seastore/cache.h"
 #include "crimson/os/seastore/transaction_manager.h"
 #include "crimson/os/seastore/segment_manager.h"
@@ -46,6 +47,7 @@ std::ostream &operator<<(std::ostream &lhs, const test_extent_record_t &rhs) {
 
 struct transaction_manager_test_t : public seastar_test_suite_t {
   std::unique_ptr<SegmentManager> segment_manager;
+  SegmentCleaner segment_cleaner;
   Journal journal;
   Cache cache;
   LBAManagerRef lba_manager;
@@ -53,11 +55,17 @@ struct transaction_manager_test_t : public seastar_test_suite_t {
 
   transaction_manager_test_t()
     : segment_manager(create_ephemeral(segment_manager::DEFAULT_TEST_EPHEMERAL)),
+      segment_cleaner(
+	SegmentCleaner::config_t::default_from_segment_manager(
+	  *segment_manager)),
       journal(*segment_manager),
       cache(*segment_manager),
       lba_manager(
 	lba_manager::create_lba_manager(*segment_manager, cache)),
-      tm(*segment_manager, journal, cache, *lba_manager) {}
+      tm(*segment_manager, segment_cleaner, journal, cache, *lba_manager) {
+    journal.set_segment_provider(&segment_cleaner);
+    segment_cleaner.set_extent_callback(&tm);
+  }
 
   seastar::future<> set_up_fut() final {
     return segment_manager->init().safe_then([this] {