rgw: data sync uses yield_spawn_window()

restore concurrency to RGWDataSyncShardCR by replacing
drain_all_but_stack_cb() with yield_spawn_window()

Fixes: https://tracker.ceph.com/issues/55000
Signed-off-by: Casey Bodley <cbodley@redhat.com>

diff --git a/src/rgw/rgw_data_sync.cc b/src/rgw/rgw_data_sync.cc
index e9bc1dd7b9c5d5d930ea6f81779d82ee6566e030..7bd1d338488565d6e076d4048abc1bc60e8f7989 100644 (file)
--- a/src/rgw/rgw_data_sync.cc
+++ b/src/rgw/rgw_data_sync.cc
@@ -1592,17 +1592,11 @@ public:
             tn->log(0, SSTR("ERROR: cannot start syncing " << iter->first << ". Duplicate entry?"));
           } else {
             // fetch remote and write locally
-            spawn(sync_single_entry(source_bs, iter->first, iter->first,
-                                    entry_timestamp, false), false);
+            yield_spawn_window(sync_single_entry(source_bs, iter->first, iter->first,
+                                                 entry_timestamp, false),
+                               spawn_window, std::nullopt);
           }
           sync_marker.marker = iter->first;
-
-          drain_all_but_stack_cb(lease_stack.get(),
-                                 [&](uint64_t stack_id, int ret) {
-                                   if (ret < 0) {
-                                     tn->log(10, "a sync operation returned error");
-                                   }
-                                 });
         }
       } while (omapvals->more);
       omapvals.reset();
@@ -1747,16 +1741,10 @@ public:
           if (!marker_tracker->start(log_iter->log_id, 0, log_iter->log_timestamp)) {
             tn->log(0, SSTR("ERROR: cannot start syncing " << log_iter->log_id << ". Duplicate entry?"));
           } else {
-            spawn(sync_single_entry(source_bs, log_iter->entry.key, log_iter->log_id,
-                                    log_iter->log_timestamp, false), false);
+            yield_spawn_window(sync_single_entry(source_bs, log_iter->entry.key, log_iter->log_id,
+                                                 log_iter->log_timestamp, false),
+                               spawn_window, std::nullopt);
           }
-
-          drain_all_but_stack_cb(lease_stack.get(),
-                                 [&](uint64_t stack_id, int ret) {
-                                   if (ret < 0) {
-                                     tn->log(10, "a sync operation returned error");
-                                   }
-                                 });
         }
 
         tn->log(20, SSTR("shard_id=" << shard_id << " sync_marker=" << sync_marker.marker