Merge pull request ceph#58868 from xxhdx1985126/wip-crimson-clean_region-based-clone-recovery

crimson/osd/pg: properly propagate snap mapper updates and do clean-region-based clone objects recovery

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

Author: Matan-B

src/crimson/osd/ec_backend.cc

@@ -32,6 +32,6 @@ ECBackend::submit_transaction(const std::set<pg_shard_t> &pg_shards,
 			      std::vector<pg_log_entry_t>&& log_entries)
 {
   // todo
-  return {seastar::now(),
-	  seastar::make_ready_future<crimson::osd::acked_peers_t>()};
+  return make_ready_future<rep_op_ret_t>(seastar::now(),
+	  seastar::make_ready_future<crimson::osd::acked_peers_t>());
 }

src/crimson/osd/object_metadata_helper.cc

@@ -1,3 +1,6 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:nil -*-
+// vim: ts=8 sw=2 smarttab expandtab
+
 #include "crimson/osd/object_metadata_helper.h"
 
 namespace {
@@ -23,12 +26,26 @@ subsets_t calc_clone_subsets(
   subsets_t subsets;
   logger().debug("{}: {} clone_overlap {} ",
                  __func__, soid, snapset.clone_overlap);
-
+  assert(missing.get_items().contains(soid));
+  const pg_missing_item &missing_item = missing.get_items().at(soid);
+  auto dirty_regions = missing_item.clean_regions.get_dirty_regions();
+  if (dirty_regions.empty()) {
+    logger().debug(
+      "{} {} not touched, no need to recover, skipping",
+      __func__,
+      soid);
+    return subsets;
+  }
   uint64_t size = snapset.clone_size[soid.snap];
   if (size) {
     subsets.data_subset.insert(0, size);
   }
 
+  // let data_subset store only the modified content of the object.
+  subsets.data_subset.intersection_of(dirty_regions);
+  logger().debug("{} {} data_subset {}",
+                 __func__, soid, subsets.data_subset);
+
   // TODO: make sure CEPH_FEATURE_OSD_CACHEPOOL is not supported in Crimson
   // Skips clone subsets if caching was enabled (allow_incomplete_clones).
 
@@ -140,7 +157,7 @@ subsets_t calc_head_subsets(
     subsets.data_subset.insert(0, obj_size);
   }
   assert(missing.get_items().contains(head));
-  const pg_missing_item missing_item = missing.get_items().at(head);
+  const pg_missing_item &missing_item = missing.get_items().at(head);
   // let data_subset store only the modified content of the object.
   subsets.data_subset.intersection_of(missing_item.clean_regions.get_dirty_regions());
   logger().debug("{} {} data_subset {}",

src/crimson/osd/ops_executer.cc

@@ -871,6 +871,7 @@ OpsExecuter::interruptible_future<> OpsExecuter::snap_map_modify(
   ceph::os::Transaction& txn)
 {
   logger().debug("{}: soid {}, snaps {}", __func__, soid, snaps);
+  // TODO: avoid seastar::async https://tracker.ceph.com/issues/67704
   return interruptor::async([soid, snaps, &snap_mapper,
                              _t=osdriver.get_transaction(&txn)]() mutable {
     assert(std::size(snaps) > 0);
@@ -974,6 +975,7 @@ std::unique_ptr<OpsExecuter::CloningContext> OpsExecuter::execute_clone(
     0
   };
   encode(cloned_snaps, cloning_ctx->log_entry.snaps);
+  cloning_ctx->log_entry.clean_regions.mark_data_region_dirty(0, initial_obs.oi.size);
 
   return cloning_ctx;
 }
@@ -1019,14 +1021,6 @@ OpsExecuter::flush_clone_metadata(
   update_clone_overlap();
   if (cloning_ctx) {
     std::move(*cloning_ctx).apply_to(log_entries, *obc);
-    const auto& coid = log_entries.front().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

src/crimson/osd/ops_executer.h

@@ -21,6 +21,7 @@
 #include "os/Transaction.h"
 #include "osd/osd_types.h"
 
+#include "crimson/common/coroutine.h"
 #include "crimson/common/errorator.h"
 #include "crimson/common/interruptible_future.h"
 #include "crimson/common/type_helpers.h"
@@ -272,6 +273,7 @@ class OpsExecuter : public seastar::enable_lw_shared_from_this<OpsExecuter> {
     OSDriver& osdriver,
     ceph::os::Transaction& txn);
 
+public:
   static interruptible_future<> snap_map_remove(
     const hobject_t& soid,
     SnapMapper& snap_mapper,
@@ -290,6 +292,7 @@ class OpsExecuter : public seastar::enable_lw_shared_from_this<OpsExecuter> {
     OSDriver& osdriver,
     ceph::os::Transaction& txn);
 
+private:
   // this gizmo could be wrapped in std::optional for the sake of lazy
   // initialization. we don't need it for ops that doesn't have effect
   // TODO: verify the init overhead of chunked_fifo
@@ -424,7 +427,7 @@ class OpsExecuter : public seastar::enable_lw_shared_from_this<OpsExecuter> {
     const std::vector<OSDOp>& ops,
     SnapMapper& snap_mapper,
     OSDriver& osdriver,
-    MutFunc&& mut_func) &&;
+    MutFunc mut_func) &&;
   std::vector<pg_log_entry_t> prepare_transaction(
     const std::vector<OSDOp>& ops);
   void fill_op_params(modified_by m);
@@ -510,60 +513,60 @@ OpsExecuter::flush_changes_n_do_ops_effects(
   const std::vector<OSDOp>& ops,
   SnapMapper& snap_mapper,
   OSDriver& osdriver,
-  MutFunc&& mut_func) &&
+  MutFunc mut_func) &&
 {
   const bool want_mutate = !txn.empty();
   // osd_op_params are instantiated by every wr-like operation.
   assert(osd_op_params || !want_mutate);
   assert(obc);
-  rep_op_fut_t maybe_mutated =
-    interruptor::make_ready_future<rep_op_fut_tuple>(
-	seastar::now(),
-	interruptor::make_interruptible(osd_op_errorator::now()));
+
+  auto submitted = interruptor::now();
+  auto all_completed =
+    interruptor::make_interruptible(osd_op_errorator::now());
+
   if (cloning_ctx) {
     ceph_assert(want_mutate);
   }
+
   if (want_mutate) {
-    maybe_mutated = flush_clone_metadata(
+    auto log_entries = co_await flush_clone_metadata(
       prepare_transaction(ops),
       snap_mapper,
       osdriver,
-      txn
-    ).then_interruptible([mut_func=std::move(mut_func),
-                          this](auto&& log_entries) mutable {
-      if (auto log_rit = log_entries.rbegin(); log_rit != log_entries.rend()) {
-        ceph_assert(log_rit->version == osd_op_params->at_version);
-      }
-      auto [submitted, all_completed] =
-        std::forward<MutFunc>(mut_func)(std::move(txn),
-                                        std::move(obc),
-                                        std::move(*osd_op_params),
-                                        std::move(log_entries));
-      return interruptor::make_ready_future<rep_op_fut_tuple>(
-	std::move(submitted),
-	osd_op_ierrorator::future<>(std::move(all_completed)));
-    });
+      txn);
+
+    if (auto log_rit = log_entries.rbegin(); log_rit != log_entries.rend()) {
+      ceph_assert(log_rit->version == osd_op_params->at_version);
+    }
+
+    auto [_submitted, _all_completed] = co_await mut_func(
+      std::move(txn),
+      std::move(obc),
+      std::move(*osd_op_params),
+      std::move(log_entries));
+
+    submitted = std::move(_submitted);
+    all_completed = std::move(_all_completed);
   }
+
   apply_stats();
 
-  if (__builtin_expect(op_effects.empty(), true)) {
-    return maybe_mutated;
-  } else {
-    return maybe_mutated.then_unpack_interruptible(
-      // need extra ref pg due to apply_stats() which can be executed after
-      // informing snap mapper
-      [this, pg=this->pg](auto&& submitted, auto&& all_completed) mutable {
-      return interruptor::make_ready_future<rep_op_fut_tuple>(
-	  std::move(submitted),
-	  all_completed.safe_then_interruptible([this, pg=std::move(pg)] {
-	    // let's do the cleaning of `op_effects` in destructor
-	    return interruptor::do_for_each(op_effects,
-	      [pg=std::move(pg)](auto& op_effect) {
-	      return op_effect->execute(pg);
-	    });
-	  }));
+  if (op_effects.size()) [[unlikely]] {
+    // need extra ref pg due to apply_stats() which can be executed after
+    // informing snap mapper
+    all_completed =
+      std::move(all_completed).safe_then_interruptible([this, pg=this->pg] {
+      // let's do the cleaning of `op_effects` in destructor
+      return interruptor::do_for_each(op_effects,
+        [pg=std::move(pg)](auto& op_effect) {
+        return op_effect->execute(pg);
+      });
     });
   }
+
+  co_return std::make_tuple(
+    std::move(submitted),
+    std::move(all_completed));
 }
 
 template <class Func>

src/crimson/osd/osd_operations/snaptrim_event.cc

@@ -263,16 +263,18 @@ SnapTrimObjSubEvent::adjust_snaps(
     ghobject_t{coid, ghobject_t::NO_GEN, shard_id_t::NO_SHARD},
     OI_ATTR,
     bl);
-  add_log_entry(
+  auto &loge = add_log_entry(
     pg_log_entry_t::MODIFY,
     coid,
     obc->obs.oi.prior_version,
     0,
     osd_reqid_t(),
     obc->obs.oi.mtime,
     0);
-  return OpsExecuter::snap_map_modify(
-    coid, new_snaps, pg->snap_mapper, pg->osdriver, txn);
+  bufferlist snapsbl;
+  encode(new_snaps, snapsbl);
+  loge.snaps.swap(snapsbl);
+  return interruptor::now();
 }
 
 void SnapTrimObjSubEvent::update_head(
@@ -400,32 +402,8 @@ SnapTrimObjSubEvent::start()
   // lock both clone's and head's obcs
   co_await pg->obc_loader.with_obc<RWState::RWWRITE>(
     coid,
-    [this](auto head_obc, auto clone_obc) {
-      logger().debug("{}: got clone_obc={}", *this, clone_obc->get_oid());
-      return enter_stage<interruptor>(
-        client_pp().process
-      ).then_interruptible(
-        [this,clone_obc=std::move(clone_obc), head_obc=std::move(head_obc)]() mutable {
-	  logger().debug("{}: processing clone_obc={}", *this, clone_obc->get_oid());
-	  return remove_or_update(
-	    clone_obc, head_obc
-	  ).safe_then_interruptible([clone_obc, this](auto&& txn) mutable {
-	    auto [submitted, all_completed] = pg->submit_transaction(
-	      std::move(clone_obc),
-	      std::move(txn),
-	      std::move(osd_op_p),
-	      std::move(log_entries));
-	    return submitted.then_interruptible(
-	      [this, all_completed=std::move(all_completed)]() mutable {
-		return enter_stage<interruptor>(
-		  client_pp().wait_repop
-		).then_interruptible([all_completed=std::move(all_completed)]() mutable{
-		  return std::move(all_completed);
-		});
-	      });
-	  });
-	});
-    },
+    std::bind(&SnapTrimObjSubEvent::process_and_submit,
+              this, std::placeholders::_1, std::placeholders::_2),
     false
   ).handle_error_interruptible(
     remove_or_update_iertr::pass_further{},
@@ -436,6 +414,33 @@ SnapTrimObjSubEvent::start()
   co_await interruptor::make_interruptible(handle.complete());
 }
 
+ObjectContextLoader::load_obc_iertr::future<>
+SnapTrimObjSubEvent::process_and_submit(ObjectContextRef head_obc,
+                                        ObjectContextRef clone_obc) {
+  logger().debug("{}: got clone_obc={}", *this, clone_obc->get_oid());
+
+  co_await enter_stage<interruptor>(client_pp().process);
+
+  logger().debug("{}: processing clone_obc={}", *this, clone_obc->get_oid());
+
+  auto txn = co_await remove_or_update(clone_obc, head_obc);
+
+  auto [submitted, all_completed] = co_await pg->submit_transaction(
+	  std::move(clone_obc),
+	  std::move(txn),
+	  std::move(osd_op_p),
+	  std::move(log_entries)
+  );
+
+  co_await std::move(submitted);
+
+  co_await enter_stage<interruptor>(client_pp().wait_repop);
+
+  co_await std::move(all_completed);
+
+  co_return;
+}
+
 void SnapTrimObjSubEvent::print(std::ostream &lhs) const
 {
   lhs << "SnapTrimObjSubEvent("

src/crimson/osd/osd_operations/snaptrim_event.h

@@ -113,6 +113,10 @@ class SnapTrimObjSubEvent : public PhasedOperationT<SnapTrimObjSubEvent> {
 private:
   object_stat_sum_t delta_stats;
 
+  ObjectContextLoader::load_obc_iertr::future<> process_and_submit(
+    ObjectContextRef head_obc,
+    ObjectContextRef clone_obc);
+
   snap_trim_obj_subevent_ret_t remove_clone(
     ObjectContextRef obc,
     ObjectContextRef head_obc,
@@ -134,7 +138,7 @@ class SnapTrimObjSubEvent : public PhasedOperationT<SnapTrimObjSubEvent> {
   remove_or_update_iertr::future<ceph::os::Transaction>
   remove_or_update(ObjectContextRef obc, ObjectContextRef head_obc);
 
-  void add_log_entry(
+  pg_log_entry_t& add_log_entry(
     int _op,
     const hobject_t& _soid,
     const eversion_t& pv,
@@ -152,6 +156,7 @@ class SnapTrimObjSubEvent : public PhasedOperationT<SnapTrimObjSubEvent> {
       mt,
       return_code);
     osd_op_p.at_version.version++;
+    return log_entries.back();
   }
 
   Ref<PG> pg;