msg/async: perform recv reset immediately if called inside EC.

Fixes: https://tracker.ceph.com/issues/43808
Signed-off-by: Radoslaw Zarzynski <rzarzyns@redhat.com>

diff --git a/src/msg/async/ProtocolV1.cc b/src/msg/async/ProtocolV1.cc
index cae647edc93aa11aa935fa74d13f46cfe058524b..a082d187a7f5e6513dd9bc53136d2c47126fd23c 100644 (file)
--- a/src/msg/async/ProtocolV1.cc
+++ b/src/msg/async/ProtocolV1.cc
@@ -1251,6 +1251,15 @@ void ProtocolV1::discard_out_queue() {
   write_in_progress = false;
 }
 
+void ProtocolV1::reset_security()
+{
+  ldout(cct, 5) << __func__ << dendl;
+
+  auth_meta.reset(new AuthConnectionMeta);
+  authorizer_more.clear();
+  session_security.reset();
+}
+
 void ProtocolV1::reset_recv_state()
 {
   ldout(cct, 5) << __func__ << dendl;
@@ -1259,18 +1268,18 @@ void ProtocolV1::reset_recv_state()
   // We need to do the warp because holding `write_lock` is not
   // enough as `write_event()` releases it just before calling
   // `write_message()`. `submit_to()` here is NOT blocking.
-  connection->center->submit_to(connection->center->get_id(), [this] {
-    ldout(cct, 5) << "reset_recv_state reseting security handlers" << dendl;
-
-    // Possibly unnecessary. See the comment in `deactivate_existing`.
-    std::lock_guard<std::mutex> l(connection->lock);
-    std::lock_guard<std::mutex> wl(connection->write_lock);
-
-    // clean up state internal variables and states
-    auth_meta.reset(new AuthConnectionMeta);
-    authorizer_more.clear();
-    session_security.reset();
-  }, /* nowait = */true);
+  if (!connection->center->in_thread()) {
+    connection->center->submit_to(connection->center->get_id(), [this] {
+      ldout(cct, 5) << "reset_recv_state (warped) reseting security handlers"
+                    << dendl;
+      // Possibly unnecessary. See the comment in `deactivate_existing`.
+      std::lock_guard<std::mutex> l(connection->lock);
+      std::lock_guard<std::mutex> wl(connection->write_lock);
+      reset_security();
+    }, /* nowait = */true);
+  } else {
+    reset_security();
+  }
 
   // clean read and write callbacks
   connection->pendingReadLen.reset();

diff --git a/src/msg/async/ProtocolV1.h b/src/msg/async/ProtocolV1.h
index 72b707fe2a6cc1dc11431dfe8909a8aca452103b..3f63a68b2e56f716dc753b144b3a8e80f1a33c35 100644 (file)
--- a/src/msg/async/ProtocolV1.h
+++ b/src/msg/async/ProtocolV1.h
@@ -204,6 +204,7 @@ protected:
   void discard_out_queue();
 
   void reset_recv_state();
+  void reset_security();
 
   ostream &_conn_prefix(std::ostream *_dout);
 

diff --git a/src/msg/async/ProtocolV2.cc b/src/msg/async/ProtocolV2.cc
index d9a5a72aeccf6c3cb81c287af8e18b8acb7e3c6d..f258eb12ee074447e164d5329b68764621ee17cd 100644 (file)
--- a/src/msg/async/ProtocolV2.cc
+++ b/src/msg/async/ProtocolV2.cc
@@ -222,27 +222,36 @@ uint64_t ProtocolV2::discard_requeued_up_to(uint64_t out_seq, uint64_t seq) {
   return count;
 }
 
+void ProtocolV2::reset_security() {
+  ldout(cct, 5) << __func__ << dendl;
+
+  auth_meta.reset(new AuthConnectionMeta);
+  session_stream_handlers.rx.reset(nullptr);
+  session_stream_handlers.tx.reset(nullptr);
+  pre_auth.rxbuf.clear();
+  pre_auth.txbuf.clear();
+}
+
 // it's expected the `write_lock` is held while calling this method.
 void ProtocolV2::reset_recv_state() {
   ldout(cct, 5) << __func__ << dendl;
 
-  // execute in the same thread that uses the rx/tx handlers. We need
-  // to do the warp because holding `write_lock` is not enough as
-  // `write_event()` unlocks it just before calling `write_message()`.
-  // `submit_to()` here is NOT blocking.
-  connection->center->submit_to(connection->center->get_id(), [this] {
-    ldout(cct, 5) << "reset_recv_state reseting crypto handlers" << dendl;
-
-    // Possibly unnecessary. See the comment in `deactivate_existing`.
-    std::lock_guard<std::mutex> l(connection->lock);
-    std::lock_guard<std::mutex> wl(connection->write_lock);
-
-    auth_meta.reset(new AuthConnectionMeta);
-    session_stream_handlers.rx.reset(nullptr);
-    session_stream_handlers.tx.reset(nullptr);
-    pre_auth.rxbuf.clear();
-    pre_auth.txbuf.clear();
-  }, /* nowait = */true);
+  if (!connection->center->in_thread()) {
+    // execute in the same thread that uses the rx/tx handlers. We need
+    // to do the warp because holding `write_lock` is not enough as
+    // `write_event()` unlocks it just before calling `write_message()`.
+    // `submit_to()` here is NOT blocking.
+    connection->center->submit_to(connection->center->get_id(), [this] {
+      ldout(cct, 5) << "reset_recv_state (warped) reseting crypto handlers"
+                    << dendl;
+      // Possibly unnecessary. See the comment in `deactivate_existing`.
+      std::lock_guard<std::mutex> l(connection->lock);
+      std::lock_guard<std::mutex> wl(connection->write_lock);
+      reset_security();
+    }, /* nowait = */true);
+  } else {
+    reset_security();
+  }
 
   // clean read and write callbacks
   connection->pendingReadLen.reset();

diff --git a/src/msg/async/ProtocolV2.h b/src/msg/async/ProtocolV2.h
index e5544f987460114bffc3c841f03166d12262892e..2dbe647ae546f5c50b60f7a5c19d969eedd28977 100644 (file)
--- a/src/msg/async/ProtocolV2.h
+++ b/src/msg/async/ProtocolV2.h
@@ -132,6 +132,7 @@ private:
   void requeue_sent();
   uint64_t discard_requeued_up_to(uint64_t out_seq, uint64_t seq);
   void reset_recv_state();
+  void reset_security();
   void reset_throttle();
   Ct<ProtocolV2> *_fault();
   void discard_out_queue();