Merge pull request ceph#55101 from Matan-B/wip-matanb-crimson-snaptrim_event-cleanup

crimson/osd/osd_operations: snaptrim_event cleanup

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

Author: Matan-B

src/crimson/common/subop_blocker.h

@@ -0,0 +1,53 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+
+#pragma once
+
+#include "osd/osd_op_util.h"
+#include "crimson/osd/osd_operation.h"
+
+namespace crimson::osd {
+
+using interruptor =
+  ::crimson::interruptible::interruptor<
+    ::crimson::osd::IOInterruptCondition>;
+
+// bases on 998cb8c141bb89aafae298a9d5e130fbd78fe5f2
+template <typename T>
+struct SubOpBlocker : crimson::BlockerT<SubOpBlocker<T>> {
+  static constexpr const char* type_name = "CompoundOpBlocker";
+
+  using id_done_t = std::pair<crimson::OperationRef, T>;
+
+  void dump_detail(Formatter *f) const final {
+    f->open_array_section("dependent_operations");
+    {
+      for (const auto &kv : subops) {
+        f->dump_unsigned("op_id", kv.first->get_id());
+      }
+    }
+    f->close_section();
+  }
+
+  template <class... Args>
+  void emplace_back(Args&&... args) {
+      subops.emplace_back(std::forward<Args>(args)...);
+  };
+
+  T interruptible_wait_completion() {
+    return interruptor::do_for_each(subops, [](auto&& kv) {
+      return std::move(kv.second);
+    });
+  }
+
+  T wait_completion() {
+    return seastar::do_for_each(subops, [](auto&& kv) {
+      return std::move(kv.second);
+    });
+  }
+
+private:
+  std::vector<id_done_t> subops;
+};
+
+} // namespace crimson::osd

src/crimson/osd/osd_operations/background_recovery.cc

@@ -12,12 +12,6 @@
 #include "crimson/osd/osd_operation_external_tracking.h"
 #include "crimson/osd/osd_operations/background_recovery.h"
 
