From: Xuehan Xu <xxhdx1985126@163.com>
Date: Sun, 12 Apr 2020 04:04:40 +0000 (+0800)
Subject: crimson/net: wait for wandering connections to close when shutting down
X-Git-Tag: v16.1.0~2096^2~2
X-Git-Url: http://git-server-git.apps.pok.os.sepia.ceph.com/?a=commitdiff_plain;h=02c0285c791b8d95c1f895f519b812c9d3bb1cec;p=ceph.git

crimson/net: wait for wandering connections to close when shutting down

Signed-off-by: Xuehan Xu <xxhdx1985126@163.com>
---

diff --git a/src/crimson/net/Protocol.cc b/src/crimson/net/Protocol.cc
index cb0ffab9ef44..78578a5959aa 100644
--- a/src/crimson/net/Protocol.cc
+++ b/src/crimson/net/Protocol.cc
@@ -51,6 +51,7 @@ void Protocol::close(bool dispatch_reset,
   // unregister_conn() drops a reference, so hold another until completion
   auto cleanup = [conn_ref = conn.shared_from_this(), this] {
       logger().debug("{} closed!", conn);
+      on_closed();
 #ifdef UNIT_TESTS_BUILT
       is_closed_clean = true;
       if (conn.interceptor) {
diff --git a/src/crimson/net/Protocol.h b/src/crimson/net/Protocol.h
index 73f64c51359c..3deb706acb4e 100644
--- a/src/crimson/net/Protocol.h
+++ b/src/crimson/net/Protocol.h
@@ -64,6 +64,8 @@ class Protocol {
 
   virtual void notify_write() {};
 
+  virtual void on_closed() {}
+
  public:
   const proto_t proto_type;
   SocketRef socket;
diff --git a/src/crimson/net/ProtocolV2.cc b/src/crimson/net/ProtocolV2.cc
index 89cdbad792ac..05f535829147 100644
--- a/src/crimson/net/ProtocolV2.cc
+++ b/src/crimson/net/ProtocolV2.cc
@@ -2160,10 +2160,19 @@ void ProtocolV2::trigger_close()
   }
 
   protocol_timer.cancel();
-
+  messenger.closing_conn(
+      seastar::static_pointer_cast<SocketConnection>(
+	conn.shared_from_this()));
   trigger_state(state_t::CLOSING, write_state_t::drop, false);
 }
 
+void ProtocolV2::on_closed()
+{
+  messenger.closed_conn(
+      seastar::static_pointer_cast<SocketConnection>(
+	conn.shared_from_this()));
+}
+
 void ProtocolV2::print(std::ostream& out) const
 {
   out << conn;
diff --git a/src/crimson/net/ProtocolV2.h b/src/crimson/net/ProtocolV2.h
index 5224d5a39732..5f0aef78fc29 100644
--- a/src/crimson/net/ProtocolV2.h
+++ b/src/crimson/net/ProtocolV2.h
@@ -19,6 +19,7 @@ class ProtocolV2 final : public Protocol {
   ~ProtocolV2() override;
   void print(std::ostream&) const final;
  private:
+  void on_closed() override;
   bool is_connected() const override;
 
   void start_connect(const entity_addr_t& peer_addr,
diff --git a/src/crimson/net/SocketMessenger.cc b/src/crimson/net/SocketMessenger.cc
index c3ec04f13931..f3d23f4aae67 100644
--- a/src/crimson/net/SocketMessenger.cc
+++ b/src/crimson/net/SocketMessenger.cc
@@ -174,6 +174,10 @@ seastar::future<> SocketMessenger::shutdown()
     return seastar::parallel_for_each(connections, [] (auto conn) {
       return conn.second->close_clean(false);
     });
+  }).then([this] {
+    return seastar::parallel_for_each(closing_conns, [] (auto conn) {
+      return conn->close_clean(false);
+    });
   }).then([this] {
     ceph_assert(connections.empty());
     shutdown_promise.set_value();
@@ -306,6 +310,23 @@ void SocketMessenger::unregister_conn(SocketConnectionRef conn)
   connections.erase(found);
 }
 
+void SocketMessenger::closing_conn(SocketConnectionRef conn)
+{
+  closing_conns.push_back(conn);
+}
+
+void SocketMessenger::closed_conn(SocketConnectionRef conn)
+{
+  for (auto it = closing_conns.begin();
+       it != closing_conns.end();) {
+    if (*it == conn) {
+      it = closing_conns.erase(it);
+    } else {
+      it++;
+    }
+  }
+}
+
 seastar::future<uint32_t>
 SocketMessenger::get_global_seq(uint32_t old)
 {
diff --git a/src/crimson/net/SocketMessenger.h b/src/crimson/net/SocketMessenger.h
index 684d62961166..e86a44d6719f 100644
--- a/src/crimson/net/SocketMessenger.h
+++ b/src/crimson/net/SocketMessenger.h
@@ -46,6 +46,7 @@ class SocketMessenger final : public Messenger {
   ChainedDispatchersRef dispatchers;
   std::map<entity_addr_t, SocketConnectionRef> connections;
   std::set<SocketConnectionRef> accepting_conns;
+  std::vector<SocketConnectionRef> closing_conns;
   ceph::net::PolicySet<Throttle> policy_set;
   // Distinguish messengers with meaningful names for debugging
   const std::string logic_name;
@@ -119,6 +120,8 @@ class SocketMessenger final : public Messenger {
   void unaccept_conn(SocketConnectionRef);
   void register_conn(SocketConnectionRef);
   void unregister_conn(SocketConnectionRef);
+  void closing_conn(SocketConnectionRef);
+  void closed_conn(SocketConnectionRef);
   seastar::shard_id shard_id() const {
     assert(seastar::this_shard_id() == master_sid);
     return master_sid;