Add flow-control and remove auto-read in netty4 HTTP pipeline #126441

mhl-b · 2025-04-08T05:15:22Z

Remove buffering and queueing in Netty4HttpRequestBodyStream and Netty4HttpHeaderValidator and replace it with netty's FlowControlHandler in HTTP pipeline. Disable channel auto-read and introduce explicit channel/ctx.read().

Add a few helper classes to detect missing reads in tests MissingReadDetector and ReadSniffer.

elasticsearchmachine · 2025-04-08T05:27:02Z

Hi @mhl-b, I've created a changelog YAML for you.

…search into netty-flow-control-handler

elasticsearchmachine · 2025-04-08T08:30:07Z

Pinging @elastic/es-distributed-coordination (Team:Distributed Coordination)

…handler

DaveCTurner

Production code looks good to me, I have mostly just commented on the tests.

DaveCTurner · 2025-04-10T17:23:47Z

modules/transport-netty4/src/test/java/org/elasticsearch/http/netty4/ReadSniffer.java

+ */
+public class ReadSniffer extends ChannelOutboundHandlerAdapter {
+
+    int readCnt;


naming nit, we can afford the extra vowels here

Suggested change

int readCnt;

int readCount;

DaveCTurner · 2025-04-10T17:26:55Z

modules/transport-netty4/src/main/java/org/elasticsearch/http/netty4/MissingReadDetector.java

+    private long lastRead;
+    private ScheduledFuture<?> checker;
+
+    MissingReadDetector(TimeProvider timer, long missingReadInterval) {


naming nit, let's record that it's milliseconds here

Suggested change

MissingReadDetector(TimeProvider timer, long missingReadInterval) {

MissingReadDetector(TimeProvider timer, long missingReadIntervalMillis) {

DaveCTurner · 2025-04-10T17:34:23Z

modules/transport-netty4/src/main/java/org/elasticsearch/http/netty4/Netty4HttpAggregator.java

@@ -48,6 +49,9 @@ public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception
        }
        if (aggregating || msg instanceof FullHttpRequest) {
            super.channelRead(ctx, msg);
+            if (msg instanceof LastHttpContent == false) {
+                ctx.read(); // HttpObjectAggregator is tricky with auto-read off, it might not call read again, calling on its behalf
+            }
        } else {
            streamContentSizeHandler.channelRead(ctx, msg);


This branch is apparently not covered by the unit test suite, could we add a test that hits it?

org.elasticsearch.http.netty4.Netty4HttpHeaderValidatorTests#testWithFlowControlAndAggregator is covering it

Also, next step is to remove aggregator from netty. There is no incentive to add more tests here, more to remove in coming days.

DaveCTurner · 2025-04-10T17:35:27Z

...ansport-netty4/src/main/java/org/elasticsearch/http/netty4/Netty4HttpContentSizeHandler.java

@@ -123,6 +123,7 @@ private void handleRequest(ChannelHandlerContext ctx, HttpRequest request) {
                isContinueExpected = true;
            } else {
                ctx.writeAndFlush(EXPECTATION_FAILED_CLOSE.retainedDuplicate()).addListener(ChannelFutureListener.CLOSE);
+                ctx.read();


The request.decoderResult().isFailure() case above is also not covered by unit tests, can we add that?

DaveCTurner · 2025-04-10T17:35:47Z

.../transport-netty4/src/main/java/org/elasticsearch/http/netty4/Netty4HttpHeaderValidator.java

+        assert ctx.channel().config().isAutoRead() == false : "auto-read should be always disabled";
+        if (msg instanceof HttpObject httpObject) {
+            if (httpObject.decoderResult().isFailure()) {
+                ctx.fireChannelRead(httpObject); // pass-through for decoding failures


This line is also not covered by a unit test

DaveCTurner · 2025-04-10T17:42:02Z

...ransport-netty4/src/main/java/org/elasticsearch/http/netty4/Netty4HttpPipeliningHandler.java

-                            serverTransport.getThreadPool().getThreadContext(),
-                            activityTracker
-                        );
+                        var contentStream = new Netty4HttpRequestBodyStream(ctx, serverTransport.getThreadPool().getThreadContext());


Apparently this is not touched by the unit tests, nor is the msg instanceof HttpContent branch below

the pipelining test is covered primarily by Netty4IncrementalRequestHandlingIT

DaveCTurner · 2025-04-10T17:46:55Z

...earch/xpack/security/transport/netty4/SecurityNetty4HttpServerTransportCloseNotifyTests.java

@@ -251,7 +251,7 @@ public void close() {
            server.dispatcher.reqQueue.forEach(r -> r.request.getHttpRequest().release());
            server.netty.stop();
            server.threadPool.shutdownNow();
-            safeAwait(client.netty.config().group().shutdownGracefully());


Should we fix the other shutdownGracefully() calls with default timeouts? Maybe in a follow-up?

yes, we should, can do follow up

DaveCTurner · 2025-04-10T17:48:43Z

...ransport-netty4/src/main/java/org/elasticsearch/http/netty4/Netty4HttpRequestBodyStream.java

    private final List<ChunkHandler> tracingHandlers = new ArrayList<>(4);
    private final ThreadContext threadContext;
-    private final ThreadWatchdog.ActivityTracker activityTracker;


We can't drop the ActivityTracker here, it's super-important for finding bugs that block the transport threads.

Sorry hang on I think I see, is it now the case that all activity is tracked by Netty4HttpPipeliningHandler, i.e. we're handling every single HTTP chunk within org.elasticsearch.http.netty4.Netty4HttpPipeliningHandler#channelRead?

Right, before stream could serve chunk from current buffer without calling channel read. Now it is always from channel read.

DaveCTurner · 2025-04-10T17:50:31Z

...ort-netty4/src/test/java/org/elasticsearch/http/netty4/Netty4HttpRequestBodyStreamTests.java

-
-    // ensures that we read from channel when no current chunks available
-    // and pass next chunk downstream without holding
-    public void testReadFromChannel() {


Can we adjust this test to match the new behaviour rather than just discarding it? It seems worth keeping to me.

Every call is channel read, pretty much every test unit and integ cover this.

DaveCTurner · 2025-04-10T17:53:31Z

...sport-netty4/src/test/java/org/elasticsearch/http/netty4/Netty4HttpHeaderValidatorTests.java

-import static org.hamcrest.Matchers.nullValue;
-import static org.hamcrest.Matchers.sameInstance;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;

 public class Netty4HttpHeaderValidatorTests extends ESTestCase {


I'm concerned that we're losing so many of these tests. They seem important.

There are many, most of them around buffering and internal state. Do you see anything particular is missing? I think current unit test small but robust. There are plenty of security tests in x-pack that run through this code too. The diff in github is ugly, I should probably create new test file and remove old one.

TLDR. Only missing one is decoder failure test as you pointed out already.

Here is list of previous tests and their fate:

testValidationPausesAndResumesData, covered by testIgnoreReadWhenValidating

testValidatorDoesNotTweakAutoReadAfterValidationComplete, not relevant, there is no autoread

testContentForwardedAfterValidation, covered by testMixedValidationResults

testContentDroppedAfterValidationFailure, covered by testMixedValidationResults (need to move few lines in current test)

testValidationErrorForwardsAsDecoderErrorMessage, covered by testMixedValidationResults

testValidationExceptionForwardsAsDecoderErrorMessage, a missing one as you pointed out

testValidationHandlesMultipleQueuedUpMessages, not relevant, there is no queue

testValidationFailureRecoversForEnqueued, there is no queue

testValidationFailureRecoversForInbound, covered by testMixedValidationResults

testValidationSuccessForLargeMessage

testValidationFailureForLargeMessage - I can extend testMixedValidationResults that writes a random size of content. But there is no content size semantic in validator, it was relevant with queueing.

testFullRequest... there is no need for these, HttpRequestDecoder does not emit full requests. Validator has to be placed after HttpObjectAggregator to handle those.

mhl-b · 2025-04-11T00:19:06Z

@DaveCTurner , addressed in 34f5884

naming changes
decoder failure unit tests
stream closure unit test
order of auto-read=off and first read in pipeline initialization

…handler

…search into netty-flow-control-handler

DaveCTurner

Ok thanks this all LGTM then. Could you fix the top-level comment to be something more appropriate for the Git commit (it's good info, you can move it into a reply, just doesn't really need to be in Git)

ywangd

This looks like a great PR that brings in a lot simplification. I like it! Thanks for requesting the reivew. I don't really feel qualified to review all the changes without spending significant more time on educating myself first on Netty though. David's approval should be sufficient. So I took this as a learning opportunity and read through the production code changes.

The code is definitely a lot earier to read. I only had minor comments. IIUC, the explicit read is sometimes called more often than needed, e.g. both Netty4HttpPipeliningHandler and Netty4HttpRequestBodyStream can call read when the stream ends? But that should not be a problem as long as we don't receive more data during header validation, for which there is an interception plus an assertion?

Thanks for working on this!

ywangd · 2025-04-11T07:33:58Z

.../transport-netty4/src/main/java/org/elasticsearch/http/netty4/Netty4HttpHeaderValidator.java

+            } else {
+                if (msg instanceof HttpRequest request) {
+                    validate(ctx, request);
+                } else if (msg instanceof HttpContent content) {


If there is no other possible type, should it be just an else branch with assert msg instanceof HttpContent?

If not bug in spotless and we could upgrade spotless version, we could use guarded patterns in switch statement. #115750. Cant use java features because of spotless, sad.

switch (msg) { case HttpObject obj when obj.decoderResult().isFailure() -> ctx.fireChannelRead(obj); case HttpRequest request -> validate(ctx, request); case HttpContent content when state == State.DROPPING -> { content.release(); ctx.read(); } case HttpContent content when state == State.PASSING -> ctx.fireChannelRead(content); default -> { assert false; } }

ywangd · 2025-04-11T07:45:27Z

.../transport-netty4/src/main/java/org/elasticsearch/http/netty4/Netty4HttpHeaderValidator.java

+    private boolean droppingContent;
+    private boolean validatingRequest;


Nit: It feels these two booleans are mutually exclusive and hence can be combined as a single enum?

ywangd · 2025-04-11T07:54:14Z

.../transport-netty4/src/main/java/org/elasticsearch/http/netty4/Netty4HttpHeaderValidator.java

-                    // this prevents thread-context changes to propagate beyond the validation, as netty worker threads are reused
-                    try (ThreadContext.StoredContext ignore = threadContext.newStoredContext()) {
-                        validator.validate(httpRequest, ctx.channel(), listener);
+        if (msg instanceof HttpObject httpObject) {


Similar here, it should not be possible to see a different type other than HttpObject? If so, we could use an assertion either here or in an else branch?

ywangd · 2025-04-11T07:56:16Z

modules/transport-netty4/src/main/java/org/elasticsearch/http/netty4/MissingReadDetector.java

+                if (now >= lastRead + interval) {
+                    logger.warn("chan-id={} haven't read from channel for [{}ms]", ctx.channel().id(), (now - lastRead));


I wonder whether it is worthwhile to warn faster if the last seen message is not either a FullHttpRequest or LastHttpContent?

What is use-case for this? Not reading from stream at anytime sounds equally bad. Our transport code that missed read at the end of request or stream handler that forgot to read chunk or close stream, both problematic.

…handler

mhl-b · 2025-04-11T18:32:34Z

Thanks @DaveCTurner and @ywangd.

Netty4HttpPipeliningHandler and Netty4HttpRequestBodyStream can call read when the stream ends

Good eyes. Yes it is. I dont think it's problematic at this time, since only bulk is handling stream and can tolerate unexpected read. But I changed it to only one read. It should be refactored in my next task to move aggregator to the rest controller. And every request in netty would be streamed.

mhl-b · 2025-04-11T21:47:12Z

For the history, previous PR description, before making it git friendly:

This PR removes HTTP content buffering from Netty4HttpRequestBodyStream and Netty4HttpHeaderValidator and replace it with netty's FlowControlHandler that sits after HTTP decoder. The flow control is required for multiple purposes - prevent content from invalid requests pass to the downstream, provide backpressure and throttling when resources are constrained.

FlowControlHandler emits exactly one message(HttpObject: HttpRequest, HttpContent, HttpLastContent) per read. Which is desirable behaviour for us in all cases.

@Override
protected void initChannel(Channel ch) throws Exception {
  // auto-read must be disabled all the time
  ch.config().setAutoRead(false);
    ...
  // from this point in pipeline every handler must call ctx or channel #read() when ready to process next HTTP part
  ch.pipeline().addLast(new FlowControlHandler());
  if (Assertions.ENABLED) {
    // missing reads are hard to catch, but we can detect absence of reads within interval
    long missingReadIntervalMs = 10_000;
    ch.pipeline().addLast(new MissingReadDetector(transport.threadPool, missingReadIntervalMs));
  }
  ...
   // make very first read call, since auto-read is disabled; following reads must come from the handlers
  ch.read();
}

But FlowControlHandler does not work when channel auto-read is enabled. Second change in this PR is replacement of auto-read by explicit read's from channel handlers. The auto-read feature in netty will read and push bytes as they come, not when requested. If downstream code cannot catch up with load, this "push" strategy can consume a lot of memory. Disabling auto-read changes strategy from "push" to "pull", making application code in charge when to read from the channel. Despite being obviously better for resource usage the auto-read=off requires extra care for reading from channel, missing read will leave channel in a non-operable state.

Also add a utility handler MissingReadDetector that detects if there no reads in given time interval when assertions are enabled. Absence of reads are hard to catch. Prints warnings:

[2025-04-10T00:45:49,104][WARN ][o.e.h.n.MissingReadDetector][node_t0][transport_worker][T#2] haven't read from channel for [12727ms]
[2025-04-10T00:45:59,104][WARN ][o.e.h.n.MissingReadDetector][node_t0][transport_worker][T#2] haven't read from channel for [22693ms]

This change removes a lot of code related to correct buffering and need to synchronize body-stream and header-validator on auto-read. It makes implementation pretty straightforward for both. Right now Netty4HttpHeaderValidator will intercept and ignore read calls when validation is still going.

public class Netty4HttpHeaderValidator extends ChannelDuplexHandler {
...
    @Override
    public void read(ChannelHandlerContext ctx) throws Exception {
        // until validation is completed we can ignore read calls,
        // once validation is finished HttpRequest will be fired and downstream can read from there
        if (validatingRequest == false) {
            ctx.read();
        }
    }
...
}

…elastic#126441)" This reverts commit c8805b8.

…ipeline (#127030) * Revert "Release buffers in netty test (#126744)" This reverts commit f9f3def. * Revert "Add flow-control and remove auto-read in netty4 HTTP pipeline (#126441)" This reverts commit c8805b8.

…elastic#126441)" This reverts commit c8805b8. Commit

…ty4 HTTP pipeline This reverts commit 2a243d8.

Re-applying elastic#126441 with the extra `FlowControlHandler` needed to ensure one-chunk-per-read semantics - see elastic#127111 for related tests.

Re-applying #126441 with the extra `FlowControlHandler` needed to ensure one-chunk-per-read semantics - see #127111 for related tests.

add flow-control and remove auto-read

3fe635b

mhl-b requested a review from DaveCTurner April 8, 2025 05:15

elasticsearchmachine added the v9.1.0 label Apr 8, 2025

non-reentrant reads

a89fec2

mhl-b added :Distributed Coordination/Network Http and internode communication implementations Team:Distributed Coordination Meta label for Distributed Coordination team >enhancement labels Apr 8, 2025

Update docs/changelog/126441.yaml

d83f45b

mhl-b changed the title ~~Add flow-control and remove auto-read in netty4 http pipeline~~ Add flow-control and remove auto-read in netty4 HTTP pipeline Apr 8, 2025

mhl-b added 4 commits April 7, 2025 22:43

Merge branch 'main' into netty-flow-control-handler

8170e19

requested reads in validator

b749dd3

Merge branch 'netty-flow-control-handler' of github.com:mhl-b/elastic…

3be681b

…search into netty-flow-control-handler

intercept readComplete

f8d7fb1

mhl-b marked this pull request as ready for review April 8, 2025 08:29

mhl-b and others added 7 commits April 8, 2025 13:23

Merge remote-tracking branch 'upstream/main' into netty-flow-control-…

e9df377

…handler

missing read detector

fea33ed

more reads and relax detector

d8c89c2

[CI] Auto commit changes from spotless

c90d4d7

more tests

55681ed

comment

2a6a792

Merge remote-tracking branch 'upstream/main' into netty-flow-control-…

294002e

…handler

mhl-b requested a review from ywangd April 10, 2025 02:38

DaveCTurner reviewed Apr 10, 2025

View reviewed changes

feedback

34f5884

mhl-b requested a review from DaveCTurner April 11, 2025 00:21

mhl-b added 2 commits April 10, 2025 17:22

Merge branch 'main' into netty-flow-control-handler

c849a98

Merge remote-tracking branch 'upstream/main' into netty-flow-control-…

9edc660

…handler

Merge branch 'netty-flow-control-handler' of github.com:mhl-b/elastic…

d6464a4

…search into netty-flow-control-handler

DaveCTurner approved these changes Apr 11, 2025

View reviewed changes

ywangd reviewed Apr 11, 2025

View reviewed changes

mhl-b added 2 commits April 11, 2025 11:28

feedback

6b3083e

Merge remote-tracking branch 'upstream/main' into netty-flow-control-…

833884b

…handler

Merge branch 'main' into netty-flow-control-handler

33ec141

mhl-b merged commit c8805b8 into elastic:main Apr 11, 2025
17 checks passed

brianseeders added a commit to brianseeders/elasticsearch that referenced this pull request Apr 17, 2025

Revert "Add flow-control and remove auto-read in netty4 HTTP pipeline (…

98570f3

…elastic#126441)" This reverts commit c8805b8.

bcully added a commit to bcully/elasticsearch that referenced this pull request Apr 17, 2025

Revert "Add flow-control and remove auto-read in netty4 HTTP pipeline (…

5385456

…elastic#126441)" This reverts commit c8805b8.

brianseeders mentioned this pull request Apr 17, 2025

Revert #126441 Add flow-control and remove auto-read in netty4 HTTP pipeline #127030

Merged

ankikuma added a commit to ankikuma/elasticsearch that referenced this pull request Apr 17, 2025

Revert "Add flow-control and remove auto-read in netty4 HTTP pipeline (…

8f2eab0

…elastic#126441)" This reverts commit c8805b8. Commit

DaveCTurner added a commit to DaveCTurner/elasticsearch that referenced this pull request Apr 18, 2025

Reinstate elastic#126441 Add flow-control and remove auto-read in net…

0190b12

…ty4 HTTP pipeline This reverts commit 2a243d8.

DaveCTurner mentioned this pull request Apr 23, 2025

Replace auto-read with proper flow-control in HTTP pipeline #127259

Merged

DaveCTurner added a commit that referenced this pull request Apr 25, 2025

Replace auto-read with proper flow-control in HTTP pipeline (#127259)

3cf7061

Re-applying #126441 with the extra `FlowControlHandler` needed to ensure one-chunk-per-read semantics - see #127111 for related tests.

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Add flow-control and remove auto-read in netty4 HTTP pipeline #126441

Add flow-control and remove auto-read in netty4 HTTP pipeline #126441

mhl-b commented Apr 8, 2025 •

edited

Loading

elasticsearchmachine commented Apr 8, 2025

elasticsearchmachine commented Apr 8, 2025

DaveCTurner left a comment

DaveCTurner Apr 10, 2025

DaveCTurner Apr 10, 2025

DaveCTurner Apr 10, 2025

mhl-b Apr 10, 2025 •

edited

Loading

mhl-b Apr 10, 2025

DaveCTurner Apr 10, 2025

DaveCTurner Apr 10, 2025

DaveCTurner Apr 10, 2025

mhl-b Apr 11, 2025

DaveCTurner Apr 10, 2025

mhl-b Apr 10, 2025

DaveCTurner Apr 10, 2025

DaveCTurner Apr 10, 2025

mhl-b Apr 10, 2025 •

edited

Loading

DaveCTurner Apr 10, 2025

mhl-b Apr 10, 2025

DaveCTurner Apr 10, 2025

mhl-b Apr 10, 2025 •

edited

Loading

mhl-b commented Apr 11, 2025

DaveCTurner left a comment

ywangd left a comment

ywangd Apr 11, 2025

mhl-b Apr 11, 2025 •

edited

Loading

ywangd Apr 11, 2025

ywangd Apr 11, 2025

ywangd Apr 11, 2025

mhl-b Apr 11, 2025 •

edited

Loading

mhl-b commented Apr 11, 2025

mhl-b commented Apr 11, 2025 •

edited

Loading

	MissingReadDetector(TimeProvider timer, long missingReadInterval) {
	MissingReadDetector(TimeProvider timer, long missingReadIntervalMillis) {

		private boolean droppingContent;
		private boolean validatingRequest;

		if (now >= lastRead + interval) {
		logger.warn("chan-id={} haven't read from channel for [{}ms]", ctx.channel().id(), (now - lastRead));

Add flow-control and remove auto-read in netty4 HTTP pipeline #126441

Add flow-control and remove auto-read in netty4 HTTP pipeline #126441

Conversation

mhl-b commented Apr 8, 2025 • edited Loading

elasticsearchmachine commented Apr 8, 2025

elasticsearchmachine commented Apr 8, 2025

DaveCTurner left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

mhl-b Apr 10, 2025 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

mhl-b Apr 10, 2025 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

mhl-b Apr 10, 2025 • edited Loading

Choose a reason for hiding this comment

mhl-b commented Apr 11, 2025

DaveCTurner left a comment

Choose a reason for hiding this comment

ywangd left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

mhl-b Apr 11, 2025 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

mhl-b Apr 11, 2025 • edited Loading

Choose a reason for hiding this comment

mhl-b commented Apr 11, 2025

mhl-b commented Apr 11, 2025 • edited Loading

mhl-b commented Apr 8, 2025 •

edited

Loading

mhl-b Apr 10, 2025 •

edited

Loading

mhl-b Apr 10, 2025 •

edited

Loading

mhl-b Apr 10, 2025 •

edited

Loading

mhl-b Apr 11, 2025 •

edited

Loading

mhl-b Apr 11, 2025 •

edited

Loading

mhl-b commented Apr 11, 2025 •

edited

Loading