-namespace {
-  seastar::logger& logger() {
-    return crimson::get_logger(ceph_subsys_osd);
-  }
-}
-
 namespace crimson {
   template <>
   struct EventBackendRegistry<osd::UrgentRecovery> {

src/crimson/osd/osd_operations/pg_advance_map.cc

@@ -21,9 +21,9 @@ namespace {
 namespace crimson::osd {
 
 PGAdvanceMap::PGAdvanceMap(
-  ShardServices &shard_services, Ref<PG> pg, epoch_t to,
+  Ref<PG> pg, ShardServices &shard_services, epoch_t to,
   PeeringCtx &&rctx, bool do_init)
-  : shard_services(shard_services), pg(pg), to(to),
+  : pg(pg), shard_services(shard_services), to(to),
     rctx(std::move(rctx)), do_init(do_init)
 {
   logger().debug("{}: created", *this);

src/crimson/osd/osd_operations/pg_advance_map.h

@@ -25,8 +25,8 @@ class PGAdvanceMap : public PhasedOperationT<PGAdvanceMap> {
   static constexpr OperationTypeCode type = OperationTypeCode::pg_advance_map;
 
 protected:
-  ShardServices &shard_services;
   Ref<PG> pg;
+  ShardServices &shard_services;
   PipelineHandle handle;
 
   std::optional<epoch_t> from;
@@ -37,7 +37,7 @@ class PGAdvanceMap : public PhasedOperationT<PGAdvanceMap> {
 
 public:
   PGAdvanceMap(
-    ShardServices &shard_services, Ref<PG> pg, epoch_t to,
+    Ref<PG> pg, ShardServices &shard_services, epoch_t to,
     PeeringCtx &&rctx, bool do_init);
   ~PGAdvanceMap();
 

src/crimson/osd/osd_operations/snaptrim_event.cc

@@ -39,31 +39,6 @@ PG::BackgroundProcessLock::lock_with_op(SnapTrimEvent &st_event) noexcept
   });
 }
 
-void SnapTrimEvent::SubOpBlocker::dump_detail(Formatter *f) const
-{
-  f->open_array_section("dependent_operations");
-  {
-    for (const auto &kv : subops) {
-      f->dump_unsigned("op_id", kv.first->get_id());
-    }
-  }
-  f->close_section();
-}
-
-template <class... Args>
-void SnapTrimEvent::SubOpBlocker::emplace_back(Args&&... args)
-{
-  subops.emplace_back(std::forward<Args>(args)...);
-};
-
-SnapTrimEvent::remove_or_update_iertr::future<>
-SnapTrimEvent::SubOpBlocker::wait_completion()
-{
-  return interruptor::do_for_each(subops, [](auto&& kv) {
-    return std::move(kv.second);
-  });
-}
-
 void SnapTrimEvent::print(std::ostream &lhs) const
 {
   lhs << "SnapTrimEvent("
@@ -85,19 +60,11 @@ CommonPGPipeline& SnapTrimEvent::client_pp()
   return pg->request_pg_pipeline;
 }
 
-SnapTrimEvent::snap_trim_ertr::future<seastar::stop_iteration>
+SnapTrimEvent::snap_trim_event_ret_t
 SnapTrimEvent::start()
 {
   ShardServices &shard_services = pg->get_shard_services();
-  IRef ref = this;
-  return interruptor::with_interruption(
-    // SnapTrimEvent is a background operation,
-    // it's lifetime is not guarnteed since the caller
-    // returned future is being ignored. We should capture
-    // a self reference thourhgout the entire execution
-    // progress (not only on finally() continuations).
-    // See: PG::on_active_actmap()
-    [&shard_services, this, ref] {
+  return interruptor::with_interruption([&shard_services, this] {
     return enter_stage<interruptor>(
       client_pp().wait_for_active
     ).then_interruptible([this] {
@@ -154,22 +121,19 @@ SnapTrimEvent::start()
         return [&shard_services, this](const auto &to_trim) {
 	  for (const auto& object : to_trim) {
 	    logger().debug("{}: trimming {}", *this, object);
-	    auto [op, fut] = shard_services.start_operation_may_interrupt<
-	      interruptor, SnapTrimObjSubEvent>(
-	      pg,
-	      object,
-	      snapid);
 	    subop_blocker.emplace_back(
-	      std::move(op),
-	      std::move(fut)
-	    );
+	      shard_services.start_operation_may_interrupt<
+	      interruptor, SnapTrimObjSubEvent>(
+	        pg,
+	        object,
+	        snapid));
 	  }
 	  return interruptor::now();
 	}(to_trim).then_interruptible([this] {
 	  return enter_stage<interruptor>(wait_subop);
 	}).then_interruptible([this] {
           logger().debug("{}: awaiting completion", *this);
-          return subop_blocker.wait_completion();
+          return subop_blocker.interruptible_wait_completion();
         }).finally([this] {
 	  pg->background_process_lock.unlock();
 	}).si_then([this] {
@@ -200,11 +164,12 @@ SnapTrimEvent::start()
         });
       });
     });
-  }, [this, ref]
-     (std::exception_ptr eptr) -> snap_trim_ertr::future<seastar::stop_iteration> {
+  }, [this](std::exception_ptr eptr) -> snap_trim_event_ret_t {
     logger().debug("{}: interrupted {}", *this, eptr);
     return crimson::ct_error::eagain::make();
-  }, pg).finally([this, ref=std::move(ref)] {
+  }, pg).finally([this] {
+    // This SnapTrimEvent op lifetime is maintained within
+    // PerShardState::start_operation() implementation.
     logger().debug("{}: exit", *this);
     handle.exit();
   });
@@ -216,7 +181,7 @@ CommonPGPipeline& SnapTrimObjSubEvent::client_pp()
   return pg->request_pg_pipeline;
 }
 
-SnapTrimObjSubEvent::remove_or_update_iertr::future<>
+SnapTrimObjSubEvent::snap_trim_obj_subevent_ret_t
 SnapTrimObjSubEvent::remove_clone(
   ObjectContextRef obc,
   ObjectContextRef head_obc,
@@ -463,7 +428,7 @@ SnapTrimObjSubEvent::remove_or_update(
   });
 }
 
-SnapTrimObjSubEvent::remove_or_update_iertr::future<>
+SnapTrimObjSubEvent::snap_trim_obj_subevent_ret_t
 SnapTrimObjSubEvent::start()
 {
   return enter_stage<interruptor>(

src/crimson/osd/osd_operations/snaptrim_event.h

@@ -8,6 +8,7 @@
 
 #include "crimson/osd/osdmap_gate.h"
 #include "crimson/osd/osd_operation.h"
+#include "crimson/common/subop_blocker.h"
 #include "crimson/osd/osd_operations/common/pg_pipeline.h"
 #include "crimson/osd/pg.h"
 #include "crimson/osd/pg_activation_blocker.h"
@@ -38,6 +39,10 @@ class SnapTrimEvent final : public PhasedOperationT<SnapTrimEvent> {
     crimson::ct_error::eagain>;
   using snap_trim_iertr = remove_or_update_iertr::extend<
     crimson::ct_error::eagain>;
+  using snap_trim_event_ret_t =
+    snap_trim_ertr::future<seastar::stop_iteration>;
+  using snap_trim_obj_subevent_ret_t =
+      remove_or_update_iertr::future<>;
 
   static constexpr OperationTypeCode type = OperationTypeCode::snaptrim_event;
 
@@ -52,27 +57,12 @@ class SnapTrimEvent final : public PhasedOperationT<SnapTrimEvent> {
 
   void print(std::ostream &) const final;
   void dump_detail(ceph::Formatter* f) const final;
-  snap_trim_ertr::future<seastar::stop_iteration> start();
+  snap_trim_event_ret_t start();
 
 private:
   CommonPGPipeline& client_pp();
 
-  // bases on 998cb8c141bb89aafae298a9d5e130fbd78fe5f2
-  struct SubOpBlocker : crimson::BlockerT<SubOpBlocker> {
-    static constexpr const char* type_name = "CompoundOpBlocker";
-
-    using id_done_t = std::pair<crimson::OperationRef,
-                                remove_or_update_iertr::future<>>;
-
-    void dump_detail(Formatter *f) const final;
-
-    template <class... Args>
-    void emplace_back(Args&&... args);
-
-    remove_or_update_iertr::future<> wait_completion();
-  private:
-    std::vector<id_done_t> subops;
-  } subop_blocker;
+  SubOpBlocker<snap_trim_obj_subevent_ret_t> subop_blocker;
 
   // we don't need to synchronize with other instances of SnapTrimEvent;
   // it's here for the sake of op tracking.
@@ -87,8 +77,8 @@ class SnapTrimEvent final : public PhasedOperationT<SnapTrimEvent> {
     static constexpr auto type_name = "SnapTrimEvent::wait_trim_timer";
   } wait_trim_timer;
 
-  PipelineHandle handle;
   Ref<PG> pg;
+  PipelineHandle handle;
   SnapMapper& snap_mapper;
   const snapid_t snapid;
   const bool needs_pause;
@@ -122,6 +112,8 @@ class SnapTrimObjSubEvent : public PhasedOperationT<SnapTrimObjSubEvent> {
   using remove_or_update_iertr =
     crimson::interruptible::interruptible_errorator<
       IOInterruptCondition, remove_or_update_ertr>;
+  using snap_trim_obj_subevent_ret_t =
+      remove_or_update_iertr::future<>;
 
   static constexpr OperationTypeCode type =
     OperationTypeCode::snaptrimobj_subevent;
@@ -137,7 +129,7 @@ class SnapTrimObjSubEvent : public PhasedOperationT<SnapTrimObjSubEvent> {
 
   void print(std::ostream &) const final;
   void dump_detail(ceph::Formatter* f) const final;
-  remove_or_update_iertr::future<> start();
+  snap_trim_obj_subevent_ret_t start();
 
   CommonPGPipeline& client_pp();
 
@@ -146,7 +138,7 @@ class SnapTrimObjSubEvent : public PhasedOperationT<SnapTrimObjSubEvent> {
    * https://tracker.ceph.com/issues/63307 */
   object_stat_sum_t delta_stats;
 
-  remove_or_update_iertr::future<> remove_clone(
+  snap_trim_obj_subevent_ret_t remove_clone(
     ObjectContextRef obc,
     ObjectContextRef head_obc,
     ceph::os::Transaction& txn);

src/crimson/osd/shard_services.cc

@@ -97,8 +97,9 @@ seastar::future<> PerShardState::broadcast_map_to_pgs(
     pgs.begin(), pgs.end(),
     [=, &shard_services](auto& pg) {
       return shard_services.start_operation<PGAdvanceMap>(
+	pg.second,
 	shard_services,
-	pg.second, epoch,
+	epoch,
 	PeeringCtx{}, false).second;
     });
 }
@@ -689,7 +690,7 @@ seastar::future<Ref<PG>> ShardServices::handle_pg_create_info(
 	    rctx.transaction);
 
 	  return start_operation<PGAdvanceMap>(
-	    *this, pg, get_map()->get_epoch(), std::move(rctx), true
+	    pg, *this, get_map()->get_epoch(), std::move(rctx), true
 	  ).second.then([pg=pg] {
 	    return seastar::make_ready_future<Ref<PG>>(pg);
 	  });