From b0d3d2a392cfa6e4e21731746ec1d0a32e7966f9 Mon Sep 17 00:00:00 2001
From: Kefu Chai <kchai@redhat.com>
Date: Tue, 10 Nov 2020 21:03:02 +0800
Subject: [PATCH] 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>
---
 src/crimson/osd/pg.cc                         | 28 +++++++++++++++++++
 src/crimson/osd/pg.h                          |  4 +++
 .../osd/replicated_recovery_backend.cc        | 15 ++--------
 3 files changed, 34 insertions(+), 13 deletions(-)

diff --git a/src/crimson/osd/pg.cc b/src/crimson/osd/pg.cc
index f05d6f4d697..d21fbdc2c5d 100644
--- 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 1a2173231f6..d51d806ffea 100644
--- 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 10f2f84f032..4f29568a1dc 100644
--- 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(
-- 
2.39.5