Skip to content

Commit 73a035c

Browse files
DaveCTurnerDaveCTurner
committed
Remove first FlowControlHandler from HTTP pipeline
Today we have a `FlowControlHandler` near the top of the Netty HTTP pipeline in order to hold back a request body while validating the request headers. This is inefficient since once we've validated the headers we can handle the body chunks as fast as they arrive, needing no more flow control. Moreover today we always fork the validation completion back onto the event loop, forcing any available chunks to be buffered in the `FlowControlHandler`. This commit moves the flow-control mechanism into `Netty4HttpHeaderValidator` itself so that we can bypass it on validated message bodies. Morever in the (common) case that validation completes immediately, e.g. because the credentials are available in cache, then with this commit we skip the flow-control-related buffering entirely.
1 parent e586a01 commit 73a035c

File tree

4 files changed

+113
-67
lines changed

4 files changed

+113
-67
lines changed

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

Lines changed: 104 additions & 60 deletions
Original file line numberDiff line numberDiff line change
@@ -15,19 +15,23 @@
1515
import io.netty.handler.codec.http.HttpContent;
1616
import io.netty.handler.codec.http.HttpObject;
1717
import io.netty.handler.codec.http.HttpRequest;
18+
import io.netty.util.ReferenceCounted;
1819

1920
import org.elasticsearch.action.ActionListener;
2021
import org.elasticsearch.action.support.ContextPreservingActionListener;
22+
import org.elasticsearch.action.support.SubscribableListener;
2123
import org.elasticsearch.common.util.concurrent.ThreadContext;
22-
import org.elasticsearch.core.Nullable;
2324
import org.elasticsearch.http.netty4.internal.HttpValidator;
2425
import org.elasticsearch.transport.Transports;
2526

