diff --git a/sdk/cosmos/azure-cosmos-tests/src/test/java/com/azure/cosmos/implementation/http/Http2PingHandlerTest.java b/sdk/cosmos/azure-cosmos-tests/src/test/java/com/azure/cosmos/implementation/http/Http2PingHandlerTest.java index 004cde861d87..a7403ce5bf57 100644 --- a/sdk/cosmos/azure-cosmos-tests/src/test/java/com/azure/cosmos/implementation/http/Http2PingHandlerTest.java +++ b/sdk/cosmos/azure-cosmos-tests/src/test/java/com/azure/cosmos/implementation/http/Http2PingHandlerTest.java @@ -16,6 +16,7 @@ import java.lang.reflect.Method; import java.util.concurrent.ScheduledFuture; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicLong; import static org.assertj.core.api.Assertions.assertThat; @@ -25,7 +26,7 @@ * Covers state transitions that do not require advancing real time -- the handler's * timeout / threshold logic reads {@code System.nanoTime()}, so the tests sidestep * the clock by pre-setting {@code pingOutstandingSinceNanos} (or - * {@code lastActivityNanos}) via reflection: + * {@code lastReadActivityNanos}) via reflection: *
+ * {@code channelReadComplete} fires at the end of each inbound read cycle on + * the child channel -- so this method stamps the parent attribute exactly once + * per batch of HEADERS/DATA frames demuxed by {@link io.netty.handler.codec.http2.Http2MultiplexHandler}, + * not once per frame. Under normal load this is plenty of granularity to + * suppress unnecessary PINGs without becoming a hot-path allocator. + *
+ * Edge cases: + *
+ * Why this exists: {@link io.netty.handler.codec.http2.Http2MultiplexHandler} dispatches H2 stream frames to + * child channels, so request/response read traffic never surfaces on the parent + * pipeline's {@link #channelRead}. Without this signal, {@link #lastReadActivityNanos} + * only observes PING ACKs / SETTINGS / GOAWAY on the parent, which means a + * connection serving thousands of QPS would still look "idle" to the PING handler + * and be PINGed unnecessarily on every interval -- adding measurable latency + * overhead under load. + *
+ * The attribute is initialized in {@link #handlerAdded} and updated via
+ * {@code accumulateAndGet(now, Math::max)} so writes are monotonic. The PING
+ * handler reads it via {@link #effectiveLastReadActivityNanos(ChannelHandlerContext)}
+ * to compute idleness as {@code max(parent-pipeline reads, child-stream reads)}.
+ */
+ static final AttributeKey
+ * Read-only: writes (our own PING sends) are deliberately NOT included here
+ * because the purpose of the PING is to detect a peer that has stopped
+ * responding -- only inbound frames are positive evidence of liveness. The
+ * separate {@link #lastPingSendNanos} field tracks our own probes for the
+ * send-throttle in {@link #maybeSendPing}; it is combined with this value
+ * there but not here.
+ *
+ * The attribute should never be null in practice (it is seeded in
+ * {@link #handlerAdded}), but defensive null-handling is cheap and avoids an NPE
+ * if the channel was somehow torn down between attribute seeding and the
+ * scheduled task firing.
+ */
+ private long effectiveLastReadActivityNanos(ChannelHandlerContext ctx) {
+ AtomicLong childReadActivity = ctx.channel().attr(LAST_CHILD_STREAM_READ_ACTIVITY_NANOS).get();
+ long childNanos = childReadActivity != null ? childReadActivity.get() : 0L;
+ return Math.max(lastReadActivityNanos, childNanos);
+ }
+
/**
* Returns {@code true} when the manual HTTP/2 PING keepalive should be active for
* a client built with the given {@link Http2ConnectionConfig}. This is the single
diff --git a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/http/ReactorNettyClient.java b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/http/ReactorNettyClient.java
index e95989c5aa3c..3e6008405869 100644
--- a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/http/ReactorNettyClient.java
+++ b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/http/ReactorNettyClient.java
@@ -223,8 +223,7 @@ private void configureChannelPipelineHandlers() {
// Duplicate handler name is the only possible cause.
}
}
- }))
- .doOnRequest((req, conn) -> {
+
// Install a @Sharable head-of-pipeline rewrap handler on each H2
// child-stream pipeline. When Http2PingHandler closes the parent
// (TCP) channel after consecutive PING-ACK timeouts or PING-send
@@ -237,21 +236,31 @@ private void configureChannelPipelineHandlers() {
// stack maps that to GATEWAY_HTTP2_PING_TIMEOUT_CHANNEL_CLOSED so
// ClientRetryPolicy can suppress region mark-down.
//
+ // doOnConnected fires once per child stream channel (the existing
+ // customHeaderCleaner install above relies on the same contract), so
+ // this install is amortized to one-time per stream rather than
+ // per-request -- avoids the operator-wrap + per-request pipeline
+ // walk that the prior .doOnRequest()-based install paid on every
+ // HTTP call.
+ //
// Gate on ch.parent() != null so this only runs on H2 child streams
- // (H1.1 connections have null parent and never need the rewrap).
- Channel ch = conn.channel();
- if (ch.parent() != null && ch.pipeline().get(Http2PingCloseRewrapHandler.HANDLER_NAME) == null) {
+ // (the parent TCP channel uses Http2ParentChannelExceptionHandler
+ // installed above; H1.1 connections never enter this branch since
+ // the entire enclosing block is guarded by isH2Enabled).
+ Channel ch = connection.channel();
+ if (ch.parent() != null
+ && channelPipeline.get(Http2PingCloseRewrapHandler.HANDLER_NAME) == null) {
try {
- ch.pipeline().addFirst(
+ channelPipeline.addFirst(
Http2PingCloseRewrapHandler.HANDLER_NAME,
Http2PingCloseRewrapHandler.INSTANCE);
} catch (IllegalArgumentException ignored) {
// TOCTOU race: between the get()==null check above and addFirst(),
- // a concurrent doOnRequest may have installed the handler.
+ // a concurrent doOnConnected may have installed the handler.
// Duplicate handler name is the only possible cause.
}
}
- });
+ }));
}
}