crimson/osd: add PG::with_head_obc()

this method replicates `PG::get_or_load_head_obc()`. but uses a different
way to ensure that the "lock" on obc is always released even if the
called func throws. it always guard the called func with a
`with_lock()`, so `lock.unlock()` is always called. the plan is to
replace `PG::get_or_load_head_obc()` with `PG::with_head_obc()` in
the following changes piecemeal.

Signed-off-by: Kefu Chai <kchai@redhat.com>

diff --git a/src/crimson/osd/pg.cc b/src/crimson/osd/pg.cc
index f05d6f4d6979efd3c211a66043863ccfc5e89320..d21fbdc2c5daf2c481db22432b9ae64946c0eafc 100644 (file)
--- a/src/crimson/osd/pg.cc
+++ b/src/crimson/osd/pg.cc
@@ -807,6 +807,34 @@ PG::get_or_load_clone_obc(hobject_t oid, ObjectContextRef head)
   }
 }
 
+template<RWState::State State>
+seastar::future<>
+PG::with_head_obc(hobject_t oid, with_obc_func_t&& func)
+{
+  assert(oid.is_head());
+  auto [obc, existed] = shard_services.obc_registry.get_cached_obc(oid);
+  return obc->with_lock<State>(
+    [oid=std::move(oid), existed=existed, obc=std::move(obc),
+     func=std::move(func), this] {
+    auto loaded = seastar::make_ready_future<ObjectContextRef>(obc);
+    if (existed) {
+      logger().debug("with_head_obc: found {} in cache", oid);
+    } else {
+      logger().debug("with_head_obc: cache miss on {}", oid);
+      loaded = obc->with_promoted_lock<RWState::RWEXCL>([this, obc] {
+        return load_head_obc(obc);
+      });
+    }
+    return loaded.then([func = std::move(func)](auto obc) {
+      return func(std::move(obc));
+    });
+  });
+}
+
+// explicitly instantiate the used instantiations
+template seastar::future<>
+PG::with_head_obc<RWState::RWREAD>(hobject_t, with_obc_func_t&&);
+
 PG::load_obc_ertr::future<
   std::pair<crimson::osd::ObjectContextRef, bool>>
 PG::get_or_load_head_obc(hobject_t oid)

diff --git a/src/crimson/osd/pg.h b/src/crimson/osd/pg.h
index 1a2173231f60bb54c4d08b7fd5f202b6be86f38a..d51d806ffea8ead30301242c7d942691eb0a39a0 100644 (file)
--- a/src/crimson/osd/pg.h
+++ b/src/crimson/osd/pg.h
@@ -510,6 +510,10 @@ public:
 
 public:
   using with_obc_func_t = std::function<seastar::future<> (ObjectContextRef)>;
+
+  template<RWState::State State>
+  seastar::future<> with_head_obc(hobject_t oid, with_obc_func_t&& func);
+
   load_obc_ertr::future<> with_locked_obc(
     Ref<MOSDOp> &m,
     const OpInfo &op_info,

diff --git a/src/crimson/osd/replicated_recovery_backend.cc b/src/crimson/osd/replicated_recovery_backend.cc
index 10f2f84f032bc1c7ef5f0f0acde4725f52144f95..4f29568a1dcd71770972c6b5e6dbd330669b2108 100644 (file)
--- a/src/crimson/osd/replicated_recovery_backend.cc
+++ b/src/crimson/osd/replicated_recovery_backend.cc
@@ -122,22 +122,11 @@ seastar::future<> ReplicatedRecoveryBackend::load_obc_for_recovery(
   if (recovery_waiter.obc) {
     return seastar::now();
   }
-  return pg.get_or_load_head_obc(soid).safe_then(
-    [&recovery_waiter](auto p) {
-    auto& [obc, existed] = p;
+  return pg.with_head_obc<RWState::RWREAD>(soid, [&recovery_waiter](auto obc) {
     logger().debug("load_obc_for_recovery: loaded obc: {}", obc->obs.oi.soid);
     recovery_waiter.obc = obc;
-    if (!existed) {
-      // obc is loaded with excl lock
-      recovery_waiter.obc->put_lock_type(RWState::RWEXCL);
-    }
     return recovery_waiter.obc->wait_recovery_read();
-  }, crimson::osd::PG::load_obc_ertr::all_same_way(
-      [this, &recovery_waiter, soid](const std::error_code& e) {
-      logger().error("load_obc_for_recovery: load failure of obc: {}", soid);
-      return seastar::make_exception_future<>(e);
-    })
-  );
+  });
 }
 
 seastar::future<> ReplicatedRecoveryBackend::push_delete(