AsyncConnection: queue_reset after stopped

Signed-off-by: Haomai Wang <haomai@xsky.com>

diff --git a/src/msg/async/AsyncConnection.cc b/src/msg/async/AsyncConnection.cc
index 2591b1017dea217d5512d22245ba68552c32cd98..84e74ca44619e7d4c5757a3ad3ce9489b2a39d83 100644 (file)
--- a/src/msg/async/AsyncConnection.cc
+++ b/src/msg/async/AsyncConnection.cc
@@ -1754,14 +1754,12 @@ ssize_t AsyncConnection::handle_connect_msg(ceph_msg_connect &connect, bufferlis
   inject_delay();
   if (existing->policy.lossy) {
     // disconnect from the Connection
-    existing->dispatch_queue->queue_reset(this);
     ldout(async_msgr->cct, 1) << __func__ << " replacing on lossy channel, failing existing" << dendl;
     existing->_stop();
+    existing->dispatch_queue->queue_reset(existing.get());
   } else {
     assert(can_write == WriteStatus::NOWRITE);
     existing->write_lock.Lock(true);
-    // queue a reset on the new connection, which we're dumping for the old
-    dispatch_queue->queue_reset(this);
 
     // reset the in_seq if this is a hard reset from peer,
     // otherwise we respect our original connection's value
@@ -1808,6 +1806,8 @@ ssize_t AsyncConnection::handle_connect_msg(ceph_msg_connect &connect, bufferlis
 
     ldout(async_msgr->cct, 1) << __func__ << " stop myself to swap existing" << dendl;
     _stop();
+    // queue a reset on the new connection, which we're dumping for the old
+    dispatch_queue->queue_reset(this);
     existing->lock.Unlock();
     return 0;
   }
@@ -2083,8 +2083,8 @@ void AsyncConnection::fault()
 
   if (policy.lossy && !(state >= STATE_CONNECTING && state < STATE_CONNECTING_READY)) {
     ldout(async_msgr->cct, 1) << __func__ << " on lossy channel, failing" << dendl;
-    dispatch_queue->queue_reset(this);
     _stop();
+    dispatch_queue->queue_reset(this);
     return ;
   }
 
@@ -2112,10 +2112,9 @@ void AsyncConnection::fault()
       state >=STATE_ACCEPTING && state <= STATE_ACCEPTING_WAIT_CONNECT_MSG_AUTH) {
     ldout(async_msgr->cct, 0) << __func__ << " with nothing to send and in the half "
                               << " accept state just closed" << dendl;
-    dispatch_queue->queue_reset(this);
-
     write_lock.Unlock();
     _stop();
+    dispatch_queue->queue_reset(this);
     return ;
   }
   reset_recv_state();

diff --git a/src/msg/async/AsyncConnection.h b/src/msg/async/AsyncConnection.h
index 741085864b6e1b6203847c9a540543fd408a942c..a4a14b55fb15f96bdf2d38572c3d3b14d285e7f5 100644 (file)
--- a/src/msg/async/AsyncConnection.h
+++ b/src/msg/async/AsyncConnection.h
@@ -369,10 +369,11 @@ class AsyncConnection : public Connection {
   void local_deliver();
   void stop() {
     lock.Lock();
-    if (state != STATE_CLOSED)
-      dispatch_queue->queue_reset(this);
+    bool need_queue_reset = (state != STATE_CLOSED);
     lock.Unlock();
     mark_down();
+    if (need_queue_reset)
+      dispatch_queue->queue_reset(this);
   }
   void cleanup_handler() {
     delete read_handler;