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 cb0ffab9ef44a107f0b4a523e690a66a03b265fd..78578a5959aa10a6034ec60ae9c08cfd1c1de4e2 100644 (file)
--- 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 73f64c51359cd7b96cb30025a190fe5c809a981c..3deb706acb4eb67abac2770d461f4621cb5a428a 100644 (file)
--- 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 89cdbad792ac1d4d6a05ff115ea3f03c5bbed4e0..05f53582914708f4c82e137f511d345e0efd62cd 100644 (file)
--- 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 5224d5a39732933cb292f4319083dab6ea990010..5f0aef78fc294964ab705957cbbbb297b669b3ef 100644 (file)
--- 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 c3ec04f13931495cf4513d76f2bd77397be00415..f3d23f4aae6700a9c11634d4cddbe1405820f77e 100644 (file)
--- 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 684d6296116651c936937c36abb7a9b113cdaf21..e86a44d6719ff891f75d4e63a0c4fef5c99657b1 100644 (file)
--- 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;