Decrement activeStreamCount earlier s.t. client view matches the server. (#12704)

src/Servers/Kestrel/Core/src/CoreStrings.resx

@@ -1,17 +1,17 @@
 <?xml version="1.0" encoding="utf-8"?>
 <root>
-  <!--
+  <!-- 
-    Microsoft ResX Schema
+    Microsoft ResX Schema 
-
+    
     Version 2.0
-
+    
-    The primary goals of this format is to allow a simple XML format
+    The primary goals of this format is to allow a simple XML format 
-    that is mostly human readable. The generation and parsing of the
+    that is mostly human readable. The generation and parsing of the 
-    various data types are done through the TypeConverter classes
+    various data types are done through the TypeConverter classes 
     associated with the data types.
-
+    
     Example:
-
+    
     ... ado.net/XML headers & schema ...
     <resheader name="resmimetype">text/microsoft-resx</resheader>
     <resheader name="version">2.0</resheader>
@@ -26,36 +26,36 @@
         <value>[base64 mime encoded string representing a byte array form of the .NET Framework object]</value>
         <comment>This is a comment</comment>
     </data>
-
+                
-    There are any number of "resheader" rows that contain simple
+    There are any number of "resheader" rows that contain simple 
     name/value pairs.
-
+    
-    Each data row contains a name, and value. The row also contains a
+    Each data row contains a name, and value. The row also contains a 
-    type or mimetype. Type corresponds to a .NET class that support
+    type or mimetype. Type corresponds to a .NET class that support 
-    text/value conversion through the TypeConverter architecture.
+    text/value conversion through the TypeConverter architecture. 
-    Classes that don't support this are serialized and stored with the
+    Classes that don't support this are serialized and stored with the 
     mimetype set.
-
+    
-    The mimetype is used for serialized objects, and tells the
+    The mimetype is used for serialized objects, and tells the 
-    ResXResourceReader how to depersist the object. This is currently not
+    ResXResourceReader how to depersist the object. This is currently not 
     extensible. For a given mimetype the value must be set accordingly:
-
+    
-    Note - application/x-microsoft.net.object.binary.base64 is the format
+    Note - application/x-microsoft.net.object.binary.base64 is the format 
-    that the ResXResourceWriter will generate, however the reader can
+    that the ResXResourceWriter will generate, however the reader can 
     read any of the formats listed below.
-
+    
     mimetype: application/x-microsoft.net.object.binary.base64
-    value   : The object must be serialized with
+    value   : The object must be serialized with 
             : System.Runtime.Serialization.Formatters.Binary.BinaryFormatter
             : and then encoded with base64 encoding.
-
+    
     mimetype: application/x-microsoft.net.object.soap.base64
-    value   : The object must be serialized with
+    value   : The object must be serialized with 
             : System.Runtime.Serialization.Formatters.Soap.SoapFormatter
             : and then encoded with base64 encoding.
 
     mimetype: application/x-microsoft.net.object.bytearray.base64
-    value   : The object must be serialized into a byte array
+    value   : The object must be serialized into a byte array 
             : using a System.ComponentModel.TypeConverter
             : and then encoded with base64 encoding.
     -->
@@ -614,4 +614,7 @@ For more information on configuring HTTPS see https://go.microsoft.com/fwlink/?l
   <data name="Http2StreamResetByApplication" xml:space="preserve">
     <value>The HTTP/2 stream was reset by the application with error code {errorCode}.</value>
   </data>
+  <data name="Http2TellClientToCalmDown" xml:space="preserve">
+    <value>A new stream was refused because this connection has too many streams that haven't finished processing. This may happen if many streams are aborted but not yet cleaned up.</value>
+  </data>
 </root>

src/Servers/Kestrel/Core/src/Internal/Http2/Http2Connection.cs

@@ -64,7 +64,8 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http2
         private bool _gracefulCloseStarted;
 
         private readonly Dictionary<int, Http2Stream> _streams = new Dictionary<int, Http2Stream>();
-        private int _activeStreamCount = 0;
+        private int _clientActiveStreamCount = 0;
+        private int _serverActiveStreamCount = 0;
 
         // The following are the only fields that can be modified outside of the ProcessRequestsAsync loop.
         private readonly ConcurrentQueue<Http2Stream> _completedStreams = new ConcurrentQueue<Http2Stream>();
@@ -219,6 +220,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http2
                         Log.Http2StreamError(ConnectionId, ex);
                         // The client doesn't know this error is coming, allow draining additional frames for now.
                         AbortStream(_incomingFrame.StreamId, new IOException(ex.Message, ex));
+
                         await _frameWriter.WriteRstStreamAsync(ex.StreamId, ex.ErrorCode);
                     }
                     finally
