librbd: new async journal promote state machine

Signed-off-by: Jason Dillaman <dillaman@redhat.com>

src/librbd/CMakeLists.txt

@@ -46,6 +46,7 @@ set(librbd_internal_srcs
   journal/RemoveRequest.cc
   journal/CreateRequest.cc
   journal/OpenRequest.cc
+  journal/PromoteRequest.cc
   journal/Replay.cc
   journal/StandardPolicy.cc
   journal/Utils.cc

src/librbd/Journal.cc

@@ -7,6 +7,7 @@
 #include "librbd/ExclusiveLock.h"
 #include "librbd/ImageCtx.h"
 #include "librbd/journal/OpenRequest.h"
+#include "librbd/journal/PromoteRequest.h"
 #include "librbd/journal/Replay.h"
 #include "cls/journal/cls_journal_types.h"
 #include "journal/Journaler.h"
@@ -540,48 +541,11 @@ int Journal<I>::promote(I *image_ctx) {
   CephContext *cct = image_ctx->cct;
   ldout(cct, 20) << __func__ << dendl;
 
-  Journaler journaler(image_ctx->md_ctx, image_ctx->id, IMAGE_CLIENT_ID, {});
+  C_SaferCond ctx;
+  auto promote_req = journal::PromoteRequest<I>::create(image_ctx, false, &ctx);
+  promote_req->send();
 
-  Mutex lock("lock");
-  journal::ImageClientMeta client_meta;
-  uint64_t tag_tid;
-  journal::TagData tag_data;
-
-  C_SaferCond open_ctx;
-  auto open_req = journal::OpenRequest<I>::create(image_ctx, &journaler, &lock,
-                                                  &client_meta, &tag_tid,
-                                                  &tag_data, &open_ctx);
-  open_req->send();
-
-  BOOST_SCOPE_EXIT_ALL(&journaler) {
-    journaler.shut_down();
-  };
-
-  int r = open_ctx.wait();
-  if (r < 0) {
-    return r;
-  }
-
-  journal::TagPredecessor predecessor;
-  if (tag_data.mirror_uuid == ORPHAN_MIRROR_UUID) {
-    // orderly promotion -- demotion epoch will have a single entry
-    // so link to our predecessor (demotion) epoch
-    predecessor = journal::TagPredecessor{
-      ORPHAN_MIRROR_UUID, true, tag_tid, 1};
-  } else {
-    // forced promotion -- create an epoch no peers can link against
-    predecessor = journal::TagPredecessor{
-      LOCAL_MIRROR_UUID, true, tag_tid, 0};
-  }
-
-  cls::journal::Tag new_tag;
-  r = allocate_journaler_tag(cct, &journaler, client_meta.tag_class,
-                             predecessor, LOCAL_MIRROR_UUID, &new_tag);
-  if (r < 0) {
-    return r;
-  }
-
-  return 0;
+  return ctx.wait();
 }
 
 template <typename I>

src/librbd/journal/PromoteRequest.cc

@@ -0,0 +1,150 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+
+#include "librbd/journal/PromoteRequest.h"
+#include "common/dout.h"
+#include "common/errno.h"
+#include "common/WorkQueue.h"
+#include "journal/Journaler.h"
+#include "journal/Settings.h"
+#include "librbd/ImageCtx.h"
+#include "librbd/Journal.h"
+#include "librbd/Utils.h"
+#include "librbd/journal/OpenRequest.h"
+
+#define dout_subsys ceph_subsys_rbd
+#undef dout_prefix
+#define dout_prefix *_dout << "librbd::journal::PromoteRequest: " << this \
+                           << " " << __func__ << ": "
+
+namespace librbd {
+namespace journal {
+
+using librbd::util::create_async_context_callback;
+using librbd::util::create_context_callback;
+
+template <typename I>
+PromoteRequest<I>::PromoteRequest(I *image_ctx, bool force, Context *on_finish)
+  : m_image_ctx(image_ctx), m_force(force), m_on_finish(on_finish),
+    m_lock("PromoteRequest::m_lock") {
+}
+
+template <typename I>
+void PromoteRequest<I>::send() {
+  send_open();
+}
+
+template <typename I>
+void PromoteRequest<I>::send_open() {
+  CephContext *cct = m_image_ctx->cct;
+  ldout(cct, 20) << dendl;
+
+  m_journaler = new Journaler(m_image_ctx->md_ctx, m_image_ctx->id,
+                              Journal<>::IMAGE_CLIENT_ID, {});
+  Context *ctx = create_async_context_callback(
+    *m_image_ctx, create_context_callback<
+      PromoteRequest<I>, &PromoteRequest<I>::handle_open>(this));
+  auto open_req = OpenRequest<I>::create(m_image_ctx, m_journaler,
+                                         &m_lock, &m_client_meta,
+                                         &m_tag_tid, &m_tag_data, ctx);
+  open_req->send();
+}
+
+template <typename I>
+void PromoteRequest<I>::handle_open(int r) {
+  CephContext *cct = m_image_ctx->cct;
+  ldout(cct, 20) << "r=" << r << dendl;
+
+  if (r < 0) {
+    m_ret_val = r;
+    lderr(cct) << "failed to open journal: " << cpp_strerror(r) << dendl;
+    shut_down();
+    return;
+  }
+
+  allocate_tag();
+}
+
+template <typename I>
+void PromoteRequest<I>::allocate_tag() {
+  CephContext *cct = m_image_ctx->cct;
+  ldout(cct, 20) << dendl;
+
+  journal::TagPredecessor predecessor;
+  if (!m_force && m_tag_data.mirror_uuid == Journal<>::ORPHAN_MIRROR_UUID) {
+    // orderly promotion -- demotion epoch will have a single entry
+    // so link to our predecessor (demotion) epoch
+    predecessor = TagPredecessor{Journal<>::ORPHAN_MIRROR_UUID, true, m_tag_tid,
+                                 1};
+  } else {
+    // forced promotion -- create an epoch no peers can link against
+    predecessor = TagPredecessor{Journal<>::LOCAL_MIRROR_UUID, true, m_tag_tid,
+                                 0};
+  }
+
+  TagData tag_data;
+  tag_data.mirror_uuid = Journal<>::LOCAL_MIRROR_UUID;
+  tag_data.predecessor = predecessor;
+
+  bufferlist tag_bl;
+  ::encode(tag_data, tag_bl);
+
+  Context *ctx = create_context_callback<
+    PromoteRequest<I>, &PromoteRequest<I>::handle_allocate_tag>(this);
+  m_journaler->allocate_tag(m_client_meta.tag_class, tag_bl, &m_tag, ctx);
+}
+
+template <typename I>
+void PromoteRequest<I>::handle_allocate_tag(int r) {
+  CephContext *cct = m_image_ctx->cct;
+  ldout(cct, 20) << "r=" << r << dendl;
+
+  if (r < 0) {
+    m_ret_val = r;
+    lderr(cct) << "failed to allocate tag: " << cpp_strerror(r) << dendl;
+  }
+
+  shut_down();
+}
+
+template <typename I>
+void PromoteRequest<I>::shut_down() {
+  CephContext *cct = m_image_ctx->cct;
+  ldout(cct, 20) << dendl;
+
+  Context *ctx = create_async_context_callback(
+    *m_image_ctx, create_context_callback<
+      PromoteRequest<I>, &PromoteRequest<I>::handle_shut_down>(this));
+  m_journaler->shut_down(ctx);
+}
+
+template <typename I>
+void PromoteRequest<I>::handle_shut_down(int r) {
+  CephContext *cct = m_image_ctx->cct;
+  ldout(cct, 20) << "r=" << r << dendl;
+
+  if (r < 0) {
+    lderr(cct) << "failed to shut down journal: " << cpp_strerror(r) << dendl;
+  }
+
+  delete m_journaler;
+  finish(r);
+}
+
+template <typename I>
+void PromoteRequest<I>::finish(int r) {
+  if (m_ret_val < 0) {
+    r = m_ret_val;
+  }
+
+  CephContext *cct = m_image_ctx->cct;
+  ldout(cct, 20) << "r=" << r << dendl;
+
+  m_on_finish->complete(r);
+  delete this;
+}
+
+} // namespace journal
+} // namespace librbd
+
+template class librbd::journal::PromoteRequest<librbd::ImageCtx>;

src/librbd/journal/PromoteRequest.h

@@ -0,0 +1,88 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+
+#ifndef CEPH_LIBRBD_JOURNAL_PROMOTE_REQUEST_H
+#define CEPH_LIBRBD_JOURNAL_PROMOTE_REQUEST_H
+
+#include "include/int_types.h"
+#include "common/Mutex.h"
+#include "cls/journal/cls_journal_types.h"
+#include "librbd/journal/Types.h"
+#include "librbd/journal/TypeTraits.h"
+
+struct Context;
+
+namespace librbd {
+
+struct ImageCtx;
+
+namespace journal {
+
+template <typename ImageCtxT = ImageCtx>
+class PromoteRequest {
+public:
+  static PromoteRequest* create(ImageCtxT *image_ctx, bool force,
+                                Context *on_finish) {
+    return new PromoteRequest(image_ctx, force, on_finish);
+  }
+
+  PromoteRequest(ImageCtxT *image_ctx, bool force, Context *on_finish);
+
+  void send();
+
+private:
+  /**
+   * @verbatim
+   *
+   * <start>
+   *    |
+   *    v
+   *  OPEN
+   *    |
+   *    v
+   * ALLOCATE_TAG
+   *    |
+   *    v
+   * SHUT_DOWN
+   *    |
+   *    v
+   * <finish>
+   *
+   * @endverbatim
+   */
+
+  typedef typename TypeTraits<ImageCtxT>::Journaler Journaler;
+
+  ImageCtxT *m_image_ctx;
+  bool m_force;
+  Context *m_on_finish;
+
+  Journaler *m_journaler = nullptr;
+  int m_ret_val = 0;
+
+  Mutex m_lock;
+  ImageClientMeta m_client_meta;
+  uint64_t m_tag_tid = 0;
+  TagData m_tag_data;
+
+  cls::journal::Tag m_tag;
+
+  void send_open();
+  void handle_open(int r);
+
+  void allocate_tag();
+  void handle_allocate_tag(int r);
+
+  void shut_down();
+  void handle_shut_down(int r);
+
+  void finish(int r);
+
+};
+
+} // namespace journal
+} // namespace librbd
+
+extern template class librbd::journal::PromoteRequest<librbd::ImageCtx>;
+
+#endif // CEPH_LIBRBD_JOURNAL_PROMOTE_REQUEST_H

src/test/journal/mock/MockJournaler.h

@@ -108,6 +108,8 @@ struct MockJournaler {
   MOCK_METHOD2(get_cached_client, int(const std::string&, cls::journal::Client*));
   MOCK_METHOD2(update_client, void(const bufferlist &, Context *));
 
+  MOCK_METHOD4(allocate_tag, void(uint64_t, const bufferlist &,
+                                  cls::journal::Tag*, Context *));
   MOCK_METHOD3(get_tag, void(uint64_t, cls::journal::Tag *, Context *));
   MOCK_METHOD3(get_tags, void(uint64_t, journal::Journaler::Tags*, Context*));
   MOCK_METHOD4(get_tags, void(uint64_t, uint64_t, journal::Journaler::Tags*,
@@ -163,9 +165,10 @@ struct MockJournalerProxy {
     return -EINVAL;
   }
 
-  void allocate_tag(uint64_t, const bufferlist &,
-                    cls::journal::Tag*, Context *on_finish) {
-    on_finish->complete(-EINVAL);
+  void allocate_tag(uint64_t tag_class, const bufferlist &tag_data,
+                    cls::journal::Tag* tag, Context *on_finish) {
+    MockJournaler::get_instance().allocate_tag(tag_class, tag_data, tag,
+                                               on_finish);
   }
 
   void init(Context *on_finish) {

src/test/librbd/CMakeLists.txt

@@ -36,6 +36,7 @@ set(unittest_librbd_srcs
   image/test_mock_RefreshRequest.cc
   image_watcher/test_mock_RewatchRequest.cc
   journal/test_mock_OpenRequest.cc
+  journal/test_mock_PromoteRequest.cc
   journal/test_mock_Replay.cc
   object_map/test_mock_InvalidateRequest.cc
   object_map/test_mock_LockRequest.cc

src/test/librbd/journal/test_mock_PromoteRequest.cc

@@ -0,0 +1,230 @@
+// -*- mode:C; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+
+#include "test/librbd/test_mock_fixture.h"
+#include "test/librbd/test_support.h"
+#include "test/librbd/mock/MockImageCtx.h"
+#include "test/journal/mock/MockJournaler.h"
+#include "librbd/journal/OpenRequest.h"
+#include "librbd/journal/PromoteRequest.h"
+
+namespace librbd {
+
+namespace {
+
+struct MockTestImageCtx : public MockImageCtx {
+  MockTestImageCtx(librbd::ImageCtx& image_ctx) : MockImageCtx(image_ctx) {
+  }
+};
+
+} // anonymous namespace
+
+namespace journal {
+
+template <>
+struct TypeTraits<MockTestImageCtx> {
+  typedef ::journal::MockJournalerProxy Journaler;
+};
+
+template <>
+struct OpenRequest<MockTestImageCtx> {
+  Context *on_finish = nullptr;
+  static OpenRequest *s_instance;
+  static OpenRequest *create(MockTestImageCtx *image_ctx,
+                             ::journal::MockJournalerProxy *journaler,
+                             Mutex *lock, ImageClientMeta *client_meta,
+                             uint64_t *tag_tid, journal::TagData *tag_data,
+                             Context *on_finish) {
+    assert(s_instance != nullptr);
+    client_meta->tag_class = 456;
+    tag_data->mirror_uuid = Journal<>::ORPHAN_MIRROR_UUID;
+    *tag_tid = 567;
+    s_instance->on_finish = on_finish;
+    return s_instance;
+  }
+
+  OpenRequest() {
+    s_instance = this;
+  }
+
+  MOCK_METHOD0(send, void());
+};
+
+OpenRequest<MockTestImageCtx> *OpenRequest<MockTestImageCtx>::s_instance = nullptr;
+
+} // namespace journal
+} // namespace librbd
+
+// template definitions
+#include "librbd/journal/PromoteRequest.cc"
+template class librbd::journal::PromoteRequest<librbd::MockTestImageCtx>;
+
+namespace librbd {
+namespace journal {
+
+using ::testing::_;
+using ::testing::InSequence;
+using ::testing::WithArg;
+
+class TestMockJournalPromoteRequest : public TestMockFixture {
+public:
+  typedef PromoteRequest<MockTestImageCtx> MockPromoteRequest;
+  typedef OpenRequest<MockTestImageCtx> MockOpenRequest;
+
+  void expect_construct_journaler(::journal::MockJournaler &mock_journaler) {
+    EXPECT_CALL(mock_journaler, construct());
+  }
+
+  void expect_open_journaler(MockTestImageCtx &mock_image_ctx,
+                             MockOpenRequest &mock_open_request, int r) {
+    EXPECT_CALL(mock_open_request, send())
+      .WillOnce(FinishRequest(&mock_open_request, r, &mock_image_ctx));
+  }
+
+  void expect_allocate_tag(::journal::MockJournaler &mock_journaler,
+                           const journal::TagPredecessor &predecessor, int r) {
+    TagData tag_data;
+    tag_data.mirror_uuid = Journal<>::LOCAL_MIRROR_UUID;
+    tag_data.predecessor = predecessor;
+
+    bufferlist tag_data_bl;
+    ::encode(tag_data, tag_data_bl);
+
+    EXPECT_CALL(mock_journaler, allocate_tag(456, ContentsEqual(tag_data_bl),
+                                             _, _))
+      .WillOnce(WithArg<3>(CompleteContext(r, NULL)));
+  }
+
+  void expect_shut_down_journaler(::journal::MockJournaler &mock_journaler,
+                                  int r) {
+    EXPECT_CALL(mock_journaler, shut_down(_))
+      .WillOnce(CompleteContext(r, NULL));
+  }
+
+};
+
+TEST_F(TestMockJournalPromoteRequest, SuccessOrderly) {
+  REQUIRE_FEATURE(RBD_FEATURE_JOURNALING);
+
+  librbd::ImageCtx *ictx;
+  ASSERT_EQ(0, open_image(m_image_name, &ictx));
+
+  MockTestImageCtx mock_image_ctx(*ictx);
+  ::journal::MockJournaler mock_journaler;
+  MockOpenRequest mock_open_request;
+
+  expect_op_work_queue(mock_image_ctx);
+
+  InSequence seq;
+  expect_construct_journaler(mock_journaler);
+  expect_open_journaler(mock_image_ctx, mock_open_request, 0);
+  expect_allocate_tag(mock_journaler,
+                      {Journal<>::ORPHAN_MIRROR_UUID, true, 567, 1}, 0);
+  expect_shut_down_journaler(mock_journaler, 0);
+
+  C_SaferCond ctx;
+  auto req = MockPromoteRequest::create(&mock_image_ctx, false, &ctx);
+  req->send();
+  ASSERT_EQ(0, ctx.wait());
+}
+
+TEST_F(TestMockJournalPromoteRequest, SuccessForced) {
+  REQUIRE_FEATURE(RBD_FEATURE_JOURNALING);
+
+  librbd::ImageCtx *ictx;
+  ASSERT_EQ(0, open_image(m_image_name, &ictx));
+
+  MockTestImageCtx mock_image_ctx(*ictx);
+  ::journal::MockJournaler mock_journaler;
+  MockOpenRequest mock_open_request;
+
+  expect_op_work_queue(mock_image_ctx);
+
+  InSequence seq;
+  expect_construct_journaler(mock_journaler);
+  expect_open_journaler(mock_image_ctx, mock_open_request, 0);
+  expect_allocate_tag(mock_journaler,
+                      {Journal<>::LOCAL_MIRROR_UUID, true, 567, 0}, 0);
+  expect_shut_down_journaler(mock_journaler, 0);
+
+  C_SaferCond ctx;
+  auto req = MockPromoteRequest::create(&mock_image_ctx, true, &ctx);
+  req->send();
+  ASSERT_EQ(0, ctx.wait());
+}
+
+TEST_F(TestMockJournalPromoteRequest, OpenError) {
+  REQUIRE_FEATURE(RBD_FEATURE_JOURNALING);
+
+  librbd::ImageCtx *ictx;
+  ASSERT_EQ(0, open_image(m_image_name, &ictx));
+
+  MockTestImageCtx mock_image_ctx(*ictx);
+  ::journal::MockJournaler mock_journaler;
+  MockOpenRequest mock_open_request;
+
+  expect_op_work_queue(mock_image_ctx);
+
+  InSequence seq;
+  expect_construct_journaler(mock_journaler);
+  expect_open_journaler(mock_image_ctx, mock_open_request, -ENOENT);
+  expect_shut_down_journaler(mock_journaler, -EINVAL);
+
+  C_SaferCond ctx;
+  auto req = MockPromoteRequest::create(&mock_image_ctx, false, &ctx);
+  req->send();
+  ASSERT_EQ(-ENOENT, ctx.wait());
+}
+
+TEST_F(TestMockJournalPromoteRequest, AllocateTagError) {
+  REQUIRE_FEATURE(RBD_FEATURE_JOURNALING);
+
+  librbd::ImageCtx *ictx;
+  ASSERT_EQ(0, open_image(m_image_name, &ictx));
+
+  MockTestImageCtx mock_image_ctx(*ictx);
+  ::journal::MockJournaler mock_journaler;
+  MockOpenRequest mock_open_request;
+
+  expect_op_work_queue(mock_image_ctx);
+
+  InSequence seq;
+  expect_construct_journaler(mock_journaler);
+  expect_open_journaler(mock_image_ctx, mock_open_request, 0);
+  expect_allocate_tag(mock_journaler,
+                      {Journal<>::LOCAL_MIRROR_UUID, true, 567, 0}, -EBADMSG);
+  expect_shut_down_journaler(mock_journaler, -EINVAL);
+
+  C_SaferCond ctx;
+  auto req = MockPromoteRequest::create(&mock_image_ctx, true, &ctx);
+  req->send();
+  ASSERT_EQ(-EBADMSG, ctx.wait());
+}
+
+TEST_F(TestMockJournalPromoteRequest, ShutDownError) {
+  REQUIRE_FEATURE(RBD_FEATURE_JOURNALING);
+
+  librbd::ImageCtx *ictx;
+  ASSERT_EQ(0, open_image(m_image_name, &ictx));
+
+  MockTestImageCtx mock_image_ctx(*ictx);
+  ::journal::MockJournaler mock_journaler;
+  MockOpenRequest mock_open_request;
+
+  expect_op_work_queue(mock_image_ctx);
+
+  InSequence seq;
+  expect_construct_journaler(mock_journaler);
+  expect_open_journaler(mock_image_ctx, mock_open_request, 0);
+  expect_allocate_tag(mock_journaler,
+                      {Journal<>::LOCAL_MIRROR_UUID, true, 567, 0}, 0);
+  expect_shut_down_journaler(mock_journaler, -EINVAL);
+
+  C_SaferCond ctx;
+  auto req = MockPromoteRequest::create(&mock_image_ctx, true, &ctx);
+  req->send();
+  ASSERT_EQ(-EINVAL, ctx.wait());
+}
+
+} // namespace journal
+} // namespace librbd