HTTP/2: add per-stream idle and lifetime timeouts to core multiplexer

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

@@ -0,0 +1,87 @@
+/*
+ * ====================================================================
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ * ====================================================================
+ *
+ * This software consists of voluntary contributions made by many
+ * individuals on behalf of the Apache Software Foundation.  For more
+ * information on the Apache Software Foundation, please see
+ * <http://www.apache.org/>.
+ *
+ */
+package org.apache.hc.core5.http2;
+
+import java.net.SocketTimeoutException;
+
+import org.apache.hc.core5.util.Timeout;
+
+/**
+ * {@link java.net.SocketTimeoutException} raised by the HTTP/2 stream
+ * multiplexer when a per-stream timeout elapses.
+ * <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.
+ * </p>
+ *
+ * @since 5.4
+ */
+public class H2StreamTimeoutException extends SocketTimeoutException {
+
+    private static final long serialVersionUID = 1L;
+
+    private final int streamId;
+    private final Timeout timeout;
+    private final boolean idleTimeout;
+
+    public H2StreamTimeoutException(final String message, final int streamId, final Timeout timeout, final boolean idleTimeout) {
+        super(message);
+        this.streamId = streamId;
+        this.timeout = timeout;
+        this.idleTimeout = idleTimeout;
+    }
+
+    public int getStreamId() {
+        return streamId;
+    }
+
+    public Timeout getTimeout() {
+        return timeout;
+    }
+
+    /**
+     * Indicates whether this timeout was triggered by idle time (no activity)
+     * rather than by stream lifetime.
+     *
+     * @return {@code true} if this is an idle timeout, {@code false} if it is a lifetime timeout.
+     */
+    public boolean isIdleTimeout() {
+        return idleTimeout;
+    }
+
+}

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

@@ -30,6 +30,7 @@
 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;
@@ -74,6 +75,7 @@
 import org.apache.hc.core5.http2.H2ConnectionException;
 import org.apache.hc.core5.http2.H2Error;
 import org.apache.hc.core5.http2.H2StreamResetException;
+import org.apache.hc.core5.http2.H2StreamTimeoutException;
 import org.apache.hc.core5.http2.config.H2Config;
 import org.apache.hc.core5.http2.config.H2Param;
 import org.apache.hc.core5.http2.config.H2Setting;
@@ -270,7 +272,12 @@ private void commitFrameInternal(final RawFrame frame) throws IOException {
         } else {
             outputQueue.addLast(frame);
         }
-        ioSession.setEvent(SelectionKey.OP_WRITE);
+        try {
+            ioSession.setEvent(SelectionKey.OP_WRITE);
+        } catch (final CancelledKeyException ex) {
+            connState = ConnectionHandshake.SHUTDOWN;
+            ioSession.close(CloseMode.IMMEDIATE);
+        }
     }
 
     private void commitFrame(final RawFrame frame) throws IOException {
@@ -412,7 +419,12 @@ private void incrementInputCapacity(
 
     void requestSessionOutput() {
         outputRequests.incrementAndGet();
-        ioSession.setEvent(SelectionKey.OP_WRITE);
+        try {
+            ioSession.setEvent(SelectionKey.OP_WRITE);
+        } catch (final CancelledKeyException ex) {
+            connState = ConnectionHandshake.SHUTDOWN;
+            ioSession.close(CloseMode.IMMEDIATE);
+        }
     }
 
     public final void onConnect() throws HttpException, IOException {
@@ -454,6 +466,9 @@ public final void onInput(final ByteBuffer src) throws HttpException, IOExceptio
                     break;
                 }
             }
+            if (connState.compareTo(ConnectionHandshake.SHUTDOWN) < 0) {
+                checkStreamTimeouts(System.nanoTime());
+            }
         }
     }
 
@@ -531,6 +546,11 @@ public final void onOutput() throws HttpException, IOException {
                 }
             }
         }
