crimson/osd: OpsExecuter retrieves PG when doing op effects.

This will necessary to spawn the upcoming `InternalClientRequest`
from the `Watch`'s timeout handler.

Signed-off-by: Radoslaw Zarzynski <rzarzyns@redhat.com>

diff --git a/src/crimson/osd/ops_executer.cc b/src/crimson/osd/ops_executer.cc
index 3bd94b915f75b40a1a925cc2f94937786fc849a2..db3078e1aac5bd2dd6875eddbabb66512f2ae6a6 100644 (file)
--- a/src/crimson/osd/ops_executer.cc
+++ b/src/crimson/osd/ops_executer.cc
@@ -180,7 +180,7 @@ OpsExecuter::watch_ierrorator::future<> OpsExecuter::do_op_watch_subop_watch(
       }
       return seastar::now();
     },
-    [] (auto&& ctx, ObjectContextRef obc) {
+    [] (auto&& ctx, ObjectContextRef obc, Ref<PG>) {
       auto [it, emplaced] = obc->watchers.try_emplace(ctx.key, nullptr);
       if (emplaced) {
         const auto& [cookie, entity] = ctx.key;
@@ -234,7 +234,7 @@ OpsExecuter::watch_ierrorator::future<> OpsExecuter::do_op_watch_subop_unwatch(
       }
       return seastar::now();
     },
-    [] (auto&& ctx, ObjectContextRef obc) {
+    [] (auto&& ctx, ObjectContextRef obc, Ref<PG>) {
       if (auto nh = obc->watchers.extract(ctx.key); !nh.empty()) {
         return seastar::do_with(std::move(nh.mapped()),
                          [ctx](auto&& watcher) {
@@ -347,7 +347,7 @@ OpsExecuter::watch_ierrorator::future<> OpsExecuter::do_op_notify(
       ceph::encode(ctx.ninfo.notify_id, osd_op.outdata);
       return seastar::now();
     },
-    [] (auto&& ctx, ObjectContextRef obc) {
+    [] (auto&& ctx, ObjectContextRef obc, Ref<PG>) {
       auto alive_watchers = obc->watchers | boost::adaptors::map_values
                                           | boost::adaptors::filtered(
         [] (const auto& w) {
@@ -396,7 +396,7 @@ OpsExecuter::watch_ierrorator::future<> OpsExecuter::do_op_notify_ack(
       }
       return watch_errorator::now();
     },
-    [] (auto&& ctx, ObjectContextRef obc) {
+    [] (auto&& ctx, ObjectContextRef obc, Ref<PG>) {
       logger().info("notify_ack watch_cookie={}, notify_id={}",
                     ctx.watch_cookie, ctx.notify_id);
       return seastar::do_for_each(obc->watchers,

diff --git a/src/crimson/osd/ops_executer.h b/src/crimson/osd/ops_executer.h
index bed8cf6966f1b14ae0801b2e63cd6e0940d05028..a007bfd0f43de0a8b6da36783fdbefe460f51a4a 100644 (file)
--- a/src/crimson/osd/ops_executer.h
+++ b/src/crimson/osd/ops_executer.h
@@ -105,7 +105,8 @@ private:
   // when operation requires this division, some variant of `with_effect()`
   // should be used.
   struct effect_t {
-    virtual osd_op_errorator::future<> execute() = 0;
+    // an effect can affect PG, i.e. create a watch timeout
+    virtual osd_op_errorator::future<> execute(Ref<PG> pg) = 0;
     virtual ~effect_t() = default;
   };
 
@@ -204,7 +205,9 @@ public:
   execute_op(class OSDOp& osd_op);
 
   template <typename MutFunc>
-  osd_op_ierrorator::future<> flush_changes(MutFunc&& mut_func) &&;
+  osd_op_ierrorator::future<> flush_changes_n_do_ops_effects(
+    Ref<PG> pg,
+    MutFunc&& mut_func) &&;
 
   const hobject_t &get_target() const {
     return obc->obs.oi.soid;
@@ -238,7 +241,7 @@ auto OpsExecuter::with_effect_on_obc(
   // lambda only when it's closureless. We enforce this restriction due
   // the fact that `flush_changes()` std::moves many executer's parts.
   using allowed_effect_func_t =
-    seastar::future<> (*)(context_t&&, ObjectContextRef);
+    seastar::future<> (*)(context_t&&, ObjectContextRef, Ref<PG>);
   static_assert(std::is_convertible_v<EffectFunc, allowed_effect_func_t>,
                 "with_effect function is not allowed to capture");
   struct task_t final : effect_t {
@@ -251,8 +254,10 @@ auto OpsExecuter::with_effect_on_obc(
          effect_func(std::move(effect_func)),
          obc(std::move(obc)) {
     }
-    osd_op_errorator::future<> execute() final {
-      return std::move(effect_func)(std::move(ctx), std::move(obc));
+    osd_op_errorator::future<> execute(Ref<PG> pg) final {
+      return std::move(effect_func)(std::move(ctx),
+                                    std::move(obc),
+                                    std::move(pg));
     }
   };
   auto task =
@@ -263,8 +268,8 @@ auto OpsExecuter::with_effect_on_obc(
 }
 
 template <typename MutFunc>
-OpsExecuter::osd_op_ierrorator::future<> OpsExecuter::flush_changes(
-  MutFunc&& mut_func) &&
+OpsExecuter::osd_op_ierrorator::future<>
+OpsExecuter::flush_changes_n_do_ops_effects(Ref<PG> pg, MutFunc&& mut_func) &&
 {
   const bool want_mutate = !txn.empty();
   // osd_op_params are instantiated by every wr-like operation.
@@ -280,10 +285,12 @@ OpsExecuter::osd_op_ierrorator::future<> OpsExecuter::flush_changes(
   if (__builtin_expect(op_effects.empty(), true)) {
     return maybe_mutated;
   } else {
-    return maybe_mutated.safe_then_interruptible([this] {
+    return maybe_mutated.safe_then_interruptible([pg=std::move(pg),
+                                                  this] () mutable {
       // let's do the cleaning of `op_effects` in destructor
-      return interruptor::do_for_each(op_effects, [] (auto& op_effect) {
-        return op_effect->execute();
+      return interruptor::do_for_each(op_effects,
+                                      [pg=std::move(pg)] (auto& op_effect) {
+        return op_effect->execute(pg);
       });
     });
   }

diff --git a/src/crimson/osd/pg.cc b/src/crimson/osd/pg.cc
index 3425c951ef09815a3a320ca137a662d837875326..298f5223a7f516aea2d900d993dcd8ee73bd4f54 100644 (file)
--- a/src/crimson/osd/pg.cc
+++ b/src/crimson/osd/pg.cc
@@ -727,7 +727,8 @@ PG::do_osd_ops(
       "do_osd_ops: {} - object {} all operations successful",
       *m,
       ox->get_target());
-    return std::move(*ox).flush_changes(
+    return std::move(*ox).flush_changes_n_do_ops_effects(
+      Ref<PG>{this},
       [this, m, &op_info] (auto&& txn,
                           auto&& obc,
                           auto&& osd_op_p,