@@ -232,7 +234,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http2
             catch (ConnectionResetException ex)
             {
                 // Don't log ECONNRESET errors when there are no active streams on the connection. Browsers like IE will reset connections regularly.
-                if (_activeStreamCount > 0)
+                if (_clientActiveStreamCount > 0)
                 {
                     Log.RequestProcessingError(ConnectionId, ex);
                 }
@@ -287,7 +289,10 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http2
                         stream.Abort(new IOException(CoreStrings.Http2StreamAborted, connectionError));
                     }
 
-                    while (_activeStreamCount > 0)
+                    // Use the server _serverActiveStreamCount to drain all requests on the server side.
+                    // Can't use _clientActiveStreamCount now as we now decrement that count earlier/
+                    // Can't use _streams.Count as we wait for RST/END_STREAM before removing the stream from the dictionary
+                    while (_serverActiveStreamCount > 0)
                     {
                         await _streamCompletionAwaitable;
                         UpdateCompletedStreams();
@@ -897,11 +902,21 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http2
                 throw new Http2StreamErrorException(_currentHeadersStream.StreamId, CoreStrings.Http2ErrorMissingMandatoryPseudoHeaderFields, Http2ErrorCode.PROTOCOL_ERROR);
             }
 
-            if (_activeStreamCount >= _serverSettings.MaxConcurrentStreams)
+            if (_clientActiveStreamCount >= _serverSettings.MaxConcurrentStreams)
             {
                 throw new Http2StreamErrorException(_currentHeadersStream.StreamId, CoreStrings.Http2ErrorMaxStreams, Http2ErrorCode.REFUSED_STREAM);
             }
 
+            // We don't use the _serverActiveRequestCount here as during shutdown, it and the dictionary
+            // counts get out of sync during shutdown. The streams still exist in the dictionary until the client responds with a RST or END_STREAM.
+            // Also, we care about the dictionary size for too much memory consumption.
+            if (_streams.Count >= _serverSettings.MaxConcurrentStreams * 2)
+            {
+                // Server is getting hit hard with connection resets.
+                // Tell client to calm down.
+                // TODO consider making when to send ENHANCE_YOUR_CALM configurable?
+                throw new Http2StreamErrorException(_currentHeadersStream.StreamId, CoreStrings.Http2TellClientToCalmDown, Http2ErrorCode.ENHANCE_YOUR_CALM);
+            }
             // This must be initialized before we offload the request or else we may start processing request body frames without it.
             _currentHeadersStream.InputRemaining = _currentHeadersStream.RequestHeaders.ContentLength;
 
@@ -911,8 +926,9 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http2
                 _currentHeadersStream.OnEndStreamReceived();
             }
 
-            _activeStreamCount++;
             _streams[_incomingFrame.StreamId] = _currentHeadersStream;
+            IncrementActiveClientStreamCount();
+            _serverActiveStreamCount++;
             // Must not allow app code to block the connection handling loop.
             ThreadPool.UnsafeQueueUserWorkItem(_currentHeadersStream, preferLocal: false);
         }
@@ -950,6 +966,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http2
         {
             if (_streams.TryGetValue(streamId, out var stream))
             {
+                stream.DecrementActiveClientStreamCount();
                 stream.Abort(error);
             }
         }
@@ -982,8 +999,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http2
 
                 if (stream.DrainExpirationTicks == default)
                 {
-                    // This is our first time checking this stream.
+                    _serverActiveStreamCount--;
-                    _activeStreamCount--;
                     stream.DrainExpirationTicks = now + Constants.RequestBodyDrainTimeout.Ticks;
                 }
 
