Added support for setting the read and write buffers on each transport

Author: davidfowl

src/Servers/Kestrel/Transport.Libuv/src/Internal/LibuvConnection.cs

@@ -39,7 +39,9 @@ public LibuvConnection(UvStreamHandle socket,
                                IPEndPoint remoteEndPoint,
                                IPEndPoint localEndPoint,
                                PipeOptions inputOptions = null,
-                               PipeOptions outputOptions = null)
+                               PipeOptions outputOptions = null,
+                               long? maxReadBufferSize = null,
+                               long? maxWriteBufferSize = null)
         {
             _socket = socket;
 
@@ -50,8 +52,11 @@ public LibuvConnection(UvStreamHandle socket,
             Log = log;
             Thread = thread;
 
-            inputOptions ??= new PipeOptions(MemoryPool, PipeScheduler.ThreadPool, Thread, useSynchronizationContext: false);
-            outputOptions ??= new PipeOptions(MemoryPool, Thread, PipeScheduler.ThreadPool, useSynchronizationContext: false);
+            maxReadBufferSize ??= 0;
+            maxWriteBufferSize ??= 0;
+
+            inputOptions ??= new PipeOptions(MemoryPool, PipeScheduler.ThreadPool, Thread, maxReadBufferSize.Value, maxReadBufferSize.Value / 2, useSynchronizationContext: false);
+            outputOptions ??= new PipeOptions(MemoryPool, Thread, PipeScheduler.ThreadPool, maxWriteBufferSize.Value, maxWriteBufferSize.Value / 2, useSynchronizationContext: false);
 
             var pair = DuplexPipe.CreateConnectionPair(inputOptions, outputOptions);
 

src/Servers/Kestrel/Transport.Libuv/src/Internal/ListenerContext.cs

@@ -87,7 +87,8 @@ protected internal void HandleConnection(UvStreamHandle socket)
                     }
                 }
 
-                var connection = new LibuvConnection(socket, TransportContext.Log, Thread, remoteEndPoint, localEndPoint, InputOptions, OutputOptions);
+                var options = TransportContext.Options;
+                var connection = new LibuvConnection(socket, TransportContext.Log, Thread, remoteEndPoint, localEndPoint, InputOptions, OutputOptions, options.MaxReadBufferSize, options.MaxWriteBufferSize);
                 connection.Start();
 
                 bool accepted = _acceptQueue.Writer.TryWrite(connection);

src/Servers/Kestrel/Transport.Libuv/src/LibuvTransportOptions.cs

@@ -3,6 +3,7 @@
 
 using System;
 using System.Buffers;
+using System.IO.Pipelines;
 using Microsoft.AspNetCore.Server.Kestrel.Transport.Abstractions.Internal;
 
 namespace Microsoft.AspNetCore.Server.Kestrel.Transport.Libuv
@@ -22,6 +23,10 @@ public class LibuvTransportOptions
 
         public bool NoDelay { get; set; }
 
+        public long? MaxReadBufferSize { get; set; } = PipeOptions.Default.PauseWriterThreshold;
+
+        public long? MaxWriteBufferSize { get; set; } = PipeOptions.Default.PauseWriterThreshold;
+
         internal Func<MemoryPool<byte>> MemoryPoolFactory { get; set; } = () => KestrelMemoryPool.Create();
 
         private static int ProcessorThreadCount

src/Servers/Kestrel/Transport.Sockets/src/Internal/SocketConnection.cs

@@ -34,7 +34,12 @@ internal sealed class SocketConnection : TransportConnection
         private volatile Exception _shutdownReason;
         private Task _task;
 