+
+        if (connState.compareTo(ConnectionHandshake.SHUTDOWN) < 0) {
+            checkStreamTimeouts(System.nanoTime());
+        }
+
         if (connState.compareTo(ConnectionHandshake.GRACEFUL_SHUTDOWN) == 0) {
             int liveStreams = 0;
             for (final Iterator<H2Stream> it = streams.iterator(); it.hasNext(); ) {
@@ -642,6 +662,7 @@ private void executeRequest(final RequestExecutionCommand requestExecutionComman
                 requestExecutionCommand.getExchangeHandler(),
                 requestExecutionCommand.getPushHandlerFactory(),
                 requestExecutionCommand.getContext()));
+        initializeStreamTimeouts(stream);
 
         if (streamListener != null) {
             final int initInputWindow = stream.getInputWindow().get();
@@ -760,10 +781,12 @@ 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");
@@ -954,6 +977,7 @@ 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) {
@@ -1359,8 +1383,20 @@ H2StreamChannel createChannel(final int streamId) {
         return new H2StreamChannelImpl(streamId, initInputWinSize, initOutputWinSize);
     }
 
-    H2Stream createStream(final H2StreamChannel channel, final H2StreamHandler streamHandler) throws H2ConnectionException {
-        return streams.createActive(channel, streamHandler);
+    private void initializeStreamTimeouts(final H2Stream stream) {
+        final Timeout streamIdleTimeout = stream.getIdleTimeout();
+        if (streamIdleTimeout == null || !streamIdleTimeout.isEnabled()) {
+            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;
     }
 
     private void recordPriorityFromHeaders(final int streamId, final List<? extends Header> headers) {
@@ -1463,6 +1499,7 @@ 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();
@@ -1579,4 +1616,50 @@ public String toString() {
 
     }
 
+    private void checkStreamTimeouts(final long nowNanos) throws IOException {
+        for (final Iterator<H2Stream> it = streams.iterator(); it.hasNext(); ) {
+            final H2Stream stream = it.next();
+            if (!stream.isActive()) {
+                continue;
+            }
+
+            final Timeout idleTimeout = stream.getIdleTimeout();
+            final Timeout lifetimeTimeout = stream.getLifetimeTimeout();
+            if ((idleTimeout == null || !idleTimeout.isEnabled())
+                    && (lifetimeTimeout == null || !lifetimeTimeout.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);
+                }
+            }
+        }
+    }
+
 }

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

@@ -44,6 +44,7 @@
 import org.apache.hc.core5.http.nio.HandlerFactory;
 import org.apache.hc.core5.http2.H2Error;
 import org.apache.hc.core5.http2.H2StreamResetException;
+import org.apache.hc.core5.util.Timeout;
 
 class H2Stream implements StreamControl {
 
@@ -59,6 +60,12 @@ 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;
         this.handler = handler;
@@ -67,6 +74,8 @@ class H2Stream implements StreamControl {
         this.transitionRef = new AtomicReference<>(State.RESERVED);
         this.released = new AtomicBoolean();
         this.cancelled = new AtomicBoolean();
+        this.createdNanos = 0L;
+        this.lastActivityNanos = 0L;
     }
 
     @Override
@@ -97,6 +106,7 @@ private void triggerClosed() {
 
     void activate() {
         reserved = false;
+        markCreatedAndActive();
         triggerOpen();
     }
 
@@ -139,6 +149,8 @@ boolean isLocalClosed() {
 
     void consumePromise(final List<Header> headers) throws HttpException, IOException {
         try {
+            touch();
+
             if (channel.isLocalReset()) {
                 return;
             }
@@ -158,6 +170,8 @@ void consumeHeader(final List<Header> headers, final boolean endOfStream) throws
             if (endOfStream) {
                 remoteClosed = true;
             }
+            touch();
+
             if (channel.isLocalReset()) {
                 return;
             }
@@ -176,6 +190,8 @@ void consumeData(final ByteBuffer src, final boolean endOfStream) throws HttpExc
             if (endOfStream) {
                 remoteClosed = true;
             }
+            touch();
+
             if (channel.isLocalReset()) {
                 return;
             }
@@ -197,13 +213,30 @@ boolean isOutputReady() {
 
     void produceOutput() throws HttpException, IOException {
         try {
+            if (channel.isLocalReset()) {
+                return;
+            }
+            if (cancelled.get()) {
+                localResetCancelled();
+                return;
+            }
+            touch();
+
             handler.produceOutput();
         } catch (final ProtocolException ex) {
             localReset(ex, H2Error.PROTOCOL_ERROR);
         }
     }
 
     void produceInputCapacityUpdate() throws IOException {
+        if (channel.isLocalReset()) {
+            return;
+        }
+        if (cancelled.get()) {
+            localResetCancelled();
+            return;
+        }
+        touch();
         handler.updateInputCapacity();
     }
 
@@ -299,4 +332,37 @@ public String toString() {
         return buf.toString();
     }
 
+    private void markCreatedAndActive() {
+        final long now = System.nanoTime();
+        this.createdNanos = now;
+        this.lastActivityNanos = now;
+    }
+
+    private void touch() {
+        this.lastActivityNanos = System.nanoTime();
+    }
+
+    long getCreatedNanos() {
+        return createdNanos;
+    }
+
+    long getLastActivityNanos() {
+        return lastActivityNanos;
+    }
+
+    Timeout getIdleTimeout() {
+        return idleTimeout;
+    }
+
+    void setIdleTimeout(final Timeout idleTimeout) {
+        this.idleTimeout = idleTimeout;
+    }
+
+    Timeout getLifetimeTimeout() {
+        return lifetimeTimeout;
+    }
+
+    void setLifetimeTimeout(final Timeout lifetimeTimeout) {
+        this.lifetimeTimeout = lifetimeTimeout;
+    }
 }