Skip to content

Commit e103380

Browse files
ok2cok2c
committed
RFC 9113 conformance: improved H2 stream creation and initialization, improved H2 stream state tracking and tracking of total stream counts
1 parent 1b0c1f5 commit e103380

6 files changed

Lines changed: 269 additions & 199 deletions

File tree

‎httpcore5-h2/src/main/java/org/apache/hc/core5/http2/impl/nio/AbstractH2StreamMultiplexer.java‎

Lines changed: 74 additions & 93 deletions
Original file line numberDiff line numberDiff line change
@@ -57,13 +57,14 @@
5757
import org.apache.hc.core5.http.impl.BasicEndpointDetails;
5858
import org.apache.hc.core5.http.impl.BasicHttpConnectionMetrics;
5959
import org.apache.hc.core5.http.impl.CharCodingSupport;
60+
import org.apache.hc.core5.http.nio.AsyncClientExchangeHandler;
6061
import org.apache.hc.core5.http.nio.AsyncPushConsumer;
6162
import org.apache.hc.core5.http.nio.AsyncPushProducer;
6263
import org.apache.hc.core5.http.nio.HandlerFactory;
6364
import org.apache.hc.core5.http.nio.command.CommandSupport;
64-
import org.apache.hc.core5.http.nio.command.ExecutableCommand;
65+
import org.apache.hc.core5.http.nio.command.RequestExecutionCommand;
6566
import org.apache.hc.core5.http.nio.command.ShutdownCommand;
66-
import org.apache.hc.core5.http.protocol.HttpCoreContext;
67+
import org.apache.hc.core5.http.protocol.HttpContext;
6768
import org.apache.hc.core5.http.protocol.HttpProcessor;
6869
import org.apache.hc.core5.http2.H2ConnectionException;
6970
import org.apache.hc.core5.http2.H2Error;
@@ -172,6 +173,18 @@ public String getId() {
172173
return ioSession.getId();
173174
}
174175

176+
BasicHttpConnectionMetrics getConnMetrics() {
177+
return connMetrics;
178+
}
179+
180+
HttpProcessor getHttpProcessor() {
181+
return httpProcessor;
182+
}
183+
184+
void submitCommand(final Command command) {
185+
ioSession.enqueue(command, Command.Priority.NORMAL);
186+
}
187+
175188
abstract void validateSetting(H2Param param, int value) throws H2ConnectionException;
176189

