Skip to content

Commit

Permalink
Fixes to QPACK configuration from SETTINGS frames. (#9728)
Browse files Browse the repository at this point in the history 
* Improved configuration upon sending/receiving SETTINGS frames.
* Improved handling of blocked frames in HTTP3StreamConnection.
* Fixed locking and reentrance in QPACK encoder/decoder.

Signed-off-by: Lachlan Roberts <[email protected]>
Signed-off-by: Simone Bordet <[email protected]>
Co-authored-by: Simone Bordet <[email protected]>
  • Loading branch information
@lachlan-roberts @sbordet
lachlan-roberts and sbordet authored May 15, 2023
1 parent 7ac49cd commit 6e8457c
Show file tree
Hide file tree
Showing 38 changed files with 883 additions and 378 deletions.
92 changes: 83 additions & 9 deletions ...ttp3-client/src/main/java/org/eclipse/jetty/http3/client/internal/ClientHTTP3Session.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,7 @@

package org.eclipse.jetty.http3.client.internal;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.CompletableFuture;

Expand Down Expand Up @@ -44,6 +45,7 @@ public class ClientHTTP3Session extends ClientProtocolSession
{
private static final Logger LOG = LoggerFactory.getLogger(ClientHTTP3Session.class);

private final HTTP3Configuration configuration;
private final HTTP3SessionClient session;
private final QpackEncoder encoder;
private final QpackDecoder decoder;
Expand All @@ -53,7 +55,8 @@ public class ClientHTTP3Session extends ClientProtocolSession
public ClientHTTP3Session(HTTP3Configuration configuration, ClientQuicSession quicSession, Session.Client.Listener listener, Promise<Session.Client> promise)
{
super(quicSession);
this.session = new HTTP3SessionClient(this, listener, promise);
this.configuration = configuration;
session = new HTTP3SessionClient(this, listener, promise);
addBean(session);
session.setStreamIdleTimeout(configuration.getStreamIdleTimeout());

Expand All @@ -63,27 +66,28 @@ public ClientHTTP3Session(HTTP3Configuration configuration, ClientQuicSession qu
long encoderStreamId = getQuicSession().newStreamId(StreamType.CLIENT_UNIDIRECTIONAL);
QuicStreamEndPoint encoderEndPoint = openInstructionEndPoint(encoderStreamId);
InstructionFlusher encoderInstructionFlusher = new InstructionFlusher(quicSession, encoderEndPoint, EncoderStreamConnection.STREAM_TYPE);
this.encoder = new QpackEncoder(new InstructionHandler(encoderInstructionFlusher), configuration.getMaxBlockedStreams());
encoder = new QpackEncoder(new InstructionHandler(encoderInstructionFlusher));
encoder.setMaxHeadersSize(configuration.getMaxRequestHeadersSize());
addBean(encoder);
if (LOG.isDebugEnabled())
LOG.debug("created encoder stream #{} on {}", encoderStreamId, encoderEndPoint);

long decoderStreamId = getQuicSession().newStreamId(StreamType.CLIENT_UNIDIRECTIONAL);
QuicStreamEndPoint decoderEndPoint = openInstructionEndPoint(decoderStreamId);
InstructionFlusher decoderInstructionFlusher = new InstructionFlusher(quicSession, decoderEndPoint, DecoderStreamConnection.STREAM_TYPE);
this.decoder = new QpackDecoder(new InstructionHandler(decoderInstructionFlusher), configuration.getMaxResponseHeadersSize());
decoder = new QpackDecoder(new InstructionHandler(decoderInstructionFlusher));
addBean(decoder);
if (LOG.isDebugEnabled())
LOG.debug("created decoder stream #{} on {}", decoderStreamId, decoderEndPoint);

long controlStreamId = getQuicSession().newStreamId(StreamType.CLIENT_UNIDIRECTIONAL);
QuicStreamEndPoint controlEndPoint = openControlEndPoint(controlStreamId);
this.controlFlusher = new ControlFlusher(quicSession, controlEndPoint, true);
controlFlusher = new ControlFlusher(quicSession, controlEndPoint, true);
addBean(controlFlusher);
if (LOG.isDebugEnabled())
LOG.debug("created control stream #{} on {}", controlStreamId, controlEndPoint);

this.messageFlusher = new MessageFlusher(quicSession.getByteBufferPool(), encoder, configuration.getMaxRequestHeadersSize(), configuration.isUseOutputDirectByteBuffers());
messageFlusher = new MessageFlusher(quicSession.getByteBufferPool(), encoder, configuration.isUseOutputDirectByteBuffers());
addBean(messageFlusher);
}

Expand All @@ -105,16 +109,86 @@ public HTTP3SessionClient getSessionClient()
@Override
protected void onStart()
{
// Queue the mandatory SETTINGS frame.
Map<Long, Long> settings = session.onPreface();
if (settings == null)
settings = Map.of();
// TODO: add default settings.
settings = settings != null ? new HashMap<>(settings) : new HashMap<>();

settings.compute(SettingsFrame.MAX_TABLE_CAPACITY, (k, v) ->
{
if (v == null)
{
v = (long)configuration.getMaxDecoderTableCapacity();
if (v == 0)
v = null;
}
return v;
});
settings.compute(SettingsFrame.MAX_FIELD_SECTION_SIZE, (k, v) ->
{
if (v == null)
{
v = (long)configuration.getMaxResponseHeadersSize();
if (v <= 0)
v = null;
}
return v;
});
settings.compute(SettingsFrame.MAX_BLOCKED_STREAMS, (k, v) ->
{
if (v == null)
{
v = (long)configuration.getMaxBlockedStreams();
if (v == 0)
v = null;
}
return v;
});

if (LOG.isDebugEnabled())
LOG.debug("configuring local {} on {}", settings, this);

settings.forEach((key, value) ->
{
if (key == SettingsFrame.MAX_TABLE_CAPACITY)
decoder.setMaxTableCapacity(value.intValue());
else if (key == SettingsFrame.MAX_FIELD_SECTION_SIZE)
decoder.setMaxHeadersSize(value.intValue());
else if (key == SettingsFrame.MAX_BLOCKED_STREAMS)
decoder.setMaxBlockedStreams(value.intValue());
});

// Queue the mandatory SETTINGS frame.
SettingsFrame frame = new SettingsFrame(settings);
if (controlFlusher.offer(frame, Callback.from(Invocable.InvocationType.NON_BLOCKING, session::onOpen, this::failControlStream)))
controlFlusher.iterate();
}

public void onSettings(SettingsFrame frame)
{
Map<Long, Long> settings = frame.getSettings();
if (LOG.isDebugEnabled())
LOG.debug("configuring encoder {} on {}", settings, this);
settings.forEach((key, value) ->
{
if (key == SettingsFrame.MAX_TABLE_CAPACITY)
{
int maxTableCapacity = value.intValue();
encoder.setMaxTableCapacity(maxTableCapacity);
encoder.setTableCapacity(Math.min(maxTableCapacity, configuration.getInitialEncoderTableCapacity()));
}
else if (key == SettingsFrame.MAX_FIELD_SECTION_SIZE)
{
// Must cap the maxHeaderSize to avoid large allocations.
int maxHeadersSize = Math.min(value.intValue(), configuration.getMaxRequestHeadersSize());
encoder.setMaxHeadersSize(maxHeadersSize);
}
else if (key == SettingsFrame.MAX_BLOCKED_STREAMS)
{
int maxBlockedStreams = value.intValue();
encoder.setMaxBlockedStreams(maxBlockedStreams);
}
});
}

private void failControlStream(Throwable failure)
{
long error = HTTP3ErrorCode.CLOSED_CRITICAL_STREAM_ERROR.code();
Expand Down
34 changes: 12 additions & 22 deletions ...ttp3-client/src/main/java/org/eclipse/jetty/http3/client/internal/HTTP3SessionClient.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@
import org.eclipse.jetty.http3.frames.Frame;
import org.eclipse.jetty.http3.frames.GoAwayFrame;
import org.eclipse.jetty.http3.frames.HeadersFrame;
import org.eclipse.jetty.http3.frames.SettingsFrame;
import org.eclipse.jetty.http3.internal.HTTP3ErrorCode;
import org.eclipse.jetty.http3.internal.HTTP3Session;
import org.eclipse.jetty.quic.common.ProtocolSession;
Expand Down Expand Up @@ -63,7 +64,7 @@ protected HTTP3StreamClient newHTTP3Stream(QuicStreamEndPoint endPoint, boolean
}

@Override
public void onHeaders(long streamId, HeadersFrame frame)
public void onHeaders(long streamId, HeadersFrame frame, boolean wasBlocked)
{
if (frame.getMetaData().isResponse())
{
Expand All @@ -76,10 +77,19 @@ public void onHeaders(long streamId, HeadersFrame frame)
}
else
{
super.onHeaders(streamId, frame);
super.onHeaders(streamId, frame, wasBlocked);
}
}

@Override
public void onSettings(SettingsFrame frame)
{
if (LOG.isDebugEnabled())
LOG.debug("received {} on {}", frame, this);
getProtocolSession().onSettings(frame);
super.onSettings(frame);
}

@Override
public CompletableFuture<Stream> newRequest(HeadersFrame frame, Stream.Client.Listener listener)
{
Expand Down Expand Up @@ -147,24 +157,4 @@ protected GoAwayFrame newGoAwayFrame(boolean graceful)
return GoAwayFrame.CLIENT_GRACEFUL;
return super.newGoAwayFrame(graceful);
}

@Override
protected void onSettingMaxTableCapacity(long value)
{
getProtocolSession().getQpackEncoder().setCapacity((int)value);
}

@Override
protected void onSettingMaxFieldSectionSize(long value)
{
getProtocolSession().getQpackDecoder().setMaxHeaderSize((int)value);
}

@Override
protected void onSettingMaxBlockedStreams(long value)
{
ClientHTTP3Session session = getProtocolSession();
session.getQpackDecoder().setMaxBlockedStreams((int)value);
session.getQpackEncoder().setMaxBlockedStreams((int)value);
}
}
Loading

0 comments on commit 6e8457c

Please sign in to comment.