From d047763f3ab7abad3b11f2675dea63c1a5ffa343 Mon Sep 17 00:00:00 2001
From: Yehuda Sadeh <yehuda@redhat.com>
Date: Fri, 30 Oct 2015 15:35:57 -0700
Subject: [PATCH] rgw: move children collection to other method

and call it again instead of just draining at the end

Signed-off-by: Yehuda Sadeh <yehuda@redhat.com>
---
 src/rgw/rgw_sync.cc | 102 ++++++++++++++++++++++++--------------------
 1 file changed, 55 insertions(+), 47 deletions(-)

diff --git a/src/rgw/rgw_sync.cc b/src/rgw/rgw_sync.cc
index b7173b57c04df..67f08e0d8a709 100644
--- a/src/rgw/rgw_sync.cc
+++ b/src/rgw/rgw_sync.cc
@@ -1055,6 +1055,56 @@ public:
     return 0;
   }
 
+  void collect_children()
+  {
+    int child_ret;
+    RGWCoroutinesStack *child;
+    while (collect_next(&child_ret, &child)) {
+      map<RGWCoroutinesStack *, string>::iterator iter = stack_to_pos.find(child);
+      if (iter == stack_to_pos.end()) {
+        /* some other stack that we don't care about */
+        continue;
+      }
+
+      string& pos = iter->second;
+
+      if (child_ret < 0) {
+        ldout(sync_env->cct, 0) << *this << ": child operation stack=" << child << " entry=" << pos << " returned " << child_ret << dendl;
+      }
+
+      map<string, string>::iterator prev_iter = pos_to_prev.find(pos);
+      assert(prev_iter != pos_to_prev.end());
+
+      /*
+       * we should get -EAGAIN for transient errors, for which we want to retry, so we don't
+       * update the marker and abort. We'll get called again for these. Permanent errors will be
+       * handled by marking the entry at the error log shard, so that we retry on it separately
+       */
+      if (child_ret == -EAGAIN) {
+        can_adjust_marker = false;
+      }
+
+      if (pos_to_prev.size() == 1) {
+        if (can_adjust_marker) {
+          sync_marker.marker = pos;
+        }
+        pos_to_prev.erase(prev_iter);
+      } else {
+        assert(pos_to_prev.size() > 1);
+        pos_to_prev.erase(prev_iter);
+        prev_iter = pos_to_prev.begin();
+        if (can_adjust_marker) {
+          sync_marker.marker = prev_iter->second;
+        }
+      }
+
+      ldout(sync_env->cct, 0) << *this << ": adjusting marker pos=" << sync_marker.marker << dendl;
+      stack_to_pos.erase(iter);
+
+      child->put();
+    }
+  }
+
   int full_sync() {
 #define OMAP_GET_MAX_ENTRIES 100
     int max_entries = OMAP_GET_MAX_ENTRIES;
@@ -1124,55 +1174,13 @@ public:
           }
           marker = iter->first;
         }
-        RGWCoroutinesStack *child;
-        int child_ret;
-        while (collect_next(&child_ret, &child)) {
-          map<RGWCoroutinesStack *, string>::iterator iter = stack_to_pos.find(child);
-          if (iter == stack_to_pos.end()) {
-            /* some other stack that we don't care about */
-            continue;
-          }
-
-          string& pos = iter->second;
-
-          if (child_ret < 0) {
-            ldout(sync_env->cct, 0) << *this << ": child operation stack=" << child << " entry=" << pos << " returned " << child_ret << dendl;
-          }
-
-          map<string, string>::iterator prev_iter = pos_to_prev.find(pos);
-          assert(prev_iter != pos_to_prev.end());
-
-          /*
-           * we should get -EAGAIN for transient errors, for which we want to retry, so we don't
-           * update the marker and abort. We'll get called again for these. Permanent errors will be
-           * handled by marking the entry at the error log shard, so that we retry on it separately
-           */
-          if (child_ret == -EAGAIN) {
-            can_adjust_marker = false;
-          }
-
-          if (pos_to_prev.size() == 1) {
-            if (can_adjust_marker) {
-              sync_marker.marker = pos;
-            }
-            pos_to_prev.erase(prev_iter);
-          } else {
-            assert(pos_to_prev.size() > 1);
-            pos_to_prev.erase(prev_iter);
-            prev_iter = pos_to_prev.begin();
-            if (can_adjust_marker) {
-              sync_marker.marker = prev_iter->second;
-            }
-          }
-
-          ldout(sync_env->cct, 0) << *this << ": adjusting marker pos=" << sync_marker.marker << dendl;
-          stack_to_pos.erase(iter);
-
-          child->put();
-        }
+        collect_children();
       } while ((int)entries.size() == max_entries && can_adjust_marker);
 
-      drain_all_but(1);
+      while (num_spawned() > 1) {
+        yield wait_for_child();
+        collect_children();
+      }
 
       if (!lost_lock) {
         yield {
-- 
2.39.5