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.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.
+ * <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 InterruptedIOException {
+
+    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

@@ -74,6 +74,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;
@@ -454,6 +455,9 @@ public final void onInput(final ByteBuffer src) throws HttpException, IOExceptio
                     break;
                 }
             }
+            if (connState.compareTo(ConnectionHandshake.SHUTDOWN) < 0) {
+                checkStreamTimeouts(System.nanoTime());
+            }
         }
     }
 
@@ -531,6 +535,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(); ) {
@@ -760,10 +769,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 +965,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 +1371,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 +1487,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 +1604,29 @@ 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();
+            if (idleTimeout == null || !idleTimeout.isEnabled()) {
+                continue;
+            }
+
+            final long last = stream.getLastActivityNanos();
+            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,6 +61,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;
@@ -69,6 +75,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
@@ -104,6 +112,7 @@ private void triggerClosed() {
 
     void activate() {
         reserved = false;
+        markCreatedAndActive();
         triggerOpen();
     }
 
@@ -146,6 +155,8 @@ boolean isLocalClosed() {
 
     void consumePromise(final List<Header> headers) throws HttpException, IOException {
         try {
+            touch();
+
             if (channel.isLocalReset()) {
                 return;
             }
@@ -165,6 +176,8 @@ void consumeHeader(final List<Header> headers, final boolean endOfStream) throws
             if (endOfStream) {
                 remoteClosed = true;
             }
+            touch();
+
             if (channel.isLocalReset()) {
                 return;
             }
@@ -183,6 +196,8 @@ void consumeData(final ByteBuffer src, final boolean endOfStream) throws HttpExc
             if (endOfStream) {
                 remoteClosed = true;
             }
+            touch();
+
             if (channel.isLocalReset()) {
                 return;
             }
@@ -204,13 +219,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();
     }
 
@@ -308,4 +340,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;
+    }
 }