Optimize release of resources associated by H2 streams: H2StreamHandler#releaseResources should now be called only once

Author: ok2c

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

@@ -40,6 +40,7 @@
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentLinkedDeque;
 import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
 
 import javax.net.ssl.SSLHandshakeException;
@@ -556,7 +557,7 @@ public final void onTimeout(final Timeout timeout) throws HttpException, IOExcep
         for (final Iterator<Map.Entry<Integer, H2Stream>> it = streamMap.entrySet().iterator(); it.hasNext(); ) {
             final Map.Entry<Integer, H2Stream> entry = it.next();
             final H2Stream stream = entry.getValue();
-            stream.reset(new H2StreamResetException(H2Error.NO_ERROR, "Timeout due to inactivity (" + timeout + ")"));
+            stream.fail(new H2StreamResetException(H2Error.NO_ERROR, "Timeout due to inactivity (" + timeout + ")"));
         }
         streamMap.clear();
     }
@@ -676,7 +677,7 @@ public final void onException(final Exception cause) {
             for (final Iterator<Map.Entry<Integer, H2Stream>> it = streamMap.entrySet().iterator(); it.hasNext(); ) {
                 final Map.Entry<Integer, H2Stream> entry = it.next();
                 final H2Stream stream = entry.getValue();
-                stream.reset(cause);
+                stream.fail(cause);
             }
             streamMap.clear();
             if (!(cause instanceof ConnectionClosedException)) {
@@ -873,9 +874,8 @@ private void consumeFrame(final RawFrame frame) throws HttpException, IOExceptio
                         throw new H2ConnectionException(H2Error.FRAME_SIZE_ERROR, "Invalid RST_STREAM frame payload");
                     }
                     final int errorCode = payload.getInt();
-                    stream.reset(new H2StreamResetException(errorCode, "Stream reset (" + errorCode + ")"));
+                    stream.fail(new H2StreamResetException(errorCode, "Stream reset (" + errorCode + ")"));
                     streamMap.remove(streamId);
-                    stream.releaseResources();
                     requestSessionOutput();
                 }
             }
@@ -1013,7 +1013,7 @@ private void consumeFrame(final RawFrame frame) throws HttpException, IOExceptio
                     for (final Iterator<Map.Entry<Integer, H2Stream>> it = streamMap.entrySet().iterator(); it.hasNext(); ) {
                         final Map.Entry<Integer, H2Stream> entry = it.next();
                         final H2Stream stream = entry.getValue();
-                        stream.reset(new H2StreamResetException(errorCode, "Connection terminated by the peer (" + errorCode + ")"));
+                        stream.fail(new H2StreamResetException(errorCode, "Connection terminated by the peer (" + errorCode + ")"));
                     }
                     streamMap.clear();
                     connState = ConnectionHandshake.SHUTDOWN;
@@ -1590,6 +1590,7 @@ static class H2Stream {
         private final H2StreamChannelImpl channel;
         private final H2StreamHandler handler;
         private final boolean remoteInitiated;
+        private final AtomicBoolean released;
 
         private H2Stream(
                 final H2StreamChannelImpl channel,
@@ -1598,6 +1599,7 @@ private H2Stream(
             this.channel = channel;
             this.handler = handler;
             this.remoteInitiated = remoteInitiated;
+            this.released = new AtomicBoolean();
         }
 
         int getId() {
@@ -1679,15 +1681,21 @@ void produceInputCapacityUpdate() throws IOException {
             handler.updateInputCapacity();
         }
 
-        void reset(final Exception cause) {
+        void fail(final Exception cause) {
             channel.setRemoteEndStream();
             channel.setLocalEndStream();
-            handler.failed(cause);
+            if (released.compareAndSet(false, true)) {
+                handler.failed(cause);
+                handler.releaseResources();
+            }
         }
 
         void localReset(final Exception cause, final int code) throws IOException {
             channel.localReset(code);
-            handler.failed(cause);
+            if (released.compareAndSet(false, true)) {
+                handler.failed(cause);
+                handler.releaseResources();
+            }
         }
 
         void localReset(final Exception cause, final H2Error error) throws IOException {
@@ -1707,17 +1715,22 @@ HandlerFactory<AsyncPushConsumer> getPushHandlerFactory() {
         }
 
         void cancel() {
-            reset(new RequestNotExecutedException());
+            fail(new RequestNotExecutedException());
         }
 
         boolean abort() {
             final boolean cancelled = channel.cancel();
-            handler.failed(new RequestNotExecutedException());
+            if (released.compareAndSet(false, true)) {
+                handler.failed(new RequestNotExecutedException());
+                handler.releaseResources();
+            }
             return cancelled;
         }
 
         void releaseResources() {
-            handler.releaseResources();
+            if (released.compareAndSet(false, true)) {
+                handler.releaseResources();
+            }
         }
 
         void appendState(final StringBuilder buf) {