add per-stream idle timeout and suppress CancelledKeyException on cancel

Author: arturobernalg

httpcore5-h2/src/main/java/org/apache/hc/core5/http2/H2StreamTimeoutException.java

@@ -26,7 +26,7 @@
  */
 package org.apache.hc.core5.http2;
 
-import java.net.SocketTimeoutException;
+import java.io.InterruptedIOException;
 
 import org.apache.hc.core5.util.Timeout;
 
@@ -51,7 +51,7 @@
  *
  * @since 5.4
  */
-public class H2StreamTimeoutException extends SocketTimeoutException {
+public class H2StreamTimeoutException extends InterruptedIOException {
 
     private static final long serialVersionUID = 1L;
 

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

@@ -30,7 +30,6 @@
 import java.net.SocketAddress;
 import java.nio.BufferOverflowException;
 import java.nio.ByteBuffer;
-import java.nio.channels.CancelledKeyException;
 import java.nio.channels.SelectionKey;
 import java.nio.charset.StandardCharsets;
 import java.util.Deque;
@@ -272,12 +271,7 @@ private void commitFrameInternal(final RawFrame frame) throws IOException {
         } else {
             outputQueue.addLast(frame);
         }
-        try {
-            ioSession.setEvent(SelectionKey.OP_WRITE);
-        } catch (final CancelledKeyException ex) {
-            connState = ConnectionHandshake.SHUTDOWN;
-            ioSession.close(CloseMode.IMMEDIATE);
-        }
+        ioSession.setEvent(SelectionKey.OP_WRITE);
     }
 
     private void commitFrame(final RawFrame frame) throws IOException {
@@ -419,12 +413,7 @@ private void incrementInputCapacity(
 
     void requestSessionOutput() {
         outputRequests.incrementAndGet();
-        try {
-            ioSession.setEvent(SelectionKey.OP_WRITE);
-        } catch (final CancelledKeyException ex) {
-            connState = ConnectionHandshake.SHUTDOWN;
-            ioSession.close(CloseMode.IMMEDIATE);
-        }
+        ioSession.setEvent(SelectionKey.OP_WRITE);
     }
 
     public final void onConnect() throws HttpException, IOException {
@@ -662,7 +651,6 @@ private void executeRequest(final RequestExecutionCommand requestExecutionComman
                 requestExecutionCommand.getExchangeHandler(),
                 requestExecutionCommand.getPushHandlerFactory(),
                 requestExecutionCommand.getContext()));
-        initializeStreamTimeouts(stream);
 
         if (streamListener != null) {
             final int initInputWindow = stream.getInputWindow().get();
@@ -671,6 +659,7 @@ private void executeRequest(final RequestExecutionCommand requestExecutionComman
             streamListener.onOutputFlowControl(this, streamId, initOutputWindow, initOutputWindow);
         }
         requestExecutionCommand.initiated(stream);
+        initializeStreamTimeouts(stream);
         if (stream.isOutputReady()) {
             stream.produceOutput();
         }
@@ -1624,42 +1613,21 @@ private void checkStreamTimeouts(final long nowNanos) throws IOException {
             }
 
             final Timeout idleTimeout = stream.getIdleTimeout();
-            final Timeout lifetimeTimeout = stream.getLifetimeTimeout();
-            if ((idleTimeout == null || !idleTimeout.isEnabled())
-                    && (lifetimeTimeout == null || !lifetimeTimeout.isEnabled())) {
+            if (idleTimeout == null || !idleTimeout.isEnabled()) {
                 continue;
             }
 
-            final long created = stream.getCreatedNanos();
             final long last = stream.getLastActivityNanos();
-
-            if (idleTimeout != null && idleTimeout.isEnabled()) {
-                final long idleNanos = idleTimeout.toNanoseconds();
-                if (idleNanos > 0 && nowNanos - last > idleNanos) {
-                    final int streamId = stream.getId();
-                    final H2StreamTimeoutException ex = new H2StreamTimeoutException(
-                            "HTTP/2 stream idle timeout (" + idleTimeout + ")",
-                            streamId,
-                            idleTimeout,
-                            true);
-                    stream.localReset(ex, H2Error.CANCEL);
-                    continue;
-                }
-            }
-
-            if (lifetimeTimeout != null && lifetimeTimeout.isEnabled()) {
-                final long lifeNanos = lifetimeTimeout.toNanoseconds();
-                if (lifeNanos > 0 && nowNanos - created > lifeNanos) {
-                    final int streamId = stream.getId();
-                    final H2StreamTimeoutException ex = new H2StreamTimeoutException(
-                            "HTTP/2 stream lifetime timeout (" + lifetimeTimeout + ")",
-                            streamId,
-                            lifetimeTimeout,
-                            false);
-                    stream.localReset(ex, H2Error.CANCEL);
-                }
+            final long idleNanos = idleTimeout.toNanoseconds();
+            if (idleNanos > 0 && nowNanos - last > idleNanos) {
+                final int streamId = stream.getId();
+                final H2StreamTimeoutException ex = new H2StreamTimeoutException(
+                        "HTTP/2 stream idle timeout (" + idleTimeout + ")",
+                        streamId,
+                        idleTimeout,
+                        true);
+                stream.localReset(ex, H2Error.CANCEL);
             }
         }
     }
-
 }

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

@@ -26,6 +26,7 @@
  */
 package org.apache.hc.core5.http2.impl.nio.bootstrap;
 
+import java.nio.channels.CancelledKeyException;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicReference;
 
@@ -48,7 +49,11 @@ public void setDependency(final Cancellable cancellable) {
         if (cancelled.get()) {
             final Cancellable dependency = dependencyRef.getAndSet(null);
             if (dependency != null) {
-                dependency.cancel();
+                try {
+                    dependency.cancel();
+                } catch (final CancelledKeyException ignore) {
+                    // Session already gone; cancellation is effectively complete.
+                }
             }
         }
     }
@@ -63,7 +68,11 @@ public boolean cancel() {
         if (cancelled.compareAndSet(false, true)) {
             final Cancellable dependency = dependencyRef.getAndSet(null);
             if (dependency != null) {
-                dependency.cancel();
+                try {
+                    dependency.cancel();
+                } catch (final CancelledKeyException ignore) {
+                    // Session already gone; treat as successfully cancelled.
+                }
             }
             return true;
         }

httpcore5-h2/src/test/java/org/apache/hc/core5/http2/impl/nio/TestAbstractH2StreamMultiplexer.java

@@ -1080,54 +1080,5 @@ void testStreamIdleTimeoutTriggersH2StreamTimeoutException() throws Exception {
         Assertions.assertTrue(stream.isLocalClosed());
         Assertions.assertTrue(stream.isClosed());
     }
-
-    @Test
-    void testStreamLifetimeTimeoutTriggersH2StreamTimeoutException() throws Exception {
-        Mockito.when(protocolIOSession.getSocketTimeout()).thenReturn(Timeout.DISABLED);
-
-        Mockito.when(protocolIOSession.write(ArgumentMatchers.any(ByteBuffer.class)))
-                .thenAnswer(invocation -> {
-                    final ByteBuffer buffer = invocation.getArgument(0, ByteBuffer.class);
-                    final int remaining = buffer.remaining();
-                    buffer.position(buffer.limit());
-                    return remaining;
-                });
-        Mockito.doNothing().when(protocolIOSession).setEvent(ArgumentMatchers.anyInt());
-        Mockito.doNothing().when(protocolIOSession).clearEvent(ArgumentMatchers.anyInt());
-
-        final H2Config h2Config = H2Config.custom().build();
-        final AbstractH2StreamMultiplexer streamMultiplexer = new H2StreamMultiplexerImpl(
-                protocolIOSession,
-                FRAME_FACTORY,
-                StreamIdGenerator.ODD,
-                httpProcessor,
-                CharCodingConfig.DEFAULT,
-                h2Config,
-                h2StreamListener,
-                () -> streamHandler);
-
-        final H2StreamChannel channel = streamMultiplexer.createChannel(3);
-        final H2Stream stream = streamMultiplexer.createStream(channel, streamHandler);
-        stream.activate();
-
-        stream.setIdleTimeout(null);
-        stream.setLifetimeTimeout(Timeout.of(1, TimeUnit.NANOSECONDS));
-
-        streamMultiplexer.onOutput();
-
-        Mockito.verify(streamHandler).failed(exceptionCaptor.capture());
-        final Exception cause = exceptionCaptor.getValue();
-        Assertions.assertInstanceOf(H2StreamTimeoutException.class, cause);
-
-        final H2StreamTimeoutException timeoutEx = (H2StreamTimeoutException) cause;
-        Assertions.assertFalse(timeoutEx.isIdleTimeout(), "Expected lifetime timeout flag");
-        Assertions.assertEquals(3, timeoutEx.getStreamId(), "Unexpected stream id");
-
-        Assertions.assertTrue(stream.isLocalClosed());
-        Assertions.assertTrue(stream.isClosed());
-    }
-
-
-
 }