librbd: make ImageCtx->object_map always present

This simplifies locking by obviating the NULL checks.  We no longer
need md_lock to protect these acceses. We can use object_map_lock
instead, to make sure no one reads an object map while its being
updated.

Keep track of whether the object map is enabled for a given snapshot
internally. In each public method, check this state, and automatically
set it correctly when refreshing the object map. During snapshot
removal, unconditionally try to remove the object map object, to
protect against bugs leaking objects, and to be consistent with image
removal.

Signed-off-by: Josh Durgin <jdurgin@redhat.com>

diff --git a/src/librbd/AioRequest.cc b/src/librbd/AioRequest.cc
index ceacc6f0e8a6eb739769a844ce7ba13cf9abd136..dff7709984bd0774bf33fa4dc7ae3f809f1bb86d 100644 (file)
--- a/src/librbd/AioRequest.cc
+++ b/src/librbd/AioRequest.cc
@@ -201,13 +201,9 @@ namespace librbd {
     ldout(m_ictx->cct, 20) << "send " << this << " " << m_oid << " " << m_object_off << "~" << m_object_len << dendl;
 
     // send read request to parent if the object doesn't exist locally
-    {
-      RWLock::RLocker l(m_ictx->md_lock);
-      if (m_ictx->object_map != NULL &&
-         !m_ictx->object_map->object_may_exist(m_object_no)) {
-       complete(-ENOENT);
-       return 0;
-      }
+    if (!m_ictx->object_map.object_may_exist(m_object_no)) {
+      complete(-ENOENT);
+      return 0;
     }
 
     librados::AioCompletion *rados_completion =
@@ -416,8 +412,7 @@ namespace librbd {
     bool lost_exclusive_lock = false;
     {
       RWLock::RLocker l(m_ictx->owner_lock);
-      RWLock::RLocker l2(m_ictx->md_lock);
-      if (m_ictx->object_map == NULL) {
+      if (!m_ictx->object_map.enabled()) {
        return false;
       }
 
@@ -435,7 +430,7 @@ namespace librbd {
         m_state = LIBRBD_AIO_WRITE_PRE; 
         FunctionContext *ctx = new FunctionContext(
           boost::bind(&AioRequest::complete, this, _1));
-        if (!m_ictx->object_map->aio_update(m_object_no, new_state,
+        if (!m_ictx->object_map.aio_update(m_object_no, new_state,
                                            current_state, ctx)) {
          // no object map update required
          return false;
@@ -454,8 +449,7 @@ namespace librbd {
                           << m_object_off << "~" << m_object_len << dendl;
 
     RWLock::RLocker l(m_ictx->owner_lock);
-    RWLock::RLocker l2(m_ictx->md_lock);
-    if (m_ictx->object_map == NULL || !post_object_map_update()) {
+    if (!m_ictx->object_map.enabled() || !post_object_map_update()) {
       return true;
     }
 
@@ -469,7 +463,7 @@ namespace librbd {
     m_state = LIBRBD_AIO_WRITE_POST;
     FunctionContext *ctx = new FunctionContext(
       boost::bind(&AioRequest::complete, this, _1));
-    if (!m_ictx->object_map->aio_update(m_object_no, OBJECT_NONEXISTENT,
+    if (!m_ictx->object_map.aio_update(m_object_no, OBJECT_NONEXISTENT,
                                        OBJECT_PENDING, ctx)) {
       // no object map update required
       return true;

diff --git a/src/librbd/AsyncResizeRequest.cc b/src/librbd/AsyncResizeRequest.cc
index 8b161d50daa8ffc2c59461649bdc55bb107949f9..1b09aae1d5aa504e6d0fdcccf1b0f1fa4958fb15 100644 (file)
--- a/src/librbd/AsyncResizeRequest.cc
+++ b/src/librbd/AsyncResizeRequest.cc
@@ -119,8 +119,7 @@ void AsyncResizeRequest::send_grow_object_map() {
   bool object_map_enabled = true;
   {
     RWLock::RLocker l(m_image_ctx.owner_lock);
-    RWLock::RLocker l2(m_image_ctx.md_lock);
-    if (m_image_ctx.object_map == NULL) {
+    if (!m_image_ctx.object_map.enabled()) {
       object_map_enabled = false;
     } else { 
       ldout(m_image_ctx.cct, 5) << this << " send_grow_object_map: "
@@ -133,7 +132,7 @@ void AsyncResizeRequest::send_grow_object_map() {
        ldout(m_image_ctx.cct, 1) << "lost exclusive lock during grow object map" << dendl;
        lost_exclusive_lock = true;
       } else {
-       m_image_ctx.object_map->aio_resize(m_new_size, OBJECT_NONEXISTENT,
+       m_image_ctx.object_map.aio_resize(m_new_size, OBJECT_NONEXISTENT,
                                           create_callback_context());
        object_map_enabled = true;
       }
@@ -152,8 +151,7 @@ bool AsyncResizeRequest::send_shrink_object_map() {
   bool lost_exclusive_lock = false;
   {
     RWLock::RLocker l(m_image_ctx.owner_lock);
-    RWLock::RLocker l2(m_image_ctx.md_lock);
-    if (m_image_ctx.object_map == NULL ||
+    if (!m_image_ctx.object_map.enabled() ||
        m_new_size > m_original_size) {
       return true;
     }
@@ -168,7 +166,7 @@ bool AsyncResizeRequest::send_shrink_object_map() {
       ldout(m_image_ctx.cct, 1) << "lost exclusive lock during shrink object map" << dendl;
       lost_exclusive_lock = true;
     } else {
-      m_image_ctx.object_map->aio_resize(m_new_size, OBJECT_NONEXISTENT,
+      m_image_ctx.object_map.aio_resize(m_new_size, OBJECT_NONEXISTENT,
                                         create_callback_context());
     }
   }

diff --git a/src/librbd/AsyncTrimRequest.cc b/src/librbd/AsyncTrimRequest.cc
index 3bbab55252582bfdf9228a387ac5e28d57ffe392..b27e791a02ebcc67a633f2c5e3d82421e3357310 100644 (file)
--- a/src/librbd/AsyncTrimRequest.cc
+++ b/src/librbd/AsyncTrimRequest.cc
@@ -34,12 +34,8 @@ public:
   }
 
   virtual int send() {
-    {
-      RWLock::RLocker l(m_image_ctx.md_lock);
-      if (m_image_ctx.object_map != NULL &&
-          !m_image_ctx.object_map->object_may_exist(m_object_no)) {
-        return 1;
-      }
+    if (!m_image_ctx.object_map.object_may_exist(m_object_no)) {
+      return 1;
     }
 
     RWLock::RLocker l(m_image_ctx.owner_lock);
@@ -163,8 +159,7 @@ void AsyncTrimRequest::send_pre_remove() {
   bool lost_exclusive_lock = false;
   {
     RWLock::RLocker l(m_image_ctx.owner_lock);
-    RWLock::RLocker l2(m_image_ctx.md_lock);
-    if (m_image_ctx.object_map == NULL) {
+    if (!m_image_ctx.object_map.enabled()) {
       remove_objects = true;
     } else {
       ldout(m_image_ctx.cct, 5) << this << " send_pre_remove: "
@@ -177,7 +172,7 @@ void AsyncTrimRequest::send_pre_remove() {
         lost_exclusive_lock = true;
       } else {
         // flag the objects as pending deletion
-        if (!m_image_ctx.object_map->aio_update(m_delete_start, m_num_objects,
+        if (!m_image_ctx.object_map.aio_update(m_delete_start, m_num_objects,
                                                OBJECT_PENDING, OBJECT_EXISTS,
                                                create_callback_context())) {
           remove_objects = true;
@@ -200,8 +195,7 @@ bool AsyncTrimRequest::send_post_remove() {
   bool lost_exclusive_lock = false;
   {
     RWLock::RLocker l(m_image_ctx.owner_lock);
-    RWLock::RLocker l2(m_image_ctx.md_lock);
-    if (m_image_ctx.object_map == NULL) {
+    if (!m_image_ctx.object_map.enabled()) {
       clean_boundary = true;
     } else {
       ldout(m_image_ctx.cct, 5) << this << " send_post_remove: "
@@ -213,7 +207,7 @@ bool AsyncTrimRequest::send_post_remove() {
         ldout(m_image_ctx.cct, 1) << "lost exclusive lock during trim" << dendl;
       } else {
         // flag the pending objects as removed
-        if (!m_image_ctx.object_map->aio_update(m_delete_start, m_num_objects,
+        if (!m_image_ctx.object_map.aio_update(m_delete_start, m_num_objects,
                                                OBJECT_NONEXISTENT,
                                                OBJECT_PENDING,
                                                create_callback_context())) {

diff --git a/src/librbd/CopyupRequest.cc b/src/librbd/CopyupRequest.cc
index 49fd599cdb1a29edf8fc922ad105a30e118c1ead..68708e8928f43b9d6680054d68c87f6d0ac0eb9f 100644 (file)
--- a/src/librbd/CopyupRequest.cc
+++ b/src/librbd/CopyupRequest.cc
@@ -172,8 +172,7 @@ namespace librbd {
     bool copyup = false;
     {
       RWLock::RLocker l(m_ictx->owner_lock);
-      RWLock::RLocker l2(m_ictx->md_lock);
-      if (m_ictx->object_map == NULL) {
+      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 copy-on-read"
@@ -181,7 +180,7 @@ namespace librbd {
        return true; 
       } else {
        m_state = STATE_OBJECT_MAP;
-        if (!m_ictx->object_map->aio_update(m_object_no, OBJECT_EXISTS,
+        if (!m_ictx->object_map.aio_update(m_object_no, OBJECT_EXISTS,
                                            boost::optional<uint8_t>(),
                                            create_callback_context())) {
          copyup = true;

diff --git a/src/librbd/ImageCtx.cc b/src/librbd/ImageCtx.cc
index 01978dcfc5fd23d709d8086047acffdc3a2503f7..f58f7ffe2fcca7097b93fc1f6912e96ba28c77fe 100644 (file)
--- a/src/librbd/ImageCtx.cc
+++ b/src/librbd/ImageCtx.cc
@@ -59,7 +59,7 @@ namespace librbd {
       object_cacher(NULL), writeback_handler(NULL), object_set(NULL),
       readahead(),
       total_bytes_read(0), copyup_finisher(NULL),
-      object_map(NULL)
+      object_map(*this)
   {
     md_ctx.dup(p);
     data_ctx.dup(p);
@@ -129,7 +129,6 @@ namespace librbd {
       delete copyup_finisher;
       copyup_finisher = NULL;
     }
-    delete object_map;
     delete[] format_string;
   }
 
@@ -299,8 +298,9 @@ namespace librbd {
       snap_exists = true;
       data_ctx.snap_set_read(snap_id);
 
-      if (object_map != NULL) {
-        object_map->refresh(in_snap_id);
+      if (object_map.enabled()) {
+       RWLock::WLocker l(object_map_lock);
+        object_map.refresh(in_snap_id);
       }
       return 0;
     }
@@ -315,8 +315,9 @@ namespace librbd {
     snap_exists = true;
     data_ctx.snap_set_read(snap_id);
 
-    if (object_map != NULL) {
-      object_map->refresh(CEPH_NOSNAP);
+    if (object_map.enabled()) {
+      RWLock::WLocker l(object_map_lock);
+      object_map.refresh(CEPH_NOSNAP);
     }
   }
 

diff --git a/src/librbd/ImageCtx.h b/src/librbd/ImageCtx.h
index 3d006982abcb3e0423f4a370e0ae00bd64b6cc44..996309fe1434c73dfc98fa704996e1e5408cf976 100644 (file)
--- a/src/librbd/ImageCtx.h
+++ b/src/librbd/ImageCtx.h
@@ -26,6 +26,7 @@
 
 #include "cls/rbd/cls_rbd_client.h"
 #include "librbd/LibrbdWriteback.h"
+#include "librbd/ObjectMap.h"
 #include "librbd/SnapInfo.h"
 #include "librbd/parent_types.h"
 
@@ -38,7 +39,6 @@ namespace librbd {
   class AsyncOperation;
   class CopyupRequest;
   class ImageWatcher;
-  class ObjectMap;
 
   struct ImageCtx {
     CephContext *cct;
@@ -81,7 +81,7 @@ namespace librbd {
     RWLock snap_lock; // protects snapshot-related member variables, features, and flags
     RWLock parent_lock; // protects parent_md and parent
     Mutex refresh_lock; // protects refresh_seq and last_refresh
-    RWLock object_map_lock; // protects object map updates
+    RWLock object_map_lock; // protects object map updates and object_map itself
     Mutex async_ops_lock; // protects async_ops
     Mutex copyup_list_lock; // protects copyup_waiting_list
 
@@ -114,7 +114,7 @@ namespace librbd {
 
     xlist<AsyncOperation*> async_ops;
 
-    ObjectMap *object_map;
+    ObjectMap object_map;
 
     atomic_t async_request_seq;
 
@@ -162,6 +162,7 @@ namespace librbd {
     bool test_features(uint64_t test_features) const;
     int get_flags(librados::snap_t in_snap_id, uint64_t *flags) const;
     bool test_flags(uint64_t test_flags) const;
+
     const parent_info* get_parent_info(librados::snap_t in_snap_id) const;
     int64_t get_parent_pool_id(librados::snap_t in_snap_id) const;
     std::string get_parent_image_id(librados::snap_t in_snap_id) const;

diff --git a/src/librbd/ImageWatcher.cc b/src/librbd/ImageWatcher.cc
index 25fbeab9e7c2ebe04dfa6e30035f43bcbe6f55fd..2fd81bd145d3ce834752a6f9afbbb88339df7627 100644 (file)
--- a/src/librbd/ImageWatcher.cc
+++ b/src/librbd/ImageWatcher.cc
@@ -311,13 +311,15 @@ int ImageWatcher::lock() {
     m_owner_client_id = get_client_id();
   }
 
-  if (m_image_ctx.object_map != NULL) {
-    r = m_image_ctx.object_map->lock();
+  if (m_image_ctx.object_map.enabled()) {
+    r = m_image_ctx.object_map.lock();
     if (r < 0 && r != -ENOENT) {
       unlock();
       return r;
     }
-    m_image_ctx.object_map->refresh(CEPH_NOSNAP);
+    RWLock::RLocker l2(m_image_ctx.snap_lock);
+    RWLock::WLocker l3(m_image_ctx.object_map_lock);
+    m_image_ctx.object_map.refresh(CEPH_NOSNAP);
   }
 
   bufferlist bl;
@@ -348,8 +350,8 @@ int ImageWatcher::unlock()
     return r;
   }
 
-  if (m_image_ctx.object_map != NULL) {
-    m_image_ctx.object_map->unlock();
+  if (m_image_ctx.object_map.enabled()) {
+    m_image_ctx.object_map.unlock();
   }
 
   FunctionContext *ctx = new FunctionContext(

diff --git a/src/librbd/LibrbdWriteback.cc b/src/librbd/LibrbdWriteback.cc
index 4c917d4adb7a376e421ba4643ff31b5224d3b984..694f2c739da67f9bdf025925579a3542c4c7df43 100644 (file)
--- a/src/librbd/LibrbdWriteback.cc
+++ b/src/librbd/LibrbdWriteback.cc
@@ -108,9 +108,7 @@ namespace librbd {
     Context *req = new C_Request(m_ictx->cct, onfinish, &m_lock);
 
     {
-      RWLock::RLocker l(m_ictx->md_lock);
-      if (m_ictx->object_map != NULL &&
-         !m_ictx->object_map->object_may_exist(object_no)) {
+      if (!m_ictx->object_map.object_may_exist(object_no)) {
        m_finisher->queue(req, -ENOENT);
        return;
       }

diff --git a/src/librbd/ObjectMap.cc b/src/librbd/ObjectMap.cc
index edd33b7e76e5064e9a96de799eb02bb24d258b31..9a277cad71a55275fc3a2a20fd7fddff5ddf8a50 100644 (file)
--- a/src/librbd/ObjectMap.cc
+++ b/src/librbd/ObjectMap.cc
@@ -17,7 +17,7 @@
 namespace librbd {
 
 ObjectMap::ObjectMap(ImageCtx &image_ctx)
-  : m_image_ctx(image_ctx)
+  : m_image_ctx(image_ctx), m_enabled(false)
 {
 }
 
@@ -33,12 +33,25 @@ std::string ObjectMap::object_map_name(const std::string &image_id,
   return oid;
 }
 
+bool ObjectMap::enabled() const
+{
+  RWLock::RLocker l(m_image_ctx.object_map_lock);
+  return m_enabled;
+}
+
 int ObjectMap::lock()
 {
   if (!m_image_ctx.test_features(RBD_FEATURE_OBJECT_MAP)) {
     return 0;
   }
 
+  {
+    RWLock::RLocker l(m_image_ctx.object_map_lock);
+    if (!m_enabled) {
+      return 0;
+    }
+  }
+
   int r;
   bool broke_lock = false;
   CephContext *cct = m_image_ctx.cct;
@@ -119,6 +132,9 @@ bool ObjectMap::object_may_exist(uint64_t object_no) const
   }
 
   RWLock::RLocker l(m_image_ctx.object_map_lock);
+  if (!m_enabled) {
+    return true;
+  }
   assert(object_no < m_object_map.size());
 
   bool exists = (m_object_map[object_no] == OBJECT_EXISTS ||
@@ -132,16 +148,20 @@ bool ObjectMap::object_may_exist(uint64_t object_no) const
 void ObjectMap::refresh(uint64_t snap_id)
 {
   assert(m_image_ctx.snap_lock.is_locked());
+  assert(m_image_ctx.object_map_lock.is_wlocked());
   uint64_t features;
-  m_image_ctx.get_features(m_image_ctx.snap_id, &features);
-  if ((features & RBD_FEATURE_OBJECT_MAP) == 0) {
+  m_image_ctx.get_features(snap_id, &features);
+  if ((features & RBD_FEATURE_OBJECT_MAP) == 0 ||
+      (m_image_ctx.snap_id == snap_id && !m_image_ctx.snap_exists)) {
+    m_object_map.clear();
+    m_enabled = false;
     return;
   }
+  m_enabled = true;
 
   CephContext *cct = m_image_ctx.cct;
   ldout(cct, 10) << &m_image_ctx << " refreshing object map" << dendl;
 
-  RWLock::WLocker l(m_image_ctx.object_map_lock);
   std::string oid(object_map_name(m_image_ctx.id, snap_id));
   int r = cls_client::object_map_load(&m_image_ctx.md_ctx, oid,
                                       &m_object_map);
@@ -178,6 +198,9 @@ void ObjectMap::rollback(uint64_t snap_id) {
   ldout(cct, 10) << &m_image_ctx << " rollback object map" << dendl;
 
   RWLock::WLocker l(m_image_ctx.object_map_lock);
+  if (!m_enabled) {
+    return;
+  }
 
   std::string snap_oid(object_map_name(m_image_ctx.id, snap_id));
   bufferlist bl;
@@ -215,15 +238,16 @@ void ObjectMap::snapshot(uint64_t snap_id) {
 
   int r;
   bufferlist bl;
-  {
-    RWLock::RLocker l(m_image_ctx.object_map_lock);
-    std::string oid(object_map_name(m_image_ctx.id, CEPH_NOSNAP));
-    r = m_image_ctx.md_ctx.read(oid, bl, 0, 0);
-    if (r < 0) {
-      lderr(cct) << "unable to load object map: " << cpp_strerror(r)
-                << dendl;
-      invalidate();
-    }
+  RWLock::WLocker l(m_image_ctx.object_map_lock);
+  if (!m_enabled) {
+    return;
+  }
+  std::string oid(object_map_name(m_image_ctx.id, CEPH_NOSNAP));
+  r = m_image_ctx.md_ctx.read(oid, bl, 0, 0);
+  if (r < 0) {
+    lderr(cct) << "unable to load object map: " << cpp_strerror(r)
+              << dendl;
+    invalidate();
   }
 
   std::string snap_oid(object_map_name(m_image_ctx.id, snap_id));
@@ -291,6 +315,7 @@ bool ObjectMap::aio_update(uint64_t start_object_no, uint64_t end_object_no,
 
 void ObjectMap::invalidate() {
   assert(m_image_ctx.snap_lock.is_wlocked());
+  assert(m_image_ctx.object_map_lock.is_wlocked());
   uint64_t flags;
   m_image_ctx.get_flags(m_image_ctx.snap_id, &flags);
   if ((flags & RBD_FLAG_OBJECT_MAP_INVALID) != 0) {
@@ -328,12 +353,8 @@ bool ObjectMap::Request::should_complete(int r) {
     }
 
     {
-      RWLock::RLocker l(m_image_ctx.md_lock);
       RWLock::WLocker l2(m_image_ctx.object_map_lock);
-      ObjectMap *object_map = m_image_ctx.object_map;
-      if (object_map != NULL) {
-       finish(object_map);
-      }
+      finish(&m_image_ctx.object_map);
     }
     return true;
 
@@ -341,7 +362,7 @@ bool ObjectMap::Request::should_complete(int r) {
     ldout(cct, 20) << "INVALIDATE" << dendl;
     if (r < 0) {
       lderr(cct) << "failed to invalidate object map: " << cpp_strerror(r)
-                << dendl; 
+                << dendl;
     }
     return true;
 

diff --git a/src/librbd/ObjectMap.h b/src/librbd/ObjectMap.h
index 2af1db2f057fc85a916cc4244b83385b9abd219e..564816f1193d6e0670091906e6be40eed3e56eb3 100644 (file)
--- a/src/librbd/ObjectMap.h
+++ b/src/librbd/ObjectMap.h
@@ -46,6 +46,8 @@ public:
   void rollback(uint64_t snap_id);
   void snapshot(uint64_t snap_id);
 
+  bool enabled() const;
+
 private:
 
   class Request : public AsyncRequest {
@@ -120,6 +122,8 @@ private:
 
   ceph::BitVector<2> m_object_map;
 
+  bool m_enabled;
+
   void invalidate();
 
 };

diff --git a/src/librbd/internal.cc b/src/librbd/internal.cc
index d3ee11696a51df2606f8ebb7a60b9c933e460095..f00cb40a142b16ce01a66fa36f518b4f29b485c8 100644 (file)
--- a/src/librbd/internal.cc
+++ b/src/librbd/internal.cc
@@ -320,9 +320,7 @@ namespace librbd {
 
     {
       RWLock::WLocker l(ictx->snap_lock);
-      if (ictx->object_map != NULL) {
-       ictx->object_map->rollback(snap_id);
-      }
+      ictx->object_map.rollback(snap_id);
     }
     return 0;
   }
@@ -584,13 +582,11 @@ namespace librbd {
       }
     }
 
-    if (ictx->object_map != NULL) {
-      r = ictx->md_ctx.remove(ObjectMap::object_map_name(ictx->id, snap_id));
-      if (r < 0 && r != -ENOENT) {
-       lderr(ictx->cct) << "snap_remove: failed to remove snapshot object map"
-                        << dendl;
-       return 0;
-      }
+    r = ictx->md_ctx.remove(ObjectMap::object_map_name(ictx->id, snap_id));
+    if (r < 0 && r != -ENOENT) {
+      lderr(ictx->cct) << "snap_remove: failed to remove snapshot object map"
+                      << dendl;
+      return 0;
     }
 
     r = rm_snap(ictx, snap_name);
@@ -1785,9 +1781,7 @@ reprotect_and_return_err:
 
     RWLock::WLocker l(ictx->snap_lock);
     if (!ictx->old_format) {
-      if (ictx->object_map != NULL) {
-       ictx->object_map->snapshot(snap_id);
-      }
+      ictx->object_map.snapshot(snap_id);
       if (lock_owner) {
        // immediately start using the new snap context if we
        // own the exclusive lock
@@ -2055,15 +2049,8 @@ reprotect_and_return_err:
        ictx->snap_exists = false;
       }
 
-      if ((ictx->features & RBD_FEATURE_OBJECT_MAP) == 0) {
-       delete ictx->object_map;
-       ictx->object_map = NULL;
-      } else {
-       ictx->object_map = new ObjectMap(*ictx);
-       if (ictx->snap_exists) {
-         ictx->object_map->refresh(ictx->snap_id);
-       }
-      }
+      RWLock::WLocker object_map_locker(ictx->object_map_lock);
+      ictx->object_map.refresh(ictx->snap_id);
 
       ictx->data_ctx.selfmanaged_snap_set_write_ctx(ictx->snapc.seq, ictx->snaps);
     } // release snap_lock