osd: prime splits/merges for any potential fabricated split/merge participant

If we have to fabricate a merge target, we need to prime any future splits
it might have.  Otherwise a sequence like

- e100 1.f merge to 1.7
- e110 1.7 split to 1.f, 1.17, 1.1f

where we process all of the above in one go at, say, e120, will lead to
a crash in register_and_wake_split_child because 1.17 and/or 1.1f aren't
primed.

Fix by making identify_splits_and_merges do a recursive scan on any
merge/split participants detected too.

Fixes: http://tracker.ceph.com/issues/38483
Signed-off-by: xie xingguo <xie.xingguo@zte.com.cn>
(cherry picked from commit 6afe4f16dbf15f84074dd4e787d182d18dbccfa6)

diff --git a/src/osd/OSD.cc b/src/osd/OSD.cc
index 5255cfc843f68b6fe02e1b792b1bdef84e3e8e54..6f79331a1b711550d405a1e0961ad8b98dfea59f 100644 (file)
--- a/src/osd/OSD.cc
+++ b/src/osd/OSD.cc
@@ -353,9 +353,11 @@ void OSDService::identify_splits_and_merges(
           << " pg_nums " << p->second << dendl;
   deque<spg_t> queue;
   queue.push_back(pgid);
+  set<spg_t> did;
   while (!queue.empty()) {
     auto cur = queue.front();
     queue.pop_front();
+    did.insert(cur);
     unsigned pgnum = old_pgnum;
     for (auto q = p->second.lower_bound(old_map->get_epoch());
         q != p->second.end() &&
@@ -371,7 +373,8 @@ void OSDService::identify_splits_and_merges(
                     << " children " << children << dendl;
            for (auto i : children) {
              split_children->insert(make_pair(i, q->first));
-             queue.push_back(i);
+              if (!did.count(i))
+               queue.push_back(i);
            }
          }
        } else if (cur.ps() < q->second) {
@@ -401,8 +404,15 @@ void OSDService::identify_splits_and_merges(
                       << " is merge source, target " << parent
                       << ", source(s) " << children << dendl;
              merge_pgs->insert(make_pair(parent, q->first));
+              if (!did.count(parent)) {
+                // queue (and re-scan) parent in case it might not exist yet
+                // and there are some future splits pending on it
+                queue.push_back(parent);
+              }
              for (auto c : children) {
                merge_pgs->insert(make_pair(c, q->first));
+                if (!did.count(c))
+                  queue.push_back(c);
              }
            }
          } else {
@@ -418,6 +428,8 @@ void OSDService::identify_splits_and_merges(
                     << " is merge target, source " << children << dendl;
            for (auto c : children) {
              merge_pgs->insert(make_pair(c, q->first));
+              if (!did.count(c))
+                queue.push_back(c);
            }
            merge_pgs->insert(make_pair(cur, q->first));
          }