From 3076020a82cc6ea441998cfa145968659baf7090 Mon Sep 17 00:00:00 2001
From: Kefu Chai <kchai@redhat.com>
Date: Thu, 28 Feb 2019 18:13:12 +0800
Subject: [PATCH] crimson/osd: wait osdmap before processing peering evt

the constraint that

> peering messages/events from peers must remain ordered by peer

is still enforced. because, the peering message from a certain peer
will block the succeeding messages from this peer, if the latest osdmap
possessed by OSD is not new enough.

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

diff --git a/src/crimson/osd/osd.cc b/src/crimson/osd/osd.cc
index 7daec781de7..9a2d9bb1919 100644
--- a/src/crimson/osd/osd.cc
+++ b/src/crimson/osd/osd.cc
@@ -621,7 +621,8 @@ seastar::future<> OSD::committed_osd_maps(version_t first,
           std::chrono::seconds(TICK_INTERVAL));
       }
     }
-
+    // yay!
+    consume_map(osdmap->get_epoch());
     if (state.is_active()) {
       logger().info("osd.{}: now active", whoami);
       if (!osdmap->exists(whoami)) {
@@ -803,13 +804,28 @@ 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)
+{
+  // 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);
+}
+
 seastar::future<>
 OSD::do_peering_event(spg_t pgid,
                       std::unique_ptr<PGPeeringEvent> evt)
 {
   if (auto pg = pgs.find(pgid); pg != pgs.end()) {
-    return advance_pg_to(pg->second, osdmap->get_epoch()).then(
-      [pg, evt=std::move(evt)]() mutable {
+    return wait_for_map(evt->get_epoch_sent()).then(
+      [pg=pg->second, this](epoch_t epoch) {
+        return advance_pg_to(pg, epoch);
+    }).then([pg, evt=std::move(evt)]() mutable {
         return pg->second->do_peering_event(std::move(evt));
     }).then([pg=pg->second, this] {
         return _send_alive(pg->get_need_up_thru());
@@ -820,6 +836,17 @@ OSD::do_peering_event(spg_t pgid,
   }
 }
 
+seastar::future<epoch_t> OSD::wait_for_map(epoch_t epoch)
+{
+  const auto mine = osdmap->get_epoch();
+  if (mine >= epoch) {
+    return seastar::make_ready_future<epoch_t>(mine);
+  } else {
+    logger().info("evt epoch is {}, i have {}, will wait", epoch, mine);
+    return waiting_peering[epoch].get_shared_future();
+  }
+}
+
 seastar::future<> OSD::advance_pg_to(Ref<PG> pg, epoch_t to)
 {
   auto from = pg->get_osdmap_epoch();
diff --git a/src/crimson/osd/osd.h b/src/crimson/osd/osd.h
index 6a405652d1b..e6677c90353 100644
--- a/src/crimson/osd/osd.h
+++ b/src/crimson/osd/osd.h
@@ -7,6 +7,7 @@
 #include <seastar/core/future.hh>
 #include <seastar/core/gate.hh>
 #include <seastar/core/shared_ptr.hh>
+#include <seastar/core/shared_future.hh>
 #include <seastar/core/timer.hh>
 
 #include "crimson/common/simple_lru.h"
@@ -142,6 +143,15 @@ private:
                                        version_t last,
                                        Ref<MOSDMap> m);
 
+  // order the promises in descending order of the waited osdmap epoch,
+  // so we can access all the waiters expecting a map whose epoch is less
+  // than a given epoch
+  using waiting_peering_t = std::map<epoch_t, seastar::shared_promise<epoch_t>,
+				     std::greater<epoch_t>>;
+  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<> 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