msg/async: add "bool unregistered" to avoid check whether in deleted_…

…conns.

We don't use deleted_lock to protect func is_unregistered. This because if
race occur, func send_message still check state of AsyncConnection and
skip this message.

Signed-off-by: Jianpeng Ma <jianpeng.ma@intel.com>

src/msg/async/AsyncConnection.h

@@ -110,6 +110,7 @@ class AsyncConnection : public Connection {
   AsyncConnection(CephContext *cct, AsyncMessenger *m, DispatchQueue *q,
 		  Worker *w, bool is_msgr2, bool local);
   ~AsyncConnection() override;
+  bool unregisted = false;
 public:
   void maybe_start_delay_thread();
 
@@ -139,6 +140,14 @@ class AsyncConnection : public Connection {
 
   int get_con_mode() const override;
 
+  bool is_unregistered() const {
+    return unregisted;
+  }
+
+  void unregister() {
+    unregisted = true;
+  }
+
  private:
   enum {
     STATE_NONE,

src/msg/async/AsyncMessenger.h

@@ -319,10 +319,13 @@ class AsyncMessenger : public SimplePolicyMessenger {
     }
 
     // lazy delete, see "deleted_conns"
-    std::lock_guard l{deleted_lock};
-    if (deleted_conns.erase(p->second)) {
-      conns.erase(p);
-      return nullref;
+    // don't worry omit, Connection::send_message can handle this case.
+    if (p->second->is_unregistered()) {
+      std::lock_guard l{deleted_lock};
+      if (deleted_conns.erase(p->second)) {
+	conns.erase(p);
+	return nullref;
+      }
     }
 
     return p->second;
@@ -410,6 +413,7 @@ class AsyncMessenger : public SimplePolicyMessenger {
     std::lock_guard l{deleted_lock};
     conn->get_perf_counter()->dec(l_msgr_active_connections);
     deleted_conns.emplace(std::move(conn));
+    conn->unregister();
 
     if (deleted_conns.size() >= ReapDeadConnectionThreshold) {
       local_worker->center.dispatch_event_external(reap_handler);