journal: optimize speed of live replay journal pruning

When streaming playback, avoid the unnecessary watch delay when
one or more entries have been pruned.

Signed-off-by: Jason Dillaman <dillaman@redhat.com>
(cherry picked from commit 08a8ee98c03b5dfb30341c8d209f0c231b2c5d27)

diff --git a/src/journal/JournalPlayer.cc b/src/journal/JournalPlayer.cc
index 750f936b20c714388d086c95895e867b3c8c5124..09f2d2ca3f324bf04909aada0a4077e9c0031b56 100644 (file)
--- a/src/journal/JournalPlayer.cc
+++ b/src/journal/JournalPlayer.cc
@@ -300,7 +300,7 @@ int JournalPlayer::process_prefetch(uint64_t object_number) {
     if (object_player->empty() && object_player->refetch_required()) {
       ldout(m_cct, 10) << "refetching potentially partially decoded object"
                        << dendl;
-      object_player->clear_refetch_required();
+      object_player->set_refetch_state(ObjectPlayer::REFETCH_STATE_NONE);
       fetch(object_player);
     } else if (!remove_empty_object_player(object_player)) {
       ldout(m_cct, 10) << "prefetch of object complete" << dendl;
@@ -482,6 +482,7 @@ void JournalPlayer::prune_tag(uint64_t tag_tid) {
     m_prune_tag_tid = tag_tid;
   }
 
+  bool pruned = false;
   for (auto &player_pair : m_object_players) {
     ObjectPlayerPtr object_player(player_pair.second);
     ldout(m_cct, 15) << __func__ << ": checking " << object_player->get_oid()
@@ -492,12 +493,25 @@ void JournalPlayer::prune_tag(uint64_t tag_tid) {
       if (entry.get_tag_tid() == tag_tid) {
         ldout(m_cct, 20) << __func__ << ": pruned " << entry << dendl;
         object_player->pop_front();
+        pruned = true;
       } else {
         break;
       }
     }
+  }
+
+  // avoid watch delay when pruning stale tags from journal objects
+  if (pruned) {
+    ldout(m_cct, 15) << __func__ << ": reseting refetch state to immediate"
+                     << dendl;
+    for (auto &player_pair : m_object_players) {
+      ObjectPlayerPtr object_player(player_pair.second);
+      object_player->set_refetch_state(ObjectPlayer::REFETCH_STATE_IMMEDIATE);
+    }
+  }
 
-    // trim empty player to prefetch the next available object
+  // trim empty player to prefetch the next available object
+  for (auto &player_pair : m_object_players) {
     remove_empty_object_player(player_pair.second);
   }
 }
@@ -565,7 +579,7 @@ bool JournalPlayer::remove_empty_object_player(const ObjectPlayerPtr &player) {
                      << "require refetch" << dendl;
     m_active_set = active_set;
     for (auto &pair : m_object_players) {
-      pair.second->set_refetch_required();
+      pair.second->set_refetch_state(ObjectPlayer::REFETCH_STATE_IMMEDIATE);
     }
     return false;
   }
@@ -641,7 +655,7 @@ void JournalPlayer::refetch(bool immediate) {
 
   ObjectPlayerPtr object_player = get_object_player();
   if (object_player->refetch_required()) {
-    object_player->clear_refetch_required();
+    object_player->set_refetch_state(ObjectPlayer::REFETCH_STATE_NONE);
     fetch(object_player);
     return;
   }
@@ -684,11 +698,13 @@ void JournalPlayer::schedule_watch(bool immediate) {
       uint64_t active_set = m_journal_metadata->get_active_set();
       uint64_t object_set = object_player->get_object_number() / splay_width;
       if (immediate ||
+          (object_player->get_refetch_state() ==
+             ObjectPlayer::REFETCH_STATE_IMMEDIATE) ||
           (object_set < active_set && object_player->refetch_required())) {
-        ldout(m_cct, 20) << __func__ << ": refetching "
+        ldout(m_cct, 20) << __func__ << ": immediately refetching "
                          << object_player->get_oid()
                          << dendl;
-        object_player->clear_refetch_required();
+        object_player->set_refetch_state(ObjectPlayer::REFETCH_STATE_NONE);
         watch_interval = 0;
       }
     }

diff --git a/src/journal/ObjectPlayer.cc b/src/journal/ObjectPlayer.cc
index 43adb5da5cd637b5d74421ad287297a405487d63..92dd702615bcd018e749bb5e56fdce688bc98018 100644 (file)
--- a/src/journal/ObjectPlayer.cc
+++ b/src/journal/ObjectPlayer.cc
@@ -122,7 +122,7 @@ int ObjectPlayer::handle_fetch_complete(int r, const bufferlist &bl,
   assert(m_fetch_in_progress);
   m_read_off += bl.length();
   m_read_bl.append(bl);
-  m_refetch_required = true;
+  m_refetch_state = REFETCH_STATE_REQUIRED;
 
   bool full_fetch = (m_max_fetch_bytes == 2U << m_order);
   bool partial_entry = false;

diff --git a/src/journal/ObjectPlayer.h b/src/journal/ObjectPlayer.h
index a9d2d9e985ec41b473f0b7b541ddd81c89cbc754..cff33dc3bcd9c37849b6de64309a9e70de3847a6 100644 (file)
--- a/src/journal/ObjectPlayer.h
+++ b/src/journal/ObjectPlayer.h
@@ -30,6 +30,12 @@ public:
   typedef std::list<Entry> Entries;
   typedef interval_set<uint64_t> InvalidRanges;
 
+  enum RefetchState {
+    REFETCH_STATE_NONE,
+    REFETCH_STATE_REQUIRED,
+    REFETCH_STATE_IMMEDIATE
+  };
+
   ObjectPlayer(librados::IoCtx &ioctx, const std::string &object_oid_prefix,
                uint64_t object_num, SafeTimer &timer, Mutex &timer_lock,
                uint8_t order, uint64_t max_fetch_bytes);
@@ -63,13 +69,13 @@ public:
   }
 
   inline bool refetch_required() const {
-    return m_refetch_required;
+    return (get_refetch_state() != REFETCH_STATE_NONE);
   }
-  inline void set_refetch_required() {
-    m_refetch_required = true;
+  inline RefetchState get_refetch_state() const {
+    return m_refetch_state;
   }
-  inline void clear_refetch_required() {
-    m_refetch_required = false;
+  inline void set_refetch_state(RefetchState refetch_state) {
+    m_refetch_state = refetch_state;
   }
 
 private:
@@ -124,7 +130,7 @@ private:
   Context *m_watch_ctx = nullptr;
 
   bool m_unwatched = false;
-  bool m_refetch_required = true;
+  RefetchState m_refetch_state = REFETCH_STATE_IMMEDIATE;
 
   int handle_fetch_complete(int r, const bufferlist &bl, bool *refetch);
 

diff --git a/src/test/journal/test_ObjectPlayer.cc b/src/test/journal/test_ObjectPlayer.cc
index 880128ee8e19abae18cafe53a1e6228539647bf9..90a3334a7aa20a61cceb63d7a2ce4f68a344f065 100644 (file)
--- a/src/test/journal/test_ObjectPlayer.cc
+++ b/src/test/journal/test_ObjectPlayer.cc
@@ -26,7 +26,8 @@ public:
   int fetch(journal::ObjectPlayerPtr object_player) {
     while (true) {
       C_SaferCond ctx;
-      object_player->clear_refetch_required();
+      object_player->set_refetch_state(
+        journal::ObjectPlayer::REFETCH_STATE_NONE);
       object_player->fetch(&ctx);
       int r = ctx.wait();
       if (r < 0 || !object_player->refetch_required()) {