add per-stream idle timeout and suppress CancelledKeyException on cancel

;

Handle cancelled SelectionKey in interestOps access

Author: arturobernalg

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

@@ -26,32 +26,20 @@
  */
 package org.apache.hc.core5.http2;
 
-import java.net.SocketTimeoutException;
+import java.io.InterruptedIOException;
 
 import org.apache.hc.core5.util.Timeout;
 
 /**
- * {@link java.net.SocketTimeoutException} raised by the HTTP/2 stream
- * multiplexer when a per-stream timeout elapses.
+ * {@link java.net.SocketTimeoutException} raised when an HTTP/2 stream exceeds its configured idle timeout.
  * <p>
- * This exception is used for timeouts that are scoped to a single HTTP/2
- * stream rather than the underlying TCP connection, for example:
- * </p>
- * <ul>
- *     <li>an idle timeout where no activity has been observed on the stream, or</li>
- *     <li>a lifetime timeout where the total age of the stream exceeds
- *     the configured limit.</li>
- * </ul>
- * <p>
- * The {@link #isIdleTimeout()} flag can be used to distinguish whether
- * the timeout was triggered by idleness or by the overall stream lifetime.
- * The affected stream id and the timeout value are exposed via
- * {@link #getStreamId()} and {@link #getTimeout()} respectively.
+ * This timeout is scoped to a single stream and is independent of the underlying connection socket timeout.
  * </p>
  *
- * @since 5.4
+ * @since 5.5
  */
-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

@@ -144,6 +144,11 @@ enum SettingsHandshake { READY, TRANSMITTED, ACKED }
     private final Map<Integer, PriorityValue> priorities = new ConcurrentHashMap<>();
     private volatile boolean peerNoRfc7540Priorities;
 
+
+    private static final long STREAM_TIMEOUT_GRANULARITY_MILLIS = 1000;
+    private long lastStreamTimeoutCheckMillis;
+
+
     AbstractH2StreamMultiplexer(
             final ProtocolIOSession ioSession,
             final FrameFactory frameFactory,
@@ -439,10 +444,6 @@ public final void onInput(final ByteBuffer src) throws HttpException, IOExceptio
             for (;;) {
                 final RawFrame frame = inputBuffer.read(src, ioSession);
                 if (frame != null) {
-                    if (connState.compareTo(ConnectionHandshake.SHUTDOWN) < 0) {
-                        checkStreamTimeouts(System.nanoTime());
-                    }
-
                     if (streamListener != null) {
                         streamListener.onFrameInput(this, frame.getStreamId(), frame);
                     }
@@ -460,7 +461,7 @@ public final void onInput(final ByteBuffer src) throws HttpException, IOExceptio
                 }
             }
             if (connState.compareTo(ConnectionHandshake.SHUTDOWN) < 0) {
-                checkStreamTimeouts(System.nanoTime());
+                validateStreamTimeouts();
             }
         }
     }
@@ -541,7 +542,7 @@ public final void onOutput() throws HttpException, IOException {
         }
 
         if (connState.compareTo(ConnectionHandshake.SHUTDOWN) < 0) {
-            checkStreamTimeouts(System.nanoTime());
+            validateStreamTimeouts();
         }
 
         if (connState.compareTo(ConnectionHandshake.GRACEFUL_SHUTDOWN) == 0) {
@@ -655,7 +656,6 @@ private void executeRequest(final RequestExecutionCommand requestExecutionComman
                 requestExecutionCommand.getExchangeHandler(),
                 requestExecutionCommand.getPushHandlerFactory(),
                 requestExecutionCommand.getContext()));
-        initializeStreamTimeouts(stream);
 
         if (streamListener != null) {
             final int initInputWindow = stream.getInputWindow().get();
@@ -774,12 +774,10 @@ private void consumeFrame(final RawFrame frame) throws HttpException, IOExceptio
                     final H2StreamChannel channel = createChannel(streamId);
                     if (connState.compareTo(ConnectionHandshake.ACTIVE) <= 0) {
                         stream = streams.createActive(channel, incomingRequest(channel));
-                        initializeStreamTimeouts(stream);
                         streams.resetIfExceedsMaxConcurrentLimit(stream, localConfig.getMaxConcurrentStreams());
                     } else {
                         channel.localReset(H2Error.REFUSED_STREAM);
                         stream = streams.createActive(channel, NoopH2StreamHandler.INSTANCE);
-                        initializeStreamTimeouts(stream);
                     }
                 } else if (stream.isLocalClosed() && stream.isRemoteClosed()) {
                     throw new H2ConnectionException(H2Error.STREAM_CLOSED, "Stream closed");
@@ -970,7 +968,6 @@ private void consumeFrame(final RawFrame frame) throws HttpException, IOExceptio
                     channel.localReset(H2Error.REFUSED_STREAM);
                     promisedStream = streams.createActive(channel, NoopH2StreamHandler.INSTANCE);
                 }
