crimson/osd: add PG::with_clone_obc()

this method replaces `PG::get_or_load_clone_obc()`. so we can
with `seastar::with_lock()` to ensure that `lock.unlock()` is always
called when accessing clone obc.

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

diff --git a/src/crimson/osd/pg.cc b/src/crimson/osd/pg.cc
index a745580b1d46d8f186de5360b592498507b3cff1..5c78a5d115b60e4c7196978147a967f63f0a2090 100644 (file)
--- a/src/crimson/osd/pg.cc
+++ b/src/crimson/osd/pg.cc
@@ -770,43 +770,6 @@ std::optional<hobject_t> PG::resolve_oid(
   }
 }
 
-PG::load_obc_ertr::future<
-  std::pair<crimson::osd::ObjectContextRef, bool>>
-PG::get_or_load_clone_obc(hobject_t oid, ObjectContextRef head)
-{
-  if (__builtin_expect(stopping, false)) {
-    throw crimson::common::system_shutdown_exception();
-  }
-
-  ceph_assert(!oid.is_head());
-  using ObjectContextRef = crimson::osd::ObjectContextRef;
-  auto coid = resolve_oid(head->get_ro_ss(), oid);
-  if (!coid) {
-    return load_obc_ertr::make_ready_future<
-      std::pair<crimson::osd::ObjectContextRef, bool>>(
-       std::make_pair(ObjectContextRef(), true)
-      );
-  }
-  auto [obc, existed] = shard_services.obc_registry.get_cached_obc(*coid);
-  if (existed) {
-    return load_obc_ertr::make_ready_future<
-      std::pair<crimson::osd::ObjectContextRef, bool>>(
-       std::make_pair(obc, true)
-      );
-  } else {
-    bool got = obc->maybe_get_excl();
-    ceph_assert(got);
-    return backend->load_metadata(*coid).safe_then(
-      [oid, obc=std::move(obc), head](auto &&md) mutable {
-       obc->set_clone_state(std::move(md->os), std::move(head));
-       return load_obc_ertr::make_ready_future<
-         std::pair<crimson::osd::ObjectContextRef, bool>>(
-           std::make_pair(obc, false)
-         );
-      });
-  }
-}
-
 template<RWState::State State>
 seastar::future<>
 PG::with_head_obc(hobject_t oid, with_obc_func_t&& func)
@@ -831,10 +794,45 @@ PG::with_head_obc(hobject_t oid, with_obc_func_t&& func)
   });
 }
 
-// explicitly instantiate the used instantiations
-template seastar::future<>
-PG::with_head_obc<RWState::RWREAD>(hobject_t, with_obc_func_t&&);
+template<RWState::State State>
+seastar::future<>
+PG::with_clone_obc(hobject_t oid, with_obc_func_t&& func)
+{
+  assert(!oid.is_head());
+  return with_head_obc<RWState::RWREAD>(oid.get_head(),
+    [oid, func=std::move(func), this](auto head) {
+    auto coid = resolve_oid(head->get_ro_ss(), oid);
+    if (!coid) {
+      // TODO: return crimson::ct_error::enoent::make();
+      logger().error("with_clone_obc: {} clone not found", coid);
+      return seastar::make_ready_future<>();
+    }
+    auto [clone, existed] = shard_services.obc_registry.get_cached_obc(*coid);
+    return clone->template with_lock<State>(
+      [coid=*coid, existed=existed,
+      head=std::move(head), clone=std::move(clone), func=std::move(func), this] {
+      auto loaded = seastar::make_ready_future<ObjectContextRef>(clone);
+      if (existed) {
+        logger().debug("with_clone_obc: found {} in cache", coid);
+      } else {
+        logger().debug("with_clone_obc: cache miss on {}", coid);
+        loaded = clone->template with_promoted_lock<RWState::RWEXCL>(
+          [coid, clone, head, this] {
+          return backend->load_metadata(coid).safe_then(
+            [coid, clone=std::move(clone), head=std::move(head)](auto md) mutable {
+            clone->set_clone_state(std::move(md->os), std::move(head));
+            return clone;
+          });
+        });
+      }
+      return loaded.then([func = std::move(func)](auto clone) {
+        return func(std::move(clone));
+      });
+    });
+  });
+}
 
+// explicitly instantiate the used instantiations
 template seastar::future<>
 PG::with_head_obc<RWState::RWNONE>(hobject_t, with_obc_func_t&&);
 
