librbd: deep_copy: resize head object map if needed

Fixes: http://tracker.ceph.com/issues/24399
Signed-off-by: Mykola Golub <mgolub@suse.com>

src/librbd/deep_copy/SnapshotCopyRequest.cc

@@ -7,8 +7,10 @@
 #include "common/errno.h"
 #include "common/WorkQueue.h"
 #include "librbd/ExclusiveLock.h"
+#include "librbd/ObjectMap.h"
 #include "librbd/Operations.h"
 #include "librbd/Utils.h"
+#include "osdc/Striper.h"
 
 #define dout_subsys ceph_subsys_rbd
 #undef dout_prefix
@@ -546,6 +548,54 @@ void SnapshotCopyRequest<I>::handle_set_head(int r) {
     return;
   }
 
+  if (handle_cancellation()) {
+    return;
+  }
+
+  send_resize_object_map();
+}
+
+template <typename I>
+void SnapshotCopyRequest<I>::send_resize_object_map() {
+  int r = 0;
+
+  if (m_snap_id_end == CEPH_NOSNAP &&
+      m_dst_image_ctx->test_features(RBD_FEATURE_OBJECT_MAP)) {
+    RWLock::RLocker owner_locker(m_dst_image_ctx->owner_lock);
+    RWLock::RLocker snap_locker(m_dst_image_ctx->snap_lock);
+
+    if (m_dst_image_ctx->object_map != nullptr &&
+        Striper::get_num_objects(m_dst_image_ctx->layout,
+                                 m_dst_image_ctx->size) !=
+          m_dst_image_ctx->object_map->size()) {
+
+      ldout(m_cct, 20) << dendl;
+
+      auto finish_op_ctx = start_lock_op(m_dst_image_ctx->owner_lock);
+      if (finish_op_ctx != nullptr) {
+        auto ctx = new FunctionContext([this, finish_op_ctx](int r) {
+            handle_resize_object_map(r);
+            finish_op_ctx->complete(0);
+          });
+
+        m_dst_image_ctx->object_map->aio_resize(m_dst_image_ctx->size,
+                                                OBJECT_NONEXISTENT, ctx);
+        return;
+      }
+
+      lderr(m_cct) << "lost exclusive lock" << dendl;
+      r = -EROFS;
+    }
+  }
+
+  finish(r);
+}
+
+template <typename I>
+void SnapshotCopyRequest<I>::handle_resize_object_map(int r) {
+  ldout(m_cct, 20) << "r=" << r << dendl;
+
+  assert(r == 0);
   finish(0);
 }
 
@@ -596,6 +646,12 @@ int SnapshotCopyRequest<I>::validate_parent(I *image_ctx,
 template <typename I>
 Context *SnapshotCopyRequest<I>::start_lock_op() {
   RWLock::RLocker owner_locker(m_dst_image_ctx->owner_lock);
+  return start_lock_op(m_dst_image_ctx->owner_lock);
+}
+
+template <typename I>
+Context *SnapshotCopyRequest<I>::start_lock_op(RWLock &owner_lock) {
+  assert(m_dst_image_ctx->owner_lock.is_locked());
   if (m_dst_image_ctx->exclusive_lock == nullptr) {
     return new FunctionContext([](int r) {});
   }

src/librbd/deep_copy/SnapshotCopyRequest.h

@@ -70,6 +70,9 @@ private:
    * SET_HEAD (skip if not needed)
    *    |
    *    v
+   * RESIZE_OBJECT_MAP (skip if not needed)
+   *    |
+   *    v
    * <finish>
    *
    * @endverbatim
@@ -114,6 +117,9 @@ private:
   void send_set_head();
   void handle_set_head(int r);
 
+  void send_resize_object_map();
+  void handle_resize_object_map(int r);
+
   bool handle_cancellation();
 
   void error(int r);
@@ -121,6 +127,7 @@ private:
   int validate_parent(ImageCtxT *image_ctx, librbd::ParentSpec *spec);
 
   Context *start_lock_op();
+  Context *start_lock_op(RWLock &owner_lock);
 
   void finish(int r);
 };

src/test/librbd/mock/MockObjectMap.h

@@ -13,6 +13,8 @@ namespace librbd {
 struct MockObjectMap {
   MOCK_CONST_METHOD1(enabled, bool(const RWLock &object_map_lock));
 
+  MOCK_CONST_METHOD0(size, uint64_t());
+
   MOCK_METHOD1(open, void(Context *on_finish));
   MOCK_METHOD1(close, void(Context *on_finish));
 

src/test/librbd/test_DeepCopy.cc

@@ -36,6 +36,12 @@ struct TestDeepCopy : public TestFixture {
     if (m_src_ictx != nullptr) {
       deep_copy();
       if (m_dst_ictx != nullptr) {
+        if (m_dst_ictx->test_features(RBD_FEATURE_LAYERING)) {
+          bool flags_set;
+          EXPECT_EQ(0, m_dst_ictx->test_flags(RBD_FLAG_OBJECT_MAP_INVALID,
+                                              &flags_set));
+          EXPECT_FALSE(flags_set);
+        }
         compare();
         close_image(m_dst_ictx);
       }
@@ -228,6 +234,32 @@ struct TestDeepCopy : public TestFixture {
     ASSERT_EQ(0, m_src_ictx->operations->resize(new_size, true, no_op));
   }
 
+  void test_clone_expand() {
+    bufferlist bl;
+    bl.append(std::string(100, '1'));
+    ASSERT_EQ(static_cast<ssize_t>(bl.length()),
+              m_src_ictx->io_work_queue->write(0, bl.length(), bufferlist{bl},
+                                               0));
+    ASSERT_EQ(0, m_src_ictx->io_work_queue->flush());
+
+    ASSERT_EQ(0, snap_create(*m_src_ictx, "snap"));
+    ASSERT_EQ(0, snap_protect(*m_src_ictx, "snap"));
+
+    std::string clone_name = get_temp_image_name();
+    int order = m_src_ictx->order;
+    uint64_t features;
+    ASSERT_EQ(0, librbd::get_features(m_src_ictx, &features));
+    ASSERT_EQ(0, librbd::clone(m_ioctx, m_src_ictx->name.c_str(), "snap",
+                               m_ioctx, clone_name.c_str(), features, &order, 0,
+                               0));
+    close_image(m_src_ictx);
+    ASSERT_EQ(0, open_image(clone_name, &m_src_ictx));
+
+    librbd::NoOpProgressContext no_op;
+    auto new_size = m_src_ictx->size << 1;
+    ASSERT_EQ(0, m_src_ictx->operations->resize(new_size, true, no_op));
+  }
+
   void test_clone() {
     bufferlist bl;
     bl.append(std::string(((1 << m_src_ictx->order) * 2) + 1, '1'));
@@ -401,6 +433,13 @@ TEST_F(TestDeepCopy, CloneShrink)
   test_clone_shrink();
 }
 
+TEST_F(TestDeepCopy, CloneExpand)
+{
+  REQUIRE_FEATURE(RBD_FEATURE_LAYERING);
+
+  test_clone_expand();
+}
+
 TEST_F(TestDeepCopy, Clone)
 {
   REQUIRE_FEATURE(RBD_FEATURE_LAYERING);