RepoMirrors/ceph
1
0
Fork 0
mirror of https://github.com/ceph/ceph synced 2025-02-21 18:17:42 +00:00
Code Issues Packages Projects Releases Wiki Activity

msg: reset ProtocolV2's frame assembler in appropriate thread

Browse Source 
`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>
This commit is contained in:
Radoslaw Zarzynski 2022-07-27 12:10:05 +00:00
parent bccd2dbb20
commit c04256bab8
1 changed files with 8 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
src/msg/async/ProtocolV2.cc
View File

@ -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();