From 2773e68f438b48fd99c37f132771ce05675be047 Mon Sep 17 00:00:00 2001
From: Jason Dillaman <dillaman@redhat.com>
Date: Wed, 13 Nov 2019 20:01:16 -0500
Subject: [PATCH] journal: properly handle immediate overflow upon advancing
 object set

While the old set is being closed, additional IO can be queued up within
the old, closed objects while the in-flight IO settles. It's therefore
possible that the queued IO that is transferred from the old set to the
new set causes an immediate overflow of the new set.

Signed-off-by: Jason Dillaman <dillaman@redhat.com>
---
 src/journal/JournalRecorder.cc | 24 +++++++++++++++++++++---
 src/journal/JournalRecorder.h  |  2 +-
 2 files changed, 22 insertions(+), 4 deletions(-)

diff --git a/src/journal/JournalRecorder.cc b/src/journal/JournalRecorder.cc
index e85fdb4ef0c..0304ae777a9 100644
--- a/src/journal/JournalRecorder.cc
+++ b/src/journal/JournalRecorder.cc
@@ -258,6 +258,7 @@ void JournalRecorder::open_object_set() {
   ldout(m_cct, 10) << "opening object set " << m_current_set << dendl;
 
   uint8_t splay_width = m_journal_metadata->get_splay_width();
+  bool overflowed = false;
 
   auto lockers{lock_object_recorders()};
   for (const auto& p : m_object_ptrs) {
@@ -267,9 +268,18 @@ void JournalRecorder::open_object_set() {
       ceph_assert(object_recorder->is_closed());
 
       // ready to close object and open object in active set
-      create_next_object_recorder(object_recorder);
+      if (create_next_object_recorder(object_recorder)) {
+        overflowed = true;
+      }
     }
   }
+  lockers.clear();
+
+  if (overflowed) {
+    ldout(m_cct, 10) << "object set " << m_current_set << " now full" << dendl;
+    ldout(m_cct, 10) << "" << dendl;
+    close_and_advance_object_set(m_current_set);
+  }
 }
 
 bool JournalRecorder::close_object_set(uint64_t active_set) {
@@ -288,6 +298,8 @@ bool JournalRecorder::close_object_set(uint64_t active_set) {
       // flush out all queued appends and hold future appends
       if (!object_recorder->close()) {
         ++m_in_flight_object_closes;
+        ldout(m_cct, 10) << "object " << object_recorder->get_oid() << " "
+                         << "close in-progress" << dendl;
       } else {
         ldout(m_cct, 10) << "object " << object_recorder->get_oid() << " closed"
                          << dendl;
@@ -310,7 +322,7 @@ ceph::ref_t<ObjectRecorder> JournalRecorder::create_object_recorder(
   return object_recorder;
 }
 
-void JournalRecorder::create_next_object_recorder(
+bool JournalRecorder::create_next_object_recorder(
     ceph::ref_t<ObjectRecorder> object_recorder) {
   ceph_assert(ceph_mutex_is_locked(m_lock));
 
@@ -336,8 +348,14 @@ void JournalRecorder::create_next_object_recorder(
       new_object_recorder->get_object_number());
   }
 
-  new_object_recorder->append(std::move(append_buffers));
+  bool object_full = new_object_recorder->append(std::move(append_buffers));
+  if (object_full) {
+    ldout(m_cct, 10) << "object " << new_object_recorder->get_oid() << " "
+                     << "now full" << dendl;
+  }
+
   m_object_ptrs[splay_offset] = std::move(new_object_recorder);
+  return object_full;
 }
 
 void JournalRecorder::handle_update() {
diff --git a/src/journal/JournalRecorder.h b/src/journal/JournalRecorder.h
index f4edd866804..680dbcdfead 100644
--- a/src/journal/JournalRecorder.h
+++ b/src/journal/JournalRecorder.h
@@ -114,7 +114,7 @@ private:
 
   ceph::ref_t<ObjectRecorder> create_object_recorder(uint64_t object_number,
                                            ceph::mutex* lock);
-  void create_next_object_recorder(ceph::ref_t<ObjectRecorder> object_recorder);
+  bool create_next_object_recorder(ceph::ref_t<ObjectRecorder> object_recorder);
 
   void handle_update();
 
-- 
2.47.3