xiangfu0 commented on code in PR #20021:
URL: https://github.com/apache/pinot/pull/20021#discussion_r4368866109
##########
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 it deserves an explicit callout. The description's "Configuration and
compatibility" section now carries a release-note paragraph that lists each key
with its default (10 s write timeout, 10 s `TCP_USER_TIMEOUT` on Epoll only, 16
MiB / 8 MiB watermarks), states they turn on for every broker on upgrade where
there were no limits before, and says how to raise or disable each
(non-positive disables the timeouts).
On the GC point, one refinement: a stop-the-world pause alone does not trip
`TCP_USER_TIMEOUT`, which bounds data that goes unacknowledged, and the paused
server's kernel keeps acknowledging. It only comes into play once the server's
receive buffer has filled, which is the same condition under which the
progress-based write timeout fires, so in practice both trip together, and only
for a pause long enough to back up a full socket buffer of requests. Agreed
that reconnecting beats hanging there; the release note says so.
##########
pinot-core/src/main/java/org/apache/pinot/core/transport/ServerChannels.java:
##########
@@ -143,10 +170,53 @@ public void sendRequest(String rawTableName,
AsyncQueryResponse asyncQueryRespon
ServerRoutingInstance serverRoutingInstance, InstanceRequest
instanceRequest, long timeoutMs)
throws Exception {
byte[] requestBytes =
THREAD_LOCAL_T_SERIALIZER.get().serialize(instanceRequest);
- ServerChannel serverChannel =
_serverToChannelMap.computeIfAbsent(serverRoutingInstance, ServerChannel::new);
- // This server is now query-carrying, whoever opened the channel. See
[#hasChannel].
- serverChannel._openedByQuery = true;
- serverChannel.sendRequest(rawTableName, asyncQueryResponse,
serverRoutingInstance, requestBytes, timeoutMs);
+ for (int attempt = 1; attempt <= MAX_SEND_ATTEMPTS; attempt++) {
Review Comment:
Fixed in 00ea1b915f rather than left as a note. `ServerChannels#sendRequest`
now takes one deadline for the whole call, and every attempt gets only what is
left of it, so a sender that races an eviction can no longer stack up three
full lock waits. Within an attempt the connect gets what the lock wait left, so
the lock wait and the connect together also stay within the caller's
`timeoutMs` (on master, after the rebase, the bounded connect used to get a
fresh full budget after the lock).
##########
pinot-core/src/main/java/org/apache/pinot/core/transport/RequestWriteTimeoutHandler.java:
##########
@@ -0,0 +1,201 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.core.transport;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelDuplexHandler;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelOutboundBuffer;
+import io.netty.channel.ChannelProgressiveFuture;
+import io.netty.channel.ChannelProgressiveFutureListener;
+import io.netty.channel.ChannelProgressivePromise;
+import io.netty.channel.ChannelPromise;
+import java.util.HashSet;
+import java.util.Set;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.TimeUnit;
+import org.apache.pinot.common.metrics.BrokerMeter;
+import org.apache.pinot.common.metrics.BrokerMetrics;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+/// Broker-side socket-idle monitor. While outbound bytes are pending, closes
a server channel if the kernel accepts
+/// no bytes for the configured timeout. Progress by the current queue head
keeps later queued writes alive; enqueue,
+/// cancellation and write failure do not reset the idle clock.
+///
+/// Install on the socket side of SSL so ciphertext progress is observable.
This does not time plaintext waiting
+/// inside SSL before ciphertext exists; SSL's handshake timeout and the
pre-connect handshake budget remain separate.
+/// Each socket write uses a progressive promise, relaying the downstream
outcome to the original promise without
+/// completing it early or releasing Netty-owned buffers. One connection timer
runs only while writes are pending.
+///
+/// Not sharable: all state is confined to the channel's event loop.
+public class RequestWriteTimeoutHandler extends ChannelDuplexHandler {
+ private static final Logger LOGGER =
LoggerFactory.getLogger(RequestWriteTimeoutHandler.class);
+
+ private final ServerRoutingInstance _serverRoutingInstance;
+ private final long _timeoutMs;
+ private final long _timeoutNanos;
+ private final Set<PendingWrite> _pendingWrites = new HashSet<>();
+ private Set<PendingWrite> _unflushedWrites = new HashSet<>();
+ private long _lastProgressNanos;
+ private ScheduledFuture<?> _timer;
+ private boolean _closed;
+
+ public RequestWriteTimeoutHandler(ServerRoutingInstance
serverRoutingInstance, long timeoutMs) {
+ _serverRoutingInstance = serverRoutingInstance;
+ _timeoutMs = timeoutMs;
+ _timeoutNanos = TimeUnit.MILLISECONDS.toNanos(timeoutMs);
+ }
+
+ @Override
+ public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise
promise) {
+ // SSL supplies regular promises for ciphertext. Void writes also need
monitoring; unvoid preserves their
+ // exception propagation while giving the relay a real completion signal.
+ PendingWrite write = new PendingWrite(ctx, promise.unvoid(), msg
instanceof ByteBuf
+ && ((ByteBuf) msg).isReadable());
+ ctx.write(msg, write._progressivePromise);
+ }
+
+ @Override
+ public void flush(ChannelHandlerContext ctx) {
+ // Netty makes flushed outbound promises uncancelable. Preserve that
contract for the caller's promise too,
+ // while cancellation before flush still cancels the queued downstream
promise so Netty releases/skips it.
+ if (!_unflushedWrites.isEmpty()) {
+ Set<PendingWrite> unflushedWrites = _unflushedWrites;
+ // Detach this first-flush batch before callbacks can complete writes or
remove the handler reentrantly.
+ _unflushedWrites = new HashSet<>();
+ for (PendingWrite write : unflushedWrites) {
+ if (_pendingWrites.contains(write) &&
!write._promise.setUncancellable()) {
+ write._progressivePromise.cancel(false);
+ }
+ }
+ }
+ ctx.flush();
+ }
+
+ @Override
+ public void channelInactive(ChannelHandlerContext ctx) {
+ stopMonitoring();
+ ctx.fireChannelInactive();
+ }
+
+ @Override
+ public void handlerRemoved(ChannelHandlerContext ctx) {
+ stopMonitoring();
+ }
+
+ private void stopMonitoring() {
+ _closed = true;
+ cancelTimer();
+ // The listeners still relay real downstream outcomes after removal; do
not complete or release anything here.
+ _pendingWrites.clear();
+ _unflushedWrites.clear();
+ }
+
+ private void cancelTimer() {
+ if (_timer != null) {
+ _timer.cancel(false);
+ _timer = null;
+ }
+ }
+
+ private void scheduleTimeout(ChannelHandlerContext ctx, long delayNanos) {
+ _timer = ctx.executor().schedule(() -> checkTimeout(ctx), delayNanos,
TimeUnit.NANOSECONDS);
+ }
+
+ private void checkTimeout(ChannelHandlerContext ctx) {
+ _timer = null;
+ if (_closed || _pendingWrites.isEmpty()) {
+ return;
+ }
+ long idleNanos = System.nanoTime() - _lastProgressNanos;
+ if (idleNanos < _timeoutNanos) {
+ scheduleTimeout(ctx, _timeoutNanos - idleNanos);
+ return;
+ }
+ _closed = true;
+ Channel channel = ctx.channel();
+ ChannelOutboundBuffer outboundBuffer = channel.unsafe().outboundBuffer();
+ long pendingBytes = outboundBuffer != null ?
outboundBuffer.totalPendingWriteBytes() : -1;
+ LOGGER.error("Request write to server: {} made no progress for {}ms ({}
bytes pending, writable: {}), closing the "
+ + "channel as the socket is stalled or half-open",
_serverRoutingInstance, _timeoutMs, pendingBytes,
+ channel.isWritable());
+
BrokerMetrics.get().addMeteredGlobalValue(BrokerMeter.NETTY_CONNECTION_WRITE_TIMEOUTS,
1);
+ ctx.close();
+ }
+
+ /// Per-write outcome relay; progress updates the shared socket clock, never
a separate write deadline.
+ private final class PendingWrite implements ChannelProgressiveFutureListener
{
+ final ChannelPromise _promise;
+ final ChannelProgressivePromise _progressivePromise;
+ final boolean _hasBytes;
+ long _progress;
+
+ PendingWrite(ChannelHandlerContext ctx, ChannelPromise promise, boolean
hasBytes) {
+ _promise = promise;
+ _hasBytes = hasBytes;
+ _progressivePromise = ctx.newProgressivePromise();
+ _pendingWrites.add(this);
+ _unflushedWrites.add(this);
+ if (_pendingWrites.size() == 1 && !_closed) {
+ _lastProgressNanos = System.nanoTime();
+ scheduleTimeout(ctx, _timeoutNanos);
+ }
+ _progressivePromise.addListener(this);
+ _promise.addListener(future -> {
Review Comment:
Added the comment in 00ea1b915f. It is safe for a slightly broader reason
than the on-loop call, which the comment spells out: the listener only cancels
`_progressivePromise`, which is thread-safe, and that promise is created from
this channel's context, so its own listener (`operationComplete`, the only code
that touches handler state) always runs on the event loop, even when the caller
later completes or cancels their promise from another thread.
--
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]
