Merge pull request ceph#49286 from rzarzynski/wip-crimson-snaptrimmer

crimson/osd, osd: bring snap trimming to crimson

Reviewed-by: Samuel Just <[email protected]>

Author: rzarzynski

src/common/ceph_mutex.h

@@ -15,6 +15,16 @@
 // naming the mutex for the purposes of the lockdep debug variant.
 
 #if defined(WITH_SEASTAR) && !defined(WITH_ALIEN)
+#include <seastar/core/condition-variable.hh>
+
+#include "crimson/common/log.h"
+#include "include/ceph_assert.h"
+
+#ifndef NDEBUG
+#define FUT_DEBUG(FMT_MSG, ...) crimson::get_logger(ceph_subsys_).trace(FMT_MSG, ##__VA_ARGS__)
+#else
+#define FUT_DEBUG(FMT_MSG, ...)
+#endif
 
 namespace ceph {
   // an empty class satisfying the mutex concept
@@ -33,11 +43,30 @@ namespace ceph {
     void unlock_shared() {}
   };
 
+  // this implementation assumes running within a seastar::thread
+  struct green_condition_variable : private seastar::condition_variable {
+    template <class LockT>
+    void wait(LockT&&) {
+      FUT_DEBUG("green_condition_variable::{}: before blocking", __func__);
+      seastar::condition_variable::wait().get();
+      FUT_DEBUG("green_condition_variable::{}: after blocking", __func__);
+    }
+
+    void notify_one() noexcept {
+      FUT_DEBUG("green_condition_variable::{}", __func__);
+      signal();
+    }
+
+    void notify_all() noexcept {
+      FUT_DEBUG("green_condition_variable::{}", __func__);
+      broadcast();
+    }
+  };
+
   using mutex = dummy_mutex;
   using recursive_mutex = dummy_mutex;
   using shared_mutex = dummy_shared_mutex;
-  // in seastar, we should use a difference interface for enforcing the
-  // semantics of condition_variable
+  using condition_variable = green_condition_variable;
 
   template <typename ...Args>
   dummy_mutex make_mutex(Args&& ...args) {

src/common/map_cacher.hpp

@@ -60,6 +60,11 @@ class StoreDriver {
     std::pair<K, V> *next    ///< [out] first key after key
     ) = 0; ///< @return 0 on success, -ENOENT if there is no next
 
+  virtual int get_next_or_current(
+    const K &key,       ///< [in] key at-which-or-after to get
+    std::pair<K, V> *next_or_current
+    ) = 0; ///< @return 0 on success, -ENOENT if there is no next
+
   virtual ~StoreDriver() {}
 };
 

src/crimson/common/errorator.h

@@ -335,6 +335,13 @@ class maybe_handle_error_t {
         // to throwing an exception by the handler.
         std::invoke(std::forward<ErrorVisitorT>(errfunc),
                     ErrorT::error_t::from_exception_ptr(std::move(ep)));
+      } else if constexpr (seastar::Future<decltype(result)>) {
+        // result is seastar::future but return_t is e.g. int. If so,
+        // the else clause cannot be used as seastar::future lacks
+        // errorator_type member.
+        result = seastar::make_ready_future<return_t>(
+          std::invoke(std::forward<ErrorVisitorT>(errfunc),
+                      ErrorT::error_t::from_exception_ptr(std::move(ep))));
       } else {
         result = FuturatorT::type::errorator_type::template make_ready_future<return_t>(
           std::invoke(std::forward<ErrorVisitorT>(errfunc),

src/crimson/common/interruptible_future.h

@@ -96,6 +96,11 @@ struct interrupt_cond_t {
   uint64_t ref_count = 0;
   void set(
     InterruptCondRef<InterruptCond>& ic) {
+    INTR_FUT_DEBUG(
+      "{}: going to set interrupt_cond: {}, ic: {}",
+      __func__,
+      (void*)interrupt_cond.get(),
+      (void*)ic.get());
     if (!interrupt_cond) {
       interrupt_cond = ic;
     }
@@ -110,13 +115,15 @@ struct interrupt_cond_t {
   void reset() {
     if (--ref_count == 0) {
       INTR_FUT_DEBUG(
-	"call_with_interruption_impl clearing interrupt_cond: {},{}",
+	"{}: clearing interrupt_cond: {},{}",
+        __func__,
 	(void*)interrupt_cond.get(),
 	typeid(InterruptCond).name());
       interrupt_cond.release();
     } else {
       INTR_FUT_DEBUG(
-	"call_with_interruption_impl end without clearing interrupt_cond: {},{}, ref_count: {}",
+	"{}: end without clearing interrupt_cond: {},{}, ref_count: {}",
+        __func__,
 	(void*)interrupt_cond.get(),
 	typeid(InterruptCond).name(),
 	ref_count);
@@ -1386,12 +1393,46 @@ struct interruptor
   template <typename Func,
 	    typename Result = futurize_t<std::invoke_result_t<Func>>>
   static inline Result async(Func&& func) {
-    return seastar::async([func=std::forward<Func>(func),
-			   interrupt_condition=interrupt_cond<InterruptCond>.interrupt_cond]
-			  () mutable {
+    auto interruption_condition = interrupt_cond<InterruptCond>.interrupt_cond;
+    INTR_FUT_DEBUG(
+      "interruptible_future_detail::async() yielding out, "
+      "interrupt_cond {},{} cleared",
+      (void*)interruption_condition.get(),
+      typeid(InterruptCond).name());
+    interrupt_cond<InterruptCond>.reset();
+    auto ret = seastar::async([func=std::forward<Func>(func),
+			       interruption_condition] () mutable {
       return non_futurized_call_with_interruption(
-	  interrupt_condition, std::forward<Func>(func));
+	  interruption_condition, std::forward<Func>(func));
     });
+    interrupt_cond<InterruptCond>.set(interruption_condition);
+    INTR_FUT_DEBUG(
+      "interruptible_future_detail::async() yield back, interrupt_cond: {},{}",
+      (void*)interrupt_cond<InterruptCond>.interrupt_cond.get(),
+      typeid(InterruptCond).name());
+    return ret;
+  }
+
+  template <class FutureT>
+  static decltype(auto) green_get(FutureT&& fut) {
+    if (fut.available()) {
+      return fut.get();
+    } else {
+      // destined to wait!
+      auto interruption_condition = interrupt_cond<InterruptCond>.interrupt_cond;
+      INTR_FUT_DEBUG(
+        "green_get() waiting, interrupt_cond: {},{}",
+        (void*)interrupt_cond<InterruptCond>.interrupt_cond.get(),
+        typeid(InterruptCond).name());
+      interrupt_cond<InterruptCond>.reset();
+      auto&& value = fut.get();
+      interrupt_cond<InterruptCond>.set(interruption_condition);
+      INTR_FUT_DEBUG(
+        "green_get() got, interrupt_cond: {},{}",
+        (void*)interrupt_cond<InterruptCond>.interrupt_cond.get(),
+        typeid(InterruptCond).name());
+      return std::move(value);
+    }
   }
 
   static void yield() {

src/crimson/osd/CMakeLists.txt

@@ -26,6 +26,7 @@ add_executable(crimson-osd
   osd_operations/logmissing_request_reply.cc
   osd_operations/background_recovery.cc
   osd_operations/recovery_subrequest.cc
+  osd_operations/snaptrim_event.cc
   pg_recovery.cc
   recovery_backend.cc
   replicated_recovery_backend.cc
@@ -45,6 +46,7 @@ add_executable(crimson-osd
   ${PROJECT_SOURCE_DIR}/src/osd/PGStateUtils.cc
   ${PROJECT_SOURCE_DIR}/src/osd/MissingLoc.cc
   ${PROJECT_SOURCE_DIR}/src/osd/PGLog.cc
+  ${PROJECT_SOURCE_DIR}/src/osd/SnapMapper.cc
   ${PROJECT_SOURCE_DIR}/src/osd/recovery_types.cc
   ${PROJECT_SOURCE_DIR}/src/osd/osd_perf_counters.cc
   watch.cc

src/crimson/osd/ops_executer.cc

@@ -19,6 +19,7 @@
 #include "crimson/osd/pg.h"
 #include "crimson/osd/watch.h"
 #include "osd/ClassHandler.h"
+#include "osd/SnapMapper.h"
 
 namespace {
   seastar::logger& logger() {
@@ -806,6 +807,9 @@ void OpsExecuter::fill_op_params_bump_pg_version()
 std::vector<pg_log_entry_t> OpsExecuter::prepare_transaction(
   const std::vector<OSDOp>& ops)
 {
+  // let's ensure we don't need to inform SnapMapper about this particular
+  // entry.
+  assert(obc->obs.oi.soid.snap >= CEPH_MAXSNAP);
   std::vector<pg_log_entry_t> log_entries;
   log_entries.emplace_back(
     obc->obs.exists ?
@@ -827,6 +831,57 @@ std::vector<pg_log_entry_t> OpsExecuter::prepare_transaction(
   return log_entries;
 }
 
+OpsExecuter::interruptible_future<> OpsExecuter::snap_map_remove(
+  const hobject_t& soid,
+  SnapMapper& snap_mapper,
+  OSDriver& osdriver,
+  ceph::os::Transaction& txn)
+{
+  logger().debug("{}: soid {}", __func__, soid);
+  return interruptor::async([soid, &snap_mapper,
+                             _t=osdriver.get_transaction(&txn)]() mutable {
+    const auto r = snap_mapper.remove_oid(soid, &_t);
+    if (r) {
+      logger().error("{}: remove_oid {} failed with {}",
+                     __func__, soid, r);
+    }
+    // On removal tolerate missing key corruption
+    assert(r == 0 || r == -ENOENT);
+  });
+}
+
+OpsExecuter::interruptible_future<> OpsExecuter::snap_map_modify(
+  const hobject_t& soid,
+  const std::set<snapid_t>& snaps,
+  SnapMapper& snap_mapper,
+  OSDriver& osdriver,
+  ceph::os::Transaction& txn)
+{
+  logger().debug("{}: soid {}, snaps {}", __func__, soid, snaps);
+  return interruptor::async([soid, snaps, &snap_mapper,
+                             _t=osdriver.get_transaction(&txn)]() mutable {
+    assert(std::size(snaps) > 0);
+    [[maybe_unused]] const auto r = snap_mapper.update_snaps(
+      soid, snaps, 0, &_t);
+    assert(r == 0);
+  });
+}
+
+OpsExecuter::interruptible_future<> OpsExecuter::snap_map_clone(
+  const hobject_t& soid,
+  const std::set<snapid_t>& snaps,
+  SnapMapper& snap_mapper,
+  OSDriver& osdriver,
+  ceph::os::Transaction& txn)
+{
+  logger().debug("{}: soid {}, snaps {}", __func__, soid, snaps);
+  return interruptor::async([soid, snaps, &snap_mapper,
+                             _t=osdriver.get_transaction(&txn)]() mutable {
+    assert(std::size(snaps) > 0);
+    snap_mapper.add_oid(soid, snaps, &_t);
+  });
+}
+
 // Defined here because there is a circular dependency between OpsExecuter and PG
 uint32_t OpsExecuter::get_pool_stripe_width() const {
   return pg->get_pgpool().info.get_stripe_width();
@@ -920,15 +975,29 @@ void OpsExecuter::CloningContext::apply_to(
   processed_obc.ssc->snapset = std::move(new_snapset);
 }
 
-void OpsExecuter::flush_clone_metadata(
-  std::vector<pg_log_entry_t>& log_entries)
+OpsExecuter::interruptible_future<std::vector<pg_log_entry_t>>
+OpsExecuter::flush_clone_metadata(
+  std::vector<pg_log_entry_t>&& log_entries,
+  SnapMapper& snap_mapper,
+  OSDriver& osdriver,
+  ceph::os::Transaction& txn)
 {
   assert(!txn.empty());
+  auto maybe_snap_mapped = interruptor::now();
   if (cloning_ctx) {
     osd_op_params->at_version = pg->next_version();
-    std::move(*cloning_ctx).apply_to(osd_op_params->at_version,
-                                     log_entries,
-                                     *obc);
+    std::move(*cloning_ctx).apply_to(
+      osd_op_params->at_version,
+      log_entries,
+      *obc);
+    const auto& coid = log_entries.back().soid;
+    const auto& cloned_snaps = obc->ssc->snapset.clone_snaps[coid.snap];
+    maybe_snap_mapped = snap_map_clone(
+      coid,
+      std::set<snapid_t>{std::begin(cloned_snaps), std::end(cloned_snaps)},
+      snap_mapper,
+      osdriver,
+      txn);
   }
   if (snapc.seq > obc->ssc->snapset.seq) {
      // update snapset with latest snap context
@@ -937,6 +1006,12 @@ void OpsExecuter::flush_clone_metadata(
   }
   logger().debug("{} done, initial snapset={}, new snapset={}",
     __func__, obc->obs.oi.soid, obc->ssc->snapset);
+  return std::move(
+    maybe_snap_mapped
+  ).then_interruptible([log_entries=std::move(log_entries)]() mutable {
+    return interruptor::make_ready_future<std::vector<pg_log_entry_t>>(
+      std::move(log_entries));
+  });
 }
 
 // TODO: make this static