RFC 9113 conformance: H2 stream can be cancelled (reset) locally from multiple threads

Author: ok2c

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

@@ -1367,6 +1367,9 @@ public void submit(final List<Header> headers, final boolean endStream) throws I
                 if (headers == null || headers.isEmpty()) {
                     throw new H2ConnectionException(H2Error.INTERNAL_ERROR, "Message headers are missing");
                 }
+                if (isLocalReset()) {
+                    return;
+                }
                 ensureNotClosed();
                 commitHeaders(id, headers, endStream);
                 if (endStream) {
@@ -1382,6 +1385,9 @@ public void push(final List<Header> headers, final AsyncPushProducer pushProduce
             acceptPushRequest();
             ioSession.getLock().lock();
             try {
+                if (isLocalReset()) {
+                    return;
+                }
                 ensureNotClosed();
                 final int promisedStreamId = streams.generateStreamId();
                 final H2StreamChannel channel = createChannel(promisedStreamId);
@@ -1405,6 +1411,9 @@ public void update(final int increment) throws IOException {
         public int write(final ByteBuffer payload) throws IOException {
             ioSession.getLock().lock();
             try {
+                if (isLocalReset()) {
+                    return 0;
+                }
                 ensureNotClosed();
                 return streamData(id, outputWindow, payload);
             } finally {
@@ -1416,6 +1425,9 @@ public int write(final ByteBuffer payload) throws IOException {
         public void endStream(final List<? extends Header> trailers) throws IOException {
             ioSession.getLock().lock();
             try {
+                if (isLocalReset()) {
+                    return;
+                }
                 ensureNotClosed();
                 localClosed = true;
                 if (trailers != null && !trailers.isEmpty()) {

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

@@ -39,7 +39,6 @@
 import org.apache.hc.core5.http.Header;
 import org.apache.hc.core5.http.HttpException;
 import org.apache.hc.core5.http.ProtocolException;
-import org.apache.hc.core5.http.StreamClosedException;
 import org.apache.hc.core5.http.nio.AsyncPushConsumer;
 import org.apache.hc.core5.http.nio.HandlerFactory;
 import org.apache.hc.core5.http2.H2Error;
@@ -56,6 +55,7 @@ enum State { RESERVED, OPEN, CLOSED }
     private final Consumer<State> stateChangeCallback;
     private final AtomicReference<State> transitionRef;
     private final AtomicBoolean released;
+    private final AtomicBoolean cancelled;
 
     private volatile boolean reserved;
     private volatile boolean remoteClosed;
@@ -67,6 +67,7 @@ enum State { RESERVED, OPEN, CLOSED }
         this.reserved = true;
         this.transitionRef = new AtomicReference<>(State.RESERVED);
         this.released = new AtomicBoolean();
+        this.cancelled = new AtomicBoolean();
     }
 
     int getId() {
@@ -136,6 +137,10 @@ void consumePromise(final List<Header> headers) throws HttpException, IOExceptio
             if (channel.isLocalReset()) {
                 return;
             }
+            if (cancelled.get()) {
+                localResetCancelled();
+                return;
+            }
             handler.consumePromise(headers);
             channel.markLocalClosed();
         } catch (final ProtocolException ex) {
@@ -151,6 +156,10 @@ void consumeHeader(final List<Header> headers, final boolean endOfStream) throws
             if (channel.isLocalReset()) {
                 return;
             }
+            if (cancelled.get()) {
+                localResetCancelled();
+                return;
+            }
             handler.consumeHeader(headers, remoteClosed);
         } catch (final ProtocolException ex) {
             localReset(ex, H2Error.PROTOCOL_ERROR);
@@ -165,6 +174,10 @@ void consumeData(final ByteBuffer src, final boolean endOfStream) throws HttpExc
             if (channel.isLocalReset()) {
                 return;
             }
+            if (cancelled.get()) {
+                localResetCancelled();
+                return;
+            }
             handler.consumeData(src, remoteClosed);
         } catch (final CharacterCodingException ex) {
             localReset(ex, H2Error.INTERNAL_ERROR);
@@ -222,6 +235,10 @@ void localReset(final H2StreamResetException ex) throws IOException {
         localReset(ex, ex.getCode());
     }
 
+    void localResetCancelled() throws IOException {
+        localReset(new H2StreamResetException(H2Error.CANCEL, "Cancelled"));
+    }
+
     void handle(final HttpException ex) throws IOException, HttpException {
         handler.handle(ex, remoteClosed);
     }
@@ -231,16 +248,12 @@ HandlerFactory<AsyncPushConsumer> getPushHandlerFactory() {
     }
 
     boolean abort() {
-        final boolean cancelled = channel.cancel();
-        if (released.compareAndSet(false, true)) {
-            try {
-                handler.failed(new StreamClosedException());
-                handler.releaseResources();
-            } finally {
-                triggerClosed();
-            }
+        if (cancelled.compareAndSet(false, true)) {
+            channel.requestOutput();
+            return true;
+        } else {
+            return false;
         }
-        return cancelled;
     }
 
     boolean abortGracefully() throws IOException {