msg: reset ProtocolV2's frame assembler in appropriate thread

`set_is_rev1()` actually resets the internal state of `FrameAssembler`,
so -- to avoid racing -- it should be performed by the same thread that
takes care about executing `ProtocolV2::write_message()`.

```cppp
void set_is_rev1(bool is_rev1) {
  m_descs.clear();
  m_flags = 0;
  m_is_rev1 = is_rev1;
}
```

See comments in the tracker for analysis.

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

diff --git a/src/msg/async/ProtocolV2.cc b/src/msg/async/ProtocolV2.cc
index 9cc56edd8bb7166dbe47c6e511fb5a411b064937..1713e1aefdf50c8f07e095810e6df9154b5b86ed 100644 (file)
--- a/src/msg/async/ProtocolV2.cc
+++ b/src/msg/async/ProtocolV2.cc
@@ -2723,14 +2723,11 @@ CtPtr ProtocolV2::reuse_connection(const AsyncConnectionRef& existing,
   exproto->pre_auth.enabled = false;
 
   if (!reconnecting) {
-    exproto->peer_supported_features = peer_supported_features;
-    exproto->tx_frame_asm.set_is_rev1(tx_frame_asm.get_is_rev1());
-    exproto->rx_frame_asm.set_is_rev1(rx_frame_asm.get_is_rev1());
-
     exproto->client_cookie = client_cookie;
     exproto->peer_name = peer_name;
     exproto->connection_features = connection_features;
     existing->set_features(connection_features);
+    exproto->peer_supported_features = peer_supported_features;
   }
   exproto->peer_global_seq = peer_global_seq;
 
@@ -2773,6 +2770,9 @@ CtPtr ProtocolV2::reuse_connection(const AsyncConnectionRef& existing,
         new_worker,
         new_center,
         exproto,
+        reconnecting=reconnecting,
+        tx_is_rev1=tx_frame_asm.get_is_rev1(),
+        rx_is_rev1=rx_frame_asm.get_is_rev1(),
         temp_stream_handlers=std::move(temp_stream_handlers),
         temp_compression_handlers=std::move(temp_compression_handlers)
       ](ConnectedSocket &cs) mutable {
@@ -2790,6 +2790,10 @@ CtPtr ProtocolV2::reuse_connection(const AsyncConnectionRef& existing,
           existing->open_write = false;
           exproto->session_stream_handlers = std::move(temp_stream_handlers);
           exproto->session_compression_handlers = std::move(temp_compression_handlers);
+          if (!reconnecting) {
+            exproto->tx_frame_asm.set_is_rev1(tx_is_rev1);
+            exproto->rx_frame_asm.set_is_rev1(rx_is_rev1);
+          }
           existing->write_lock.unlock();
           if (exproto->state == NONE) {
             existing->shutdown_socket();