From e2bc6ad7346e496c87de32cb6cb0634337763887 Mon Sep 17 00:00:00 2001
From: Kefu Chai <kchai@redhat.com>
Date: Fri, 8 Mar 2019 21:01:07 +0800
Subject: [PATCH] crimson/osd: advance pg in consume_map()

Signed-off-by: Kefu Chai <kchai@redhat.com>
---
 src/crimson/osd/osd.cc | 25 +++++++++++++++----------
 src/crimson/osd/osd.h  |  2 +-
 2 files changed, 16 insertions(+), 11 deletions(-)

diff --git a/src/crimson/osd/osd.cc b/src/crimson/osd/osd.cc
index 9a2d9bb1919..c291d1de907 100644
--- a/src/crimson/osd/osd.cc
+++ b/src/crimson/osd/osd.cc
@@ -622,7 +622,8 @@ seastar::future<> OSD::committed_osd_maps(version_t first,
       }
     }
     // yay!
-    consume_map(osdmap->get_epoch());
+    return consume_map(osdmap->get_epoch());
+  }).then([m, this] {
     if (state.is_active()) {
       logger().info("osd.{}: now active", whoami);
       if (!osdmap->exists(whoami)) {
@@ -804,17 +805,20 @@ seastar::future<> OSD::handle_pg_log(ceph::net::ConnectionRef conn,
   return do_peering_event(m->get_spg(), std::move(evt));
 }
 
-
-void OSD::consume_map(epoch_t epoch)
+seastar::future<> OSD::consume_map(epoch_t epoch)
 {
   // todo: m-to-n: broadcast this news to all shards
-  auto first = waiting_peering.lower_bound(epoch);
-  auto last = waiting_peering.end();
-  std::for_each(first, last,
-                [epoch, this](auto& blocked_requests) {
-                  blocked_requests.second.set_value(epoch);
-                });
-  waiting_peering.erase(first, last);
+  return seastar::parallel_for_each(pgs.begin(), pgs.end(), [=](auto& pg) {
+    return advance_pg_to(pg.second, epoch);
+  }).then([epoch, this] {
+    auto first = waiting_peering.lower_bound(epoch);
+    auto last = waiting_peering.end();
+    std::for_each(first, last, [epoch, this](auto& blocked_requests) {
+      blocked_requests.second.set_value(epoch);
+    });
+    waiting_peering.erase(first, last);
+    return seastar::now();
+  });
 }
 
 seastar::future<>
@@ -831,6 +835,7 @@ OSD::do_peering_event(spg_t pgid,
         return _send_alive(pg->get_need_up_thru());
     });
   } else {
+    logger().warn("pg not found: {}", pgid);
     // todo: handle_pg_query_nopg()
     return seastar::now();
   }
diff --git a/src/crimson/osd/osd.h b/src/crimson/osd/osd.h
index e6677c90353..fbd0341261d 100644
--- a/src/crimson/osd/osd.h
+++ b/src/crimson/osd/osd.h
@@ -151,7 +151,7 @@ private:
   waiting_peering_t waiting_peering;
   // wait for an osdmap whose epoch is greater or equal to given epoch
   seastar::future<epoch_t> wait_for_map(epoch_t epoch);
-  void consume_map(epoch_t epoch);
+  seastar::future<> consume_map(epoch_t epoch);
   seastar::future<> do_peering_event(spg_t pgid,
 				     std::unique_ptr<PGPeeringEvent> evt);
   seastar::future<> advance_pg_to(Ref<PG> pg, epoch_t to);
-- 
2.39.5