crimson: use make_interruptible, coroutine, and RAII releaser for recover_object_with_throttle

791772f used with_throttle here in a way which caused
then_interruptible in PGRecovery::recover_object to be called outside of
an interruptible context.

Instead of using a wrapper taking a lambda, rephrase as an RAII releaser
suitable for use in a coroutine.  This avoids needing to structure
with_throttle to deal correctly with both interruptible and
non-interruptible contexts.

Fixes: https://tracker.ceph.com/issues/70939
Signed-off-by: Samuel Just <[email protected]>

Author: athanatos

src/crimson/osd/osd_operation.h

@@ -310,6 +310,30 @@ class OperationThrottler : public BlockerT<OperationThrottler>,
     return !max_in_progress || in_progress < max_in_progress;
   }
 
+  class ThrottleReleaser {
+    OperationThrottler *parent = nullptr;
+  public:
+    ThrottleReleaser(OperationThrottler *parent) : parent(parent) {}
+    ThrottleReleaser(const ThrottleReleaser &) = delete;
+    ThrottleReleaser(ThrottleReleaser &&rhs) noexcept {
+      std::swap(parent, rhs.parent);
+    }
+
+    ~ThrottleReleaser() {
+      if (parent) {
+	parent->release_throttle();
+      }
+    }
+  };
+
+  auto get_throttle(crimson::osd::scheduler::params_t params) {
+    return acquire_throttle(
+      params
+    ).then([this] {
+      return ThrottleReleaser{this};
+    });
+  }
+
   template <typename F>
   auto with_throttle(
     crimson::osd::scheduler::params_t params,

src/crimson/osd/pg_recovery.cc

@@ -5,6 +5,7 @@
 #include <fmt/ostream.h>
 #include <fmt/ranges.h>
 
+#include "crimson/common/coroutine.h"
 #include "crimson/common/log.h"
 #include "crimson/common/type_helpers.h"
 #include "crimson/osd/backfill_facades.h"
@@ -525,22 +526,19 @@ void PGRecovery::request_primary_scan(
 
 PGRecovery::interruptible_future<>
 PGRecovery::recover_object_with_throttle(
-  const hobject_t &soid,
+  hobject_t soid,
   eversion_t need)
 {
   LOG_PREFIX(PGRecovery::recover_object_with_throttle);
-  crimson::osd::scheduler::params_t params =
-    {1, 0, crimson::osd::scheduler::scheduler_class_t::background_best_effort};
-  auto &ss = pg->get_shard_services();
   DEBUGDPP("{} {}", pg->get_dpp(), soid, need);
-  return ss.with_throttle(
-    std::move(params),
-    [FNAME, this, soid, need] {
-    DEBUGDPP("got throttle: {} {}", pg->get_dpp(), soid, need);
-    auto backend = pg->get_recovery_backend();
-    assert(backend);
-    return backend->recover_object(soid, need);
-  });
+  auto releaser = co_await interruptor::make_interruptible(
+    pg->get_shard_services().get_throttle(
+      crimson::osd::scheduler::params_t{
+	1, 0, crimson::osd::scheduler::scheduler_class_t::background_best_effort
+      }));
+  DEBUGDPP("got throttle: {} {}", pg->get_dpp(), soid, need);
+  co_await pg->get_recovery_backend()->recover_object(soid, need);
+  co_return;
 }
 
 void PGRecovery::enqueue_push(

src/crimson/osd/pg_recovery.h

@@ -25,6 +25,9 @@ class PGBackend;
 
 class PGRecovery : public crimson::osd::BackfillState::BackfillListener {
 public:
+  using interruptor =
+    ::crimson::interruptible::interruptor<
+      ::crimson::osd::IOInterruptCondition>;
   template <typename T = void>
   using interruptible_future = RecoveryBackend::interruptible_future<T>;
   PGRecovery(PGRecoveryListener* pg) : pg(pg) {}
@@ -100,7 +103,7 @@ class PGRecovery : public crimson::osd::BackfillState::BackfillListener {
   friend class crimson::osd::UrgentRecovery;
 
   interruptible_future<> recover_object_with_throttle(
-    const hobject_t &soid,
+    hobject_t soid,
     eversion_t need);
 
   interruptible_future<> recover_object(

src/crimson/osd/shard_services.h

@@ -594,6 +594,7 @@ class ShardServices : public OSDMapService {
 
   FORWARD_TO_OSD_SINGLETON(get_pool_info)
   FORWARD(with_throttle, with_throttle, local_state.throttler)
+  FORWARD(get_throttle, get_throttle, local_state.throttler)
 
   FORWARD_TO_OSD_SINGLETON(build_incremental_map_msg)
   FORWARD_TO_OSD_SINGLETON(send_incremental_map)