Skip to content

Defer connectionEstablished observer callback and H2 auto-read until #3461

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
daschl wants to merge 2 commits into
base: main
Choose a base branch
from
Open

Defer connectionEstablished observer callback and H2 auto-read until #3461

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
c76ec97
Defer connectionEstablished observer callback and H2 auto-read until
daschl Mar 23, 2026
385ba70
Fix comments
daschl Apr 28, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
11 changes: 8 additions & 3 deletions ...etalk-http-netty/src/main/java/io/servicetalk/http/netty/DeferredServerChannelBinder.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -69,11 +69,16 @@ static Single<HttpServerContext> bind(final HttpExecutionContext executionContex

return TcpServerBinder.bind(listenAddress, tcpConfig, executionContext, connectionAcceptor, channelInit,
serverConnection -> {
// Start processing requests on http/1.1 connection:
// Notify connection established after all acceptors have completed:
if (serverConnection instanceof NettyHttpServerConnection) {
((NettyHttpServerConnection) serverConnection).process(true);
((NettyHttpServerConnection) serverConnection).notifyConnectionEstablishedAndProcess();
} else if (serverConnection instanceof H2ServerParentConnectionContext) {
((H2ServerParentConnectionContext) serverConnection)
.notifyConnectionEstablishedAndEnableAutoRead();
} else {
throw new IllegalStateException("Unexpected connection type: " +
serverConnection.getClass().getName());
}
Comment thread
daschl marked this conversation as resolved.
// Nothing to do otherwise as h2 uses auto read on the parent channel
}, earlyConnectionAcceptor, lateConnectionAcceptor)
.map(delegate -> {
LOGGER.debug("Started HTTP server with ALPN for address {}", delegate.listenAddress());
Expand Down
3 changes: 2 additions & 1 deletion ...k-http-netty/src/main/java/io/servicetalk/http/netty/H2ClientParentConnectionContext.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -187,7 +187,8 @@ private static final class DefaultH2ClientParentConnection extends AbstractH2Par
HttpHeadersFactory headersFactory,
StreamingHttpRequestResponseFactory reqRespFactory,
ConnectionObserver observer) {
super(connection, delayedCancellable, waitForSslHandshake, observer);
super(connection, delayedCancellable, waitForSslHandshake, observer,
false /* deferAutoRead: client enables auto-read immediately */);
this.subscriber = requireNonNull(subscriber);
this.headersFactory = requireNonNull(headersFactory);
this.reqRespFactory = requireNonNull(reqRespFactory);
Expand Down
9 changes: 6 additions & 3 deletions ...icetalk-http-netty/src/main/java/io/servicetalk/http/netty/H2ParentConnectionContext.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -194,15 +194,18 @@ abstract static class AbstractH2ParentConnection extends ChannelInboundHandlerAd
final boolean waitForSslHandshake;
private final DelayedCancellable delayedCancellable;
final ConnectionObserver observer;
private final boolean deferAutoRead;

AbstractH2ParentConnection(H2ParentConnectionContext parentContext,
DelayedCancellable delayedCancellable,
boolean waitForSslHandshake,
ConnectionObserver observer) {
ConnectionObserver observer,
boolean deferAutoRead) {
this.parentContext = parentContext;
this.delayedCancellable = delayedCancellable;
this.waitForSslHandshake = waitForSslHandshake;
this.observer = observer;
this.deferAutoRead = deferAutoRead;
}

abstract void tryCompleteSubscriber();
Expand All @@ -226,8 +229,8 @@ public final void handlerAdded(ChannelHandlerContext ctx) {
if (channel.isActive()) {
doChannelActive(ctx);
}
if (!channel.config().isAutoRead()) {
// auto read is required for h2
if (!deferAutoRead && !channel.config().isAutoRead()) {
// auto read is required for h2, but might be deferred due to connection acceptors
channel.config().setAutoRead(true);
}
}
Expand Down
52 changes: 44 additions & 8 deletions ...k-http-netty/src/main/java/io/servicetalk/http/netty/H2ServerParentConnectionContext.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -44,6 +44,7 @@
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.EventLoop;
import io.netty.handler.codec.http2.Http2SettingsFrame;
import io.netty.handler.codec.http2.Http2StreamChannel;
import org.slf4j.Logger;
Expand All @@ -66,15 +67,21 @@
final class H2ServerParentConnectionContext extends H2ParentConnectionContext implements ServerContext {
private static final Logger LOGGER = LoggerFactory.getLogger(H2ServerParentConnectionContext.class);
private final SocketAddress listenAddress;
@Nullable
private ConnectionObserver observer;
@Nullable
private DefaultH2ServerParentConnection parentConnectionHandler;
private H2ServerParentConnectionContext(final Channel channel, final HttpExecutionContext executionContext,
Comment thread
daschl marked this conversation as resolved.
final FlushStrategy flushStrategy,
final long idleTimeoutMs,
@Nullable final SslConfig sslConfig,
@Nullable final SSLSession sslSession,
final SocketAddress listenAddress,
final KeepAliveManager keepAliveManager) {
final KeepAliveManager keepAliveManager,
final ConnectionObserver observer) {
super(channel, executionContext, flushStrategy, idleTimeoutMs, sslConfig, sslSession, keepAliveManager);
this.listenAddress = requireNonNull(listenAddress);
this.observer = requireNonNull(observer);
}

@Override
Expand All @@ -87,6 +94,29 @@ public SocketAddress listenAddress() {
return listenAddress;
}

/**
* Notifies the observer and enables auto-read. Must be called after all connection acceptors complete.
* Both actions are combined because the {@code multiplexedObserver} must be set before any frames arrive,
* and enabling auto-read is what triggers frame delivery. Always dispatches to the event loop to ensure
* the observer write and {@code setAutoRead(true)} both execute on the thread that processes frames.
*/
void notifyConnectionEstablishedAndEnableAutoRead() {
final EventLoop eventLoop = nettyChannel().eventLoop();
if (eventLoop.inEventLoop()) {
doNotifyAndEnableAutoRead();
} else {
eventLoop.execute(this::doNotifyAndEnableAutoRead);
}
}

private void doNotifyAndEnableAutoRead() {
assert observer != null && parentConnectionHandler != null;
parentConnectionHandler.multiplexedObserver = observer.multiplexedConnectionEstablished(this);
observer = null;
parentConnectionHandler = null;
nettyChannel().config().setAutoRead(true);
}

static Single<HttpServerContext> bind(final HttpExecutionContext executionContext,
final ReadOnlyHttpServerConfig config,
final SocketAddress listenAddress,
Expand All @@ -98,11 +128,12 @@ static Single<HttpServerContext> bind(final HttpExecutionContext executionContex
return failed(newH2ConfigException());
}
final ReadOnlyTcpServerConfig tcpServerConfig = config.tcpConfig();
// Called AFTER all connection acceptors have completed.
return TcpServerBinder.bind(listenAddress, tcpServerConfig, executionContext, connectionAcceptor,
(channel, connectionObserver) -> initChannel(listenAddress, channel, executionContext, config,
new TcpServerChannelInitializer(tcpServerConfig, connectionObserver, executionContext), service,
connectionObserver),
serverConnection -> { /* nothing to do as h2 uses auto read on the parent channel */ },
H2ServerParentConnectionContext::notifyConnectionEstablishedAndEnableAutoRead,
earlyConnectionAcceptor, lateConnectionAcceptor)
.map(delegate -> {
LOGGER.debug("Started HTTP/2 server with prior-knowledge for address {}", delegate.listenAddress());
Expand Down Expand Up @@ -148,11 +179,13 @@ protected void handleSubscribe(final Subscriber<? super H2ServerParentConnection
H2ServerParentConnectionContext connection = new H2ServerParentConnectionContext(channel,
httpExecutionContext, config.tcpConfig().flushStrategy(),
config.tcpConfig().idleTimeoutMs(), sslConfig, sslSession, listenAddress,
new KeepAliveManager(channel, h2ServerConfig.keepAlivePolicy()));
new KeepAliveManager(channel, h2ServerConfig.keepAlivePolicy()), observer);
channel.attr(CHANNEL_CLOSEABLE_KEY).set(connection);
delayedCancellable = new DelayedCancellable();
parentChannelInitializer = new DefaultH2ServerParentConnection(connection, subscriber,
delayedCancellable, shouldWaitForSslHandshake(sslSession, sslConfig), observer);
delayedCancellable, shouldWaitForSslHandshake(sslSession, sslConfig), observer,
true /* deferAutoRead */);
connection.parentConnectionHandler = parentChannelInitializer;

new H2ServerParentChannelInitializer(h2ServerConfig,
new io.netty.channel.ChannelInitializer<Http2StreamChannel>() {
Expand Down Expand Up @@ -190,7 +223,7 @@ protected void initChannel(final Http2StreamChannel streamChannel) {
// ServiceTalk HTTP service handler
new NettyHttpServerConnection(streamConnection, service, HTTP_2_0,
h2ServerConfig.headersFactory(),
config.allowDropTrailersReadFromTransport()).process(false);
config.allowDropTrailersReadFromTransport(), null).process(false);
}
}).init(channel);
} catch (Throwable cause) {
Expand Down Expand Up @@ -221,8 +254,9 @@ private static final class DefaultH2ServerParentConnection extends AbstractH2Par
final Subscriber<? super H2ServerParentConnectionContext> subscriber,
final DelayedCancellable delayedCancellable,
final boolean waitForSslHandshake,
final ConnectionObserver observer) {
super(parentContext, delayedCancellable, waitForSslHandshake, observer);
final ConnectionObserver observer,
final boolean deferAutoRead) {
super(parentContext, delayedCancellable, waitForSslHandshake, observer, deferAutoRead);
this.subscriber = requireNonNull(subscriber);
}

Expand All @@ -231,7 +265,9 @@ void tryCompleteSubscriber() {
if (subscriber != null) {
Subscriber<? super H2ServerParentConnectionContext> subscriberCopy = subscriber;
subscriber = null;
multiplexedObserver = observer.multiplexedConnectionEstablished(parentContext);
// multiplexedObserver will be set via notifyConnectionEstablishedAndEnableAutoRead() in
// connectionConsumer,
// after all connection acceptors pass. Safe default: NoopMultiplexedObserver.INSTANCE
subscriberCopy.onSuccess((H2ServerParentConnectionContext) parentContext);
}
}
Expand Down
27 changes: 19 additions & 8 deletions servicetalk-http-netty/src/main/java/io/servicetalk/http/netty/NettyHttpServer.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -63,7 +63,6 @@
import io.servicetalk.transport.netty.internal.FlushStrategy;
import io.servicetalk.transport.netty.internal.FlushStrategyHolder;
import io.servicetalk.transport.netty.internal.InfluencerConnectionAcceptor;
import io.servicetalk.transport.netty.internal.NettyConnection;
import io.servicetalk.transport.netty.internal.NettyConnectionContext;
import io.servicetalk.transport.netty.internal.NettyConnectionContext.FlushStrategyProvider;

Expand Down Expand Up @@ -144,7 +143,7 @@ static Single<HttpServerContext> bind(final HttpExecutionContext executionContex
(channel, connectionObserver) -> initChannel(channel, executionContext, config,
new TcpServerChannelInitializer(tcpServerConfig, connectionObserver, executionContext), service,
connectionObserver),
serverConnection -> serverConnection.process(true),
serverConnection -> serverConnection.notifyConnectionEstablishedAndProcess(),
earlyConnectionAcceptor, lateConnectionAcceptor)
.map(delegate -> {
LOGGER.debug("Started HTTP/1.1 server for address {}.", delegate.listenAddress());
Expand Down Expand Up @@ -185,10 +184,10 @@ private static Single<NettyHttpServerConnection> initChannel(final Channel chann
closeHandler, tcpConfig.flushStrategy(), tcpConfig.idleTimeoutMs(), tcpConfig.sslConfig(),
initializer.andThen(getChannelInitializer(
getByteBufAllocator(builderExecutionContext.bufferAllocator()), h1Config, closeHandler)),
HTTP_1_1, observer, false, __ -> false)
HTTP_1_1, observer, false, __ -> false, true)
.map(conn -> new NettyHttpServerConnection(conn, service,
HTTP_1_1, h1Config.headersFactory(),
config.allowDropTrailersReadFromTransport())),
HTTP_1_1, h1Config.headersFactory(),
config.allowDropTrailersReadFromTransport(), observer)),
HTTP_1_1, channel);
}

Expand Down Expand Up @@ -265,17 +264,20 @@ public String toString() {

static final class NettyHttpServerConnection extends HttpServiceContext implements NettyConnectionContext {
private final StreamingHttpService service;
private final NettyConnection<Object, Object> connection;
private final DefaultNettyConnection<Object, Object> connection;
private final HttpHeadersFactory headersFactory;
private final HttpExecutionContext executionContext;
private final ChangingFlushStrategy flushStrategy;
private final boolean requireTrailerHeader;
@Nullable
private ConnectionObserver observer;

NettyHttpServerConnection(final NettyConnection<Object, Object> connection,
NettyHttpServerConnection(final DefaultNettyConnection<Object, Object> connection,
final StreamingHttpService service,
final HttpProtocolVersion version,
final HttpHeadersFactory headersFactory,
final boolean requireTrailerHeader) {
final boolean requireTrailerHeader,
@Nullable final ConnectionObserver observer) {
super(headersFactory,
new DefaultHttpResponseFactory(headersFactory, connection.executionContext().bufferAllocator(),
version),
Expand All @@ -292,6 +294,7 @@ static final class NettyHttpServerConnection extends HttpServiceContext implemen
this.flushStrategy = new ChangingFlushStrategy(new FlushStrategyHolder(connection.defaultFlushStrategy()));
connection.updateFlushStrategy((current, isCurrentOriginal) -> flushStrategy);
this.requireTrailerHeader = requireTrailerHeader;
this.observer = observer;
}

void process(final boolean handleMultipleRequests) {
Expand All @@ -309,6 +312,14 @@ void process(final boolean handleMultipleRequests) {
.subscribe(new ErrorLoggingHttpSubscriber(this));
}

void notifyConnectionEstablishedAndProcess() {
if (observer != null) {
connection.notifyConnectionEstablished(observer);
observer = null;
}
process(true);
}

@Override
public Cancellable updateFlushStrategy(final FlushStrategyProvider strategyProvider) {
return flushStrategy.updateFlushStrategy(strategyProvider);
Expand Down
9 changes: 8 additions & 1 deletion servicetalk-http-netty/src/main/java/io/servicetalk/http/netty/OptionalSslNegotiator.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -102,7 +102,14 @@ static Single<HttpServerContext> bind(final HttpExecutionContext executionContex

final Consumer<NettyConnectionContext> connectionConsumer = serverConnection -> {
if (serverConnection instanceof NettyHttpServer.NettyHttpServerConnection) {
((NettyHttpServer.NettyHttpServerConnection) serverConnection).process(true);
((NettyHttpServer.NettyHttpServerConnection) serverConnection)
.notifyConnectionEstablishedAndProcess();
} else if (serverConnection instanceof H2ServerParentConnectionContext) {
((H2ServerParentConnectionContext) serverConnection)
.notifyConnectionEstablishedAndEnableAutoRead();
} else {
throw new IllegalStateException("Unexpected connection type: " +
serverConnection.getClass().getName());
}
};

Expand Down
Loading
Loading