177190
abstract H2Setting[] generateSettings(H2Config localConfig);
@@ -182,18 +195,18 @@ public String getId() {
182195

183196
abstract void acceptPushFrame() throws H2ConnectionException;
184197

185-
abstract H2StreamHandler createRemotelyInitiatedStream(
186-
H2StreamChannel channel,
198+
abstract H2StreamHandler incomingRequest(H2StreamChannel channel) throws IOException;
199+
200+
abstract H2StreamHandler incomingPushPromise(H2StreamChannel channel,
201+
HandlerFactory<AsyncPushConsumer> pushHandlerFactory) throws IOException;
187202

188-
HttpProcessor httpProcessor,
189-
BasicHttpConnectionMetrics connMetrics,
190-
HandlerFactory<AsyncPushConsumer> pushHandlerFactory) throws IOException;
203+
abstract H2StreamHandler outgoingRequest(H2StreamChannel channel,
204+
AsyncClientExchangeHandler exchangeHandler,
205+
HandlerFactory<AsyncPushConsumer> pushHandlerFactory,
206+
HttpContext context) throws IOException;
191207

192-
abstract H2StreamHandler createLocallyInitiatedStream(
193-
ExecutableCommand command,
194-
H2StreamChannel channel,
195-
HttpProcessor httpProcessor,
196-
BasicHttpConnectionMetrics connMetrics) throws IOException;
208+
abstract H2StreamHandler outgoingPushPromise(H2StreamChannel channel,
209+
AsyncPushProducer pushProducer) throws IOException;
197210

198211
abstract boolean allowGracefulAbort(H2Stream stream);
199212

@@ -495,20 +508,20 @@ public final void onOutput() throws HttpException, IOException {
495508
executeShutdown((ShutdownCommand) command);
496509
} else if (command instanceof PingCommand) {
497510
executePing((PingCommand) command);
498-
} else if (command instanceof ExecutableCommand) {
499-
executeRequest((ExecutableCommand) command);
500-
if (!outputQueue.isEmpty()) {
501-
return;
502-
}
511+
} else if (command instanceof RequestExecutionCommand) {
512+
executeRequest((RequestExecutionCommand) command);
513+
}
514+
if (!outputQueue.isEmpty()) {
515+
return;
503516
}
504517
}
505518
}
506519
if (connState.compareTo(ConnectionHandshake.GRACEFUL_SHUTDOWN) == 0) {
507520
int liveStreams = 0;
508521
for (final Iterator<H2Stream> it = streams.iterator(); it.hasNext(); ) {
509522
final H2Stream stream = it.next();
510-
if (stream.isLocalClosed() && stream.isRemoteClosed()) {
511-
streams.release(stream);
523+
if (stream.isClosedPastLingerDeadline()) {
524+
streams.dropStreamId(stream.getId());
512525
it.remove();
513526
} else {
514527
if (streams.isSameSide(stream.getId()) || stream.getId() <= streams.getLastRemoteId()) {
@@ -555,10 +568,10 @@ public final void onTimeout(final Timeout timeout) throws HttpException, IOExcep
555568
final RawFrame goAway;
556569
if (localSettingState != SettingsHandshake.ACKED) {
557570
goAway = frameFactory.createGoAway(streams.getLastRemoteId(), H2Error.SETTINGS_TIMEOUT,
558-
"Setting timeout (" + timeout + ")");
571+
"Setting timeout (" + timeout + ")");
559572
} else {
560573
goAway = frameFactory.createGoAway(streams.getLastRemoteId(), H2Error.NO_ERROR,
561-
"Timeout due to inactivity (" + timeout + ")");
574+
"Timeout due to inactivity (" + timeout + ")");
562575
}
563576
commitFrame(goAway);
564577
for (final Iterator<H2Stream> it = streams.iterator(); it.hasNext(); ) {
@@ -601,14 +614,13 @@ private void executePing(final PingCommand pingCommand) throws IOException {
601614
commitFrame(ping);
602615
}
603616

604-
private void executeRequest(final ExecutableCommand executableCommand) throws IOException, HttpException {
617+
private void executeRequest(final RequestExecutionCommand requestExecutionCommand) throws IOException, HttpException {
605618
final int streamId = streams.generateStreamId();
606-
final H2StreamChannel channel = createChannel(streamId, true);
607-
final H2StreamHandler streamHandler = createLocallyInitiatedStream(
608-
executableCommand, channel, httpProcessor, connMetrics);
609-
610-
final H2Stream stream = new H2Stream(channel, streamHandler);
611-
streams.addLocallyInitiated(stream);
619+
final H2StreamChannel channel = createChannel(streamId);
620+
final H2Stream stream = streams.createActive(channel, outgoingRequest(channel,
621+
requestExecutionCommand.getExchangeHandler(),
622+
requestExecutionCommand.getPushHandlerFactory(),
623+
requestExecutionCommand.getContext()));
612624

613625
if (streamListener != null) {
614626
final int initInputWindow = stream.getInputWindow().get();
@@ -620,23 +632,12 @@ private void executeRequest(final ExecutableCommand executableCommand) throws IO
620632
if (stream.isOutputReady()) {
621633
stream.produceOutput();
622634
}
623-
final CancellableDependency cancellableDependency = executableCommand.getCancellableDependency();
635+
final CancellableDependency cancellableDependency = requestExecutionCommand.getCancellableDependency();
624636
if (cancellableDependency != null) {
625637
cancellableDependency.setDependency(stream::abort);
626638
}
627639
}
628640

629-
public void executePush(final int promisedStreamId, final AsyncPushProducer pushProducer) {
630-
final H2StreamChannel channel = createChannel(promisedStreamId, true);
631-
final HttpCoreContext context = HttpCoreContext.create();
632-
context.setSSLSession(getSSLSession());
633-
context.setEndpointDetails(getEndpointDetails());
634-
final H2StreamHandler streamHandler = new ServerPushH2StreamHandler(
635-
channel, httpProcessor, connMetrics, pushProducer, context);
636-
final H2Stream stream = new H2Stream(channel, streamHandler);
637-
streams.addLocallyInitiated(stream);
638-
}
639-
640641
public final void onException(final Exception cause) {
641642
try {
642643
for (;;) {
@@ -702,8 +703,8 @@ private void consumeFrame(final RawFrame frame) throws HttpException, IOExceptio
702703
stream.localReset(ex, ex.getCause() != null ? H2Error.INTERNAL_ERROR : H2Error.CANCEL);
703704
}
704705

705-
if (stream.isTerminated()) {
706-
streams.release(stream);
706+
if (stream.isClosed()) {
707+
stream.releaseResources();
707708
requestSessionOutput();
708709
}
709710
}
@@ -722,26 +723,20 @@ private void consumeFrame(final RawFrame frame) throws HttpException, IOExceptio
722723
throw new H2ConnectionException(H2Error.PROTOCOL_ERROR, "GOAWAY received");
723724
}
724725

725-
final H2StreamChannel channel = createChannel(streamId, false);
726-
final H2StreamHandler streamHandler;
726+
final H2StreamChannel channel = createChannel(streamId);
727727
if (connState.compareTo(ConnectionHandshake.ACTIVE) <= 0) {
728-
streamHandler = createRemotelyInitiatedStream(channel, httpProcessor, connMetrics, null);
728+
stream = streams.createActive(channel, incomingRequest(channel));
729729
} else {
730-
streamHandler = NoopH2StreamHandler.INSTANCE;
731-
channel.markLocalClosed();
732-
}
733-
734-
stream = new H2Stream(channel, streamHandler);
735-
if (stream.isOutputReady()) {
736-
stream.produceOutput();
730+
channel.localReset(H2Error.REFUSED_STREAM);
731+
stream = streams.createActive(channel, NoopH2StreamHandler.INSTANCE);
737732
}
738-
streams.addRemotelyInitiated(stream);
739733
} else if (stream.isLocalClosed() && stream.isRemoteClosed()) {
740734
throw new H2ConnectionException(H2Error.STREAM_CLOSED, "Stream closed");
735+
} else if (stream.isReserved()) {
736+
stream.activate();
741737
}
742738
try {
743739
consumeHeaderFrame(frame, stream);
744-
745740
if (stream.isOutputReady()) {
746741
stream.produceOutput();
747742
}
@@ -753,8 +748,8 @@ private void consumeFrame(final RawFrame frame) throws HttpException, IOExceptio
753748
stream.handle(ex);
754749
}
755750

756-
if (stream.isTerminated()) {
757-
streams.release(stream);
751+
if (stream.isClosed()) {
752+
stream.releaseResources();
758753
requestSessionOutput();
759754
}
760755
}
@@ -779,8 +774,8 @@ private void consumeFrame(final RawFrame frame) throws HttpException, IOExceptio
779774
stream.localReset(ex, ex.getCause() != null ? H2Error.INTERNAL_ERROR : H2Error.CANCEL);
780775
}
781776

782-
if (stream.isTerminated()) {
783-
streams.release(stream);
777+
if (stream.isClosed()) {
778+
stream.releaseResources();
784779
requestSessionOutput();
785780
}
786781
}
@@ -829,7 +824,6 @@ private void consumeFrame(final RawFrame frame) throws HttpException, IOExceptio
829824
requestSessionOutput();
830825
} else {
831826
stream.fail(new H2StreamResetException(errorCode, "Stream reset (" + errorCode + ")"));
832-
streams.release(stream);
833827
requestSessionOutput();
834828
}
835829
}
@@ -900,7 +894,7 @@ private void consumeFrame(final RawFrame frame) throws HttpException, IOExceptio
900894
throw new H2ConnectionException(H2Error.PROTOCOL_ERROR, "Push is disabled");
901895
}
902896

903-
final H2Stream stream = streams.lookup(streamId);
897+
final H2Stream stream = streams.lookupValid(streamId);
904898
if (stream.isRemoteClosed()) {
905899
stream.localReset(new H2StreamResetException(H2Error.STREAM_CLOSED, "Stream closed"));
906900
break;
@@ -918,19 +912,14 @@ private void consumeFrame(final RawFrame frame) throws HttpException, IOExceptio
918912
throw new H2ConnectionException(H2Error.PROTOCOL_ERROR, "Stream already open: " + promisedStreamId);
919913
}
920914

921-
final H2StreamChannel channel = createChannel(promisedStreamId, false);
922-
final H2StreamHandler streamHandler;
915+
final H2StreamChannel channel = createChannel(promisedStreamId);
916+
final H2Stream promisedStream;
923917
if (connState.compareTo(ConnectionHandshake.ACTIVE) <= 0) {
924-
streamHandler = createRemotelyInitiatedStream(channel, httpProcessor, connMetrics,
925-
stream.getPushHandlerFactory());
918+
promisedStream = streams.createReserved(channel, incomingPushPromise(channel, stream.getPushHandlerFactory()));
926919
} else {
927-
streamHandler = NoopH2StreamHandler.INSTANCE;
928-
channel.markLocalClosed();
920+
channel.localReset(H2Error.REFUSED_STREAM);
921+
promisedStream = streams.createActive(channel, NoopH2StreamHandler.INSTANCE);
929922
}
930-
931-
final H2Stream promisedStream = new H2Stream(channel, streamHandler);
932-
streams.addRemotelyInitiated(promisedStream);
933-
934923
try {
935924
consumePushPromiseFrame(frame, payload, promisedStream);
936925
} catch (final H2StreamResetException ex) {
@@ -1133,9 +1122,10 @@ private void produceOutput() throws HttpException, IOException {
11331122
if (!stream.isLocalClosed() && stream.getOutputWindow().get() > 0) {
11341123
stream.produceOutput();
11351124
}
1136-
if (stream.isTerminated()) {
1125+
if (stream.isClosedPastLingerDeadline()) {
1126+
streams.dropStreamId(stream.getId());
1127+
stream.releaseResources();
11371128
it.remove();
1138-
streams.release(stream);
11391129
requestSessionOutput();
11401130
}
11411131
if (!outputQueue.isEmpty()) {
@@ -1303,12 +1293,12 @@ ByteBuffer getContent() {
13031293

13041294
}
13051295

1306-
H2StreamChannel createChannel(final int streamId, final boolean idle) {
1307-
return new H2StreamChannelImpl(streamId, idle, initInputWinSize, initOutputWinSize);
1296+
H2StreamChannel createChannel(final int streamId) {
1297+
return new H2StreamChannelImpl(streamId, initInputWinSize, initOutputWinSize);
13081298
}
13091299

1310-
void addStream(final H2Stream stream) throws H2ConnectionException {
1311-
streams.addLocallyInitiated(stream);
1300+
H2Stream createStream(final H2StreamChannel channel, final H2StreamHandler streamHandler) throws H2ConnectionException {
1301+
return streams.createActive(channel, streamHandler);
13121302
}
13131303

13141304
class H2StreamChannelImpl implements H2StreamChannel {
@@ -1317,14 +1307,11 @@ class H2StreamChannelImpl implements H2StreamChannel {
13171307
private final AtomicInteger inputWindow;
13181308
private final AtomicInteger outputWindow;
13191309

1320-
private volatile boolean idle;
13211310
private volatile boolean localClosed;
1322-
13231311
private volatile long localResetTime;
13241312

1325-
H2StreamChannelImpl(final int id, final boolean idle, final int initialInputWindowSize, final int initialOutputWindowSize) {
1313+
H2StreamChannelImpl(final int id, final int initialInputWindowSize, final int initialOutputWindowSize) {
13261314
this.id = id;
1327-
this.idle = idle;
13281315
this.inputWindow = new AtomicInteger(initialInputWindowSize);
13291316
this.outputWindow = new AtomicInteger(initialOutputWindowSize);
13301317
}
@@ -1358,7 +1345,6 @@ public void submit(final List<Header> headers, final boolean endStream) throws I
13581345
throw new H2ConnectionException(H2Error.INTERNAL_ERROR, "Message headers are missing");
13591346
}
13601347
ensureNotClosed();
1361-
idle = false;
13621348
commitHeaders(id, headers, endStream);
13631349
if (endStream) {
13641350
localClosed = true;
@@ -1371,15 +1357,14 @@ public void submit(final List<Header> headers, final boolean endStream) throws I
13711357
@Override
13721358
public void push(final List<Header> headers, final AsyncPushProducer pushProducer) throws HttpException, IOException {
13731359
acceptPushRequest();
1374-
13751360
ioSession.getLock().lock();
13761361
try {
13771362
ensureNotClosed();
1378-
13791363
final int promisedStreamId = streams.generateStreamId();
1380-
executePush(promisedStreamId, pushProducer);
1364+
final H2StreamChannel channel = createChannel(promisedStreamId);
1365+
streams.createReserved(channel, outgoingPushPromise(channel, pushProducer));
1366+
13811367
commitPushPromise(id, promisedStreamId, headers);
1382-
idle = false;
13831368
} finally {
13841369
ioSession.getLock().unlock();
13851370
}
@@ -1450,12 +1435,9 @@ public boolean localReset(final int code) throws IOException {
14501435
localClosed = true;
14511436
localResetTime = System.currentTimeMillis();
14521437

1453-
if (!idle) {
1454-
final RawFrame resetStream = frameFactory.createResetStream(id, code);
1455-
commitFrameInternal(resetStream);
1456-
return true;
1457-
}
1458-
return false;
1438+
final RawFrame resetStream = frameFactory.createResetStream(id, code);
1439+
commitFrameInternal(resetStream);
1440+
return true;
14591441
} finally {
14601442
ioSession.getLock().unlock();
14611443
}
@@ -1484,8 +1466,7 @@ public String toString() {
14841466
.append(", inputWindow=").append(inputWindow)
14851467
.append(", outputWindow=").append(outputWindow)
14861468
.append(", localClosed=").append(localClosed)
1487-
.append(", idle=").append(idle);
1488-
buf.append("]");
1469+
.append("]");
14891470
return buf.toString();
14901471
}
14911472

0 commit comments

Comments
 (0)