@@ -1022,13 +1038,13 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http2
 
                 Log.Http2ConnectionClosing(_context.ConnectionId);
 
-                if (_gracefulCloseInitiator == GracefulCloseInitiator.Server && _activeStreamCount > 0)
+                if (_gracefulCloseInitiator == GracefulCloseInitiator.Server && _clientActiveStreamCount > 0)
                 {
                     _frameWriter.WriteGoAwayAsync(int.MaxValue, Http2ErrorCode.NO_ERROR);
                 }
             }
 
-            if (_activeStreamCount == 0)
+            if (_clientActiveStreamCount == 0)
             {
                 if (_gracefulCloseStarted)
                 {
@@ -1235,6 +1251,16 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http2
             return false;
         }
 
+        public void IncrementActiveClientStreamCount()
+        {
+            Interlocked.Increment(ref _clientActiveStreamCount);
+        }
+
+        public void DecrementActiveClientStreamCount()
+        {
+            Interlocked.Decrement(ref _clientActiveStreamCount);
+        }
+
         private class StreamCloseAwaitable : ICriticalNotifyCompletion
         {
             private static readonly Action _callbackCompleted = () => { };

src/Servers/Kestrel/Core/src/Internal/Http2/Http2OutputProducer.cs

@@ -164,6 +164,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http2
                 if (appCompleted && !_startedWritingDataFrames && (_stream.ResponseTrailers == null || _stream.ResponseTrailers.Count == 0))
                 {
                     _streamEnded = true;
+                    _stream.DecrementActiveClientStreamCount();
                     http2HeadersFrame = Http2HeadersFrameFlags.END_STREAM;
                 }
                 else
@@ -378,6 +379,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http2
                         }
 
                         _stream.ResponseTrailers.SetReadOnly();
+                        _stream.DecrementActiveClientStreamCount();
                         flushResult = await _frameWriter.WriteResponseTrailers(_streamId, _stream.ResponseTrailers);
                     }
                     else if (readResult.IsCompleted && _streamEnded)
@@ -392,7 +394,12 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http2
                     }
                     else
                     {
-                        flushResult = await _frameWriter.WriteDataAsync(_streamId, _flowControl, readResult.Buffer, endStream: readResult.IsCompleted);
+                        var endStream = readResult.IsCompleted;
+                        if (endStream)
+                        {
+                            _stream.DecrementActiveClientStreamCount();
+                        }
+                        flushResult = await _frameWriter.WriteDataAsync(_streamId, _flowControl, readResult.Buffer, endStream);
                     }
 
                     _pipeReader.AdvanceTo(readResult.Buffer.End);

src/Servers/Kestrel/Core/src/Internal/Http2/Http2Stream.cs

@@ -24,6 +24,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http2
         private readonly StreamInputFlowControl _inputFlowControl;
         private readonly StreamOutputFlowControl _outputFlowControl;
 
+        private bool _decrementCalled;
         public Pipe RequestBodyPipe { get; }
 
         internal long DrainExpirationTicks { get; set; }
@@ -97,6 +98,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http2
                     var (oldState, newState) = ApplyCompletionFlag(StreamCompletionFlags.Aborted);
                     if (oldState != newState)
                     {
+                        Debug.Assert(!_decrementCalled);
                         // Don't block on IO. This never faults.
                         _ = _http2Output.WriteRstStreamAsync(Http2ErrorCode.NO_ERROR);
                         RequestBodyPipe.Writer.Complete();
@@ -419,6 +421,9 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http2
 
         public void AbortRstStreamReceived()
         {
+            // Client sent a reset stream frame, decrement total count.
+            DecrementActiveClientStreamCount();
+
             ApplyCompletionFlag(StreamCompletionFlags.RstStreamReceived);
             Abort(new IOException(CoreStrings.Http2StreamResetByClient));
         }
@@ -460,6 +465,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http2
 
             Log.Http2StreamResetAbort(TraceIdentifier, error, abortReason);
 
+            DecrementActiveClientStreamCount();
             // Don't block on IO. This never faults.
             _ = _http2Output.WriteRstStreamAsync(error);
 
@@ -481,6 +487,23 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http2
             _inputFlowControl.Abort();
         }
 
