Fix socket timeouts, connection lifecycle and session pool accounting - #11
Fix socket timeouts, connection lifecycle and session pool accounting#11CritasWang wants to merge 1 commit into
Conversation
13c8e1d to
349836a
Compare
- Bound every blocking socket read/write after the TCP handshake with a new socket_timeout (SO_RCVTIMEO/SO_SNDTIMEO via socket2), applied before the TLS handshake: RPC reads, drop-time closeSession and the TLS handshake can no longer block forever; SO_KEEPALIVE is enabled. connect_timeout is now one total budget per endpoint across all resolved addresses. - Fail over at the handshake level in Session::open (connect + openSession + requestStatementId), mirroring reconnect(). - Mark a connection broken on transport-level failures so is_open() stops lying; pools discard the session instead of handing it out again, and fetch_results (which cannot be retried) also marks it. - SessionPool: spend the acquire_timeout budget on growth/hand-out failures instead of failing instantly, decrement live under the state lock (no lost Condvar wakeups), wake waiters from close() before the blocking closeSession calls, and treat acquire_timeout=Duration::MAX as "wait without deadline" instead of panicking. - Skip the reconnect pacing sleeps for pooled sessions so a pool slot is not held for the full reconnect walk. - Redirect: normalize endpoint matching (case/brackets/loopback), prefer the newest hint when idle sessions hold conflicting hints (process-wide insertion seq), make cache TTL/capacity configurable, and expose TableSessionPool::acquire_for_device. - Add regression tests using silent/fake listeners for every fix above. Closes #4, closes #5, closes #6, closes #7, closes #8, closes #9, closes #10.
349836a to
93271ed
Compare
PDGGK
left a comment
PDGGK
left a comment
There was a problem hiding this comment.
Thanks for asking me to look, and for taking the whole of the earlier review. I read the change at
93271ed2 and built a fake-server harness so I could run things rather than argue from the source.
Four items I would fix before merging, three I would only raise. Each one below was reproduced at
runtime with a control that passes, so a null result would have shown up as a failing control
instead of a quiet pass.
Some things I went looking for and did not find. The shared connect budget holds: one deadline for
the endpoint, each resolved address getting the remainder. The endpoint loop in Session::open
drops a connection before trying the next one, and a session left half-open by a failed
requestStatementId gets cleaned up server-side when the TCP connection goes away. I also expected
a gap around the pool's database catch-up, and found the comment at pool.rs:403 explaining that
new sessions already start in the pool's current database, so the first hand-out needs no USE.
That one is closed.
Before merging
1. broken = true does not stop the session from using that connection
with_retry and fetch_results set self.broken = true, and the pool honours it: release()
gates on is_open(), which is connection.is_some() && !broken. Inside the session it has no
effect, because connection_mut() only checks self.connection.as_mut(). It never looks at
broken. Two paths then use a connection the code has already decided is unusable:
close_query(), whichSessionDataSet's destructor calls atdataset.rs:209, goes through
connection_mut()and sendscloseOperation.Session::close()takes the connection and sendscloseSessionwhateverbrokensays.
I ran this against a fake server with socket_timeout = 400ms. The server answers openSession
and requestStatementId normally, then answers fetchResultsV2 with a valid frame header claiming
1000 bytes, sends 10 of them, and stops.
fetch_results -> Err(transport error) 401.124 ms
is_open() after that -> false
close_query -> wrote 68 bytes 400.133 ms
CONTROL (honest peer) -> Ok 112.792 µs
The test asserts that nothing arrived between the server going quiet and close_query running, so
those 68 bytes can only be the closeOperation, written onto a stream sitting mid-frame. It then
waits out a second full timeout. With socket_timeout = None that wait has no bound.
The same run turned up a smaller inconsistency. With auto-reconnect off, is_open() returns false
while open() returns session already open, because open() at session.rs:253 checks
connection.is_some() and is_open() checks is_some() && !broken.
What I would do is put the decision in one place:
fn mark_broken(&mut self) {
self.connection = None;
self.session_id = -1;
self.statement_id = -1;
self.broken = true;
}called from everywhere a connection is written off, including fetch_results, with
connection_mut() rejecting broken as a second line of defence. Dropping the TCP connection
without sending closeSession is safe: RPCServiceThriftHandler.deleteContext calls
handleClientExit, which closes the current session (ClientRPCServiceImpl.java:3619). I checked
that on apache/iotdb master and did not check the older servers listed in COMPATIBILITY.md.
The harness is 273 lines, two cases, nothing beyond the crate's existing dependencies. I can open
it as a PR against your branch if that saves you rebuilding it.
2. After open_session() fails, acquire() parks without re-checking
The growth path in pool.rs:
// Spend the remaining acquire_timeout budget waiting
// for a release instead of failing immediately; the
// loop re-checks closed/idle/deadline at the top.
state = self.wait_for_change(state, deadline);The comment describes the behaviour I would want, but the call parks first and the re-check only
happens after it wakes. open_session() runs with the lock released, so a notify_one from a
release, or the notify_all from close(), arrives while nobody is waiting and is dropped. The
thread parks anyway, with a session already idle or the pool already closed.
With acquire_timeout = 6s, a thread inside open_session() against an unroutable endpoint, and
close() called 250 ms in, acquire() returned after 6.000320959 s. It ran to its own deadline
and ignored the close entirely. Under Duration::MAX there would have been nothing to wake it at
all.
Checking closed, then idle, then the deadline before parking covers both cases. Two
barrier-driven tests would pin it: a release during the open failure, and a close() during it.
3. acquire_timeout = Duration::ZERO now refuses a session that is sitting idle
The deadline check moved above the sweep and idle.pop_front(). On main the order is
closed → sweep → idle → growth → deadline; here it is closed → deadline → sweep → idle → growth.
With zero, the first iteration returns before the queue is ever examined:
idle_count() == 1
acquire() -> Err(pool exhausted: no session available within 0ns (1 live, max 4))
control, same pool with a 2 s timeout -> Ok
Zero used to mean "do not block". It now means "always fail", which looks more like a side effect
of the move than a decision.
4. Endpoint::equivalent is not transitive, and treats distinct hosts as the same one
Running the function's own logic:
equivalent(LOCALHOST, localhost) = true
equivalent(localhost, 127.0.0.1) = true
equivalent(LOCALHOST, 127.0.0.1) = false <- not transitive
equivalent(127.0.0.1, 127.0.0.2) = true <- two different nodes
equivalent([::1], 127.0.0.1) = false <- the test asserts 127.0.0.1 == ::1
Brackets and case are stripped for the text comparison, but is_loopback_host receives the
original string, so [::1] fails to parse as an IpAddr and falls through.
The 127.0.0.1 against 127.0.0.2 case is the one with consequences. acquire_for_device finds
the newest hint for a device and then uses equivalent to pick which idle session to hand out. Two
DataNodes bound to different loopback addresses on the same port is an ordinary local multi-node
setup, and there a hint pointing at one node can return a session connected to the other.
Keeping the normalised text comparison and dropping the blanket loopback equivalence is the
conservative version: a missed match falls back to the normal acquire path, whereas a false match
sends the request to the wrong node. Storing the connected peer's SocketAddr on Connection and
comparing against that would handle aliases properly.
Raising, not blocking
5. Every thrift::Error is treated as transport damage
Error::Thrift wraps all four thrift::Error variants, and with_retry reads any of them as a
reason to throw the connection away and replay the operation. That is wrong for a remote
TApplicationException. The generated code calls read_message_end() before returning
thrift::Error::Application, so the frame has been consumed and the connection is fine. The same
shape appears 207 times in src/protocol/client.rs. Retrying there sends the operation twice and
can bury the application error under whatever the retry produces.
A fake server that logs which connection each RPC arrives on, answers the handshake normally, then
replies to executeUpdateStatementV2 with a complete, well-formed TApplicationException:
[server] conn#1 exec#1 method=executeUpdateStatementV2 seqid=3
[server] conn#2 exec#2 method=executeUpdateStatementV2 seqid=3
executeUpdateStatementV2 received: 2
The client threw away a clean connection, reconnected, and sent the statement again.
I would be careful with the fix here. "Never reconnect on Application" is the obvious reading and
it would introduce a different problem. Application has three other sources in thrift 0.23 —
verify_expected_sequence_number, verify_expected_service_call, verify_expected_message_type —
and all three return before read_message_end(), on a frame that really is half-read. Those are
the cases where the stream is genuinely out of step, and treating them as clean would leave a dirty
connection in the pool.
Discriminating on ApplicationErrorKind covers both: BadSequenceId, WrongMethodName and
InvalidMessageType mean broken, everything else means a clean remote error. If a server ever sent
an exception carrying one of those kinds on an intact frame, the cost is one unnecessary reconnect,
and the mistake never runs the other way.
I hit one of those three by accident while building the harness. A reply with the wrong sequence id
produced ApplicationError { kind: BadSequenceId, message: "expected 3 got 0" } with unread bytes
still sitting in the framed transport.
6. Two places where the documentation claims more than the code does
Both are wording; I am not suggesting the behaviour should change.
socket_timeout bounds each socket read and write, not each RPC. SO_RCVTIMEO restarts on
every recv. A framed read is read_i32 for the header and read_exact for the body
(thrift 0.23 transport/framed.rs:47,73), so a peer sending a byte at a time keeps every syscall
inside the bound while the RPC itself runs long. Measured through a real TFramedReadTransport
using the same 300 ms your own test uses:
silent peer -> Err(WouldBlock) 301 ms <- what the PR's test covers, bounded
trickle peer -> Ok(20 bytes) 4.689 s no error
Nothing here regressed; it was unbounded before either way. The sentence beside each claim already
gets the scope right ("a peer that accepts the connection and then goes silent"), and that is also
exactly what socket_timeout_bounds_reads_after_handshake exercises. Only the "every RPC read"
enumeration reaches past it. Five places carry that phrasing: README.md, README_ZH.md,
session.rs, connection/mod.rs, and the PR description. The case worth naming is a peer that is
slow rather than dead, which can still hold a pool slot for as long as it likes.
The retry note in the README. The new paragraph says the session reconnects and retries the
operation once after a transport-level failure. For reads that is accurate. execute_non_query
takes arbitrary SQL and goes through the same wrapper, and a reply lost after the server has
already run the statement looks identical, from the client, to a request that never arrived.
Replaying DELETE, DDL, or INSERT ... VALUES(now(), ...) applies it a second time. The pool
itself depends on this path for the USE {db} catch-up.
Here the server logs that it executed the statement and then withholds the reply:
[server] conn#1 exec#1 method=executeUpdateStatementV2 seqid=3
[server] executed, but withholding the reply
[server] conn#2 exec#2 method=executeUpdateStatementV2 seqid=3
CREATE DATABASE root.replay_probe received: 2
The retry predates this PR and I am not asking you to redesign it. What changed is the reach: with
a 60 s socket_timeout by default, a server that is merely slow is enough to trigger a replay,
where before it took a connection that had actually broken, and slow is the case where the request
almost certainly arrived. A line in the README saying the retry is unsafe for non-idempotent
statements, plus an issue for the real fix, would be about the right size.
7. TableSessionBuilder::socket_timeout cannot express "no bound", and zero inverts
The builder always stores Some(timeout), so the documented None cannot be reached through it.
Duration::ZERO reaches connect_one's .filter(|t| !t.is_zero()) and turns the timeout off
altogether. Someone writing .socket_timeout(Duration::ZERO) and expecting to fail immediately
gets a connection that never times out; I tried it, and an RPC against a silent peer had not
returned after three seconds. The zero-means-none rule is written on connect_one but not on the
builder, which is the side a caller reads.
Scope
All of the above runs on a clean checkout of 93271ed2 against fake servers, each with a control
that passes. Timings come from one machine with thrift 0.23.0, so the shapes are the point rather
than the exact milliseconds.
I did not read session.rs beyond open, with_retry, reconnect, is_open, Drop, and the
execute and insert signatures. Whether replaying an insert with a caller-supplied timestamp
does any harm depends on IoTDB's same-timestamp overwrite semantics, which I did not check, so
item 6 stays scoped to execute_non_query. The server-side cleanup in item 1 was verified on
apache/iotdb master and not on the older servers in COMPATIBILITY.md.
2 participants
Summary
One combined PR for the session-pool / connection-lifecycle / redirect review findings, split into
issues #4–#10 for tracking. The root cause (#4) is fixed by a new socket-level I/O timeout, and the
other fixes are the follow-on work each issue describes.
What changed
socket_timeoutonConnectionOptions/SessionConfig(Some(60s)by default,Nonerestores the oldbehaviour, zero treated as
None). It is applied viasocket2right after connect andbefore the TLS handshake, so the TLS handshake, every RPC read, and the best-effort
closeSessiononDropare all bounded.SO_KEEPALIVEis now enabled on the socket.Wedged reads surface as
Error::Thrift, sowith_retry/reconnect actually engages.Session::open. The endpoint loop now retriesconnect +
openSession+requestStatementIdtogether, mirroringreconnect().acquire()spends itsacquire_timeoutbudget on growth /hand-out failures instead of failing instantly;
liveis now decremented under the state lock(no lost Condvar wakeups);
close()wakes waiters before the blockingcloseSessioncalls;acquire_timeout = Duration::MAXmeans "wait without a deadline" viachecked_add.connect_streamshares one totalconnect_timeoutacross every resolved address of an endpoint.
rejection and socket timeouts, and
fetch_resultswhich cannot be retried) marks theconnection broken;
is_open()then reports false and pools discard the session. Note: the16,384,000-byte frame cap itself cannot be raised without a thrift 0.23 API change
(
TFramedReadTransportexposes no config setter), so this PR fixes the desync/reuse half.and treats loopback spellings as equivalent (general hostname-vs-IP DNS resolution is deliberately
not done on the acquire path); the newest hint wins when idle sessions hold conflicting hints
(the cache insertion seq is now process-wide);
SessionConfiggainsredirect_cache_ttl/redirect_cache_max_entries;TableSessionPool::acquire_for_deviceis exposed.
reconnect, so a pool slot is no longer held for the full C#-style reconnect walk.
Tests
New regression tests use fake/silent listeners:
connection::tests::socket_timeout_bounds_reads_after_handshakeconnection::tls_tests::tls_handshake_times_out_against_silent_peersession::tests::open_fails_over_to_next_endpoint_when_authentication_fails(deterministic fake Thrift handshake server)no_reconnect_when_disablednow asserts the broken-session behaviourpool::tests::growth_failure_spends_acquire_timeout_budget,close_wakes_waiters_promptly,acquire_timeout_max_waits_until_closed_without_panicking,acquire_for_device_prefers_newest_conflicting_hintconnection::tests::endpoint_equivalent_normalizes_and_matches_loopbackVerified locally:
cargo fmt --check,./tools/check-license.sh,cargo clippy --all-targets -- -D warnings(default and--features tls),cargo test120 passed,cargo test --features tls132 passed(live-server tests ran against a local IoTDB instance).
Closes #4, closes #5, closes #6, closes #7, closes #8, closes #9, closes #10.