27+
import java.util.ArrayDeque;
28+
2629
public class Netty4HttpHeaderValidator extends ChannelDuplexHandler {
2730

2831
private final HttpValidator validator;
2932
private final ThreadContext threadContext;
30-
private State state;
33+
private State state = State.PASSING;
34+
private final ArrayDeque<Object> buffer = new ArrayDeque<>();
3135

3236
public Netty4HttpHeaderValidator(HttpValidator validator, ThreadContext threadContext) {
3337
this.validator = validator;
@@ -36,80 +40,120 @@ public Netty4HttpHeaderValidator(HttpValidator validator, ThreadContext threadCo
3640

3741
@Override
3842
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
43+
if (state == State.VALIDATING || buffer.size() > 0) {
44+
// there's already some buffered messages that need to be processed before this one, so queue this one up behind them
45+
buffer.offerLast(msg);
46+
return;
47+
}
48+
3949
assert msg instanceof HttpObject;
40-
var httpObject = (HttpObject) msg;
50+
final var httpObject = (HttpObject) msg;
4151
if (httpObject.decoderResult().isFailure()) {
4252
ctx.fireChannelRead(httpObject); // pass-through for decoding failures
53+
} else if (msg instanceof HttpRequest httpRequest) {
54+
validate(ctx, httpRequest);
55+
} else if (state == State.PASSING) {
56+
assert msg instanceof HttpContent;
57+
ctx.fireChannelRead(msg);
4358
} else {
44-
if (msg instanceof HttpRequest request) {
45-
validate(ctx, request);
46-
} else {
47-
assert msg instanceof HttpContent;
48-
var content = (HttpContent) msg;
49-
if (state == State.DROPPING) {
50-
content.release();
51-
ctx.read();
52-
} else {
53-
assert state == State.PASSING : "unexpected content before validation completed";
54-
ctx.fireChannelRead(content);
55-
}
56-
}
59+
assert state == State.DROPPING : state;
60+
assert msg instanceof HttpContent;
61+
final var httpContent = (HttpContent) msg;
62+
httpContent.release();
63+
ctx.read();
5764
}
5865
}
5966

67+
@Override
68+
public void channelReadComplete(ChannelHandlerContext ctx) {
69+
if (buffer.size() == 0) {
70+
ctx.fireChannelReadComplete();
71+
} // else we're buffering messages so will manage the read-complete messages ourselves
72+
}
73+
6074
@Override
6175
public void read(ChannelHandlerContext ctx) throws Exception {
62-
// until validation is completed we can ignore read calls,
63-
// once validation is finished HttpRequest will be fired and downstream can read from there
76+
assert ctx.channel().eventLoop().inEventLoop();
6477
if (state != State.VALIDATING) {
65-
ctx.read();
78+
if (buffer.size() > 0) {
79+
final var message = buffer.pollFirst();
80+
if (message instanceof HttpRequest httpRequest) {
81+
validate(ctx, httpRequest);
82+
} else {
83+
assert message instanceof HttpContent;
84+
assert state == State.PASSING : state; // DROPPING releases any buffered chunks up-front
85+
ctx.fireChannelRead(message);
86+
ctx.fireChannelReadComplete(); // downstream will have to call read() again when it's ready
87+
}
88+
} else {
89+
ctx.read();
90+
}
6691
}
6792
}
6893

69-
void validate(ChannelHandlerContext ctx, HttpRequest request) {
70-
assert Transports.assertDefaultThreadContext(threadContext);
71-
state = State.VALIDATING;
72-
ActionListener.run(
73-
// this prevents thread-context changes to propagate to the validation listener
74-
// atm, the validation listener submits to the event loop executor, which doesn't know about the ES thread-context,
75-
// so this is just a defensive play, in case the code inside the listener changes to not use the event loop executor
76-
ActionListener.assertOnce(
77-
new ContextPreservingActionListener<Void>(
78-
threadContext.wrapRestorable(threadContext.newStoredContext()),
79-
new ActionListener<>() {
80-
@Override
81-
public void onResponse(Void unused) {
82-
handleValidationResult(ctx, request, null);
83-
}
84-
85-
@Override
86-
public void onFailure(Exception e) {
87-
handleValidationResult(ctx, request, e);
88-
}
89-
}
90-
)
91-
),
92-
listener -> {
93-
// this prevents thread-context changes to propagate beyond the validation, as netty worker threads are reused
94-
try (ThreadContext.StoredContext ignore = threadContext.newStoredContext()) {
95-
validator.validate(request, ctx.channel(), listener);
96-
}
97-
}
98-
);
94+
void validate(ChannelHandlerContext ctx, HttpRequest httpRequest) {
95+
final var validationResultListener = new ValidationResultListener(ctx, httpRequest);
96+
SubscribableListener.newForked(validationResultListener::doValidate)
97+
.addListener(
98+
validationResultListener,
99+
// dispatch back to event loop unless validation completed already in which case we can just continue on this thread
100+
// straight away, avoiding the need to buffer any subsequent messages
101+
ctx.channel().eventLoop(),
102+
null
103+
);
99104
}
100105

101-
void handleValidationResult(ChannelHandlerContext ctx, HttpRequest request, @Nullable Exception validationError) {
102-
assert Transports.assertDefaultThreadContext(threadContext);
103-
// Always explicitly dispatch back to the event loop to prevent reentrancy concerns if we are still on event loop
104-
ctx.channel().eventLoop().execute(() -> {
105-
if (validationError != null) {
106-
request.setDecoderResult(DecoderResult.failure(validationError));
107-
state = State.DROPPING;
108-
} else {
109-
state = State.PASSING;
106+
private class ValidationResultListener implements ActionListener<Void> {
107+
108+
private final ChannelHandlerContext ctx;
109+
private final HttpRequest httpRequest;
110+
111+
ValidationResultListener(ChannelHandlerContext ctx, HttpRequest httpRequest) {
112+
this.ctx = ctx;
113+
this.httpRequest = httpRequest;
114+
}
115+
116+
void doValidate(ActionListener<Void> listener) {
117+
assert Transports.assertDefaultThreadContext(threadContext);
118+
assert ctx.channel().eventLoop().inEventLoop();
119+
assert state == State.PASSING || state == State.DROPPING : state;
120+
state = State.VALIDATING;
121+
try (var ignore = threadContext.newEmptyContext()) {
122+
validator.validate(
123+
httpRequest,
124+
ctx.channel(),
125+
new ContextPreservingActionListener<>(threadContext::newEmptyContext, listener)
126+
);
110127
}
111-
ctx.fireChannelRead(request);
112-
});
128+
}
129+
130+
@Override
131+
public void onResponse(Void unused) {
132+
assert Transports.assertDefaultThreadContext(threadContext);
133+
assert ctx.channel().eventLoop().inEventLoop();
134+
assert state == State.VALIDATING : state;
135+
state = State.PASSING;
136+
fireChannelRead();
137+
}
138+
139+
@Override
140+
public void onFailure(Exception e) {
141+
assert Transports.assertDefaultThreadContext(threadContext);
142+
assert ctx.channel().eventLoop().inEventLoop();
143+
assert state == State.VALIDATING : state;
144+
httpRequest.setDecoderResult(DecoderResult.failure(e));
145+
state = State.DROPPING;
146+
while (buffer.isEmpty() == false && buffer.peekFirst() instanceof HttpRequest == false) {
147+
assert buffer.peekFirst() instanceof HttpContent;
148+
((ReferenceCounted) buffer.pollFirst()).release();
149+
}
150+
fireChannelRead();
151+
}
152+
153+
private void fireChannelRead() {
154+
ctx.fireChannelRead(httpRequest);
155+
ctx.fireChannelReadComplete(); // downstream needs to read() again
156+
}
113157
}
114158

115159
private enum State {

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

Lines changed: 0 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -371,7 +371,6 @@ protected HttpMessage createMessage(String[] initialLine) throws Exception {
371371
ch.pipeline().addLast("decoder", decoder); // parses the HTTP bytes request into HTTP message pieces
372372

373373
// from this point in pipeline every handler must call ctx or channel #read() when ready to process next HTTP part
374-
ch.pipeline().addLast(new FlowControlHandler());
375374
if (Assertions.ENABLED) {
376375
// missing reads are hard to catch, but we can detect absence of reads within interval
377376
long missingReadIntervalMs = 10_000;

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

Lines changed: 1 addition & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -143,8 +143,7 @@ public void testIgnoreReadWhenValidating() {
143143
asInstanceOf(LastHttpContent.class, channel.readInbound()).release();
144144
}
145145

146-
public void testWithFlowControlAndAggregator() {
147-
channel.pipeline().addFirst(new FlowControlHandler());
146+
public void testWithAggregator() {
148147
channel.pipeline().addLast(new Netty4HttpAggregator(8192, (req) -> true, new HttpRequestDecoder()));
149148

150149
channel.writeInbound(newHttpRequest());

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

Lines changed: 8 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -63,6 +63,7 @@
6363
import org.elasticsearch.common.transport.BoundTransportAddress;
6464
import org.elasticsearch.common.transport.TransportAddress;
6565
import org.elasticsearch.common.unit.ByteSizeValue;
66+
import org.elasticsearch.common.util.concurrent.EsExecutors;
6667
import org.elasticsearch.common.util.concurrent.ThreadContext;
6768
import org.elasticsearch.core.TimeValue;
6869
import org.elasticsearch.core.Tuple;
@@ -120,6 +121,7 @@
120121
import static org.hamcrest.Matchers.emptyIterable;
121122
import static org.hamcrest.Matchers.equalTo;
122123
import static org.hamcrest.Matchers.hasSize;
124+
import static org.hamcrest.Matchers.in;
123125
import static org.hamcrest.Matchers.instanceOf;
124126
import static org.hamcrest.Matchers.is;
125127
import static org.hamcrest.Matchers.iterableWithSize;
@@ -976,7 +978,7 @@ public void testMultipleValidationsOnTheSameChannel() throws InterruptedExceptio
976978
final HttpServerTransport.Dispatcher dispatcher = new HttpServerTransport.Dispatcher() {
977979
@Override
978980
public void dispatchRequest(final RestRequest request, final RestChannel channel, final ThreadContext threadContext) {
979-
assertThat(okURIs.contains(request.uri()), is(true));
981+
assertThat(request.uri(), in(okURIs));
980982
// assert validated request is dispatched
981983
okURIs.remove(request.uri());
982984
channel.sendResponse(new RestResponse(OK, RestResponse.TEXT_CONTENT_TYPE, new BytesArray("dispatch OK")));
@@ -985,7 +987,7 @@ public void dispatchRequest(final RestRequest request, final RestChannel channel
985987
@Override
986988
public void dispatchBadRequest(final RestChannel channel, final ThreadContext threadContext, final Throwable cause) {
987989
// assert unvalidated request is NOT dispatched
988-
assertThat(nokURIs.contains(channel.request().uri()), is(true));
990+
assertThat(channel.request().uri(), in(nokURIs));
989991
nokURIs.remove(channel.request().uri());
990992
try {
991993
channel.sendResponse(new RestResponse(channel, (Exception) ((ElasticsearchWrapperException) cause).getCause()));
@@ -1000,9 +1002,11 @@ public void dispatchBadRequest(final RestChannel channel, final ThreadContext th
10001002
assertThat(channelSetOnce.get(), is(channel));
10011003
// some requests are validated while others are not
10021004
if (httpPreRequest.uri().contains("X-Auth=OK")) {
1003-
validationListener.onResponse(null);
1005+
randomFrom(EsExecutors.DIRECT_EXECUTOR_SERVICE, channel.eventLoop()).execute(() -> validationListener.onResponse(null));
10041006
} else if (httpPreRequest.uri().contains("X-Auth=NOK")) {
1005-
validationListener.onFailure(new ElasticsearchSecurityException("Boom", UNAUTHORIZED));
1007+
randomFrom(EsExecutors.DIRECT_EXECUTOR_SERVICE, channel.eventLoop()).execute(
1008+
() -> validationListener.onFailure(new ElasticsearchSecurityException("Boom", UNAUTHORIZED))
1009+
);
10061010
} else {
10071011
throw new AssertionError("Unrecognized URI");
10081012
}

0 commit comments

Comments
 (0)