+        public void DecrementActiveClientStreamCount()
+        {
+            // Decrement can be called twice, via calling CompleteAsync and then Abort on the HttpContext.
+            // Only decrement once total.
+            lock (_completionLock)
+            {
+                if (_decrementCalled)
+                {
+                    return;
+                }
+
+                _decrementCalled = true;
+            }
+          
+            _context.StreamLifetimeHandler.DecrementActiveClientStreamCount();
+        }
+
         private Pipe CreateRequestBodyPipe(uint windowSize)
             => new Pipe(new PipeOptions
             (

src/Servers/Kestrel/Core/src/Internal/Http2/IHttp2StreamLifetimeHandler.cs

@@ -6,5 +6,6 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http2
     internal interface IHttp2StreamLifetimeHandler
     {
         void OnStreamCompleted(Http2Stream stream);
+        void DecrementActiveClientStreamCount();
     }
 }

src/Servers/Kestrel/test/InMemory.FunctionalTests/Http2/Http2ConnectionTests.cs

@@ -916,6 +916,66 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Tests
                 });
         }
 
+        [Fact]
+        public async Task Frame_MultipleStreams_CanBeCreatedIfClientCountIsLessThanActualMaxStreamCount()
+        {
+            _serviceContext.ServerOptions.Limits.Http2.MaxStreamsPerConnection = 1; 
+            var firstRequestBlock = new TaskCompletionSource<object>();
+            var firstRequestReceived = new TaskCompletionSource<object>();
+            var makeFirstRequestWait = false;
+            await InitializeConnectionAsync(async context =>
+            {
+                if (!makeFirstRequestWait)
+                {
+                    makeFirstRequestWait = true;
+                    firstRequestReceived.SetResult(null);
+                    await firstRequestBlock.Task.DefaultTimeout();
+                }
+            });
+
+            await StartStreamAsync(1, _browserRequestHeaders, endStream: true);
+            await SendRstStreamAsync(1);
+
+            await firstRequestReceived.Task.DefaultTimeout();
+
+            await StartStreamAsync(3, _browserRequestHeaders, endStream: true);
+
+            await ExpectAsync(Http2FrameType.HEADERS,
+                withLength: 55,
+                withFlags: (byte)(Http2HeadersFrameFlags.END_HEADERS | Http2HeadersFrameFlags.END_STREAM),
+                withStreamId: 3);
+
+            firstRequestBlock.SetResult(null);
+
+            await StopConnectionAsync(3, ignoreNonGoAwayFrames: false);
+        }
+
+        [Fact]
+        public async Task Frame_MultipleStreams_RequestsNotFinished_EnhanceYourCalm()
+        {
+            _serviceContext.ServerOptions.Limits.Http2.MaxStreamsPerConnection = 1;
+            var tcs = new TaskCompletionSource<object>();
+            await InitializeConnectionAsync(async context =>
+            {
+                await tcs.Task.DefaultTimeout();
+            });
+
+            await StartStreamAsync(1, _browserRequestHeaders, endStream: false);
+            await SendRstStreamAsync(1);
+            await StartStreamAsync(3, _browserRequestHeaders, endStream: true);
+            await SendRstStreamAsync(3);
+            await StartStreamAsync(5, _browserRequestHeaders, endStream: true);
+
+            await WaitForStreamErrorAsync(
+                expectedStreamId: 5,
+                expectedErrorCode: Http2ErrorCode.ENHANCE_YOUR_CALM,
+                expectedErrorMessage: CoreStrings.Http2TellClientToCalmDown);
+
+            tcs.SetResult(null);
+
+            await StopConnectionAsync(5, ignoreNonGoAwayFrames: false);
+        }
+
         [Fact]
         public async Task DATA_Received_StreamClosedImplicitly_ConnectionError()
         {