-                initializeStreamTimeouts(promisedStream);
                 try {
                     consumePushPromiseFrame(frame, payload, promisedStream);
                 } catch (final H2StreamResetException ex) {
@@ -1376,16 +1373,8 @@ H2StreamChannel createChannel(final int streamId) {
         return new H2StreamChannelImpl(streamId, initInputWinSize, initOutputWinSize);
     }
 
-    private void initializeStreamTimeouts(final H2Stream stream) {
-        final Timeout socketTimeout = ioSession.getSocketTimeout();
-        if (socketTimeout != null && socketTimeout.isEnabled()) {
-            stream.setIdleTimeout(socketTimeout);
-        }
-    }
-
     H2Stream createStream(final H2StreamChannel channel, final H2StreamHandler streamHandler) {
         final H2Stream stream = streams.createActive(channel, streamHandler);
-        initializeStreamTimeouts(stream);
         return stream;
     }
 
@@ -1489,7 +1478,6 @@ public void push(final List<Header> headers, final AsyncPushProducer pushProduce
                 final int promisedStreamId = streams.generateStreamId();
                 final H2StreamChannel channel = createChannel(promisedStreamId);
                 final H2Stream stream = streams.createReserved(channel, outgoingPushPromise(channel, pushProducer));
-                initializeStreamTimeouts(stream);
 
                 commitPushPromise(id, promisedStreamId, headers);
                 stream.markRemoteClosed();
@@ -1614,42 +1602,29 @@ 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);
-                    // Once reset due to idle timeout, we do not care about lifetime anymore
-                    continue;
-                }
+            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);
             }
+        }
+    }
 
-            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);
-                }
-            }
+    private void validateStreamTimeouts() throws IOException {
+        final long nowMillis = System.currentTimeMillis();
+        if ((nowMillis - lastStreamTimeoutCheckMillis) >= STREAM_TIMEOUT_GRANULARITY_MILLIS) {
+            lastStreamTimeoutCheckMillis = nowMillis;
+            checkStreamTimeouts(System.nanoTime());
         }
     }
 

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

@@ -61,11 +61,9 @@ class H2Stream implements StreamControl {
     private volatile boolean reserved;
     private volatile boolean remoteClosed;
 
-    private volatile long createdNanos;
     private volatile long lastActivityNanos;
 
     private volatile Timeout idleTimeout;
-    private volatile Timeout lifetimeTimeout;
 
     H2Stream(final H2StreamChannel channel, final H2StreamHandler handler, final Consumer<State> stateChangeCallback) {
         this.channel = channel;
@@ -75,7 +73,6 @@ class H2Stream implements StreamControl {
         this.transitionRef = new AtomicReference<>(State.RESERVED);
         this.released = new AtomicBoolean();
         this.cancelled = new AtomicBoolean();
-        this.createdNanos = 0L;
         this.lastActivityNanos = 0L;
     }
 
@@ -91,7 +88,7 @@ public State getState() {
 
     @Override
     public void setTimeout(final Timeout timeout) {
-        // not supported
+        this.idleTimeout = timeout;
     }
 
     boolean isReserved() {
@@ -219,16 +216,13 @@ boolean isOutputReady() {
 
     void produceOutput() throws HttpException, IOException {
         try {
-            touch();
-
             handler.produceOutput();
         } catch (final ProtocolException ex) {
             localReset(ex, H2Error.PROTOCOL_ERROR);
         }
     }
 
     void produceInputCapacityUpdate() throws IOException {
-        touch();
         handler.updateInputCapacity();
     }
 
@@ -327,19 +321,13 @@ public String toString() {
     }
 
     private void markCreatedAndActive() {
-        final long now = System.nanoTime();
-        this.createdNanos = now;
-        this.lastActivityNanos = now;
+        this.lastActivityNanos = System.nanoTime();
     }
 
     private void touch() {
         this.lastActivityNanos = System.nanoTime();
     }
 
-    long getCreatedNanos() {
-        return createdNanos;
-    }
-
     long getLastActivityNanos() {
         return lastActivityNanos;
     }
@@ -348,15 +336,4 @@ Timeout getIdleTimeout() {
         return idleTimeout;
     }
 
-    void setIdleTimeout(final Timeout idleTimeout) {
-        this.idleTimeout = idleTimeout;
-    }
-
-    Timeout getLifetimeTimeout() {
-        return lifetimeTimeout;
-    }
-
-    void setLifetimeTimeout(final Timeout lifetimeTimeout) {
-        this.lifetimeTimeout = lifetimeTimeout;
-    }
 }