leizhiyuan opened a new pull request, #1454: URL: https://github.com/apache/rocketmq-clients/pull/1454
### Which Issue(s) This PR Fixes Fixes #1448 ### Brief Description `SessionManager` held one `tokio::Mutex` over the session map while connecting a new endpoint, starting its telemetry stream, syncing settings and reconnecting, so one slow broker stalled every RPC to every endpoint. Java looks sessions up under a read lock and never blocks under the write lock. - `session_map` is now a `parking_lot::Mutex<HashMap<String, Arc<tokio::Mutex<Session>>>>`. The map lock is synchronous and never held across `.await`; each session has its own async lock, so I/O on one session only waits for that session. - `get_or_create_session`: lookup without I/O; creation outside the map lock, single-flighted per endpoint (a per-endpoint creation lock), so concurrent callers connect once. A failed creation registers nothing. - `sync_settings`, `clear_dead_and_idle_sessions`: work on a snapshot; a session is only removed if it is still the registered one (`Arc::ptr_eq`). - `reconnect_session`: reconnects under the session's own lock instead of taking the session out of the map, so the session stays registered (and visible to other callers) during the reconnect, as before on failure. #### Notes for reviewers - Behavior of eviction / reconnect / settings sync is unchanged; the existing tests for them pass with only the map access in the tests adapted. - No conflict with the other PRs of #1445. ### How Did You Test This Change? **Unit tests**: - `test_creating_a_slow_session_does_not_block_other_endpoints`: `Session::new` of one endpoint takes 500 ms; looking up an existing session of another endpoint must take < 100 ms (master: 449 ms). - `test_sync_settings_does_not_block_other_endpoints`: `update_settings` of one session takes 500 ms; lookup of another session must take < 100 ms (master: 449 ms). - Guards for the new code: `test_concurrent_creation_of_the_same_session_connects_once` (4 concurrent callers, 1 `Session::new`), `test_failed_session_creation_is_not_registered`. **End-to-end** (mock proxy over real gRPC, 8 senders to a healthy broker, one send to a new broker whose telemetry answers after 3 s, 3 runs): | healthy broker, during those 3 s | `master` | this PR | |---|---|---| | sends completed | 47 | **59,966** | | send latency p99 | 3,002 ms | **0.62 ms** | ``` $ cargo test # lib: 154 passed, 0 failed (master: 150) $ cargo clippy --all-features -- -D warnings # clean $ cargo fmt --check # clean $ RUSTDOCFLAGS='--cfg docsrs' cargo doc --no-deps --all-features # ok $ cargo build --examples --all-features # ok $ cargo +1.74.0 check --all-features --tests --examples # ok (MSRV) ``` --- Part of #1445 (overview, end-to-end benchmark and merge order). -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
