xiangfu0 commented on code in PR #20022:
URL: https://github.com/apache/pinot/pull/20022#discussion_r4366307520
##########
pinot-core/src/main/java/org/apache/pinot/core/transport/ServerChannels.java:
##########
@@ -227,35 +418,99 @@ protected void initChannel(SocketChannel ch) {
_serverRoutingInstance));
}
});
+ if (isTcpUserTimeoutApplied()) {
+ _bootstrap.option(EpollChannelOption.TCP_USER_TIMEOUT,
_tcpUserTimeoutMs);
+ }
}
void closeChannel() {
- if (_channel != null) {
- _channel.close();
+ Channel channel = _channel;
+ if (channel != null) {
+ channel.close();
}
}
+ /// See [ServerChannels#evictChannel]. Flags the entry first, then closes
whatever channel it holds; no lock is
+ /// taken so that the routing manager's Helix callback is never blocked
behind a connect or write in progress.
+ void evict() {
+ _evicted = true;
+ Channel channel = _channel;
+ if (channel != null) {
+ closeLocally(channel);
+ }
+ }
+
+ /// Closes a channel the broker decided to drop (eviction after a restart,
replacement of a suspect channel).
+ /// The queries still in flight on it fail through `channelInactive` as on
any closed channel, but the server is
+ /// not reported for it: the close was the broker's decision, not evidence
against the server.
+ private void closeLocally(Channel channel) {
+ DataTableHandler dataTableHandler =
channel.pipeline().get(DataTableHandler.class);
+ if (dataTableHandler != null) {
+ dataTableHandler.markClosedByBroker();
+ }
+ channel.close();
+ }
+
+ void recordResponse() {
+ _lastResponseTimeMs = System.currentTimeMillis();
+ _unansweredQueries.set(0);
+ }
+
+ /// Runs on the liveness thread a grace period after a query written at
`submitTimeMs` timed out. See
+ /// [ServerChannels#recordUnansweredQuery].
+ void judgeUnansweredQuery(long submitTimeMs) {
+ // Answered something since the query was written (a fresh channel
counts as answered from its connect time, so
+ // queries written to the channel it replaced do not count against it),
or no longer in use. Strictly after:
+ // the connect and the first write can share a millisecond, while a real
answer never shares one with its
+ // request.
+ if (_evicted || _lastResponseTimeMs > submitTimeMs) {
+ return;
+ }
+ int unansweredQueries = _unansweredQueries.incrementAndGet();
+ if (unansweredQueries < _unansweredQueriesThreshold) {
+ return;
+ }
+ _unansweredQueries.set(0);
+ _suspect = true;
Review Comment:
Fixed, and a little more than the re-read: the liveness bookkeeping
(`_lastResponseTimeMs`, the unanswered count, the pending judgment) now sits
behind a small lock on the channel entry, taken briefly by `recordResponse` on
the event loop, by the connect path when it resets a fresh channel, and by the
judgment on the liveness thread. The judgment re-checks the last response time
and flips `_suspect` under that lock, so a response landing while it runs is
never lost: either it is seen and the judgment bails, or it runs after and
zeroes the count. The lock is uncontended in practice (one short critical
section per response).
##########
pinot-common/src/main/java/org/apache/pinot/common/config/NettyConfig.java:
##########
@@ -18,13 +18,56 @@
*/
package org.apache.pinot.common.config;
+import com.google.common.base.Preconditions;
import org.apache.pinot.spi.env.PinotConfiguration;
/// Container object for netty configuration of pinot clients and servers
(netty, grizzly, etc.)
+///
+/// Every setting is read as `<namespace>.<suffix>`, where the namespace is
`pinot.broker.netty` on the broker and
+/// `pinot.server.netty` on the server. The channel liveness settings below
only apply to the client side, i.e. the
+/// broker-to-server channels opened by `ServerChannels`; the server-side
`QueryServer` ignores them.
public class NettyConfig {
private static final String NATIVE_TRANSPORTS_ENABLED =
"native.transports.enabled";
+
+ /// Upper bound on how long a request write may stay pending before the
channel is closed, in milliseconds. A write
+ /// is pending while the kernel has not accepted the bytes, which on a
half-open or stalled socket is forever (the
+ /// peer stops acknowledging, the send buffer fills, and `isActive()` keeps
reporting `true` until the kernel gives
+ /// up on retransmits, typically after many minutes). Closing the channel
fails the in-flight queries on it fast
+ /// and lets the failure detector quarantine the server. Non-positive
disables the check. Broker key:
+ /// `pinot.broker.netty.write.timeout.ms`.
+ public static final String WRITE_TIMEOUT_MS = "write.timeout.ms";
+ public static final long DEFAULT_WRITE_TIMEOUT_MS = 10_000L;
+
+ /// Linux `TCP_USER_TIMEOUT`, in milliseconds: how long transmitted data may
remain unacknowledged before the
+ /// kernel closes the connection, replacing the default retransmission
timeout of several minutes. Covers the
+ /// case the write timeout cannot see: small writes that the kernel accepted
into a send buffer that the peer never
+ /// drains. Only applied when the broker uses the Epoll transport
(`native.transports.enabled` on Linux); NIO and
+ /// KQueue have no such socket option. Non-positive leaves the kernel
default. Broker key:
+ /// `pinot.broker.netty.tcp.user.timeout.ms`.
+ public static final String TCP_USER_TIMEOUT_MS = "tcp.user.timeout.ms";
+ public static final int DEFAULT_TCP_USER_TIMEOUT_MS = 10_000;
Review Comment:
Agreed that this deserves to be explicit, so #20021 now carries a "Release
note" section covering the three new defaults, what each does, when it fires,
and how to raise or disable it, plus the rolling-upgrade note (broker-only, no
server or wire change, so brokers can be upgraded in any order and rolled back
independently).
Why on by default, briefly: all three only act on a socket whose peer has
stopped taking bytes. The write timeout is now on progress rather than
completion (see the thread above): it fires only when the kernel has accepted
nothing for 10 s, which on a healthy link needs a receiver that stopped
reading, not a slow one. `TCP_USER_TIMEOUT` (Epoll only) fires only when
transmitted data has gone unacknowledged for 10 s; on a lossy link Linux
retransmits well within that, and a path that cannot deliver an ACK in 10 s has
already failed every query on it at the default 10 s broker timeout. The
watermark rejection needs 16 MB queued beyond what the kernel accepted, which a
healthy peer drains in milliseconds. The failure each prevents is silent and
long (minutes of lost queries, direct-memory exhaustion that takes healthy
channels down with it), while a false positive costs one query and a reconnect,
so the defaults err towards acting. Operators on unusual links can raise each
one, or set
it non-positive to disable it, as the release note says.
##########
pinot-core/src/main/java/org/apache/pinot/core/transport/ServerChannels.java:
##########
@@ -169,6 +273,71 @@ public void connect(ServerRoutingInstance
serverRoutingInstance)
_serverToChannelMap.computeIfAbsent(serverRoutingInstance,
ServerChannel::new).connect();
}
+ /// Marks the cached channel to the given routing instance suspect: the
server stopped answering queries on it
+ /// while the socket still looked active. The next query to it, or the next
[#probe], replaces it with a fresh
+ /// connection instead of trusting `isActive()`.
+ public void markSuspect(ServerRoutingInstance serverRoutingInstance) {
+ ServerChannel serverChannel =
_serverToChannelMap.get(serverRoutingInstance);
+ if (serverChannel != null) {
+ serverChannel._suspect = true;
+ }
+ }
+
+ /// The failure detector's retry probe for one channel. Does nothing and
returns `false` when this broker has not
+ /// queried the server over this channel (see [#hasChannel]); otherwise
reconnects the channel if it is down,
+ /// replaces it with a fresh connection if it was marked suspect, and
returns `true`, throwing if the connection
+ /// cannot be established.
+ public boolean probe(ServerRoutingInstance serverRoutingInstance)
+ throws InterruptedException, TimeoutException {
+ ServerChannel serverChannel =
_serverToChannelMap.get(serverRoutingInstance);
+ if (serverChannel == null || !serverChannel._openedByQuery) {
+ return false;
+ }
+ serverChannel.connect();
+ return true;
+ }
+
+ /// Records a response from the server on the given channel: it is alive,
whatever it answered.
+ public void recordResponse(ServerRoutingInstance serverRoutingInstance) {
+ ServerChannel serverChannel =
_serverToChannelMap.get(serverRoutingInstance);
+ if (serverChannel != null) {
+ serverChannel.recordResponse();
+ }
+ }
+
+ /// Records that a query written to the given channel at `submitTimeMs`
timed out unanswered. The verdict waits
+ /// out the grace period: the query only counts against the channel if by
then the server has still sent nothing
+ /// on it since the query was written. That lets the server's own late
timeout response through, so a burst of
+ /// slow queries, or a broker pause that delays reading the responses, does
not look like a dead channel: only a
+ /// channel the server has gone silent on does. At the threshold the channel
is marked suspect and the server is
+ /// reported through [QueryRouter#reportUnansweredServer]. No-op when the
check is disabled.
+ public void recordUnansweredQuery(ServerRoutingInstance
serverRoutingInstance, long submitTimeMs) {
Review Comment:
Good call. The judgments queued per channel are now capped at `max(16, 2 ×
threshold)`: a timed-out query beyond that is simply not scheduled, because the
ones already queued are enough to cross the threshold on their own, so the
outcome under a dead server at high QPS is unchanged while the liveness queue
stays at a few dozen tasks per channel instead of the inflight count. Each
judgment still gets its own full grace period, which is what keeps the burst
case honest, and a channel already marked suspect is not reported again until
it has been replaced, so one burst means one report. Covered by
`ServerChannelsTest#testPendingJudgmentsPerChannelAreCapped`.
##########
pinot-core/src/main/java/org/apache/pinot/core/transport/ServerChannels.java:
##########
@@ -264,17 +519,40 @@ void sendRequest(String rawTableName, AsyncQueryResponse
asyncQueryResponse,
}
}
+ private void checkNotEvicted() {
+ if (_evicted) {
+ closeChannel();
+ throw new IllegalStateException("Channel to server: " +
_serverRoutingInstance + " has been evicted");
+ }
+ }
+
/// Lazy query path: opens the TCP connection only. Any TLS handshake is
left to proceed
/// asynchronously so the channel lock is released as soon as the socket
is up, keeping the first
/// query's critical section short. Startup pre-connect uses
[#preConnectWithoutLocking(long)]
/// instead, which additionally pays the handshake.
+ ///
+ /// Also replaces a channel marked suspect: the server stopped answering
on it while the socket still looked
+ /// active, which is what a half-open socket looks like until the kernel
gives up on it, so `isActive()` is not
+ /// trusted. The suspect channel is dropped before the connect, so a
failed connect does not leave it in place,
+ /// and whatever is still in flight on it fails without the server being
reported (nothing on it was being
+ /// answered anyway).
void connectWithoutLocking()
throws InterruptedException {
- if (_channel == null || !_channel.isActive()) {
- long startTime = System.currentTimeMillis();
- _channel = _bootstrap.connect().sync().channel();
- recordConnectTime(System.currentTimeMillis() - startTime);
+ Channel channel = _channel;
+ if (channel != null && channel.isActive() && !_suspect) {
+ return;
}
+ if (channel != null && _suspect) {
+ LOGGER.info("Replacing suspect channel to server: {} with a fresh
connection", _serverRoutingInstance);
+ _channel = null;
+ closeLocally(channel);
+ }
+ _suspect = false;
+ long startTime = System.currentTimeMillis();
+ _channel = _bootstrap.connect().sync().channel();
Review Comment:
Right, and I kept it that way deliberately. The lazy path has always paid
the connect under `_channelLock` (first query, and every reconnect after a
channel went inactive), and a suspect replacement is the same reconnect taken a
little earlier. Concurrent senders wait at most their own `tryLock(timeoutMs)`,
after which they fail with the channel-lock timeout rather than hanging, and
since the server has usually been reported and taken out of routing by then
there are few of them. Moving the connect out from under the lock would need a
second connection-establishment path with its own races for a case that lasts
one round trip; not worth it in this PR. Bounding `CONNECT_TIMEOUT_MILLIS` on
the lazy path (Netty's 30 s default today) would shorten the worst case for
everyone and is a reasonable small follow-up.
--
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]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
