diff --git a/java/org/apache/coyote/http2/Http2AsyncUpgradeHandler.java b/java/org/apache/coyote/http2/Http2AsyncUpgradeHandler.java index 758b63e66648..bedc8788d912 100644 --- a/java/org/apache/coyote/http2/Http2AsyncUpgradeHandler.java +++ b/java/org/apache/coyote/http2/Http2AsyncUpgradeHandler.java @@ -156,7 +156,7 @@ void sendStreamReset(StreamStateMachine state, StreamException se) throws IOExce boolean active = state.isActive(); state.sendReset(); if (active) { - decrementActiveRemoteStreamCount(); + decrementActiveRemoteStreamCount(getStream(se.getStreamId())); } } diff --git a/java/org/apache/coyote/http2/Http2UpgradeHandler.java b/java/org/apache/coyote/http2/Http2UpgradeHandler.java index a50f62287d54..2ac1b969b3eb 100644 --- a/java/org/apache/coyote/http2/Http2UpgradeHandler.java +++ b/java/org/apache/coyote/http2/Http2UpgradeHandler.java @@ -288,8 +288,8 @@ protected void processStreamOnContainerThread(Stream stream) { } - protected void decrementActiveRemoteStreamCount() { - setConnectionTimeoutForStreamCount(activeRemoteStreamCount.decrementAndGet()); + protected void decrementActiveRemoteStreamCount(Stream stream) { + setConnectionTimeoutForStreamCount(stream.decrementAndGetActiveRemoteStreamCount()); } @@ -596,7 +596,7 @@ void sendStreamReset(StreamStateMachine state, StreamException se) throws IOExce boolean active = state.isActive(); state.sendReset(); if (active) { - decrementActiveRemoteStreamCount(); + decrementActiveRemoteStreamCount(getStream(se.getStreamId())); } } socketWrapper.write(true, rstFrame, 0, rstFrame.length); @@ -839,7 +839,7 @@ void writeBody(Stream stream, ByteBuffer data, int len, boolean finished) throws protected void sentEndOfStream(Stream stream) { stream.sentEndOfStream(); if (!stream.isActive()) { - decrementActiveRemoteStreamCount(); + decrementActiveRemoteStreamCount(stream); } } @@ -1221,7 +1221,7 @@ private int allocate(AbstractStream stream, int allocation) { } - private Stream getStream(int streamId) { + Stream getStream(int streamId) { Integer key = Integer.valueOf(streamId); AbstractStream result = streams.get(key); if (result instanceof Stream) { @@ -1590,6 +1590,7 @@ public HeaderEmitter headersStart(int streamId, boolean headersEndStream) throws Stream stream = getStream(streamId, false); if (stream == null) { stream = createRemoteStream(streamId); + activeRemoteStreamCount.incrementAndGet(); } if (streamId < maxActiveRemoteStreamId) { throw new ConnectionException(sm.getString("upgradeHandler.stream.old", Integer.valueOf(streamId), @@ -1668,9 +1669,8 @@ public void headersEnd(int streamId, boolean endOfStream) throws Http2Exception Stream stream = (Stream) abstractNonZeroStream; if (stream.isActive()) { if (stream.receivedEndOfHeaders()) { - - if (localSettings.getMaxConcurrentStreams() < activeRemoteStreamCount.incrementAndGet()) { - decrementActiveRemoteStreamCount(); + if (localSettings.getMaxConcurrentStreams() < activeRemoteStreamCount.get()) { + decrementActiveRemoteStreamCount(stream); // Ignoring maxConcurrentStreams increases the overhead count increaseOverheadCount(FrameType.HEADERS); throw new StreamException( @@ -1714,7 +1714,7 @@ public void receivedEndOfStream(int streamId) throws ConnectionException { private void receivedEndOfStream(Stream stream) throws ConnectionException { stream.receivedEndOfStream(); if (!stream.isActive()) { - decrementActiveRemoteStreamCount(); + decrementActiveRemoteStreamCount(stream); } } @@ -1740,7 +1740,7 @@ public void reset(int streamId, long errorCode) throws Http2Exception { boolean active = stream.isActive(); stream.receiveReset(errorCode); if (active) { - decrementActiveRemoteStreamCount(); + decrementActiveRemoteStreamCount(stream); } } } diff --git a/java/org/apache/coyote/http2/Stream.java b/java/org/apache/coyote/http2/Stream.java index 1a74f3b84c2a..95b596a00431 100644 --- a/java/org/apache/coyote/http2/Stream.java +++ b/java/org/apache/coyote/http2/Stream.java @@ -28,6 +28,7 @@ import java.util.Locale; import java.util.Map; import java.util.Set; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.locks.Lock; import java.util.concurrent.locks.ReentrantLock; import java.util.function.Supplier; @@ -92,6 +93,7 @@ class Stream extends AbstractNonZeroStream implements HeaderEmitter { private final StreamInputBuffer inputBuffer; private final StreamOutputBuffer streamOutputBuffer = new StreamOutputBuffer(); private final Http2OutputBuffer http2OutputBuffer = new Http2OutputBuffer(coyoteResponse, streamOutputBuffer); + private final AtomicBoolean removedFromActiveCount = new AtomicBoolean(false); // State machine would be too much overhead private int headerState = HEADER_STATE_START; @@ -883,6 +885,20 @@ public void setIncremental(boolean incremental) { } + int decrementAndGetActiveRemoteStreamCount() { + /* + * Protect against mis-counting of active streams. This method should only be called once per stream but since + * the count of active streams is used to enforce the maximum concurrent streams limit, make sure each stream is + * only removed from the active count exactly once. + */ + if (removedFromActiveCount.compareAndSet(false, true)) { + return handler.activeRemoteStreamCount.decrementAndGet(); + } else { + return handler.activeRemoteStreamCount.get(); + } + } + + private static void push(final Http2UpgradeHandler handler, final Request request, final Stream stream) throws IOException { if (org.apache.coyote.Constants.IS_SECURITY_ENABLED) { diff --git a/webapps/docs/changelog.xml b/webapps/docs/changelog.xml index dce08e2919a5..76b0fb554b33 100644 --- a/webapps/docs/changelog.xml +++ b/webapps/docs/changelog.xml @@ -165,6 +165,10 @@ Connector element, similar to the Executor element, for consistency. (remm) + + Make counting of active HTTP/2 streams per connection more robust. + (markt) +