Merged PR 3614: Port 2.1 fixes to 2.2

Placeholder

Author: BrennanConroy

eng/PatchConfig.props

@@ -85,6 +85,8 @@ Later on, this will be checked using this condition:
   </PropertyGroup>
   <PropertyGroup Condition=" '$(VersionPrefix)' == '2.2.9' ">
     <PackagesInPatch>
+      Microsoft.AspNetCore.Http.Connections;
+      Microsoft.AspNetCore.SignalR.Core;
     </PackagesInPatch>
   </PropertyGroup>
 </Project>

src/SignalR/common/Http.Connections/src/Internal/HttpConnectionContext.cs

@@ -28,6 +28,8 @@ public class HttpConnectionContext : ConnectionContext,
                                          IHttpTransportFeature,
                                          IConnectionInherentKeepAliveFeature
     {
+        private static long _tenSeconds = TimeSpan.FromSeconds(10).Ticks;
+
         private readonly object _itemsLock = new object();
         private readonly object _heartbeatLock = new object();
         private List<(Action<object> handler, object state)> _heartbeatHandlers;
@@ -36,6 +38,12 @@ public class HttpConnectionContext : ConnectionContext,
         private IDuplexPipe _application;
         private IDictionary<object, object> _items;
 
+        private CancellationTokenSource _sendCts;
+        private bool _activeSend;
+        private long _startedSendTime;
+        private readonly object _sendingLock = new object();
+        internal CancellationToken SendingToken { get; private set; }
+
         // This tcs exists so that multiple calls to DisposeAsync all wait asynchronously
         // on the same task
         private readonly TaskCompletionSource<object> _disposeTcs = new TaskCompletionSource<object>(TaskCreationOptions.RunContinuationsAsynchronously);
@@ -240,8 +248,26 @@ private async Task WaitOnTasks(Task applicationTask, Task transportTask, bool cl
                     }
                     else
                     {
-                        // The other transports don't close their own output, so we can do it here safely
-                        Application?.Output.Complete();
+                        // Normally it isn't safe to try and acquire this lock because the Send can hold onto it for a long time if there is backpressure
+                        // It is safe to wait for this lock now because the Send will be in one of 4 states
+                        // 1. In the middle of a write which is in the middle of being canceled by the CancelPendingFlush above, when it throws
+                        //    an OperationCanceledException it will complete the PipeWriter which will make any other Send waiting on the lock
+                        //    throw an InvalidOperationException if they call Write
+                        // 2. About to write and see that there is a pending cancel from the CancelPendingFlush, go to 1 to see what happens
+                        // 3. Enters the Send and sees the Dispose state from DisposeAndRemoveAsync and releases the lock
+                        // 4. No Send in progress
+                        await WriteLock.WaitAsync();
+                        try
+                        {
+                            // Complete the applications read loop
+                            Application?.Output.Complete();
+                        }
+                        finally
+                        {
+                            WriteLock.Release();
+                        }
+
+                        Application?.Input.CancelPendingRead();
                     }
                 }
 
@@ -309,6 +335,40 @@ private async Task WaitOnTasks(Task applicationTask, Task transportTask, bool cl
             }
         }
 
