rbd-mirror: fix potential infinite loop when formatting status message

The improvements include:

- tag_tid values should always be increasing, so loop only if
  master.tag_tid > mirror_tag_tid in calculate_behind_master_or_send_update;
- in send_update_tag_cache don't refetch a tag if it is already in the
  cache;
- make fake tags with tag_data.predecessor.tag_tid set to zero;
- make sure the new tag is inserted to the cache if an old entry
  with this id happens to exist.

Fixes: http://tracker.ceph.com/issues/22932
Signed-off-by: Mykola Golub <mgolub@suse.com>
(cherry picked from commit 2fb99b5ed39c36879d1b6180ac47629bc5a3b315)

diff --git a/src/tools/rbd_mirror/image_replayer/ReplayStatusFormatter.cc b/src/tools/rbd_mirror/image_replayer/ReplayStatusFormatter.cc
index c3a04007bcb8641e16781ef0cc5d502f3432761c..9d11db3374003699d53515121804da18f02d27ab 100644 (file)
--- a/src/tools/rbd_mirror/image_replayer/ReplayStatusFormatter.cc
+++ b/src/tools/rbd_mirror/image_replayer/ReplayStatusFormatter.cc
@@ -111,7 +111,7 @@ bool ReplayStatusFormatter<I>::calculate_behind_master_or_send_update() {
   cls::journal::ObjectPosition master = m_master_position;
   uint64_t mirror_tag_tid = m_mirror_position.tag_tid;
 
-  while (master.tag_tid != mirror_tag_tid) {
+  while (master.tag_tid > mirror_tag_tid) {
     auto tag_it = m_tag_cache.find(master.tag_tid);
     if (tag_it == m_tag_cache.end()) {
       send_update_tag_cache(master.tag_tid, mirror_tag_tid);
@@ -119,10 +119,12 @@ bool ReplayStatusFormatter<I>::calculate_behind_master_or_send_update() {
     }
     librbd::journal::TagData &tag_data = tag_it->second;
     m_entries_behind_master += master.entry_tid;
-    master = cls::journal::ObjectPosition(0, tag_data.predecessor.tag_tid,
-                                         tag_data.predecessor.entry_tid);
+    master = {0, tag_data.predecessor.tag_tid, tag_data.predecessor.entry_tid};
+  }
+  if (master.tag_tid == mirror_tag_tid &&
+      master.entry_tid > m_mirror_position.entry_tid) {
+    m_entries_behind_master += master.entry_tid - m_mirror_position.entry_tid;
   }
-  m_entries_behind_master += master.entry_tid - m_mirror_position.entry_tid;
 
   dout(20) << "clearing tags not needed any more (below mirror position)"
           << dendl;
@@ -151,11 +153,8 @@ bool ReplayStatusFormatter<I>::calculate_behind_master_or_send_update() {
 template <typename I>
 void ReplayStatusFormatter<I>::send_update_tag_cache(uint64_t master_tag_tid,
                                                     uint64_t mirror_tag_tid) {
-
-  dout(20) << "master_tag_tid=" << master_tag_tid << ", mirror_tag_tid="
-          << mirror_tag_tid << dendl;
-
-  if (master_tag_tid == mirror_tag_tid) {
+  if (master_tag_tid <= mirror_tag_tid ||
+      m_tag_cache.find(master_tag_tid) != m_tag_cache.end()) {
     Context *on_finish = nullptr;
     {
       Mutex::Locker locker(m_lock);
@@ -167,6 +166,9 @@ void ReplayStatusFormatter<I>::send_update_tag_cache(uint64_t master_tag_tid,
     return;
   }
 
+  dout(20) << "master_tag_tid=" << master_tag_tid << ", mirror_tag_tid="
+          << mirror_tag_tid << dendl;
+
   FunctionContext *ctx = new FunctionContext(
     [this, master_tag_tid, mirror_tag_tid](int r) {
       handle_update_tag_cache(master_tag_tid, mirror_tag_tid, r);
@@ -200,16 +202,12 @@ void ReplayStatusFormatter<I>::handle_update_tag_cache(uint64_t master_tag_tid,
       tag_data.predecessor.mirror_uuid !=
         librbd::Journal<>::ORPHAN_MIRROR_UUID) {
     dout(20) << "hit remote image non-primary epoch" << dendl;
-    tag_data.predecessor.tag_tid = mirror_tag_tid;
-  } else if (tag_data.predecessor.tag_tid == 0) {
-    // We failed. Don't consider this fatal, just terminate retrieving.
-    dout(20) << "making fake tag" << dendl;
-    tag_data.predecessor.tag_tid = mirror_tag_tid;
+    tag_data.predecessor = {};
   }
 
   dout(20) << "decoded tag " << master_tag_tid << ": " << tag_data << dendl;
 
-  m_tag_cache.insert(std::make_pair(master_tag_tid, tag_data));
+  m_tag_cache[master_tag_tid] = tag_data;
   send_update_tag_cache(tag_data.predecessor.tag_tid, mirror_tag_tid);
 }