From 172ad649fb8a2410bdc078a5752b2ca09dc03ac2 Mon Sep 17 00:00:00 2001
From: Jason Dillaman <dillaman@redhat.com>
Date: Wed, 12 Jun 2019 13:36:24 -0400
Subject: [PATCH] librbd: new rbd_journal_object_writethrough_until_flush
 option

When set to true, the journal will not attempt to batch appends until
after it receives the the first flush request from the user.

Signed-off-by: Jason Dillaman <dillaman@redhat.com>
---
 src/common/options.cc                |  6 ++++
 src/librbd/Journal.cc                | 30 +++++++++++++---
 src/librbd/Journal.h                 |  4 +++
 src/librbd/io/ImageRequest.cc        |  1 +
 src/test/librbd/mock/MockJournal.h   |  2 ++
 src/test/librbd/test_mock_Journal.cc | 52 +++++++++++++++++++++-------
 6 files changed, 79 insertions(+), 16 deletions(-)

diff --git a/src/common/options.cc b/src/common/options.cc
index 2b02dbfa907d8..167bb45044ec4 100644
--- a/src/common/options.cc
+++ b/src/common/options.cc
@@ -7304,6 +7304,12 @@ static std::vector<Option> get_rbd_options() {
     .set_default(5)
     .set_description("commit time interval, seconds"),
 
+    Option("rbd_journal_object_writethrough_until_flush", Option::TYPE_BOOL, Option::LEVEL_ADVANCED)
+    .set_default(false)
+    .set_description("when enabled, the rbd_journal_object_flush* configuration "
+                     "options are ignored until the first flush so that batched "
+                     "journal IO is known to be safe for consistency"),
+
     Option("rbd_journal_object_flush_interval", Option::TYPE_UINT, Option::LEVEL_ADVANCED)
     .set_default(0)
     .set_description("maximum number of pending commits per journal object"),
diff --git a/src/librbd/Journal.cc b/src/librbd/Journal.cc
index 11b2f23fa0fc9..66abdf9020733 100644
--- a/src/librbd/Journal.cc
+++ b/src/librbd/Journal.cc
@@ -714,6 +714,26 @@ void Journal<I>::flush_commit_position(Context *on_finish) {
   m_journaler->flush_commit_position(on_finish);
 }
 
+template <typename I>
+void Journal<I>::user_flushed() {
+  if (m_state == STATE_READY && !m_user_flushed.exchange(true) &&
+      m_image_ctx.config.template get_val<bool>("rbd_journal_object_writethrough_until_flush")) {
+    Mutex::Locker locker(m_lock);
+    if (m_state == STATE_READY) {
+      CephContext *cct = m_image_ctx.cct;
+      ldout(cct, 5) << this << " " << __func__ << dendl;
+
+      ceph_assert(m_journaler != nullptr);
+      m_journaler->set_append_batch_options(
+        m_image_ctx.config.template get_val<uint64_t>("rbd_journal_object_flush_interval"),
+        m_image_ctx.config.template get_val<Option::size_t>("rbd_journal_object_flush_bytes"),
+        m_image_ctx.config.template get_val<double>("rbd_journal_object_flush_age"));
+    } else {
+      m_user_flushed = false;
+    }
+  }
+}
+
 template <typename I>
 uint64_t Journal<I>::append_write_event(uint64_t offset, size_t length,
                                         const bufferlist &bl,
@@ -1172,10 +1192,12 @@ void Journal<I>::start_append() {
 
   m_journaler->start_append(
     m_image_ctx.config.template get_val<uint64_t>("rbd_journal_object_max_in_flight_appends"));
-  m_journaler->set_append_batch_options(
-    m_image_ctx.config.template get_val<uint64_t>("rbd_journal_object_flush_interval"),
-    m_image_ctx.config.template get_val<Option::size_t>("rbd_journal_object_flush_bytes"),
-    m_image_ctx.config.template get_val<double>("rbd_journal_object_flush_age"));
+  if (!m_image_ctx.config.template get_val<bool>("rbd_journal_object_writethrough_until_flush")) {
+    m_journaler->set_append_batch_options(
+      m_image_ctx.config.template get_val<uint64_t>("rbd_journal_object_flush_interval"),
+      m_image_ctx.config.template get_val<Option::size_t>("rbd_journal_object_flush_bytes"),
+      m_image_ctx.config.template get_val<double>("rbd_journal_object_flush_age"));
+  }
 
   transition_state(STATE_READY, 0);
 }
diff --git a/src/librbd/Journal.h b/src/librbd/Journal.h
index 88979b6252b21..e63cc4a71377e 100644
--- a/src/librbd/Journal.h
+++ b/src/librbd/Journal.h
@@ -130,6 +130,8 @@ public:
 
   void flush_commit_position(Context *on_finish);
 
+  void user_flushed();
+
   uint64_t append_write_event(uint64_t offset, size_t length,
                               const bufferlist &bl,
                               bool flush_entry);
@@ -291,6 +293,8 @@ private:
   uint64_t m_event_tid;
   Events m_events;
 
+  std::atomic<bool> m_user_flushed = false;
+
   std::atomic<uint64_t> m_op_tid = { 0 };
   TidToFutures m_op_futures;
 
diff --git a/src/librbd/io/ImageRequest.cc b/src/librbd/io/ImageRequest.cc
index ce064f1f07a17..166c8702308b7 100644
--- a/src/librbd/io/ImageRequest.cc
+++ b/src/librbd/io/ImageRequest.cc
@@ -686,6 +686,7 @@ void ImageFlushRequest<I>::send_request() {
     ceph_assert(image_ctx.journal != NULL);
     journal_tid = image_ctx.journal->append_io_event(
       journal::EventEntry(journal::AioFlushEvent()), 0, 0, false, 0);
+    image_ctx.journal->user_flushed();
   }
 
   auto object_dispatch_spec = ObjectDispatchSpec::create_flush(
diff --git a/src/test/librbd/mock/MockJournal.h b/src/test/librbd/mock/MockJournal.h
index dd932830707cf..31806217af165 100644
--- a/src/test/librbd/mock/MockJournal.h
+++ b/src/test/librbd/mock/MockJournal.h
@@ -65,6 +65,8 @@ struct MockJournal {
 
   MOCK_METHOD0(allocate_op_tid, uint64_t());
 
+  MOCK_METHOD0(user_flushed, void());
+
   MOCK_METHOD3(append_op_event_mock, void(uint64_t, const journal::EventEntry&,
                                           Context *));
   void append_op_event(uint64_t op_tid, journal::EventEntry &&event_entry,
diff --git a/src/test/librbd/test_mock_Journal.cc b/src/test/librbd/test_mock_Journal.cc
index 00367e2214e89..86e31f905d778 100644
--- a/src/test/librbd/test_mock_Journal.cc
+++ b/src/test/librbd/test_mock_Journal.cc
@@ -399,8 +399,13 @@ public:
     EXPECT_CALL(mock_journaler, start_append(_));
   }
 
-  void expect_set_append_batch_options(::journal::MockJournaler &mock_journaler) {
-    EXPECT_CALL(mock_journaler, set_append_batch_options(_, _, _));
+  void expect_set_append_batch_options(MockJournalImageCtx &mock_image_ctx,
+                                       ::journal::MockJournaler &mock_journaler,
+                                       bool user_flushed) {
+    if (mock_image_ctx.image_ctx->config.get_val<bool>("rbd_journal_object_writethrough_until_flush") ==
+          user_flushed) {
+      EXPECT_CALL(mock_journaler, set_append_batch_options(_, _, _));
+    }
   }
 
   void expect_stop_append(::journal::MockJournaler &mock_journaler, int r) {
@@ -522,7 +527,7 @@ public:
     expect_committed(mock_journaler, 0);
     expect_flush_commit_position(mock_journaler);
     expect_start_append(mock_journaler);
-    expect_set_append_batch_options(mock_journaler);
+    expect_set_append_batch_options(mock_image_ctx, mock_journaler, false);
     ASSERT_EQ(0, when_open(mock_journal));
   }
 
@@ -590,7 +595,7 @@ TEST_F(TestMockJournal, StateTransitions) {
   expect_flush_commit_position(mock_journaler);
 
   expect_start_append(mock_journaler);
-  expect_set_append_batch_options(mock_journaler);
+  expect_set_append_batch_options(mock_image_ctx, mock_journaler, false);
 
   ASSERT_EQ(0, when_open(mock_journal));
 
@@ -668,7 +673,7 @@ TEST_F(TestMockJournal, ReplayCompleteError) {
   expect_shut_down_replay(mock_image_ctx, mock_journal_replay, 0);
   expect_flush_commit_position(mock_journaler);
   expect_start_append(mock_journaler);
-  expect_set_append_batch_options(mock_journaler);
+  expect_set_append_batch_options(mock_image_ctx, mock_journaler, false);
   ASSERT_EQ(0, when_open(mock_journal));
 
   expect_stop_append(mock_journaler, 0);
@@ -726,7 +731,7 @@ TEST_F(TestMockJournal, FlushReplayError) {
   expect_shut_down_replay(mock_image_ctx, mock_journal_replay, 0);
   expect_flush_commit_position(mock_journaler);
   expect_start_append(mock_journaler);
-  expect_set_append_batch_options(mock_journaler);
+  expect_set_append_batch_options(mock_image_ctx, mock_journaler, false);
   ASSERT_EQ(0, when_open(mock_journal));
 
   expect_stop_append(mock_journaler, 0);
@@ -781,7 +786,7 @@ TEST_F(TestMockJournal, CorruptEntry) {
   expect_shut_down_replay(mock_image_ctx, mock_journal_replay, 0);
   expect_flush_commit_position(mock_journaler);
   expect_start_append(mock_journaler);
-  expect_set_append_batch_options(mock_journaler);
+  expect_set_append_batch_options(mock_image_ctx, mock_journaler, false);
   ASSERT_EQ(0, when_open(mock_journal));
 
   expect_stop_append(mock_journaler, -EINVAL);
@@ -820,7 +825,7 @@ TEST_F(TestMockJournal, StopError) {
   expect_shut_down_replay(mock_image_ctx, mock_journal_replay, 0);
   expect_flush_commit_position(mock_journaler);
   expect_start_append(mock_journaler);
-  expect_set_append_batch_options(mock_journaler);
+  expect_set_append_batch_options(mock_image_ctx, mock_journaler, false);
   ASSERT_EQ(0, when_open(mock_journal));
 
   expect_stop_append(mock_journaler, -EINVAL);
@@ -886,7 +891,7 @@ TEST_F(TestMockJournal, ReplayOnDiskPreFlushError) {
   expect_shut_down_replay(mock_image_ctx, mock_journal_replay, 0);
   expect_flush_commit_position(mock_journaler);
   expect_start_append(mock_journaler);
-  expect_set_append_batch_options(mock_journaler);
+  expect_set_append_batch_options(mock_image_ctx, mock_journaler, false);
 
   C_SaferCond ctx;
   mock_journal.open(&ctx);
@@ -969,7 +974,7 @@ TEST_F(TestMockJournal, ReplayOnDiskPostFlushError) {
   expect_shut_down_replay(mock_image_ctx, mock_journal_replay, 0);
   expect_flush_commit_position(mock_journaler);
   expect_start_append(mock_journaler);
-  expect_set_append_batch_options(mock_journaler);
+  expect_set_append_batch_options(mock_image_ctx, mock_journaler, false);
 
   C_SaferCond ctx;
   mock_journal.open(&ctx);
@@ -1284,7 +1289,7 @@ TEST_F(TestMockJournal, ExternalReplay) {
   InSequence seq;
   expect_stop_append(mock_journaler, 0);
   expect_start_append(mock_journaler);
-  expect_set_append_batch_options(mock_journaler);
+  expect_set_append_batch_options(mock_image_ctx, mock_journaler, false);
   expect_shut_down_journaler(mock_journaler);
 
   C_SaferCond start_ctx;
@@ -1316,7 +1321,7 @@ TEST_F(TestMockJournal, ExternalReplayFailure) {
   InSequence seq;
   expect_stop_append(mock_journaler, -EINVAL);
   expect_start_append(mock_journaler);
-  expect_set_append_batch_options(mock_journaler);
+  expect_set_append_batch_options(mock_image_ctx, mock_journaler, false);
   expect_shut_down_journaler(mock_journaler);
 
   C_SaferCond start_ctx;
@@ -1500,4 +1505,27 @@ TEST_F(TestMockJournal, ForcePromoted) {
   ASSERT_EQ(0, listener.ctx.wait());
 }
 
+TEST_F(TestMockJournal, UserFlushed) {
+  REQUIRE_FEATURE(RBD_FEATURE_JOURNALING);
+
+  librbd::ImageCtx *ictx;
+  ASSERT_EQ(0, open_image(m_image_name, &ictx));
+
+  MockJournalImageCtx mock_image_ctx(*ictx);
+  MockJournal mock_journal(mock_image_ctx);
+  MockObjectDispatch mock_object_dispatch;
+  ::journal::MockJournaler mock_journaler;
+  MockJournalOpenRequest mock_open_request;
+  open_journal(mock_image_ctx, mock_journal, mock_object_dispatch,
+               mock_journaler, mock_open_request);
+  BOOST_SCOPE_EXIT_ALL(&) {
+    close_journal(mock_image_ctx, mock_journal, mock_journaler);
+  };
+
+  expect_set_append_batch_options(mock_image_ctx, mock_journaler, true);
+  mock_journal.user_flushed();
+
+  expect_shut_down_journaler(mock_journaler);
+}
+
 } // namespace librbd
-- 
2.39.5