Use the default pipe options for backpressure (#21001)

* Use the default pipe options for backpressure
- This controls memory so clients aren't easily overwhelmed. With the changes made to Pipe to no longer throw if the pause threshold is exceeded makes this work well.
- Remove PipeReaderFactory
- Implement cancellation in SSE

Contributes to #17797

Author: davidfowl

src/SignalR/clients/csharp/Client/test/UnitTests/ServerSentEventsTransportTests.cs

@@ -43,7 +43,7 @@ public async Task CanStartStopSSETransport()
                         .Setup(s => s.CopyToAsync(It.IsAny<Stream>(), It.IsAny<int>(), It.IsAny<CancellationToken>()))
                         .Returns(copyToAsyncTcs.Task);
                     mockStream.Setup(s => s.CanRead).Returns(true);
-                    return new HttpResponseMessage {Content = new StreamContent(mockStream.Object)};
+                    return new HttpResponseMessage { Content = new StreamContent(mockStream.Object) };
                 });
 
             try
@@ -76,15 +76,17 @@ public async Task SSETransportStopsSendAndReceiveLoopsWhenTransportStopped()
                 {
                     var mockStream = new Mock<Stream>();
                     mockStream
-                        .Setup(s => s.CopyToAsync(It.IsAny<Stream>(), It.IsAny<int>(), It.IsAny<CancellationToken>()))
-                        .Returns<Stream, int, CancellationToken>(async (stream, bufferSize, t) =>
+                        .Setup(s => s.ReadAsync(It.IsAny<Memory<byte>>(), It.IsAny<CancellationToken>()))
+                        .Returns<Memory<byte>, CancellationToken>(async (data, t) =>
                         {
-                            var buffer = Encoding.ASCII.GetBytes("data: 3:abc\r\n\r\n");
-                            while (!t.IsCancellationRequested)
+                            if (t.IsCancellationRequested)
                             {
-                                await stream.WriteAsync(buffer, 0, buffer.Length).OrTimeout();
-                                await Task.Delay(100);
+                                return 0;
                             }
+
+                            int count = Encoding.ASCII.GetBytes("data: 3:abc\r\n\r\n", data.Span);
+                            await Task.Delay(100);
+                            return count;
                         });
                     mockStream.Setup(s => s.CanRead).Returns(true);
 
@@ -120,6 +122,7 @@ await sseTransport.StartAsync(
         public async Task SSETransportStopsWithErrorIfServerSendsIncompleteResults()
         {
             var mockHttpHandler = new Mock<HttpMessageHandler>();
+            var calls = 0;
             mockHttpHandler.Protected()
                 .Setup<Task<HttpResponseMessage>>("SendAsync", ItExpr.IsAny<HttpRequestMessage>(), ItExpr.IsAny<CancellationToken>())
                 .Returns<HttpRequestMessage, CancellationToken>(async (request, cancellationToken) =>
@@ -128,11 +131,15 @@ public async Task SSETransportStopsWithErrorIfServerSendsIncompleteResults()
 
                     var mockStream = new Mock<Stream>();
                     mockStream
-                        .Setup(s => s.CopyToAsync(It.IsAny<Stream>(), It.IsAny<int>(), It.IsAny<CancellationToken>()))
-                        .Returns<Stream, int, CancellationToken>(async (stream, bufferSize, t) =>
+                        .Setup(s => s.ReadAsync(It.IsAny<Memory<byte>>(), It.IsAny<CancellationToken>()))
+                        .Returns<Memory<byte>, CancellationToken>((data, t) =>
                         {
-                            var buffer = Encoding.ASCII.GetBytes("data: 3:a");
-                            await stream.WriteAsync(buffer, 0, buffer.Length);
+                            if (calls == 0)
+                            {
+                                calls++;
+                                return new ValueTask<int>(Encoding.ASCII.GetBytes("data: 3:a", data.Span));
+                            }
+                            return new ValueTask<int>(0);
                         });
                     mockStream.Setup(s => s.CanRead).Returns(true);
 
@@ -165,7 +172,7 @@ bool ExpectedErrors(WriteContext writeContext)
             }
 
             var eventStreamTcs = new TaskCompletionSource<object>();
-            var copyToAsyncTcs = new TaskCompletionSource<int>();
+            var readTcs = new TaskCompletionSource<int>();
 
             var mockHttpHandler = new Mock<HttpMessageHandler>();
             mockHttpHandler.Protected()
@@ -182,8 +189,14 @@ bool ExpectedErrors(WriteContext writeContext)
                         // returns unfinished task to block pipelines
                         var mockStream = new Mock<Stream>();
                         mockStream
-                            .Setup(s => s.CopyToAsync(It.IsAny<Stream>(), It.IsAny<int>(), It.IsAny<CancellationToken>()))
-                            .Returns(copyToAsyncTcs.Task);
+                            .Setup(s => s.ReadAsync(It.IsAny<Memory<byte>>(), It.IsAny<CancellationToken>()))
+                            .Returns<Memory<byte>, CancellationToken>(async (data, ct) =>
+                            {
+                                using (ct.Register(() => readTcs.TrySetCanceled()))
+                                {
+                                    return await readTcs.Task;
+                                }
+                            });
                         mockStream.Setup(s => s.CanRead).Returns(true);
                         return new HttpResponseMessage { Content = new StreamContent(mockStream.Object) };
                     }
@@ -214,7 +227,7 @@ await sseTransport.StartAsync(
         public async Task SSETransportStopsIfChannelClosed()
         {
             var eventStreamTcs = new TaskCompletionSource<object>();
-            var copyToAsyncTcs = new TaskCompletionSource<int>();
+            var readTcs = new TaskCompletionSource<int>();
 
             var mockHttpHandler = new Mock<HttpMessageHandler>();
             mockHttpHandler.Protected()
@@ -229,8 +242,14 @@ public async Task SSETransportStopsIfChannelClosed()
                     // returns unfinished task to block pipelines
                     var mockStream = new Mock<Stream>();
                     mockStream
-                        .Setup(s => s.CopyToAsync(It.IsAny<Stream>(), It.IsAny<int>(), It.IsAny<CancellationToken>()))
-                        .Returns(copyToAsyncTcs.Task);
+                            .Setup(s => s.ReadAsync(It.IsAny<Memory<byte>>(), It.IsAny<CancellationToken>()))
+                            .Returns<Memory<byte>, CancellationToken>(async (data, ct) =>
+                            {
+                                using (ct.Register(() => readTcs.TrySetCanceled()))
+                                {
+                                    return await readTcs.Task;
+                                }
+                            });
                     mockStream.Setup(s => s.CanRead).Returns(true);
                     return new HttpResponseMessage { Content = new StreamContent(mockStream.Object) };
                 });
@@ -281,7 +300,7 @@ await sseTransport.StartAsync(
         public async Task SSETransportCancelsSendOnStop()
         {
             var eventStreamTcs = new TaskCompletionSource<object>();
-            var copyToAsyncTcs = new TaskCompletionSource<object>();
+            var readTcs = new TaskCompletionSource<object>();
             var sendSyncPoint = new SyncPoint();
 
             var mockHttpHandler = new Mock<HttpMessageHandler>();
@@ -299,10 +318,10 @@ public async Task SSETransportCancelsSendOnStop()
                         // returns unfinished task to block pipelines
                         var mockStream = new Mock<Stream>();
                         mockStream
-                            .Setup(s => s.CopyToAsync(It.IsAny<Stream>(), It.IsAny<int>(), It.IsAny<CancellationToken>()))
-                            .Returns<Stream, int, CancellationToken>(async (stream, bufferSize, t) =>
+                            .Setup(s => s.ReadAsync(It.IsAny<Memory<byte>>(), It.IsAny<CancellationToken>()))
+                            .Returns(async () =>
                             {
-                                await copyToAsyncTcs.Task;
+                                await readTcs.Task;
 
                                 throw new TaskCanceledException();
                             });
@@ -332,7 +351,7 @@ await sseTransport.StartAsync(
 
                 var stopTask = sseTransport.StopAsync();
 
-                copyToAsyncTcs.SetResult(null);
+                readTcs.SetResult(null);
                 sendSyncPoint.Continue();
 
                 await stopTask;

src/SignalR/clients/csharp/Http.Connections.Client/src/Internal/ClientPipeOptions.cs

@@ -7,6 +7,6 @@ namespace Microsoft.AspNetCore.Http.Connections.Client.Internal
 {
     internal static class ClientPipeOptions
     {
-        public static PipeOptions DefaultOptions = new PipeOptions(writerScheduler: PipeScheduler.ThreadPool, readerScheduler: PipeScheduler.ThreadPool, useSynchronizationContext: false, pauseWriterThreshold: 0, resumeWriterThreshold: 0);
+        public static PipeOptions DefaultOptions = new PipeOptions(writerScheduler: PipeScheduler.ThreadPool, readerScheduler: PipeScheduler.ThreadPool, useSynchronizationContext: false);
     }
 }

src/SignalR/clients/csharp/Http.Connections.Client/src/Internal/PipeReaderFactory.cs

@@ -129,12 +129,15 @@ private async Task ProcessAsync(Uri url, HttpResponseMessage response)
         private async Task ProcessEventStream(HttpResponseMessage response, CancellationToken cancellationToken)
         {
             Log.StartReceive(_logger);
+            
+            static void CancelReader(object state) => ((PipeReader)state).CancelPendingRead();
 
             using (response)
             using (var stream = await response.Content.ReadAsStreamAsync())
             {
-                var options = new PipeOptions(pauseWriterThreshold: 0, resumeWriterThreshold: 0);
-                var reader = PipeReaderFactory.CreateFromStream(options, stream, cancellationToken);
+                var reader = PipeReader.Create(stream);
+
+                using var registration = cancellationToken.Register(CancelReader, reader);
 
                 try
                 {

src/SignalR/clients/csharp/Http.Connections.Client/src/Internal/ServerSentEventsTransport.cs

@@ -329,15 +329,30 @@ public async Task ConnectionCanSendAndReceiveDifferentMessageSizesWebSocketsTran
                     logger.LogInformation("Started connection to {url}", url);
 
                     var bytes = Encoding.UTF8.GetBytes(message);
-                    logger.LogInformation("Sending {length} byte message", bytes.Length);
-                    await connection.Transport.Output.WriteAsync(bytes).OrTimeout();
-                    logger.LogInformation("Sent message");
 
-                    logger.LogInformation("Receiving message");
-                    // Big timeout here because it can take a while to receive all the bytes
-                    var receivedData = await connection.Transport.Input.ReadAsync(bytes.Length).OrTimeout(TimeSpan.FromMinutes(2));
-                    Assert.Equal(message, Encoding.UTF8.GetString(receivedData));
-                    logger.LogInformation("Completed receive");
+                    async Task SendMessage()
+                    {
+                        logger.LogInformation("Sending {length} byte message", bytes.Length);
+                        await connection.Transport.Output.WriteAsync(bytes).OrTimeout();
+                        logger.LogInformation("Sent message");
+                    }
+
+                    async Task ReceiveMessage()
+                    {
+                        logger.LogInformation("Receiving message");
+                        // Big timeout here because it can take a while to receive all the bytes
+                        var receivedData = await connection.Transport.Input.ReadAsync(bytes.Length).OrTimeout(TimeSpan.FromMinutes(2));
+                        Assert.Equal(message, Encoding.UTF8.GetString(receivedData));
+                        logger.LogInformation("Completed receive");
+                    }
+
+                    // Send the receive concurrently so that back pressure is released
+                    // for server -> client sends
+                    var sendingTask = SendMessage();
+                    var receivingTask = ReceiveMessage();
+
+                    await sendingTask;
+                    await receivingTask;
                 }
                 catch (Exception ex)
                 {