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,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

@@ -439,10 +439,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);
                     }
@@ -655,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();
@@ -774,12 +769,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 +963,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 +1368,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 +1473,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,43 +1597,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);
-                    // Once reset due to idle timeout, we do not care about lifetime anymore
-                    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/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;
-    }
 }