librbd: fix recursive locking issues

Signed-off-by: Jason Dillaman <dillaman@redhat.com>
(cherry picked from commit 1b57cc1da7a51e6f8ffea689b94ef843732c20a4)

diff --git a/src/librbd/AioRequest.cc b/src/librbd/AioRequest.cc
index 7f7641352bf455aea99dc19508e2e3234d46a17a..b6c4a7f265f2d894b564fb19f6c075a7cf83ceba 100644 (file)
--- a/src/librbd/AioRequest.cc
+++ b/src/librbd/AioRequest.cc
@@ -237,20 +237,23 @@ namespace librbd {
 
   void AioRead::send_copyup()
   {
-    RWLock::RLocker snap_locker(m_ictx->snap_lock);
-    RWLock::RLocker parent_locker(m_ictx->parent_lock);
+    {
+      RWLock::RLocker snap_locker(m_ictx->snap_lock);
+      RWLock::RLocker parent_locker(m_ictx->parent_lock);
+      if (!compute_parent_extents()) {
+        return;
+      }
+    }
+
     Mutex::Locker copyup_locker(m_ictx->copyup_list_lock);
     map<uint64_t, CopyupRequest*>::iterator it =
       m_ictx->copyup_list.find(m_object_no);
     if (it == m_ictx->copyup_list.end()) {
-      if (compute_parent_extents()) {
-        // create and kick off a CopyupRequest
-        CopyupRequest *new_req = new CopyupRequest(m_ictx, m_oid,
-                                                   m_object_no,
-                                                  m_parent_extents);
-        m_ictx->copyup_list[m_object_no] = new_req;
-        new_req->queue_send();
-      }
+      // create and kick off a CopyupRequest
+      CopyupRequest *new_req = new CopyupRequest(m_ictx, m_oid, m_object_no,
+                                                m_parent_extents);
+      m_ictx->copyup_list[m_object_no] = new_req;
+      new_req->queue_send();
     }
   }
 
@@ -321,11 +324,15 @@ namespace librbd {
       ldout(m_ictx->cct, 20) << "WRITE_CHECK_GUARD" << dendl;
 
       if (r == -ENOENT) {
-       RWLock::RLocker l(m_ictx->snap_lock);
-       RWLock::RLocker l2(m_ictx->parent_lock);
+        bool has_parent;
+        {
+         RWLock::RLocker snap_locker(m_ictx->snap_lock);
+         RWLock::RLocker parent_locker(m_ictx->parent_lock);
+          has_parent = compute_parent_extents();
+        }
 
        // If parent still exists, overlap might also have changed.
-       if (compute_parent_extents()) {
+       if (has_parent) {
           send_copyup();
        } else {
           // parent may have disappeared -- send original write again

diff --git a/src/librbd/CopyupRequest.cc b/src/librbd/CopyupRequest.cc
index 2ed3c4e843134c5bfdd260603d5cf35259ec13bb..1535cde8516f8e4a4f3bef6671823c91aa64c72e 100644 (file)
--- a/src/librbd/CopyupRequest.cc
+++ b/src/librbd/CopyupRequest.cc
@@ -177,32 +177,32 @@ namespace librbd {
   }
 
   bool CopyupRequest::send_object_map() {
-    bool copyup = false;
+    bool copyup = true;
     {
-      RWLock::RLocker l(m_ictx->owner_lock);
-      if (!m_ictx->object_map.enabled()) {
-       copyup = true;
-      } else if (!m_ictx->image_watcher->is_lock_owner()) {
-       ldout(m_ictx->cct, 20) << "exclusive lock not held for copyup request"
-                              << dendl;
-        assert(m_pending_requests.empty());
-        return true;
-      } else {
+      RWLock::RLocker owner_locker(m_ictx->owner_lock);
+      RWLock::RLocker snap_locker(m_ictx->snap_lock);
+      if (m_ictx->object_map.enabled()) {
+        if (!m_ictx->image_watcher->is_lock_owner()) {
+         ldout(m_ictx->cct, 20) << "exclusive lock not held for copyup request"
+                                << dendl;
+          assert(m_pending_requests.empty());
+          return true;
+        }
+
         RWLock::WLocker object_map_locker(m_ictx->object_map_lock);
         if (m_ictx->object_map[m_object_no] != OBJECT_EXISTS) {
           ldout(m_ictx->cct, 20) << __func__ << " " << this
                                 << ": oid " << m_oid
                                  << ", extents " << m_image_extents
                                  << dendl;
-         m_state = STATE_OBJECT_MAP;
+          m_state = STATE_OBJECT_MAP;
 
           Context *ctx = create_callback_context();
           bool sent = m_ictx->object_map.aio_update(m_object_no, OBJECT_EXISTS,
                                                     boost::optional<uint8_t>(),
                                                     ctx);
           assert(sent);
-        } else {
-          copyup = true;
+          copyup = false;
         }
       }
     }

diff --git a/src/librbd/ImageCtx.cc b/src/librbd/ImageCtx.cc
index 47133e91705223d39bdb1425a7e3cb87a7a1cb92..ec2ce18cb170469cdaf3b4fab8d94dffae33117c 100644 (file)
--- a/src/librbd/ImageCtx.cc
+++ b/src/librbd/ImageCtx.cc
@@ -695,6 +695,8 @@ public:
 
   void ImageCtx::shutdown_cache() {
     flush_async_operations();
+
+    RWLock::RLocker owner_locker(owner_lock);
     invalidate_cache();
     object_cacher->stop();
   }
@@ -805,21 +807,15 @@ public:
   }
 
   void ImageCtx::flush_async_operations(Context *on_finish) {
-    bool complete = false;
-    {
-      Mutex::Locker l(async_ops_lock);
-      if (async_ops.empty()) {
-        complete = true;
-      } else {
-        ldout(cct, 20) << "flush async operations: " << on_finish << " "
-                       << "count=" << async_ops.size() << dendl;
-        async_ops.front()->add_flush_context(on_finish);
-      }
+    Mutex::Locker l(async_ops_lock);
+    if (async_ops.empty()) {
+      op_work_queue->queue(on_finish, 0);
+      return;
     }
 
-    if (complete) {
-      on_finish->complete(0);
-    }
+    ldout(cct, 20) << "flush async operations: " << on_finish << " "
+                   << "count=" << async_ops.size() << dendl;
+    async_ops.front()->add_flush_context(on_finish);
   }
 
   void ImageCtx::cancel_async_requests() {

diff --git a/src/librbd/ImageWatcher.cc b/src/librbd/ImageWatcher.cc
index 53bbd3034c98d89b1610bcba15aff49f21835669..600c5da80b33a69ef9d28cdaddc609d2267e3c10 100644 (file)
--- a/src/librbd/ImageWatcher.cc
+++ b/src/librbd/ImageWatcher.cc
@@ -3,6 +3,7 @@
 #include "librbd/ImageWatcher.h"
 #include "librbd/AioCompletion.h"
 #include "librbd/ImageCtx.h"
+#include "librbd/internal.h"
 #include "librbd/ObjectMap.h"
 #include "librbd/TaskFinisher.h"
 #include "cls/lock/cls_lock_client.h"
@@ -300,9 +301,10 @@ int ImageWatcher::lock() {
   ldout(m_image_ctx.cct, 10) << "acquired exclusive lock" << dendl;
   m_lock_owner_state = LOCK_OWNER_STATE_LOCKED;
 
+  ClientId owner_client_id = get_client_id();
   {
     Mutex::Locker l(m_owner_client_id_lock);
-    m_owner_client_id = get_client_id();
+    m_owner_client_id = owner_client_id;
   }
 
   if (m_image_ctx.object_map.enabled()) {

diff --git a/src/librbd/internal.cc b/src/librbd/internal.cc
index 749ebd38ca30657ec12dd9287f50681040cf234f..f7821f32f761f7afb36862fa12595866cc5e73e1 100644 (file)
--- a/src/librbd/internal.cc
+++ b/src/librbd/internal.cc
@@ -1626,9 +1626,7 @@ reprotect_and_return_err:
       }
       assert(watchers.size() == 1);
 
-      ictx->md_lock.get_read();
       trim_image(ictx, 0, prog_ctx);
-      ictx->md_lock.put_read();
 
       ictx->parent_lock.get_read();
       // struct assignment
@@ -2134,14 +2132,13 @@ reprotect_and_return_err:
     if (r < 0)
       return r;
 
-    RWLock::RLocker l(ictx->owner_lock);
+    RWLock::RLocker owner_locker(ictx->owner_lock);
     snap_t snap_id;
     uint64_t new_size;
     {
-      RWLock::WLocker l2(ictx->md_lock);
       {
        // need to drop snap_lock before invalidating cache
-       RWLock::RLocker l3(ictx->snap_lock);
+       RWLock::RLocker snap_locker(ictx->snap_lock);
        if (!ictx->snap_exists) {
          return -ENOENT;
        }
@@ -2173,6 +2170,7 @@ reprotect_and_return_err:
       // need to flush any pending writes before resizing and rolling back -
       // writes might create new snapshots. Rolling back will replace
       // the current version, so we have to invalidate that too.
+      RWLock::WLocker md_locker(ictx->md_lock);
       ictx->flush_async_operations();
       r = ictx->invalidate_cache();
       if (r < 0) {