@@ -868,60 +866,29 @@ PG::with_locked_obc(Ref<MOSDOp> &m, const OpInfo &op_info,
   if (__builtin_expect(stopping, false)) {
     throw crimson::common::system_shutdown_exception();
   }
-  RWState::State type = get_lock_type(op_info);
-  return get_locked_obc(op, get_oid(*m), type)
-    .safe_then([f=std::move(f), type=type](auto obc) {
-    return f(obc).finally([obc, type=type] {
-      obc->put_lock_type(type);
-      return load_obc_ertr::now();
-    });
-  });
-}
-
-PG::load_obc_ertr::future<crimson::osd::ObjectContextRef>
-PG::get_locked_obc(
-  Operation *op, const hobject_t &oid, RWState::State type)
-{
-  if (__builtin_expect(stopping, false)) {
-    throw crimson::common::system_shutdown_exception();
-  }
-
-  return get_or_load_head_obc(oid.get_head()).safe_then(
-    [this, op, oid, type](auto p) -> load_obc_ertr::future<ObjectContextRef>{
-      auto &[head_obc, head_existed] = p;
-      if (oid.is_head()) {
-       if (head_existed) {
-         return head_obc->get_lock_type(op, type).then([head_obc=head_obc] {
-           ceph_assert(head_obc->loaded);
-           return load_obc_ertr::make_ready_future<ObjectContextRef>(head_obc);
-         });
-       } else {
-         head_obc->degrade_excl_to(type);
-         return load_obc_ertr::make_ready_future<ObjectContextRef>(head_obc);
-       }
-      } else {
-       return head_obc->get_lock_type(op, RWState::RWREAD).then(
-         [this, head_obc=head_obc, oid] {
-           ceph_assert(head_obc->loaded);
-           return get_or_load_clone_obc(oid, head_obc);
-         }).safe_then([head_obc=head_obc, op, oid, type](auto p) {
-             auto &[obc, existed] = p;
-             if (existed) {
-               return load_obc_ertr::future<>(
-                 obc->get_lock_type(op, type)).safe_then([obc=obc] {
-                 ceph_assert(obc->loaded);
-                 return load_obc_ertr::make_ready_future<ObjectContextRef>(obc);
-               });
-             } else {
-               obc->degrade_excl_to(type);
-               return load_obc_ertr::make_ready_future<ObjectContextRef>(obc);
-             }
-         }).safe_then([head_obc=head_obc](auto obc) {
-           head_obc->put_lock_type(RWState::RWREAD);
-           return load_obc_ertr::make_ready_future<ObjectContextRef>(obc);
-         });
-      }
-    });
+  const hobject_t oid = get_oid(*m);
+  switch (get_lock_type(op_info)) {
+  case RWState::RWREAD:
+    if (oid.is_head()) {
+      return with_head_obc<RWState::RWREAD>(oid, std::move(f));
+    } else {
+      return with_clone_obc<RWState::RWREAD>(oid, std::move(f));
+    }
+  case RWState::RWWRITE:
+    if (oid.is_head()) {
+      return with_head_obc<RWState::RWWRITE>(oid, std::move(f));
+    } else {
+      return with_clone_obc<RWState::RWWRITE>(oid, std::move(f));
+    }
+  case RWState::RWEXCL:
+    if (oid.is_head()) {
+      return with_head_obc<RWState::RWWRITE>(oid, std::move(f));
+    } else {
+      return with_clone_obc<RWState::RWWRITE>(oid, std::move(f));
+    }
+  default:
+    assert(0);
+  };
 }
 
 seastar::future<> PG::handle_rep_op(Ref<MOSDRepOp> req)

diff --git a/src/crimson/osd/pg.h b/src/crimson/osd/pg.h
index 993e4cc9dcc956d79e19e28a5ac4c94fa4fc2c7e..33782da01d3efe70e54e7c0edc23a31d49acb0bd 100644 (file)
--- a/src/crimson/osd/pg.h
+++ b/src/crimson/osd/pg.h
@@ -496,10 +496,6 @@ public:
 
   using load_obc_ertr = crimson::errorator<
     crimson::ct_error::object_corrupted>;
-  load_obc_ertr::future<
-    std::pair<crimson::osd::ObjectContextRef, bool>>
-  get_or_load_clone_obc(
-    hobject_t oid, crimson::osd::ObjectContextRef head_obc);
 
   load_obc_ertr::future<crimson::osd::ObjectContextRef>
   load_head_obc(ObjectContextRef obc);
@@ -524,6 +520,9 @@ public:
   void dump_primary(Formatter*);
 
 private:
+  template<RWState::State State>
+  seastar::future<> with_clone_obc(hobject_t oid, with_obc_func_t&& func);
+
   load_obc_ertr::future<ObjectContextRef> get_locked_obc(
     Operation *op,
     const hobject_t &oid,