+        internal void StartSendCancellation()
+        {
+            lock (_sendingLock)
+            {
+                if (_sendCts == null || _sendCts.IsCancellationRequested)
+                {
+                    _sendCts = new CancellationTokenSource();
+                    SendingToken = _sendCts.Token;
+                }
+                _startedSendTime = DateTime.UtcNow.Ticks;
+                _activeSend = true;
+            }
+        }
+        internal void TryCancelSend(long currentTicks)
+        {
+            lock (_sendingLock)
+            {
+                if (_activeSend)
+                {
+                    if (currentTicks - _startedSendTime > _tenSeconds)
+                    {
+                        _sendCts.Cancel();
+                    }
+                }
+            }
+        }
+        internal void StopSendCancellation()
+        {
+            lock (_sendingLock)
+            {
+                _activeSend = false;
+            }
+        }
+
         private static class Log
         {
             private static readonly Action<ILogger, string, Exception> _disposingConnection =

src/SignalR/common/Http.Connections/src/Internal/HttpConnectionDispatcher.cs

@@ -144,7 +144,7 @@ private async Task ExecuteAsync(HttpContext context, ConnectionDelegate connecti
                 connection.SupportedFormats = TransferFormat.Text;
 
                 // We only need to provide the Input channel since writing to the application is handled through /send.
-                var sse = new ServerSentEventsTransport(connection.Application.Input, connection.ConnectionId, _loggerFactory);
+                var sse = new ServerSentEventsTransport(connection.Application.Input, connection.ConnectionId, connection, _loggerFactory);
 
                 await DoPersistentConnection(connectionDelegate, sse, context, connection);
             }
@@ -267,7 +267,7 @@ private async Task ExecuteAsync(HttpContext context, ConnectionDelegate connecti
                         context.Response.RegisterForDispose(timeoutSource);
                         context.Response.RegisterForDispose(tokenSource);
 
-                        var longPolling = new LongPollingTransport(timeoutSource.Token, connection.Application.Input, _loggerFactory);
+                        var longPolling = new LongPollingTransport(timeoutSource.Token, connection.Application.Input, _loggerFactory, connection);
 
                         // Start the transport
                         connection.TransportTask = longPolling.ProcessRequestAsync(context, tokenSource.Token);
@@ -294,7 +294,9 @@ private async Task ExecuteAsync(HttpContext context, ConnectionDelegate connecti
                         connection.Transport.Output.Complete(connection.ApplicationTask.Exception);
 
                         // Wait for the transport to run
-                        await connection.TransportTask;
+                        // Ignore exceptions, it has been logged if there is one and the application has finished
+                        // So there is no one to give the exception to
+                        await connection.TransportTask.NoThrow();
 
                         // If the status code is a 204 it means the connection is done
                         if (context.Response.StatusCode == StatusCodes.Status204NoContent)
@@ -310,14 +312,15 @@ private async Task ExecuteAsync(HttpContext context, ConnectionDelegate connecti
                             pollAgain = false;
                         }
                     }
-                    else if (resultTask.IsFaulted)
+                    else if (resultTask.IsFaulted || resultTask.IsCanceled)
                     {
                         // Cancel current request to release any waiting poll and let dispose acquire the lock
                         currentRequestTcs.TrySetCanceled();
-
-                        // transport task was faulted, we should remove the connection
+                        // We should be able to safely dispose because there's no more data being written
+                        // We don't need to wait for close here since we've already waited for both sides
                         await _manager.DisposeAndRemoveAsync(connection, closeGracefully: false);
 
+                        // Don't poll again if we've removed the connection completely
                         pollAgain = false;
                     }
                     else if (context.Response.StatusCode == StatusCodes.Status204NoContent)
@@ -524,6 +527,14 @@ private async Task ProcessSend(HttpContext context, HttpConnectionDispatcherOpti
 
                         context.Response.StatusCode = StatusCodes.Status404NotFound;
                         context.Response.ContentType = "text/plain";
+
+                        // There are no writes anymore (since this is the write "loop")
+                        // So it is safe to complete the writer
+                        // We complete the writer here because we already have the WriteLock acquired
+                        // and it's unsafe to complete outside of the lock
+                        // Other code isn't guaranteed to be able to acquire the lock before another write
+                        // even if CancelPendingFlush is called, and the other write could hang if there is backpressure
+                        connection.Application.Output.Complete();
                         return;
                     }
                     catch (IOException ex)
@@ -571,11 +582,8 @@ private async Task ProcessDeleteAsync(HttpContext context)
 
             Log.TerminatingConection(_logger);
 
-            // Complete the receiving end of the pipe
-            connection.Application.Output.Complete();
-
-            // Dispose the connection gracefully, but don't wait for it. We assign it here so we can wait in tests
-            connection.DisposeAndRemoveTask = _manager.DisposeAndRemoveAsync(connection, closeGracefully: true);
+            // Dispose the connection, but don't wait for it. We assign it here so we can wait in tests
+            connection.DisposeAndRemoveTask = _manager.DisposeAndRemoveAsync(connection, closeGracefully: false);
 
             context.Response.StatusCode = StatusCodes.Status202Accepted;
             context.Response.ContentType = "text/plain";

src/SignalR/common/Http.Connections/src/Internal/HttpConnectionManager.cs

@@ -30,13 +30,19 @@ public partial class HttpConnectionManager
         private readonly TimerAwaitable _nextHeartbeat;
         private readonly ILogger<HttpConnectionManager> _logger;
         private readonly ILogger<HttpConnectionContext> _connectionLogger;
+        private readonly bool _useSendTimeout = true;
 
         public HttpConnectionManager(ILoggerFactory loggerFactory, IApplicationLifetime appLifetime)
         {
             _logger = loggerFactory.CreateLogger<HttpConnectionManager>();
             _connectionLogger = loggerFactory.CreateLogger<HttpConnectionContext>();
             _nextHeartbeat = new TimerAwaitable(_heartbeatTickRate, _heartbeatTickRate);
 
+            if (AppContext.TryGetSwitch("Microsoft.AspNetCore.Http.Connections.DoNotUseSendTimeout", out var timeoutDisabled))
+            {
+                _useSendTimeout = !timeoutDisabled;
+            }
+
             // Register these last as the callbacks could run immediately
             appLifetime.ApplicationStarted.Register(() => Start());
             appLifetime.ApplicationStopping.Register(() => CloseConnections());
@@ -153,20 +159,26 @@ public async Task ScanAsync()
                     connection.StateLock.Release();
                 }
 
+                var utcNow = DateTimeOffset.UtcNow;
                 // Once the decision has been made to dispose we don't check the status again
                 // But don't clean up connections while the debugger is attached.
-                if (!Debugger.IsAttached && status == HttpConnectionStatus.Inactive && (DateTimeOffset.UtcNow - lastSeenUtc).TotalSeconds > 5)
+                if (!Debugger.IsAttached && status == HttpConnectionStatus.Inactive && (utcNow - lastSeenUtc).TotalSeconds > 5)
                 {
                     Log.ConnectionTimedOut(_logger, connection.ConnectionId);
                     HttpConnectionsEventSource.Log.ConnectionTimedOut(connection.ConnectionId);
 
                     // This is most likely a long polling connection. The transport here ends because
-                    // a poll completed and has been inactive for > 5 seconds so we wait for the 
+                    // a poll completed and has been inactive for > 5 seconds so we wait for the
                     // application to finish gracefully
                     _ = DisposeAndRemoveAsync(connection, closeGracefully: true);
                 }
                 else
                 {
+                    if (!Debugger.IsAttached && _useSendTimeout)
+                    {
+                        connection.TryCancelSend(utcNow.Ticks);
+                    }
+
                     // Tick the heartbeat, if the connection is still active
                     connection.TickHeartbeat();
                 }

src/SignalR/common/Http.Connections/src/Internal/TaskExtensions.cs

@@ -0,0 +1,24 @@
+// Copyright (c) .NET Foundation. All rights reserved.
+// Licensed under the Apache License, Version 2.0. See License.txt in the project root for license information.
+using System.Runtime.CompilerServices;
+namespace System.Threading.Tasks
+{
+    internal static class TaskExtensions
+    {
+        public static async Task NoThrow(this Task task)
+        {
+            await new NoThrowAwaiter(task);
+        }
+    }
+    internal readonly struct NoThrowAwaiter : ICriticalNotifyCompletion
+    {
+        private readonly Task _task;
+        public NoThrowAwaiter(Task task) { _task = task; }
+        public NoThrowAwaiter GetAwaiter() => this;
+        public bool IsCompleted => _task.IsCompleted;
+        // Observe exception
+        public void GetResult() { _ = _task.Exception; }
+        public void OnCompleted(Action continuation) => _task.GetAwaiter().OnCompleted(continuation);
+        public void UnsafeOnCompleted(Action continuation) => OnCompleted(continuation);
+    }
+}

src/SignalR/common/Http.Connections/src/Internal/Transports/LongPollingTransport.cs

@@ -16,6 +16,7 @@ public class LongPollingTransport : IHttpTransport
         private readonly PipeReader _application;
         private readonly ILogger _logger;
         private readonly CancellationToken _timeoutToken;
+        private readonly HttpConnectionContext _connection;
 
         public LongPollingTransport(CancellationToken timeoutToken, PipeReader application, ILoggerFactory loggerFactory)
         {
@@ -24,14 +25,20 @@ public LongPollingTransport(CancellationToken timeoutToken, PipeReader applicati
             _logger = loggerFactory.CreateLogger<LongPollingTransport>();
         }
 
+        internal LongPollingTransport(CancellationToken timeoutToken, PipeReader application, ILoggerFactory loggerFactory, HttpConnectionContext connection)
+            : this(timeoutToken, application, loggerFactory)
+        {
+            _connection = connection;
+        }
+
         public async Task ProcessRequestAsync(HttpContext context, CancellationToken token)
         {
             try
             {
                 var result = await _application.ReadAsync(token);
                 var buffer = result.Buffer;
 
-                if (buffer.IsEmpty && result.IsCompleted)
+                if (buffer.IsEmpty && (result.IsCompleted || result.IsCanceled))
                 {
                     Log.LongPolling204(_logger);
                     context.Response.ContentType = "text/plain";
@@ -49,19 +56,22 @@ public async Task ProcessRequestAsync(HttpContext context, CancellationToken tok
 
                 try
                 {
-                    await context.Response.Body.WriteAsync(buffer);
+                    _connection?.StartSendCancellation();
+                    await context.Response.Body.WriteAsync(buffer, _connection?.SendingToken ?? default);
                 }
                 finally
                 {
+                    _connection?.StopSendCancellation();
                     _application.AdvanceTo(buffer.End);
                 }
             }
             catch (OperationCanceledException)
             {
-                // 3 cases:
+                // 4 cases:
                 // 1 - Request aborted, the client disconnected (no response)
                 // 2 - The poll timeout is hit (204)
-                // 3 - A new request comes in and cancels this request (204)
+                // 3 - SendingToken was canceled, abort the connection
+                // 4 - A new request comes in and cancels this request (204)
 
                 // Case 1
                 if (context.RequestAborted.IsCancellationRequested)
@@ -79,9 +89,16 @@ public async Task ProcessRequestAsync(HttpContext context, CancellationToken tok
                     context.Response.ContentType = "text/plain";
                     context.Response.StatusCode = StatusCodes.Status200OK;
                 }
-                else
+                else if (_connection?.SendingToken.IsCancellationRequested == true)
                 {
                     // Case 3
+                    context.Response.ContentType = "text/plain";
+                    context.Response.StatusCode = StatusCodes.Status204NoContent;
+                    throw;
+                }
+                else
+                {
+                    // Case 4
                     Log.LongPolling204(_logger);
                     context.Response.ContentType = "text/plain";
                     context.Response.StatusCode = StatusCodes.Status204NoContent;

src/SignalR/common/Http.Connections/src/Internal/Transports/ServerSentEventsMessageFormatter.cs

@@ -4,6 +4,7 @@
 using System;
 using System.Buffers;
 using System.IO;
+using System.Threading;
 using System.Threading.Tasks;
 
 namespace Microsoft.AspNetCore.Http.Connections.Internal
@@ -15,19 +16,24 @@ public static class ServerSentEventsMessageFormatter
 
         private const byte LineFeed = (byte)'\n';
 
-        public static async Task WriteMessageAsync(ReadOnlySequence<byte> payload, Stream output)
+        public static Task WriteMessageAsync(ReadOnlySequence<byte> payload, Stream output)
+        {
+            return WriteMessageAsync(payload, output, default);
+        }
+
+        internal static async Task WriteMessageAsync(ReadOnlySequence<byte> payload, Stream output, CancellationToken token)
         {
             // Payload does not contain a line feed so write it directly to output
             if (payload.PositionOf(LineFeed) == null)
             {
                 if (payload.Length > 0)
                 {
-                    await output.WriteAsync(DataPrefix, 0, DataPrefix.Length);
-                    await output.WriteAsync(payload);
-                    await output.WriteAsync(Newline, 0, Newline.Length);
+                    await output.WriteAsync(DataPrefix, 0, DataPrefix.Length, token);
+                    await output.WriteAsync(payload, token);
+                    await output.WriteAsync(Newline, 0, Newline.Length, token);
                 }
 
-                await output.WriteAsync(Newline, 0, Newline.Length);
+                await output.WriteAsync(Newline, 0, Newline.Length, token);
                 return;
             }
 
@@ -37,7 +43,7 @@ public static async Task WriteMessageAsync(ReadOnlySequence<byte> payload, Strea
             await WriteMessageToMemory(ms, payload);
             ms.Position = 0;
 
-            await ms.CopyToAsync(output);
+            await ms.CopyToAsync(output, bufferSize: 81920, token);
         }
 
         /// <summary>