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>

diff --git a/src/osd/OSD.cc b/src/osd/OSD.cc
index 40afda69d33ea7184ea0fffe95e6f0b8195a677a..3f7b28d3b554d5f131fefbd588e3d98c22cbf9e4 100644 (file)
--- a/src/osd/OSD.cc
+++ b/src/osd/OSD.cc
@@ -346,9 +346,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() &&
@@ -364,7 +366,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) {
@@ -394,8 +397,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 {
@@ -411,6 +421,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));
          }