Limit size of memory buffer when reading request (#304)

- Added property `KestrelServerOptions.MaxRequestBufferSize`
 - Default is 1,048,576 bytes (1MB)
 - If value is null, the size of the request buffer is unlimited.
- Fixed bug in `IConnectionControl.Resume()` where `_socket.ReadStart()` can throw if the socket is already disconnected.
- Made `UvStreamHandle.ReadStop()` idempotent, to match `uv_read_stop()`.

src/Microsoft.AspNetCore.Server.Kestrel/Filter/Internal/FilteredStreamAdapter.cs

@@ -24,9 +24,10 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Filter.Internal
             Stream filteredStream,
             MemoryPool memory,
             IKestrelTrace logger,
-            IThreadPool threadPool)
+            IThreadPool threadPool,
+            IBufferSizeControl bufferSizeControl)
         {
-            SocketInput = new SocketInput(memory, threadPool);
+            SocketInput = new SocketInput(memory, threadPool, bufferSizeControl);
             SocketOutput = new StreamSocketOutput(connectionId, filteredStream, memory, logger);
 
             _connectionId = connectionId;

src/Microsoft.AspNetCore.Server.Kestrel/Internal/Http/BufferSizeControl.cs

@@ -0,0 +1,83 @@
+using System.Diagnostics;
+
+namespace Microsoft.AspNetCore.Server.Kestrel.Internal.Http
+{
+    public class BufferSizeControl : IBufferSizeControl
+    {
+        private readonly long _maxSize;
+        private readonly IConnectionControl _connectionControl;
+        private readonly KestrelThread _connectionThread;
+
+        private readonly object _lock = new object();
+
+        private long _size;
+        private bool _connectionPaused;
+
+        public BufferSizeControl(long maxSize, IConnectionControl connectionControl, KestrelThread connectionThread)
+        {
+            _maxSize = maxSize;
+            _connectionControl = connectionControl;
+            _connectionThread = connectionThread;
+        }
+
+        private long Size
+        {
+            get
+            {
+                return _size;
+            }
+            set
+            {
+                // Caller should ensure that bytes are never consumed before the producer has called Add()
+                Debug.Assert(value >= 0);
+                _size = value;
+            }
+        }
+
+        public void Add(int count)
+        {
+            Debug.Assert(count >= 0);
+
+            if (count == 0)
+            {
+                // No-op and avoid taking lock to reduce contention
+                return;
+            }
+
+            lock (_lock)
+            {
+                Size += count;
+                if (!_connectionPaused && Size >= _maxSize)
+                {
+                    _connectionPaused = true;
+                    _connectionThread.Post(
+                        (connectionControl) => ((IConnectionControl)connectionControl).Pause(),
+                        _connectionControl);
+                }
+            }
+        }
+
+        public void Subtract(int count)
+        {
+            Debug.Assert(count >= 0);
+
+            if (count == 0)
+            {
+                // No-op and avoid taking lock to reduce contention
+                return;
+            }
+
+            lock (_lock)
+            {
+                Size -= count;
+                if (_connectionPaused && Size < _maxSize)
+                {
+                    _connectionPaused = false;
+                    _connectionThread.Post(
+                        (connectionControl) => ((IConnectionControl)connectionControl).Resume(),
+                        _connectionControl);
+                }
+            }
+        }
+    }
+}

src/Microsoft.AspNetCore.Server.Kestrel/Internal/Http/Connection.cs

@@ -2,6 +2,7 @@
 // Licensed under the Apache License, Version 2.0. See License.txt in the project root for license information.
 
 using System;
+using System.Diagnostics;
 using System.Threading;
 using System.Threading.Tasks;
 using Microsoft.AspNetCore.Server.Kestrel.Filter;
@@ -41,6 +42,8 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Internal.Http
         private ConnectionState _connectionState;
         private TaskCompletionSource<object> _socketClosedTcs;
 
+        private BufferSizeControl _bufferSizeControl;
+
         public Connection(ListenerContext context, UvStreamHandle socket) : base(context)
         {
             _socket = socket;
@@ -49,7 +52,12 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Internal.Http
 
             ConnectionId = GenerateConnectionId(Interlocked.Increment(ref _lastConnectionId));
 
-            _rawSocketInput = new SocketInput(Memory, ThreadPool);
+            if (ServerOptions.MaxRequestBufferSize.HasValue)
+            {
+                _bufferSizeControl = new BufferSizeControl(ServerOptions.MaxRequestBufferSize.Value, this, Thread);
+            }
+
+            _rawSocketInput = new SocketInput(Memory, ThreadPool, _bufferSizeControl);
             _rawSocketOutput = new SocketOutput(Thread, _socket, Memory, this, ConnectionId, Log, ThreadPool, WriteReqPool);
         }
 
@@ -217,7 +225,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Internal.Http
 
                     if (_filterContext.Connection != _libuvStream)
                     {
-                        _filteredStreamAdapter = new FilteredStreamAdapter(ConnectionId, _filterContext.Connection, Memory, Log, ThreadPool);
+                        _filteredStreamAdapter = new FilteredStreamAdapter(ConnectionId, _filterContext.Connection, Memory, Log, ThreadPool, _bufferSizeControl);
 
                         SocketInput = _filteredStreamAdapter.SocketInput;
                         SocketOutput = _filteredStreamAdapter.SocketOutput;
@@ -316,7 +324,17 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Internal.Http
         void IConnectionControl.Resume()
         {
             Log.ConnectionResume(ConnectionId);
-            _socket.ReadStart(_allocCallback, _readCallback, this);
+            try
+            {
+                _socket.ReadStart(_allocCallback, _readCallback, this);
+            }
+            catch (UvException)
+            {
+                // ReadStart() can throw a UvException in some cases (e.g. socket is no longer connected).
+                // This should be treated the same as OnRead() seeing a "normalDone" condition.
+                Log.ConnectionReadFin(ConnectionId);
+                _rawSocketInput.IncomingComplete(0, null);
+            }
         }
 
         void IConnectionControl.End(ProduceEndType endType)

src/Microsoft.AspNetCore.Server.Kestrel/Internal/Http/IBufferSizeControl.cs

@@ -0,0 +1,13 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Threading.Tasks;
+
+namespace Microsoft.AspNetCore.Server.Kestrel.Internal.Http
+{
+    public interface IBufferSizeControl
+    {
+        void Add(int count);
+        void Subtract(int count);
+    }
+}

src/Microsoft.AspNetCore.Server.Kestrel/Internal/Http/SocketInput.cs

@@ -18,6 +18,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Internal.Http
 
         private readonly MemoryPool _memory;
         private readonly IThreadPool _threadPool;
+        private readonly IBufferSizeControl _bufferSizeControl;
         private readonly ManualResetEventSlim _manualResetEvent = new ManualResetEventSlim(false, 0);
 
         private Action _awaitableState;
@@ -32,10 +33,11 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Internal.Http
         private bool _consuming;
         private bool _disposed;
 
-        public SocketInput(MemoryPool memory, IThreadPool threadPool)
+        public SocketInput(MemoryPool memory, IThreadPool threadPool, IBufferSizeControl bufferSizeControl = null)
         {
             _memory = memory;
             _threadPool = threadPool;
+            _bufferSizeControl = bufferSizeControl;
             _awaitableState = _awaitableIsNotCompleted;
         }
 
@@ -63,6 +65,9 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Internal.Http
         {
             lock (_sync)
             {
+                // Must call Add() before bytes are available to consumer, to ensure that Length is >= 0
+                _bufferSizeControl?.Add(count);
+
                 if (count > 0)
                 {
                     if (_tail == null)
@@ -93,6 +98,9 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Internal.Http
         {
             lock (_sync)
             {
+                // Must call Add() before bytes are available to consumer, to ensure that Length is >= 0
+                _bufferSizeControl?.Add(count);
+
                 if (_pinned != null)
                 {
                     _pinned.End += count;
@@ -189,10 +197,21 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Internal.Http
                 {
                     if (!consumed.IsDefault)
                     {
+                        // Compute lengthConsumed before modifying _head or consumed
+                        var lengthConsumed = 0;
+                        if (_bufferSizeControl != null)
+                        {
+                            lengthConsumed = new MemoryPoolIterator(_head).GetLength(consumed);
+                        }
+
                         returnStart = _head;
                         returnEnd = consumed.Block;
                         _head = consumed.Block;
                         _head.Start = consumed.Index;
+
+                        // Must call Subtract() after _head has been advanced, to avoid producer starting too early and growing
+                        // buffer beyond max length.
+                        _bufferSizeControl?.Subtract(lengthConsumed);
                     }
 
                     if (!examined.IsDefault &&

src/Microsoft.AspNetCore.Server.Kestrel/Internal/Networking/UvStreamHandle.cs

@@ -105,16 +105,16 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Internal.Networking
             }
         }
 
+        // UvStreamHandle.ReadStop() should be idempotent to match uv_read_stop()
         public void ReadStop()
         {
-            if (!_readVitality.IsAllocated)
+            if (_readVitality.IsAllocated)
             {
-                throw new InvalidOperationException("TODO: ReadStart must be called before ReadStop may be called");
+                _readVitality.Free();
             }
             _allocCallback = null;
             _readCallback = null;
             _readState = null;
-            _readVitality.Free();
             _uv.read_stop(this);
         }
 

src/Microsoft.AspNetCore.Server.Kestrel/KestrelServerOptions.cs

@@ -8,17 +8,41 @@ namespace Microsoft.AspNetCore.Server.Kestrel
 {
     public class KestrelServerOptions
     {
-        public IServiceProvider ApplicationServices { get; set; }
+        // Matches the default client_max_body_size in nginx.  Also large enough that most requests
-
+        // should be under the limit.
-        public IConnectionFilter ConnectionFilter { get; set; }
+        private long? _maxRequestBufferSize = 1024 * 1024;
-
-        public bool NoDelay { get; set; } = true;
 
         /// <summary>
         /// Gets or sets whether the <c>Server</c> header should be included in each response.
         /// </summary>
         public bool AddServerHeader { get; set; } = true;
 
+        public IServiceProvider ApplicationServices { get; set; }
+
+        public IConnectionFilter ConnectionFilter { get; set; }
+
+        /// <summary>
+        /// Maximum size of the request buffer.  Default is 1,048,576 bytes (1 MB).
+        /// If value is null, the size of the request buffer is unlimited.
+        /// </summary>
+        public long? MaxRequestBufferSize
+        {
+            get
+            {
+                return _maxRequestBufferSize;
+            }
+            set
+            {
+                if (value.HasValue && value.Value <= 0)
+                {
+                    throw new ArgumentOutOfRangeException("value", "Value must be null or a positive integer.");
+                }
+                _maxRequestBufferSize = value;
+            }
+        }
+
+        public bool NoDelay { get; set; } = true;
+
         /// <summary>
         /// The amount of time after the server begins shutting down before connections will be forcefully closed.
         /// By default, Kestrel will wait 5 seconds for any ongoing requests to complete before terminating

test/Microsoft.AspNetCore.Server.Kestrel.FunctionalTests/IWebHostPortExtensions.cs

@@ -10,16 +10,35 @@ namespace Microsoft.AspNetCore.Hosting
 {
     public static class IWebHostPortExtensions
     {
+        public static string GetHost(this IWebHost host)
+        {
+            return host.GetUris().First().Host;
+        }
+
         public static int GetPort(this IWebHost host)
         {
             return host.GetPorts().First();
         }
 
+        public static int GetPort(this IWebHost host, string scheme)
+        {
+            return host.GetUris()
+                .Where(u => u.Scheme.Equals(scheme, StringComparison.OrdinalIgnoreCase))
+                .Select(u => u.Port)
+                .First();
+        }
+
         public static IEnumerable<int> GetPorts(this IWebHost host)
+        {
+            return host.GetUris()
+                .Select(u => u.Port);
+        }
+
+        public static IEnumerable<Uri> GetUris(this IWebHost host)
         {
             return host.ServerFeatures.Get<IServerAddressesFeature>().Addresses
                 .Select(a => a.Replace("://+", "://localhost"))
-                .Select(a => (new Uri(a)).Port);
+                .Select(a => new Uri(a));
         }
     }
 }

test/Microsoft.AspNetCore.Server.Kestrel.FunctionalTests/MaxRequestBufferSizeTests.cs

@@ -0,0 +1,263 @@
+using System;
+using System.Collections.Generic;
+using System.IO;
+using System.Linq;
+using System.Net;
+using System.Net.Security;
+using System.Net.Sockets;
+using System.Security.Authentication;
+using System.Text;
+using System.Threading;
+using System.Threading.Tasks;
+using Microsoft.AspNetCore.Builder;
+using Microsoft.AspNetCore.Hosting;
+using Microsoft.AspNetCore.Http;
+using Xunit;
+
+namespace Microsoft.AspNetCore.Server.Kestrel.FunctionalTests
+{
+    public class MaxRequestBufferSizeTests
+    {
+        private const int _dataLength = 20 * 1024 * 1024;
+
+        public static IEnumerable<object[]> LargeUploadData
+        {
+            get
+            {
+                var maxRequestBufferSizeValues = new Tuple<long?, bool>[] {
+                    // Smallest allowed buffer.  Server should call pause/resume between each read.
+                    Tuple.Create((long?)1, true),
+
+                    // Small buffer, but large enough to hold all request headers.
+                    Tuple.Create((long?)16 * 1024, true),
+
+                    // Default buffer.
+                    Tuple.Create((long?)1024 * 1024, true),
+
+                    // Larger than default, but still significantly lower than data, so client should be paused.
+                    // On Windows, the client is usually paused around (MaxRequestBufferSize + 700,000).
+                    // On Linux, the client is usually paused around (MaxRequestBufferSize + 10,000,000).
+                    Tuple.Create((long?)5 * 1024 * 1024, true),
+
+                    // Even though maxRequestBufferSize < _dataLength, client should not be paused since the
+                    // OS-level buffers in client and/or server will handle the overflow.
+                    Tuple.Create((long?)_dataLength - 1, false),
+
+                    // Buffer is exactly the same size as data.  Exposed race condition where
+                    // IConnectionControl.Resume() was called after socket was disconnected.
+                    Tuple.Create((long?)_dataLength, false),
+
+                    // Largest possible buffer, should never trigger backpressure.
+                    Tuple.Create((long?)long.MaxValue, false),
+
+                    // Disables all code related to computing and limiting the size of the input buffer.
+                    Tuple.Create((long?)null, false)
+                };
+                var sendContentLengthHeaderValues = new[] { true, false };
+                var sslValues = new[] { true, false };
+
+                return from maxRequestBufferSize in maxRequestBufferSizeValues
+                       from sendContentLengthHeader in sendContentLengthHeaderValues
+                       from ssl in sslValues
+                       select new object[] {
+                           maxRequestBufferSize.Item1,
+                           sendContentLengthHeader,
+                           ssl,
+                           maxRequestBufferSize.Item2
+                       };
+            }
+        }
+
+        [Theory]
+        [MemberData("LargeUploadData")]
+        public async Task LargeUpload(long? maxRequestBufferSize, bool sendContentLengthHeader, bool ssl, bool expectPause)
+        {
+            // Parameters
+            var data = new byte[_dataLength];
+            var bytesWrittenTimeout = TimeSpan.FromMilliseconds(100);
+            var bytesWrittenPollingInterval = TimeSpan.FromMilliseconds(bytesWrittenTimeout.TotalMilliseconds / 10);
+            var maxSendSize = 4096;
+
+            // Initialize data with random bytes
+            (new Random()).NextBytes(data);
+
+            var startReadingRequestBody = new ManualResetEvent(false);
+            var clientFinishedSendingRequestBody = new ManualResetEvent(false);
+            var lastBytesWritten = DateTime.MaxValue;
+
+            using (var host = StartWebHost(maxRequestBufferSize, data, startReadingRequestBody, clientFinishedSendingRequestBody))
+            {
+                var port = host.GetPort(ssl ? "https" : "http");
+                using (var socket = CreateSocket(port))
+                using (var stream = await CreateStreamAsync(socket, ssl, host.GetHost()))
+                {
+                    await WritePostRequestHeaders(stream, sendContentLengthHeader ? (int?)data.Length : null);
+
+                    var bytesWritten = 0;
+
+                    Func<Task> sendFunc = async () =>
+                    {
+                        while (bytesWritten < data.Length)
+                        {
+                            var size = Math.Min(data.Length - bytesWritten, maxSendSize);
+                            await stream.WriteAsync(data, bytesWritten, size);
+                            bytesWritten += size;
+                            lastBytesWritten = DateTime.Now;
+                        }
+
+                        Assert.Equal(data.Length, bytesWritten);
+                        socket.Shutdown(SocketShutdown.Send);
+                        clientFinishedSendingRequestBody.Set();
+                    };
+
+                    var sendTask = sendFunc();
+
+                    if (expectPause)
+                    {
+                        // The minimum is (maxRequestBufferSize - maxSendSize + 1), since if bytesWritten is
+                        // (maxRequestBufferSize - maxSendSize) or smaller, the client should be able to
+                        // complete another send.
+                        var minimumExpectedBytesWritten = maxRequestBufferSize.Value - maxSendSize + 1;
+
+                        // The maximum is harder to determine, since there can be OS-level buffers in both the client
+                        // and server, which allow the client to send more than maxRequestBufferSize before getting
+                        // paused.  We assume the combined buffers are smaller than the difference between
+                        // data.Length and maxRequestBufferSize.                          
+                        var maximumExpectedBytesWritten = data.Length - 1;
+
+                        // Block until the send task has gone a while without writing bytes AND
+                        // the bytes written exceeds the minimum expected.  This indicates the server buffer
+                        // is full.
+                        // 
+                        // If the send task is paused before the expected number of bytes have been
+                        // written, keep waiting since the pause may have been caused by something else
+                        // like a slow machine.
+                        while ((DateTime.Now - lastBytesWritten) < bytesWrittenTimeout ||
+                               bytesWritten < minimumExpectedBytesWritten)
+                        {
+                            await Task.Delay(bytesWrittenPollingInterval);
+                        }
+
+                        // Verify the number of bytes written before the client was paused.
+                        Assert.InRange(bytesWritten, minimumExpectedBytesWritten, maximumExpectedBytesWritten);
+
+                        // Tell server to start reading request body
+                        startReadingRequestBody.Set();
+
+                        // Wait for sendTask to finish sending the remaining bytes
+                        await sendTask;
+                    }
+                    else
+                    {
+                        // Ensure all bytes can be sent before the server starts reading
+                        await sendTask;
+
+                        // Tell server to start reading request body
+                        startReadingRequestBody.Set();
+                    }
+
+                    using (var reader = new StreamReader(stream, Encoding.ASCII))
+                    {
+                        var response = reader.ReadToEnd();
+                        Assert.Contains($"bytesRead: {data.Length}", response);
+                    }
+                }
+            }
+        }
+
+        private static IWebHost StartWebHost(long? maxRequestBufferSize, byte[] expectedBody, ManualResetEvent startReadingRequestBody,
+            ManualResetEvent clientFinishedSendingRequestBody)
+        {
+            var host = new WebHostBuilder()
+                .UseKestrel(options =>
+                {
+                    options.MaxRequestBufferSize = maxRequestBufferSize;
+                    options.UseHttps(@"TestResources/testCert.pfx", "testPassword");
+                })
+                .UseUrls("http://127.0.0.1:0/", "https://127.0.0.1:0/")
+                .UseContentRoot(Directory.GetCurrentDirectory())
+                .Configure(app => app.Run(async context =>
+                {
+                    startReadingRequestBody.WaitOne();
+
+                    var buffer = new byte[expectedBody.Length];
+                    var bytesRead = 0;
+                    while (bytesRead < buffer.Length)
+                    {
+                        bytesRead += await context.Request.Body.ReadAsync(buffer, bytesRead, buffer.Length - bytesRead);
+                    }
+
+                    clientFinishedSendingRequestBody.WaitOne();
+
+                    // Verify client didn't send extra bytes
+                    if (context.Request.Body.ReadByte() != -1)
+                    {
+                        context.Response.StatusCode = 500;
+                        await context.Response.WriteAsync("Client sent more bytes than expectedBody.Length");
+                        return;
+                    }
+
+                    // Verify bytes received match expectedBody
+                    for (int i = 0; i < expectedBody.Length; i++)
+                    {
+                        if (buffer[i] != expectedBody[i])
+                        {
+                            context.Response.StatusCode = 500;
+                            await context.Response.WriteAsync($"Bytes received do not match expectedBody at position {i}");
+                            return;
+                        }
+                    }
+
+                    await context.Response.WriteAsync($"bytesRead: {bytesRead.ToString()}");
+                }))
+                .Build();
+
+            host.Start();
+
+            return host;
+        }
+
+        private static Socket CreateSocket(int port)
+        {
+            var socket = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
+
+            // Timeouts large enough to prevent false positives, but small enough to fail quickly.
+            socket.SendTimeout = 10 * 1000;
+            socket.ReceiveTimeout = 10 * 1000;
+
+            socket.Connect(IPAddress.Loopback, port);
+
+            return socket;
+        }
+
+        private static async Task WritePostRequestHeaders(Stream stream, int? contentLength)
+        {
+            using (var writer = new StreamWriter(stream, Encoding.ASCII, bufferSize: 1024, leaveOpen: true))
+            {
+                await writer.WriteAsync("POST / HTTP/1.0\r\n");
+                if (contentLength.HasValue)
+                {
+                    await writer.WriteAsync($"Content-Length: {contentLength.Value}\r\n");
+                }
+                await writer.WriteAsync("\r\n");
+            }
+        }
+
+        private static async Task<Stream> CreateStreamAsync(Socket socket, bool ssl, string targetHost)
+        {
+            var networkStream = new NetworkStream(socket);
+            if (ssl)
+            {
+                var sslStream = new SslStream(networkStream, leaveInnerStreamOpen: false,
+                    userCertificateValidationCallback: (a, b, c, d) => true);
+                await sslStream.AuthenticateAsClientAsync(targetHost, clientCertificates: null,
+                    enabledSslProtocols: SslProtocols.Tls11 | SslProtocols.Tls12, checkCertificateRevocation: false);
+                return sslStream;
+            }
+            else
+            {
+                return networkStream;
+            }
+        }
+    }
+}

test/Microsoft.AspNetCore.Server.Kestrel.FunctionalTests/project.json

@@ -6,6 +6,7 @@
     "Microsoft.AspNetCore.Server.Kestrel": "1.0.0-*",
     "Microsoft.AspNetCore.Server.Kestrel.Https": "1.0.0-*",
     "Microsoft.AspNetCore.Testing": "1.0.0-*",
+    "Microsoft.Extensions.Logging.Console": "1.0.0-*",
     "Newtonsoft.Json": "9.0.1-beta1",
     "xunit": "2.1.0"
   },
@@ -36,7 +37,15 @@
     }
   },
   "buildOptions": {
-    "allowUnsafe": true
+    "allowUnsafe": true,
+    "copyToOutput": {
+      "include": "TestResources/testCert.pfx"
+    }
   },
-  "testRunner": "xunit"
+  "testRunner": "xunit",
+  "publishOptions": {
+    "include": [
+      "TestResources/testCert.pfx"
+    ]
+  }
 }

test/Microsoft.AspNetCore.Server.KestrelTests/KestrelServerOptionsTests.cs

@@ -11,6 +11,33 @@ namespace Microsoft.AspNetCore.Server.KestrelTests
 {
     public class KestrelServerInformationTests
     {
+        [Fact]
+        public void MaxRequestBufferSizeDefault()
+        {
+            Assert.Equal(1024 * 1024, (new KestrelServerOptions()).MaxRequestBufferSize);
+        }
+
+        [Theory]
+        [InlineData(-1)]
+        [InlineData(0)]
+        public void MaxRequestBufferSizeInvalid(int value)
+        {
+            Assert.Throws<ArgumentOutOfRangeException>(() =>
+            {
+                (new KestrelServerOptions()).MaxRequestBufferSize = value;
+            });
+        }
+
+        [Theory]
+        [InlineData(null)]
+        [InlineData(1)]
+        public void MaxRequestBufferSizeValid(int? value)
+        {
+            var o = new KestrelServerOptions();
+            o.MaxRequestBufferSize = value;
+            Assert.Equal(value, o.MaxRequestBufferSize);
+        }
+
         [Fact]
         public void SetThreadCountUsingProcessorCount()
         {

test/Microsoft.AspNetCore.Server.KestrelTests/SocketInputTests.cs

@@ -3,16 +3,65 @@
 
 using System;
 using System.Threading.Tasks;
-using Microsoft.AspNetCore.Server.Kestrel;
 using Microsoft.AspNetCore.Server.Kestrel.Internal;
 using Microsoft.AspNetCore.Server.Kestrel.Internal.Http;
 using Microsoft.AspNetCore.Server.Kestrel.Internal.Infrastructure;
+using Microsoft.AspNetCore.Server.KestrelTests.TestHelpers;
+using Moq;
 using Xunit;
 
 namespace Microsoft.AspNetCore.Server.KestrelTests
 {
     public class SocketInputTests
     {
+        public static readonly TheoryData<Mock<IBufferSizeControl>> MockBufferSizeControlData =
+            new TheoryData<Mock<IBufferSizeControl>>() { new Mock<IBufferSizeControl>(), null };
+
+        [Theory]
+        [MemberData("MockBufferSizeControlData")]
+        public void IncomingDataCallsBufferSizeControlAdd(Mock<IBufferSizeControl> mockBufferSizeControl)
+        {
+            using (var memory = new MemoryPool())
+            using (var socketInput = new SocketInput(memory, null, mockBufferSizeControl?.Object))
+            {
+                socketInput.IncomingData(new byte[5], 0, 5);
+                mockBufferSizeControl?.Verify(b => b.Add(5));
+            }
+        }
+
+        [Theory]
+        [MemberData("MockBufferSizeControlData")]
+        public void IncomingCompleteCallsBufferSizeControlAdd(Mock<IBufferSizeControl> mockBufferSizeControl)
+        {
+            using (var memory = new MemoryPool())
+            using (var socketInput = new SocketInput(memory, null, mockBufferSizeControl?.Object))
+            {
+                socketInput.IncomingComplete(5, null);
+                mockBufferSizeControl?.Verify(b => b.Add(5));
+            }
+        }
+
+        [Theory]
+        [MemberData("MockBufferSizeControlData")]
+        public void ConsumingCompleteCallsBufferSizeControlSubtract(Mock<IBufferSizeControl> mockBufferSizeControl)
+        {
+            using (var kestrelEngine = new KestrelEngine(new MockLibuv(), new TestServiceContext()))
+            {
+                kestrelEngine.Start(1);
+
+                using (var memory = new MemoryPool())
+                using (var socketInput = new SocketInput(memory, null, mockBufferSizeControl?.Object))
+                {
+                    socketInput.IncomingData(new byte[20], 0, 20);
+
+                    var iterator = socketInput.ConsumingStart();
+                    iterator.Skip(5);
+                    socketInput.ConsumingComplete(iterator, iterator);
+                    mockBufferSizeControl?.Verify(b => b.Subtract(5));
+                }
+            }
+        }
+
         [Fact]
         public async Task ConcurrentReadsFailGracefully()
         {

test/Microsoft.AspNetCore.Server.KestrelTests/UvStreamHandleTests.cs

@@ -0,0 +1,26 @@
+using Microsoft.AspNetCore.Server.Kestrel.Internal.Infrastructure;
+using Microsoft.AspNetCore.Server.Kestrel.Internal.Networking;
+using Microsoft.AspNetCore.Server.KestrelTests.TestHelpers;
+using Moq;
+using Xunit;
+
+namespace Microsoft.AspNetCore.Server.KestrelTests
+{
+    public class UvStreamHandleTests
+    {
+        [Fact]
+        public void ReadStopIsIdempotent()
+        {
+            var mockKestrelTrace = Mock.Of<IKestrelTrace>();
+            var mockUvLoopHandle = new Mock<UvLoopHandle>(mockKestrelTrace).Object;
+            mockUvLoopHandle.Init(new MockLibuv());
+
+            // Need to mock UvTcpHandle instead of UvStreamHandle, since the latter lacks an Init() method
+            var mockUvStreamHandle = new Mock<UvTcpHandle>(mockKestrelTrace).Object;
+            mockUvStreamHandle.Init(mockUvLoopHandle, null);
+
+            mockUvStreamHandle.ReadStop();
+            mockUvStreamHandle.ReadStop();
+        }
+    }
+}

test/Microsoft.AspNetCore.Server.KestrelTests/project.json

@@ -20,7 +20,8 @@
         "System.Net.Http": "4.1.0-*",
         "System.Net.Http.WinHttpHandler": "4.0.0-*",
         "System.Net.Sockets": "4.1.0-*",
-        "System.Runtime.Handles": "4.0.1-*"
+        "System.Runtime.Handles": "4.0.1-*",
+        "moq.netcore": "4.4.0-beta8"
       },
       "imports": [
         "dnxcore50",
@@ -29,7 +30,8 @@
     },
     "net451": {
       "dependencies": {
-        "xunit.runner.console": "2.1.0"
+        "xunit.runner.console": "2.1.0",
+        "Moq": "4.2.1312.1622"
       },
       "frameworkAssemblies": {
         "System.Net.Http": "4.0.0.0"