librbd: always respond to "release lock" request if lock owner

Fixes: http://tracker.ceph.com/issues/17030
Signed-off-by: Jason Dillaman <dillaman@redhat.com>

src/librbd/ExclusiveLock.cc

@@ -135,7 +135,7 @@ void ExclusiveLock<I>::shut_down(Context *on_shut_down) {
   }
 
   // if stalled in request state machine -- abort
-  handle_lock_released();
+  handle_peer_notification();
 }
 
 template <typename I>
@@ -209,7 +209,7 @@ void ExclusiveLock<I>::handle_watch_registered() {
 }
 
 template <typename I>
-void ExclusiveLock<I>::handle_lock_released() {
+void ExclusiveLock<I>::handle_peer_notification() {
   Mutex::Locker locker(m_lock);
   if (m_state != STATE_WAITING_FOR_PEER) {
     return;

src/librbd/ExclusiveLock.h

@@ -42,7 +42,7 @@ class ExclusiveLock {
   void release_lock(Context *on_released);
 
   void handle_watch_registered();
-  void handle_lock_released();
+  void handle_peer_notification();
 
   void assert_header_locked(librados::ObjectWriteOperation *op);
 

src/librbd/ImageWatcher.cc

@@ -474,7 +474,7 @@ void ImageWatcher<I>::handle_request_lock(int r) {
                               << dendl;
 
     // treat this is a dead client -- so retest acquiring the lock
-    m_image_ctx.exclusive_lock->handle_lock_released();
+    m_image_ctx.exclusive_lock->handle_peer_notification();
   } else if (r < 0) {
     lderr(m_image_ctx.cct) << this << " error requesting lock: "
                            << cpp_strerror(r) << dendl;
@@ -617,6 +617,11 @@ bool ImageWatcher<I>::handle_payload(const AcquiredLockPayload &payload,
   }
 
   RWLock::RLocker owner_locker(m_image_ctx.owner_lock);
+  if (m_image_ctx.exclusive_lock != nullptr) {
+    // potentially wake up the exclusive lock state machine now that
+    // a lock owner has advertised itself
+    m_image_ctx.exclusive_lock->handle_peer_notification();
+  }
   if (cancel_async_requests &&
       (m_image_ctx.exclusive_lock == nullptr ||
        !m_image_ctx.exclusive_lock->is_lock_owner())) {
@@ -654,7 +659,7 @@ bool ImageWatcher<I>::handle_payload(const ReleasedLockPayload &payload,
   if (m_image_ctx.exclusive_lock != nullptr &&
       !m_image_ctx.exclusive_lock->is_lock_owner()) {
     m_task_finisher->cancel(TASK_CODE_REQUEST_LOCK);
-    m_image_ctx.exclusive_lock->handle_lock_released();
+    m_image_ctx.exclusive_lock->handle_peer_notification();
   }
   return true;
 }
@@ -668,24 +673,23 @@ bool ImageWatcher<I>::handle_payload(const RequestLockPayload &payload,
   }
 
   RWLock::RLocker l(m_image_ctx.owner_lock);
-  if (m_image_ctx.exclusive_lock != nullptr) {
-    int r;
-    if (m_image_ctx.exclusive_lock->accept_requests(&r)) {
-      // need to send something back so the client can detect a missing leader
-      ::encode(ResponseMessage(0), ack_ctx->out);
-
-      {
-        Mutex::Locker owner_client_id_locker(m_owner_client_id_lock);
-        if (!m_owner_client_id.is_valid()) {
-	  return true;
-        }
+  if (m_image_ctx.exclusive_lock != nullptr &&
+      m_image_ctx.exclusive_lock->is_lock_owner()) {
+    int r = 0;
+    bool accept_request = m_image_ctx.exclusive_lock->accept_requests(&r);
+
+    // need to send something back so the client can detect a missing leader
+    ::encode(ResponseMessage(r), ack_ctx->out);
+
+    if (accept_request) {
+      Mutex::Locker owner_client_id_locker(m_owner_client_id_lock);
+      if (!m_owner_client_id.is_valid()) {
+        return true;
       }
 
       ldout(m_image_ctx.cct, 10) << this << " queuing release of exclusive lock"
                                  << dendl;
       m_image_ctx.get_exclusive_lock_policy()->lock_requested(payload.force);
-    } else if (r < 0) {
-      ::encode(ResponseMessage(r), ack_ctx->out);
     }
   }
   return true;

src/test/librbd/test_mock_ExclusiveLock.cc

@@ -142,7 +142,7 @@ class TestMockExclusiveLock : public TestMockFixture {
                                   MockExclusiveLock &mock_exclusive_lock) {
     EXPECT_CALL(*mock_image_ctx.image_watcher, notify_request_lock())
                   .WillRepeatedly(Invoke(&mock_exclusive_lock,
-                                         &MockExclusiveLock::handle_lock_released));
+                                         &MockExclusiveLock::handle_peer_notification));
   }
 
   void expect_notify_acquired_lock(MockExclusiveLockImageCtx &mock_image_ctx) {