-        internal SocketConnection(Socket socket, MemoryPool<byte> memoryPool, PipeScheduler scheduler, ISocketsTrace trace)
+        internal SocketConnection(Socket socket,
+                                  MemoryPool<byte> memoryPool,
+                                  PipeScheduler scheduler,
+                                  ISocketsTrace trace,
+                                  long? maxReadBufferSize = null,
+                                  long? maxWriteBufferSize = null)
         {
             Debug.Assert(socket != null);
             Debug.Assert(memoryPool != null);
@@ -58,8 +63,11 @@ internal SocketConnection(Socket socket, MemoryPool<byte> memoryPool, PipeSchedu
             _receiver = new SocketReceiver(_socket, awaiterScheduler);
             _sender = new SocketSender(_socket, awaiterScheduler);
 
-            var inputOptions = new PipeOptions(MemoryPool, PipeScheduler.ThreadPool, awaiterScheduler, useSynchronizationContext: false);
-            var outputOptions = new PipeOptions(MemoryPool, awaiterScheduler, PipeScheduler.ThreadPool, useSynchronizationContext: false);
+            maxReadBufferSize ??= 0;
+            maxWriteBufferSize ??= 0;
+
+            var inputOptions = new PipeOptions(MemoryPool, PipeScheduler.ThreadPool, awaiterScheduler, maxReadBufferSize.Value, maxReadBufferSize.Value / 2, useSynchronizationContext: false);
+            var outputOptions = new PipeOptions(MemoryPool, awaiterScheduler, PipeScheduler.ThreadPool, maxWriteBufferSize.Value, maxWriteBufferSize.Value / 2, useSynchronizationContext: false);
 
             var pair = DuplexPipe.CreateConnectionPair(inputOptions, outputOptions);
 

src/Servers/Kestrel/Transport.Sockets/src/SocketConnectionListener.cs

@@ -100,7 +100,7 @@ public async ValueTask<ConnectionContext> AcceptAsync(CancellationToken cancella
                 var acceptSocket = await _listenSocket.AcceptAsync();
                 acceptSocket.NoDelay = _options.NoDelay;
 
-                var connection = new SocketConnection(acceptSocket, _memoryPool, _schedulers[_schedulerIndex], _trace);
+                var connection = new SocketConnection(acceptSocket, _memoryPool, _schedulers[_schedulerIndex], _trace, _options.MaxReadBufferSize, _options.MaxWriteBufferSize);
 
                 connection.Start();
 

src/Servers/Kestrel/Transport.Sockets/src/SocketTransportOptions.cs

@@ -3,6 +3,7 @@
 
 using System;
 using System.Buffers;
+using System.IO.Pipelines;
 using Microsoft.AspNetCore.Server.Kestrel.Transport.Abstractions.Internal;
 
 namespace Microsoft.AspNetCore.Server.Kestrel.Transport.Sockets
@@ -19,6 +20,10 @@ public class SocketTransportOptions
 
         public bool NoDelay { get; set; }
 
+        public long? MaxReadBufferSize { get; set; } = PipeOptions.Default.PauseWriterThreshold;
+
+        public long? MaxWriteBufferSize { get; set; } = PipeOptions.Default.PauseWriterThreshold;
+
         internal Func<MemoryPool<byte>> MemoryPoolFactory { get; set; } = () => KestrelMemoryPool.Create();
     }
 }

src/Servers/Kestrel/shared/test/TransportTestHelpers/TestServer.cs

@@ -61,7 +61,7 @@ public TestServer(RequestDelegate app, TestServiceContext context, Action<Kestre
             _app = app;
             Context = context;
 
-            _host = TransportSelector.GetWebHostBuilder(context.MemoryPoolFactory)
+            _host = TransportSelector.GetWebHostBuilder(context.MemoryPoolFactory, context.ServerOptions.Limits.MaxRequestBufferSize)
                 .UseKestrel(options =>
                 {
                     configureKestrel(options);

src/Servers/Kestrel/test/FunctionalTests/MaxRequestBufferSizeTests.cs

@@ -289,7 +289,7 @@ private async Task<IWebHost> StartWebHost(long? maxRequestBufferSize,
             TaskCompletionSource<object> clientFinishedSendingRequestBody,
             Func<MemoryPool<byte>> memoryPoolFactory = null)
         {
-            var host = TransportSelector.GetWebHostBuilder(memoryPoolFactory)
+            var host = TransportSelector.GetWebHostBuilder(memoryPoolFactory, maxRequestBufferSize)
                 .ConfigureServices(AddTestLogging)
                 .UseKestrel(options =>
                 {

src/Servers/Kestrel/test/Libuv.FunctionalTests/TransportSelector.cs

@@ -1,4 +1,4 @@
-// Copyright (c) .NET Foundation. All rights reserved.
+// 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;
@@ -9,9 +9,14 @@ namespace Microsoft.AspNetCore.Server.Kestrel.FunctionalTests
 {
     public static class TransportSelector
     {
-        public static IWebHostBuilder GetWebHostBuilder(Func<MemoryPool<byte>> memoryPoolFactory = null)
+        public static IWebHostBuilder GetWebHostBuilder(Func<MemoryPool<byte>> memoryPoolFactory = null,
+                                                        long? maxReadBufferSize = null)
         {
-            return new WebHostBuilder().UseLibuv(options => { options.MemoryPoolFactory = memoryPoolFactory ?? options.MemoryPoolFactory; });
+            return new WebHostBuilder().UseLibuv(options =>
+            {
+                options.MemoryPoolFactory = memoryPoolFactory ?? options.MemoryPoolFactory;
+                options.MaxReadBufferSize = maxReadBufferSize;
+            });
         }
     }
 }

src/Servers/Kestrel/test/Sockets.FunctionalTests/TransportSelector.cs

@@ -1,4 +1,4 @@
-// Copyright (c) .NET Foundation. All rights reserved.
+// 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;
@@ -9,9 +9,14 @@ namespace Microsoft.AspNetCore.Server.Kestrel.FunctionalTests
 {
     public static class TransportSelector
     {
-        public static IWebHostBuilder GetWebHostBuilder(Func<MemoryPool<byte>> memoryPoolFactory = null)
+        public static IWebHostBuilder GetWebHostBuilder(Func<MemoryPool<byte>> memoryPoolFactory = null,
+                                                        long? maxReadBufferSize = null)
         {
-            return new WebHostBuilder().UseSockets(options => { options.MemoryPoolFactory = memoryPoolFactory ?? options.MemoryPoolFactory; });
+            return new WebHostBuilder().UseSockets(options =>
+            {
+                options.MemoryPoolFactory = memoryPoolFactory ?? options.MemoryPoolFactory;
+                options.MaxReadBufferSize = maxReadBufferSize;
+            });
         }
     }
 }