Merge pull request #6348 from dillaman/wip-13567-hammer

librbd: potential assertion failure during cache read

Reviewed-by: Sage Weil <sage@redhat.com>

src/librbd/LibrbdWriteback.cc

@@ -45,23 +45,27 @@ namespace librbd {
    * @param c context to finish
    * @param l mutex to lock
    */
-  class C_Request : public Context {
+  class C_ReadRequest : public Context {
   public:
-    C_Request(CephContext *cct, Context *c, Mutex *l)
-      : m_cct(cct), m_ctx(c), m_lock(l) {}
-    virtual ~C_Request() {}
+    C_ReadRequest(CephContext *cct, Context *c, RWLock *owner_lock,
+                  Mutex *cache_lock)
+      : m_cct(cct), m_ctx(c), m_owner_lock(owner_lock),
+        m_cache_lock(cache_lock) {
+    }
     virtual void finish(int r) {
       ldout(m_cct, 20) << "aio_cb completing " << dendl;
       {
-	Mutex::Locker l(*m_lock);
+        RWLock::RLocker owner_locker(*m_owner_lock);
+        Mutex::Locker cache_locker(*m_cache_lock);
 	m_ctx->complete(r);
       }
       ldout(m_cct, 20) << "aio_cb finished" << dendl;
     }
   private:
     CephContext *m_cct;
     Context *m_ctx;
-    Mutex *m_lock;
+    RWLock *m_owner_lock;
+    Mutex *m_cache_lock;
   };
 
   class C_OrderedWrite : public Context {
@@ -105,7 +109,8 @@ namespace librbd {
 			     __u32 trunc_seq, int op_flags, Context *onfinish)
   {
     // on completion, take the mutex and then call onfinish.
-    Context *req = new C_Request(m_ictx->cct, onfinish, &m_lock);
+    Context *req = new C_ReadRequest(m_ictx->cct, onfinish, &m_ictx->owner_lock,
+                                     &m_lock);
 
     {
       if (!m_ictx->object_map.object_may_exist(object_no)) {

src/librbd/internal.cc

@@ -3563,6 +3563,8 @@ int invoke_async_request(ImageCtx *ictx, const std::string& request_type,
       return;
     }
 
+    RWLock::RLocker owner_locker(ictx->owner_lock);
+
     // readahead
     const md_config_t *conf = ictx->cct->_conf;
     if (ictx->object_cacher && conf->rbd_readahead_max_bytes > 0 &&

src/test/librbd/test_librbd.cc

@@ -2734,3 +2734,55 @@ TEST_F(TestLibRBD, ExclusiveLockTransition)
   ASSERT_PASSED(validate_object_map, image2);
   ASSERT_PASSED(validate_object_map, image3);
 }
+
+TEST_F(TestLibRBD, CacheMayCopyOnWrite) {
+  REQUIRE_FEATURE(RBD_FEATURE_LAYERING);
+
+  librados::IoCtx ioctx;
+  ASSERT_EQ(0, _rados.ioctx_create(m_pool_name.c_str(), ioctx));
+
+  librbd::RBD rbd;
+  std::string name = get_temp_image_name();
+
+  uint64_t size = 1 << 18;
+  int order = 12;
+  ASSERT_EQ(0, create_image_pp(rbd, ioctx, name.c_str(), size, &order));
+
+  librbd::Image image;
+  ASSERT_EQ(0, rbd.open(ioctx, image, name.c_str(), NULL));
+  ASSERT_EQ(0, image.snap_create("one"));
+  ASSERT_EQ(0, image.snap_protect("one"));
+
+  std::string clone_name = this->get_temp_image_name();
+  ASSERT_EQ(0, rbd.clone(ioctx, name.c_str(), "one", ioctx, clone_name.c_str(),
+                         RBD_FEATURE_LAYERING, &order));
+
+  librbd::Image clone;
+  ASSERT_EQ(0, rbd.open(ioctx, clone, clone_name.c_str(), NULL));
+  ASSERT_EQ(0, clone.flush());
+
+  bufferlist expect_bl;
+  expect_bl.append(std::string(1024, '\0'));
+
+  // test double read path
+  bufferlist read_bl;
+  uint64_t offset = 0;
+  ASSERT_EQ(1024, clone.read(offset + 2048, 1024, read_bl));
+  ASSERT_TRUE(expect_bl.contents_equal(read_bl));
+
+  bufferlist write_bl;
+  write_bl.append(std::string(1024, '1'));
+  ASSERT_EQ(1024, clone.write(offset, write_bl.length(), write_bl));
+
+  read_bl.clear();
+  ASSERT_EQ(1024, clone.read(offset + 2048, 1024, read_bl));
+  ASSERT_TRUE(expect_bl.contents_equal(read_bl));
+
+  // test read retry path
+  offset = 1 << order;
+  ASSERT_EQ(1024, clone.write(offset, write_bl.length(), write_bl));
+
+  read_bl.clear();
+  ASSERT_EQ(1024, clone.read(offset + 2048, 1024, read_bl));
+  ASSERT_TRUE(expect_bl.contents_equal(read_bl));
+}