Added the ability to configure the buffer sizes on the client (#35107)

davidfowl · web-flow · commit 3535cfb61dd9 · 2021-08-10T07:06:59.000Z
* Added the ability to configure the buffer sizes on the client
- When data comes in from the transport or sent to the transport, we use the default pipe buffer size on the client (which is 65K). This change makes it possible to configure both the transport buffer and the application buffer so that it can be configured based on users scenarios.
- Raise the default limit from 65K to 1MB on the client side and raise it from 32K (old pipe limit) to 65K on the server side.
- Added tests
diff --git a/src/SignalR/clients/csharp/Client/test/UnitTests/HttpConnectionFactoryTests.cs b/src/SignalR/clients/csharp/Client/test/UnitTests/HttpConnectionFactoryTests.cs
@@ -98,6 +98,8 @@ public void ShallowCopyHttpConnectionOptionsCopiesAllPublicProperties()
                 { $"{nameof(HttpConnectionOptions.DefaultTransferFormat)}", TransferFormat.Text },
                 { $"{nameof(HttpConnectionOptions.WebSocketConfiguration)}", webSocketConfig },
                 { $"{nameof(HttpConnectionOptions.WebSocketFactory)}", webSocketFactory },
+                { $"{nameof(HttpConnectionOptions.ApplicationMaxBufferSize)}", 1L * 1024 * 1024 },
+                { $"{nameof(HttpConnectionOptions.TransportMaxBufferSize)}", 1L * 1024 * 1024 },
             };
 
             var options = new HttpConnectionOptions();
diff --git a/src/SignalR/clients/csharp/Client/test/UnitTests/HttpConnectionTests.ConnectionLifecycle.cs b/src/SignalR/clients/csharp/Client/test/UnitTests/HttpConnectionTests.ConnectionLifecycle.cs
@@ -337,7 +337,7 @@ bool ExpectedErrors(WriteContext writeContext)
                         return Task.FromResult(ResponseUtils.CreateResponse(HttpStatusCode.InternalServerError));
                     });
 
-                    var sse = new ServerSentEventsTransport(new HttpClient(httpHandler), LoggerFactory);
+                    var sse = new ServerSentEventsTransport(new HttpClient(httpHandler), loggerFactory: LoggerFactory);
 
                     await WithConnectionAsync(
                         CreateConnection(httpHandler, loggerFactory: LoggerFactory, transport: sse),
@@ -363,7 +363,7 @@ public async Task SSEWaitsForResponseToStart()
                         return ResponseUtils.CreateResponse(HttpStatusCode.Accepted);
                     });
 
