crimson: Implement a lock_excl_wait wrapper

The wrapper returns a future immediately if the lock
is available otherwise wait asynchronously

Fixes: https://tracker.ceph.com/issues/71861
Signed-off-by: Mohit Agrawal <moagrawa@redhat.com>

diff --git a/src/crimson/osd/object_context_loader.h b/src/crimson/osd/object_context_loader.h
index be6444f0496ebdaa01e38787aa4fa5fb3665d49c..2e830406de1e1c31e3216880bc3a12ee6067859e 100644 (file)
--- a/src/crimson/osd/object_context_loader.h
+++ b/src/crimson/osd/object_context_loader.h
@@ -60,6 +60,18 @@ public:
        state = RWState::RWEXCL;
       }
 
+      seastar::future<> lock_excl_wait() {
+        // Try immediate lock first
+        if (obc->lock.try_lock_for_excl()) {
+          state = RWState::RWEXCL;
+          return seastar::now();
+        }
+        // Otherwise, wait asynchronously
+        return obc->lock.lock_for_excl().then([this] {
+          state = RWState::RWEXCL;
+        });
+      }
+
       void demote_excl_to(RWState::State lock_type) {
        assert(state == RWState::RWEXCL);
        switch (lock_type) {
@@ -168,6 +180,10 @@ public:
       target_state.lock_excl_sync();
     }
 
+    seastar::future<> lock_excl_wait() {
+      return target_state.lock_excl_wait();
+    }
+
     ObjectContextRef &get_obc() {
       ceph_assert(!target_state.is_empty());
       ceph_assert(target_state.obc->is_loaded());

diff --git a/src/crimson/osd/replicated_recovery_backend.cc b/src/crimson/osd/replicated_recovery_backend.cc
index 226b66f1db2ffb9ddd28ecaab8a77b1ce59abca2..9956780a0f86c8dfc0cd0fdc7bf318b7102d5941 100644 (file)
--- a/src/crimson/osd/replicated_recovery_backend.cc
+++ b/src/crimson/osd/replicated_recovery_backend.cc
@@ -899,7 +899,9 @@ ReplicatedRecoveryBackend::_handle_pull_response(
     pull_info.stat.num_objects_recovered++;
     auto manager = pg.obc_loader.get_obc_manager(
       recovery_waiter.obc);
-    manager.lock_excl_sync(); /* cannot already be locked */
+    // Call lock_excl_wait if the lock is already held by load_and_lock_(head|clone)
+    // and has not been released yet.
+    co_await interruptor::make_interruptible(manager.lock_excl_wait());
     co_await pg.get_recovery_handler()->on_local_recover(
       push_op.soid, get_recovering(push_op.soid).pull_info->recovery_info,
       false, t
@@ -1024,7 +1026,9 @@ ReplicatedRecoveryBackend::handle_push(
     auto ptiter = replica_push_targets.find(push_op.recovery_info.soid);
     ceph_assert(ptiter != replica_push_targets.end());
     auto manager = pg.obc_loader.get_obc_manager(ptiter->second);
-    manager.lock_excl_sync(); /* cannot already be locked */
+    // Call lock_excl_wait if the lock is already held by load_and_lock_(head|clone)
+    // and has not been released yet.
+    co_await interruptor::make_interruptible(manager.lock_excl_wait());
 
     co_await pg.get_recovery_handler()->on_local_recover(
       push_op.recovery_info.soid, push_op.recovery_info,