-                    var sse = new ServerSentEventsTransport(new HttpClient(httpHandler), LoggerFactory);
+                    var sse = new ServerSentEventsTransport(new HttpClient(httpHandler), loggerFactory: LoggerFactory);
 
                     await WithConnectionAsync(
                         CreateConnection(httpHandler, loggerFactory: LoggerFactory, transport: sse),
@@ -434,7 +434,8 @@ public async Task CancellationTokenFromStartPassedToTransport()
 
                     await WithConnectionAsync(
                         CreateConnection(httpHandler,
-                        transport: new TestTransport(onTransportStart: () => {
+                        transport: new TestTransport(onTransportStart: () =>
+                        {
                             // Cancel the token when the transport is starting  which will fail the startTask.
                             cts.Cancel();
                             return Task.CompletedTask;
@@ -462,7 +463,8 @@ public async Task CanceledCancellationTokenPassedToStartThrows()
 
                     await WithConnectionAsync(
                         CreateConnection(httpHandler,
-                        transport: new TestTransport(onTransportStart: () => {
+                        transport: new TestTransport(onTransportStart: () =>
+                        {
                             transportStartCalled = true;
                             return Task.CompletedTask;
                         })),
@@ -493,7 +495,7 @@ bool ExpectedErrors(WriteContext writeContext)
                         throw new OperationCanceledException("Cancel SSE Start.");
                     });
 
-                    var sse = new ServerSentEventsTransport(new HttpClient(httpHandler), LoggerFactory);
+                    var sse = new ServerSentEventsTransport(new HttpClient(httpHandler), loggerFactory: LoggerFactory);
 
                     await WithConnectionAsync(
                         CreateConnection(httpHandler, loggerFactory: LoggerFactory, transport: sse, transportType: HttpTransportType.ServerSentEvents),
diff --git a/src/SignalR/clients/csharp/Client/test/UnitTests/HttpConnectionTests.cs b/src/SignalR/clients/csharp/Client/test/UnitTests/HttpConnectionTests.cs
@@ -7,6 +7,7 @@
 using System.Net.Http;
 using System.Security.Cryptography.X509Certificates;
 using System.Threading.Tasks;
+using Microsoft.AspNetCore.Connections;
 using Microsoft.AspNetCore.Http.Connections;
 using Microsoft.AspNetCore.Http.Connections.Client;
 using Microsoft.AspNetCore.SignalR.Tests;
@@ -21,6 +22,25 @@ namespace Microsoft.AspNetCore.SignalR.Client.Tests
 {
     public partial class HttpConnectionTests : VerifiableLoggedTest
     {
+        [Fact]
+        public void HttpConnectionOptionsDefaults()
+        {
+            var httpOptions = new HttpConnectionOptions();
+            Assert.Equal(1024 * 1024, httpOptions.TransportMaxBufferSize);
+            Assert.Equal(1024 * 1024, httpOptions.ApplicationMaxBufferSize);
+            Assert.Equal(TimeSpan.FromSeconds(5), httpOptions.CloseTimeout);
+            Assert.Equal(TransferFormat.Binary, httpOptions.DefaultTransferFormat);
+            Assert.Equal(HttpTransports.All, httpOptions.Transports);
+        }
+
+        [Fact]
+        public void HttpConnectionOptionsNegativeBufferSizeThrows()
+        {
+            var httpOptions = new HttpConnectionOptions();
+            Assert.Throws<ArgumentOutOfRangeException>(() => httpOptions.TransportMaxBufferSize = -1);
+            Assert.Throws<ArgumentOutOfRangeException>(() => httpOptions.ApplicationMaxBufferSize = -1);
+        }
+
         [Fact]
         public void CannotCreateConnectionWithNullUrl()
         {
diff --git a/src/SignalR/clients/csharp/Client/test/UnitTests/LongPollingTransportTests.cs b/src/SignalR/clients/csharp/Client/test/UnitTests/LongPollingTransportTests.cs
@@ -254,7 +254,7 @@ public async Task StopTransportWhenConnectionAlreadyStoppedOnServer()
             {
                 using (var httpClient = new HttpClient(mockHttpHandler.Object))
                 {
-                    var longPollingTransport = new LongPollingTransport(httpClient, LoggerFactory);
+                    var longPollingTransport = new LongPollingTransport(httpClient, loggerFactory: LoggerFactory);
 
                     await longPollingTransport.StartAsync(TestUri, TransferFormat.Binary).DefaultTimeout();
 
diff --git a/src/SignalR/clients/csharp/Client/test/UnitTests/ServerSentEventsTransportTests.cs b/src/SignalR/clients/csharp/Client/test/UnitTests/ServerSentEventsTransportTests.cs
@@ -52,7 +52,7 @@ public async Task CanStartStopSSETransport()
                 using (var httpClient = new HttpClient(mockHttpHandler.Object))
                 using (StartVerifiableLog())
                 {
-                    var sseTransport = new ServerSentEventsTransport(httpClient, LoggerFactory);
+                    var sseTransport = new ServerSentEventsTransport(httpClient, loggerFactory: LoggerFactory);
                     await sseTransport.StartAsync(
                         new Uri("http://fakeuri.org"), TransferFormat.Text).DefaultTimeout();
 
@@ -97,7 +97,7 @@ public async Task SSETransportStopsSendAndReceiveLoopsWhenTransportStopped()
             using (var httpClient = new HttpClient(mockHttpHandler.Object))
             using (StartVerifiableLog())
             {
-                var sseTransport = new ServerSentEventsTransport(httpClient, LoggerFactory);
+                var sseTransport = new ServerSentEventsTransport(httpClient, loggerFactory: LoggerFactory);
 
                 Task transportActiveTask;
                 try
@@ -150,7 +150,7 @@ public async Task SSETransportStopsWithErrorIfServerSendsIncompleteResults()
             using (var httpClient = new HttpClient(mockHttpHandler.Object))
             using (StartVerifiableLog())
             {
-                var sseTransport = new ServerSentEventsTransport(httpClient, LoggerFactory);
+                var sseTransport = new ServerSentEventsTransport(httpClient, loggerFactory: LoggerFactory);
 
                 await sseTransport.StartAsync(
                     new Uri("http://fakeuri.org"), TransferFormat.Text).DefaultTimeout();
@@ -208,7 +208,7 @@ bool ExpectedErrors(WriteContext writeContext)
             using (var httpClient = new HttpClient(mockHttpHandler.Object))
             using (StartVerifiableLog(expectedErrorsFilter: ExpectedErrors))
             {
-                var sseTransport = new ServerSentEventsTransport(httpClient, LoggerFactory);
+                var sseTransport = new ServerSentEventsTransport(httpClient, loggerFactory: LoggerFactory);
 
                 await sseTransport.StartAsync(
                     new Uri("http://fakeuri.org"), TransferFormat.Text).DefaultTimeout();
@@ -258,7 +258,7 @@ public async Task SSETransportStopsIfChannelClosed()
             using (var httpClient = new HttpClient(mockHttpHandler.Object))
             using (StartVerifiableLog())
             {
-                var sseTransport = new ServerSentEventsTransport(httpClient, LoggerFactory);
+                var sseTransport = new ServerSentEventsTransport(httpClient, loggerFactory: LoggerFactory);
 
                 await sseTransport.StartAsync(
                     new Uri("http://fakeuri.org"), TransferFormat.Text).DefaultTimeout();
@@ -285,7 +285,7 @@ public async Task SSETransportStopsIfTheServerClosesTheStream()
             using (var httpClient = new HttpClient(mockHttpHandler.Object))
             using (StartVerifiableLog())
             {
-                var sseTransport = new ServerSentEventsTransport(httpClient, LoggerFactory);
+                var sseTransport = new ServerSentEventsTransport(httpClient, loggerFactory: LoggerFactory);
 
                 await sseTransport.StartAsync(
                     new Uri("http://fakeuri.org"), TransferFormat.Text).DefaultTimeout();
@@ -339,7 +339,7 @@ public async Task SSETransportCancelsSendOnStop()
             using (var httpClient = new HttpClient(mockHttpHandler.Object))
             using (StartVerifiableLog())
             {
-                var sseTransport = new ServerSentEventsTransport(httpClient, LoggerFactory);
+                var sseTransport = new ServerSentEventsTransport(httpClient, loggerFactory: LoggerFactory);
 
                 await sseTransport.StartAsync(
                     new Uri("http://fakeuri.org"), TransferFormat.Text).DefaultTimeout();
@@ -374,7 +374,7 @@ public async Task SSETransportDoesNotSupportBinary()
             using (var httpClient = new HttpClient(mockHttpHandler.Object))
             using (StartVerifiableLog())
             {
-                var sseTransport = new ServerSentEventsTransport(httpClient, LoggerFactory);
+                var sseTransport = new ServerSentEventsTransport(httpClient, loggerFactory: LoggerFactory);
 
                 var ex = await Assert.ThrowsAsync<ArgumentException>(() => sseTransport.StartAsync(new Uri("http://fakeuri.org"), TransferFormat.Binary).DefaultTimeout());
 
@@ -401,7 +401,7 @@ public async Task SSETransportThrowsForInvalidTransferFormat(TransferFormat tran
             using (var httpClient = new HttpClient(mockHttpHandler.Object))
             using (StartVerifiableLog())
             {
-                var sseTransport = new ServerSentEventsTransport(httpClient, LoggerFactory);
+                var sseTransport = new ServerSentEventsTransport(httpClient, loggerFactory: LoggerFactory);
                 var exception = await Assert.ThrowsAsync<ArgumentException>(() =>
                     sseTransport.StartAsync(new Uri("http://fakeuri.org"), transferFormat));
 
diff --git a/src/SignalR/clients/csharp/Http.Connections.Client/src/HttpConnectionFactory.cs b/src/SignalR/clients/csharp/Http.Connections.Client/src/HttpConnectionFactory.cs
@@ -92,6 +92,8 @@ internal static HttpConnectionOptions ShallowCopyHttpConnectionOptions(HttpConne
                 AccessTokenProvider = options.AccessTokenProvider,
                 CloseTimeout = options.CloseTimeout,
                 DefaultTransferFormat = options.DefaultTransferFormat,
+                ApplicationMaxBufferSize = options.ApplicationMaxBufferSize,
+                TransportMaxBufferSize = options.TransportMaxBufferSize
             };
 
             if (!OperatingSystem.IsBrowser())
diff --git a/src/SignalR/clients/csharp/Http.Connections.Client/src/HttpConnectionOptions.cs b/src/SignalR/clients/csharp/Http.Connections.Client/src/HttpConnectionOptions.cs
@@ -3,6 +3,7 @@
 
 using System;
 using System.Collections.Generic;
+using System.IO.Pipelines;
 using System.Net;
 using System.Net.Http;
 using System.Net.WebSockets;
@@ -26,6 +27,15 @@ public class HttpConnectionOptions
         private IWebProxy? _proxy;
         private bool? _useDefaultCredentials;
         private Action<ClientWebSocketOptions>? _webSocketConfiguration;
+        private PipeOptions? _transportPipeOptions;
+        private PipeOptions? _appPipeOptions;
+        private long _transportMaxBufferSize;
+        private long _applicationMaxBufferSize;
+
+        // Selected because of the number of client connections is usually much lower than
+        // server connections and therefore willing to use more memory. We'll default
+        // to a maximum of 1MB buffer;
+        private const int DefaultBufferSize = 1 * 1024 * 1024;
 
         /// <summary>
         /// Initializes a new instance of the <see cref="HttpConnectionOptions"/> class.
@@ -43,6 +53,8 @@ public HttpConnectionOptions()
             _cookies = new CookieContainer();
 
             Transports = HttpTransports.All;
+            TransportMaxBufferSize = DefaultBufferSize;
+            ApplicationMaxBufferSize = DefaultBufferSize;
         }
 
         /// <summary>
@@ -66,6 +78,59 @@ public IDictionary<string, string> Headers
             set => _headers = value ?? throw new ArgumentNullException(nameof(value));
         }
 
+        /// <summary>
+        /// Gets or sets the maximum buffer size for data read by the application before backpressure is applied.
+        /// </summary>
+        /// <remarks>
+        /// The default value is 1MB.
+        /// </remarks>
+        /// <value>
+        /// The default value is 1MB.
+        /// </value>
+        public long TransportMaxBufferSize
+        {
+            get => _transportMaxBufferSize;
+            set
+            {
+                if (value < 0)
+                {
+                    throw new ArgumentOutOfRangeException(nameof(value));
+                }
+
+                _transportMaxBufferSize = value;
+            }
+        }
+
+        /// <summary>
+        /// Gets or sets the maximum buffer size for data written by the application before backpressure is applied.
+        /// </summary>
+        /// <remarks>
+        /// The default value is 1MB.
+        /// </remarks>
+        /// <value>
+        /// The default value is 1MB.
+        /// </value>
+        public long ApplicationMaxBufferSize
+        {
+            get => _applicationMaxBufferSize;
+            set
+            {
+                if (value < 0)
+                {
+                    throw new ArgumentOutOfRangeException(nameof(value));
+                }
+
+                _applicationMaxBufferSize = value;
+            }
+        }
+
+        // We initialize these lazily based on the state of the options specified here.
+        // Though these are mutable it's extremely rare that they would be mutated past the
+        // call to initialize the routerware.
+        internal PipeOptions TransportPipeOptions => _transportPipeOptions ??= new PipeOptions(pauseWriterThreshold: TransportMaxBufferSize, resumeWriterThreshold: TransportMaxBufferSize / 2, readerScheduler: PipeScheduler.ThreadPool, useSynchronizationContext: false);
+
+        internal PipeOptions AppPipeOptions => _appPipeOptions ??= new PipeOptions(pauseWriterThreshold: ApplicationMaxBufferSize, resumeWriterThreshold: ApplicationMaxBufferSize / 2, readerScheduler: PipeScheduler.ThreadPool, useSynchronizationContext: false);
+
         /// <summary>
         /// Gets or sets a collection of client certificates that will be sent with HTTP requests.
         /// </summary>
diff --git a/src/SignalR/clients/csharp/Http.Connections.Client/src/Internal/DefaultTransportFactory.cs b/src/SignalR/clients/csharp/Http.Connections.Client/src/Internal/DefaultTransportFactory.cs
@@ -49,13 +49,13 @@ public ITransport CreateTransport(HttpTransportType availableServerTransports)
             if ((availableServerTransports & HttpTransportType.ServerSentEvents & _requestedTransportType) == HttpTransportType.ServerSentEvents)
             {
                 // We don't need to give the transport the accessTokenProvider because the HttpClient has a message handler that does the work for us.
-                return new ServerSentEventsTransport(_httpClient!, _loggerFactory);
+                return new ServerSentEventsTransport(_httpClient!, _httpConnectionOptions, _loggerFactory);
             }
 
             if ((availableServerTransports & HttpTransportType.LongPolling & _requestedTransportType) == HttpTransportType.LongPolling)
             {
                 // We don't need to give the transport the accessTokenProvider because the HttpClient has a message handler that does the work for us.
-                return new LongPollingTransport(_httpClient!, _loggerFactory);
+                return new LongPollingTransport(_httpClient!, _httpConnectionOptions, _loggerFactory);
             }
 
             throw new InvalidOperationException("No requested transports available on the server.");
diff --git a/src/SignalR/clients/csharp/Http.Connections.Client/src/Internal/LongPollingTransport.cs b/src/SignalR/clients/csharp/Http.Connections.Client/src/Internal/LongPollingTransport.cs
@@ -18,6 +18,7 @@ internal partial class LongPollingTransport : ITransport
     {
         private readonly HttpClient _httpClient;
         private readonly ILogger _logger;
+        private readonly HttpConnectionOptions _httpConnectionOptions;
         private IDuplexPipe? _application;
         private IDuplexPipe? _transport;
         // Volatile so that the poll loop sees the updated value set from a different thread
@@ -31,14 +32,11 @@ internal partial class LongPollingTransport : ITransport
 
         public PipeWriter Output => _transport!.Output;
 
-        public LongPollingTransport(HttpClient httpClient)
-            : this(httpClient, null)
-        { }
-
-        public LongPollingTransport(HttpClient httpClient, ILoggerFactory? loggerFactory)
+        public LongPollingTransport(HttpClient httpClient, HttpConnectionOptions? httpConnectionOptions = null, ILoggerFactory? loggerFactory = null)
         {
             _httpClient = httpClient;
             _logger = (loggerFactory ?? NullLoggerFactory.Instance).CreateLogger<LongPollingTransport>();
+            _httpConnectionOptions = httpConnectionOptions ?? new();
         }
 
         public async Task StartAsync(Uri url, TransferFormat transferFormat, CancellationToken cancellationToken = default)
@@ -59,8 +57,7 @@ public async Task StartAsync(Uri url, TransferFormat transferFormat, Cancellatio
             }
 
             // Create the pipe pair (Application's writer is connected to Transport's reader, and vice versa)
-            var options = ClientPipeOptions.DefaultOptions;
-            var pair = DuplexPipe.CreateConnectionPair(options, options);
+            var pair = DuplexPipe.CreateConnectionPair(_httpConnectionOptions.TransportPipeOptions, _httpConnectionOptions.AppPipeOptions);
 
             _transport = pair.Transport;
             _application = pair.Application;
diff --git a/src/SignalR/clients/csharp/Http.Connections.Client/src/Internal/ServerSentEventsTransport.cs b/src/SignalR/clients/csharp/Http.Connections.Client/src/Internal/ServerSentEventsTransport.cs
@@ -18,7 +18,7 @@ internal partial class ServerSentEventsTransport : ITransport
     {
         private readonly HttpClient _httpClient;
         private readonly ILogger _logger;
-
+        private readonly HttpConnectionOptions _httpConnectionOptions;
         // Volatile so that the SSE loop sees the updated value set from a different thread
         private volatile Exception? _error;
         private readonly CancellationTokenSource _transportCts = new CancellationTokenSource();
@@ -33,11 +33,7 @@ internal partial class ServerSentEventsTransport : ITransport
 
         public PipeWriter Output => _transport!.Output;
 
-        public ServerSentEventsTransport(HttpClient httpClient)
-            : this(httpClient, null)
-        { }
-
-        public ServerSentEventsTransport(HttpClient httpClient, ILoggerFactory? loggerFactory)
+        public ServerSentEventsTransport(HttpClient httpClient, HttpConnectionOptions? httpConnectionOptions = null, ILoggerFactory? loggerFactory = null)
         {
             if (httpClient == null)
             {
@@ -46,6 +42,7 @@ public ServerSentEventsTransport(HttpClient httpClient, ILoggerFactory? loggerFa
 
             _httpClient = httpClient;
             _logger = (loggerFactory ?? NullLoggerFactory.Instance).CreateLogger<ServerSentEventsTransport>();
+            _httpConnectionOptions = httpConnectionOptions ?? new();
         }
 
         public async Task StartAsync(Uri url, TransferFormat transferFormat, CancellationToken cancellationToken = default)
@@ -77,8 +74,7 @@ public async Task StartAsync(Uri url, TransferFormat transferFormat, Cancellatio
             }
 
             // Create the pipe pair (Application's writer is connected to Transport's reader, and vice versa)
-            var options = ClientPipeOptions.DefaultOptions;
-            var pair = DuplexPipe.CreateConnectionPair(options, options);
+            var pair = DuplexPipe.CreateConnectionPair(_httpConnectionOptions.TransportPipeOptions, _httpConnectionOptions.AppPipeOptions);
 
             _transport = pair.Transport;
             _application = pair.Application;
diff --git a/src/SignalR/clients/csharp/Http.Connections.Client/src/Internal/WebSocketsTransport.cs b/src/SignalR/clients/csharp/Http.Connections.Client/src/Internal/WebSocketsTransport.cs
diff --git a/src/SignalR/clients/csharp/Http.Connections.Client/src/PublicAPI.Unshipped.txt b/src/SignalR/clients/csharp/Http.Connections.Client/src/PublicAPI.Unshipped.txt
diff --git a/src/SignalR/common/Http.Connections/src/HttpConnectionDispatcherOptions.cs b/src/SignalR/common/Http.Connections/src/HttpConnectionDispatcherOptions.cs
diff --git a/src/SignalR/common/Http.Connections/test/HttpConnectionManagerTests.cs b/src/SignalR/common/Http.Connections/test/HttpConnectionManagerTests.cs

Original file line number	Diff line number	Diff line change
`@@ -254,7 +254,7 @@ public async Task StopTransportWhenConnectionAlreadyStoppedOnServer()`
`254`	`254`	`{`
`255`	`255`	`using (var httpClient = new HttpClient(mockHttpHandler.Object))`
`256`	`256`	`{`
`257`		`- var longPollingTransport = new LongPollingTransport(httpClient, LoggerFactory);`
	`257`	`+ var longPollingTransport = new LongPollingTransport(httpClient, loggerFactory: LoggerFactory);`
`258`	`258`
`259`	`259`	`await longPollingTransport.StartAsync(TestUri, TransferFormat.Binary).DefaultTimeout();`
`260`	`260`
Original file line number	Diff line number	Diff line change
`@@ -49,13 +49,13 @@ public ITransport CreateTransport(HttpTransportType availableServerTransports)`
`49`	`49`	`if ((availableServerTransports & HttpTransportType.ServerSentEvents & _requestedTransportType) == HttpTransportType.ServerSentEvents)`
`50`	`50`	`{`
`51`	`51`	`// We don't need to give the transport the accessTokenProvider because the HttpClient has a message handler that does the work for us.`
`52`		`- return new ServerSentEventsTransport(_httpClient!, _loggerFactory);`
	`52`	`+ return new ServerSentEventsTransport(_httpClient!, _httpConnectionOptions, _loggerFactory);`
`53`	`53`	`}`
`54`	`54`
`55`	`55`	`if ((availableServerTransports & HttpTransportType.LongPolling & _requestedTransportType) == HttpTransportType.LongPolling)`
`56`	`56`	`{`
`57`	`57`	`// We don't need to give the transport the accessTokenProvider because the HttpClient has a message handler that does the work for us.`
`58`		`- return new LongPollingTransport(_httpClient!, _loggerFactory);`
	`58`	`+ return new LongPollingTransport(_httpClient!, _httpConnectionOptions, _loggerFactory);`
`59`	`59`	`}`
`60`	`60`
`61`	`61`	`throw new InvalidOperationException("No requested transports available on the server.");`