Improve keep-alive timeout.

- Track time more accurately - Control timeout in Connection instead of Frame
aspnet · 1a273f5a34cae8ea39d99bd5e7ff64eb6057d142 · cesarbs committed Sep 12, 2016 · Sep 12, 2016 · 1a273f5
1 parent 84efe62
commit 1a273f5a34cae8ea39d99bd5e7ff64eb6057d142
@@ -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.IO;
 using System.Threading;
 using System.Threading.Tasks;
@@ -38,6 +39,9 @@ public class Connection : ConnectionContext, IConnectionControl
         private TaskCompletionSource<object> _socketClosedTcs = new TaskCompletionSource<object>();
         private BufferSizeControl _bufferSizeControl;
 
+        private long _lastTimestamp;
+        private long _timeoutTimestamp = long.MaxValue;
+
         public Connection(ListenerContext context, UvStreamHandle socket) : base(context)
         {
             _socket = socket;
@@ -62,6 +66,8 @@ public Connection(ListenerContext context, UvStreamHandle socket) : base(context
             }
 
             _frame = FrameFactory(this);
+
+            _lastTimestamp = Thread.Loop.Now();
         }
 
         // Internal for testing
@@ -156,9 +162,14 @@ public virtual void OnSocketClosed()
         }
 
         // Called on Libuv thread
-        public void Tick()
+        public void Tick(long timestamp)
         {
-            _frame.Tick();
+            if (timestamp > _timeoutTimestamp)
+            {
+                StopAsync();
+            }
+
+            Interlocked.Exchange(ref _lastTimestamp, timestamp);
         }
 
         private void ApplyConnectionFilter()
@@ -282,9 +293,17 @@ void IConnectionControl.End(ProduceEndType endType)
             }
         }
 
-        void IConnectionControl.Stop()
+        void IConnectionControl.SetTimeout(long milliseconds)
+        {
+            Debug.Assert(_timeoutTimestamp == long.MaxValue, "Concurrent timeouts are not supported");
+
+            // Add KestrelThread.HeartbeatMilliseconds extra milliseconds since this can be called right before the next heartbeat.
+            Interlocked.Exchange(ref _timeoutTimestamp, _lastTimestamp + milliseconds + KestrelThread.HeartbeatMilliseconds);
+        }
+
+        void IConnectionControl.CancelTimeout()
         {
-            StopAsync();
+            Interlocked.Exchange(ref _timeoutTimestamp, long.MaxValue);
         }
 
         private static unsafe string GenerateConnectionId(long id)
@@ -69,14 +69,15 @@ public abstract partial class Frame : ConnectionContext, IFrameControl
         private int _remainingRequestHeadersBytesAllowed;
         private int _requestHeadersParsed;
 
-        private int _secondsSinceLastRequest;
+        protected readonly long _keepAliveMilliseconds;
 
         public Frame(ConnectionContext context)
             : base(context)
         {
             _pathBase = context.ServerAddress.PathBase;
 
             FrameControl = this;
+            _keepAliveMilliseconds = (long)ServerOptions.Limits.KeepAliveTimeout.TotalMilliseconds;
         }
 
         public string ConnectionIdFeature { get; set; }
@@ -805,6 +806,8 @@ public RequestLineStatus TakeStartLine(SocketInput input)
                     return RequestLineStatus.Empty;
                 }
 
+                ConnectionControl.CancelTimeout();
+
                 _requestProcessingStatus = RequestProcessingStatus.RequestStarted;
 
                 int bytesScanned;
@@ -1269,25 +1272,6 @@ protected void ReportApplicationError(Exception ex)
             Log.ApplicationError(ConnectionId, ex);
         }
 
-        public void Tick()
-        {
-            // we're in between requests and not about to start processing a new one
-            if (_requestProcessingStatus == RequestProcessingStatus.RequestPending && !SocketInput.IsCompleted)
-            {
-                if (_secondsSinceLastRequest > ServerOptions.Limits.KeepAliveTimeout.TotalSeconds)
-                {
-                    ConnectionControl.Stop();
-                }
-
-                _secondsSinceLastRequest++;
-            }
-        }
-
-        public void RequestFinished()
-        {
-            _secondsSinceLastRequest = 0;
-        }
-
         public enum RequestLineStatus
         {
             Empty,
@@ -32,6 +32,8 @@ public class Frame<TContext> : Frame
             {
                 while (!_requestProcessingStopping)
                 {
+                    ConnectionControl.SetTimeout(_keepAliveMilliseconds);
+
                     while (!_requestProcessingStopping && TakeStartLine(SocketInput) != RequestLineStatus.Done)
                     {
                         if (SocketInput.CheckFinOrThrow())
@@ -8,6 +8,7 @@ public interface IConnectionControl
         void Pause();
         void Resume();
         void End(ProduceEndType endType);
-        void Stop();
+        void SetTimeout(long milliseconds);
+        void CancelTimeout();
     }
 }
@@ -172,7 +172,6 @@ public override ValueTask<int> ReadAsyncImplementation(ArraySegment<byte> buffer
                 var limit = buffer.Array == null ? inputLengthLimit : Math.Min(buffer.Count, inputLengthLimit);
                 if (limit == 0)
                 {
-                    _context.RequestFinished();
                     return new ValueTask<int>(0);
                 }
 
@@ -189,11 +188,6 @@ public override ValueTask<int> ReadAsyncImplementation(ArraySegment<byte> buffer
                         _context.RejectRequest(RequestRejectionReason.UnexpectedEndOfRequestContent);
                     }
 
-                    if (_inputLength == 0)
-                    {
-                        _context.RequestFinished();
-                    }
-
                     return new ValueTask<int>(actual);
                 }
                 else
@@ -212,11 +206,6 @@ public override ValueTask<int> ReadAsyncImplementation(ArraySegment<byte> buffer
                     _context.RejectRequest(RequestRejectionReason.UnexpectedEndOfRequestContent);
                 }
 
-                if (_inputLength == 0)
-                {
-                    _context.RequestFinished();
-                }
-
                 return actual;
             }
         }
@@ -368,8 +357,6 @@ public override ValueTask<int> ReadAsyncImplementation(ArraySegment<byte> buffer
                     _mode = Mode.Complete;
                 }
 
-                _context.RequestFinished();
-
                 return 0;
             }
 
@@ -19,6 +19,8 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Internal
     /// </summary>
     public class KestrelThread
     {
+        public const long HeartbeatMilliseconds = 1000;
+
         private static readonly Action<object, object> _postCallbackAdapter = (callback, state) => ((Action<object>)callback).Invoke(state);
         private static readonly Action<object, object> _postAsyncCallbackAdapter = (callback, state) => ((Action<object>)callback).Invoke(state);
 
@@ -27,9 +29,6 @@ public class KestrelThread
         // otherwise it needs to wait till the next pass of the libuv loop
         private readonly int _maxLoops = 8;
 
-        // how often the heartbeat timer will tick connections
-        private const int _heartbeatMilliseconds = 1000;
-
         private readonly KestrelEngine _engine;
         private readonly IApplicationLifetime _appLifetime;
         private readonly Thread _thread;
@@ -280,7 +279,7 @@ private void ThreadStart(object parameter)
                     _loop.Init(_engine.Libuv);
                     _post.Init(_loop, OnPost, EnqueueCloseHandle);
                     _heartbeatTimer.Init(_loop, EnqueueCloseHandle);
-                    _heartbeatTimer.Start(OnHeartbeat, timeout: 1000, repeat: 1000);
+                    _heartbeatTimer.Start(OnHeartbeat, timeout: HeartbeatMilliseconds, repeat: HeartbeatMilliseconds);
                     _initCompleted = true;
                     tcs.SetResult(0);
                 }
@@ -337,10 +336,12 @@ private void OnPost()
 
         private void OnHeartbeat(UvTimerHandle timer)
         {
+            var now = Loop.Now();
+
             Walk(ptr =>
             {
                 var handle = UvMemory.FromIntPtr<UvHandle>(ptr);
-                (handle as UvStreamHandle)?.Connection?.Tick();
+                (handle as UvStreamHandle)?.Connection?.Tick(now);
             });
         }
 
@@ -56,6 +56,7 @@ public Libuv()
             _uv_timer_init = NativeMethods.uv_timer_init;
             _uv_timer_start = NativeMethods.uv_timer_start;
             _uv_timer_stop = NativeMethods.uv_timer_stop;
+            _uv_now = NativeMethods.uv_now;
         }
 
         // Second ctor that doesn't set any fields only to be used by MockLibuv
@@ -434,6 +435,13 @@ unsafe public void timer_stop(UvTimerHandle handle)
             ThrowIfErrored(_uv_timer_stop(handle));
         }
 
+        protected Func<UvLoopHandle, long> _uv_now;
+        unsafe public long now(UvLoopHandle loop)
+        {
+            loop.Validate();
+            return _uv_now(loop);
+        }
+
         public delegate int uv_tcp_getsockname_func(UvTcpHandle handle, out SockAddr addr, ref int namelen);
         protected uv_tcp_getsockname_func _uv_tcp_getsockname;
         public void tcp_getsockname(UvTcpHandle handle, out SockAddr addr, ref int namelen)
@@ -640,6 +648,9 @@ private static class NativeMethods
             [DllImport("libuv", CallingConvention = CallingConvention.Cdecl)]
             unsafe public static extern int uv_timer_stop(UvTimerHandle handle);
 
+            [DllImport("libuv", CallingConvention = CallingConvention.Cdecl)]
+            unsafe public static extern long uv_now(UvLoopHandle loop);
+
             [DllImport("WS2_32.dll", CallingConvention = CallingConvention.Winapi)]
             unsafe public static extern int WSAIoctl(
                 IntPtr socket,
@@ -33,6 +33,11 @@ public void Stop()
             _uv.stop(this);
         }
 
+        public long Now()
+        {
+            return _uv.now(this);
+        }
+
         unsafe protected override bool ReleaseHandle()
         {
             var memory = handle;
@@ -23,8 +23,8 @@ public class KestrelServerLimits
         // Matches the default LimitRequestFields in Apache httpd.
         private int _maxRequestHeaderCount = 100;
 
-        // Matches the default http.sys keep-alive timouet.
-        private TimeSpan _keepAliveTimeout = TimeSpan.FromMinutes(2);
+        // Matches the default http.sys connection timeout.
+        private TimeSpan _connectionTimeout = TimeSpan.FromMinutes(2);
 
         /// <summary>
         /// Gets or sets the maximum size of the response buffer before write
@@ -146,17 +146,17 @@ public int MaxRequestHeaderCount
         /// Gets or sets the keep-alive timeout.
         /// </summary>
         /// <remarks>
-        /// Defaults to 2 minutes. Timeout granularity is in seconds. Sub-second values will be rounded to the next second.
+        /// Defaults to 2 minutes.
         /// </remarks>
         public TimeSpan KeepAliveTimeout
         {
             get
             {
-                return _keepAliveTimeout;
+                return _connectionTimeout;
             }
             set
             {
-                _keepAliveTimeout = TimeSpan.FromSeconds(Math.Ceiling(value.TotalSeconds));
+                _connectionTimeout = value;
             }
         }
     }
@@ -3,8 +3,11 @@
 
 using System;
 using System.IO;
+using System.Text;
+using System.Threading;
 using System.Threading.Tasks;
 using Microsoft.AspNetCore.Http;
+using Microsoft.AspNetCore.Http.Features;
 using Microsoft.AspNetCore.Testing;
 using Xunit;
 
@@ -13,23 +16,25 @@ namespace Microsoft.AspNetCore.Server.Kestrel.FunctionalTests
     public class KeepAliveTimeoutTests
     {
         private static readonly TimeSpan KeepAliveTimeout = TimeSpan.FromSeconds(10);
-        private static readonly int LongDelay = (int)TimeSpan.FromSeconds(30).TotalMilliseconds;
-        private static readonly int ShortDelay = LongDelay / 10;
+        private static readonly TimeSpan LongDelay = TimeSpan.FromSeconds(30);
+        private static readonly TimeSpan ShortDelay = TimeSpan.FromSeconds(LongDelay.TotalSeconds / 10);
 
         [Fact]
         public async Task TestKeepAliveTimeout()
         {
-            using (var server = CreateServer())
+            var longRunningCancellationTokenSource = new CancellationTokenSource();
+            var upgradeCancellationTokenSource = new CancellationTokenSource();
+
+            using (var server = CreateServer(longRunningCancellationTokenSource.Token, upgradeCancellationTokenSource.Token))
             {
                 var tasks = new[]
                 {
                     ConnectionClosedWhenKeepAliveTimeoutExpires(server),
-                    ConnectionClosedWhenKeepAliveTimeoutExpiresAfterChunkedRequest(server),
-                    KeepAliveTimeoutResetsBetweenContentLengthRequests(server),
-                    KeepAliveTimeoutResetsBetweenChunkedRequests(server),
-                    KeepAliveTimeoutNotTriggeredMidContentLengthRequest(server),
-                    KeepAliveTimeoutNotTriggeredMidChunkedRequest(server),
-                    ConnectionTimesOutWhenOpenedButNoRequestSent(server)
+                    ConnectionKeptAliveBetweenRequests(server),
+                    ConnectionNotTimedOutWhileRequestBeingSent(server),
+                    ConnectionNotTimedOutWhileAppIsRunning(server, longRunningCancellationTokenSource),
+                    ConnectionTimesOutWhenOpenedButNoRequestSent(server),
+                    KeepAliveTimeoutDoesNotApplyToUpgradedConnections(server, upgradeCancellationTokenSource)
                 };
 
                 await Task.WhenAll(tasks);
@@ -59,35 +64,7 @@ public class KeepAliveTimeoutTests
             }
         }
 
-        private async Task ConnectionClosedWhenKeepAliveTimeoutExpiresAfterChunkedRequest(TestServer server)
-        {
-            using (var connection = new TestConnection(server.Port))
-            {
-                await connection.Send(
-                        "POST / HTTP/1.1",
-                        "Transfer-Encoding: chunked",
-                        "",
-                        "5", "hello",
-                        "6", " world",
-                        "0",
-                         "",
-                         "");
-                await ReceiveResponse(connection, server.Context);
-
-                await Task.Delay(LongDelay);
-
-                await Assert.ThrowsAsync<IOException>(async () =>
-                {
-                    await connection.Send(
-                        "GET / HTTP/1.1",
-                        "",
-                        "");
-                    await ReceiveResponse(connection, server.Context);
-                });
-            }
-        }
-
-        private async Task KeepAliveTimeoutResetsBetweenContentLengthRequests(TestServer server)
+        private async Task ConnectionKeptAliveBetweenRequests(TestServer server)
         {
             using (var connection = new TestConnection(server.Port))
             {
@@ -107,62 +84,56 @@ public class KeepAliveTimeoutTests
             }
         }
 
-        private async Task KeepAliveTimeoutResetsBetweenChunkedRequests(TestServer server)
+        private async Task ConnectionNotTimedOutWhileRequestBeingSent(TestServer server)
         {
             using (var connection = new TestConnection(server.Port))
             {
-                for (var i = 0; i < 10; i++)
-                {
-                    await connection.Send(
+                var cts = new CancellationTokenSource();
+                cts.CancelAfter(LongDelay);
+
+                await connection.Send(
                         "POST / HTTP/1.1",
                         "Transfer-Encoding: chunked",
                         "",
-                        "5", "hello",
-                        "6", " world",
-                        "0",
-                         "",
-                         "");
-                    await Task.Delay(ShortDelay);
-                }
+                        "");
 
-                for (var i = 0; i < 10; i++)
+                while (!cts.IsCancellationRequested)
                 {
-                    await ReceiveResponse(connection, server.Context);
+                    await connection.Send(
+                        "1",
+                        "a",
+                        "");
                 }
-            }
-        }
 
-        private async Task KeepAliveTimeoutNotTriggeredMidContentLengthRequest(TestServer server)
-        {
-            using (var connection = new TestConnection(server.Port))
-            {
                 await connection.Send(
-                    "POST / HTTP/1.1",
-                    "Content-Length: 8",
-                    "",
-                    "a");
-                await Task.Delay(LongDelay);
-                await connection.Send("bcdefgh");
+                        "0",
+                        "",
+                        "");
                 await ReceiveResponse(connection, server.Context);
             }
         }
 
-        private async Task KeepAliveTimeoutNotTriggeredMidChunkedRequest(TestServer server)
+        private async Task ConnectionNotTimedOutWhileAppIsRunning(TestServer server, CancellationTokenSource cts)
         {
             using (var connection = new TestConnection(server.Port))
             {
                 await connection.Send(
-                        "POST / HTTP/1.1",
-                        "Transfer-Encoding: chunked",
-                        "",
-                        "5", "hello",
-                        "");
-                await Task.Delay(LongDelay);
+                    "GET /longrunning HTTP/1.1",
+                    "",
+                    "");
+                cts.CancelAfter(LongDelay);
+
+                while (!cts.IsCancellationRequested)
+                {
+                    await Task.Delay(1000);
+                }
+
+                await ReceiveResponse(connection, server.Context);
+
                 await connection.Send(
-                        "6", " world",
-                        "0",
-                         "",
-                         "");
+                    "GET / HTTP/1.1",
+                    "",
+                    "");
                 await ReceiveResponse(connection, server.Context);
             }
         }
@@ -182,9 +153,34 @@ public class KeepAliveTimeoutTests
             }
         }
 
-        private TestServer CreateServer()
+        private async Task KeepAliveTimeoutDoesNotApplyToUpgradedConnections(TestServer server, CancellationTokenSource cts)
         {
-            return new TestServer(App, new TestServiceContext
+            using (var connection = new TestConnection(server.Port))
+            {
+                await connection.Send(
+                    "GET /upgrade HTTP/1.1",
+                    "",
+                    "");
+                await connection.Receive(
+                    "HTTP/1.1 101 Switching Protocols",
+                    "Connection: Upgrade",
+                    $"Date: {server.Context.DateHeaderValue}",
+                    "",
+                    "");
+                cts.CancelAfter(LongDelay);
+
+                while (!cts.IsCancellationRequested)
+                {
+                    await Task.Delay(1000);
+                }
+
+                await connection.Receive("hello, world");
+            }
+        }
+
+        private TestServer CreateServer(CancellationToken longRunningCt, CancellationToken upgradeCt)
+        {
+            return new TestServer(httpContext => App(httpContext, longRunningCt, upgradeCt), new TestServiceContext
             {
                 ServerOptions = new KestrelServerOptions
                 {
@@ -197,21 +193,50 @@ private TestServer CreateServer()
             });
         }
 
-        private async Task App(HttpContext httpContext)
+        private async Task App(HttpContext httpContext, CancellationToken longRunningCt, CancellationToken upgradeCt)
         {
-            const string response = "hello, world";
-            httpContext.Response.ContentLength = response.Length;
-            await httpContext.Response.WriteAsync(response);
+            var ct = httpContext.RequestAborted;
+
+            if (httpContext.Request.Path == "/longrunning")
+            {
+                while (!longRunningCt.IsCancellationRequested)
+                {
+                    await Task.Delay(1000);
+                }
+
+                await httpContext.Response.WriteAsync("hello, world");
+            }
+            else if (httpContext.Request.Path == "/upgrade")
+            {
+                using (var stream = await httpContext.Features.Get<IHttpUpgradeFeature>().UpgradeAsync())
+                {
+                    while (!upgradeCt.IsCancellationRequested)
+                    {
+                        await Task.Delay(LongDelay);
+                    }
+
+                    var responseBytes = Encoding.ASCII.GetBytes("hello, world");
+                    await stream.WriteAsync(responseBytes, 0, responseBytes.Length);
+                }
+            }
+            else
+            {
+                await httpContext.Response.WriteAsync("hello, world");
+            }
         }
 
         private async Task ReceiveResponse(TestConnection connection, TestServiceContext testServiceContext)
         {
             await connection.Receive(
                 "HTTP/1.1 200 OK",
                 $"Date: {testServiceContext.DateHeaderValue}",
-                "Content-Length: 12",
+                "Transfer-Encoding: chunked",
+                "",
+                "c",
+                "hello, world",
+                "0",
                 "",
-                "hello, world");
+                "");
         }
     }
 }
@@ -1,4 +1,8 @@
-using System.Threading;
+// 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.Threading;
+using System.Threading.Tasks;
 using Microsoft.AspNetCore.Http;
 using Microsoft.AspNetCore.Server.Kestrel;
 using Microsoft.AspNetCore.Server.Kestrel.Internal;
@@ -14,7 +18,7 @@ namespace Microsoft.AspNetCore.Server.KestrelTests
     public class ConnectionTests
     {
         [Fact]
-        public void DoesNotEndConnectionOnZeroRead()
+        public async Task DoesNotEndConnectionOnZeroRead()
         {
             var mockLibuv = new MockLibuv();
 
@@ -30,14 +34,19 @@ public void DoesNotEndConnectionOnZeroRead()
                     ServerAddress = ServerAddress.FromUrl("http://127.0.0.1:0"),
                     Thread = engine.Threads[0]
                 };
-                var socket = new MockSocket(mockLibuv, Thread.CurrentThread.ManagedThreadId, trace);
-                var connection = new Connection(context, socket);
-                connection.Start();
 
-                Libuv.uv_buf_t ignored;
-                mockLibuv.AllocCallback(socket.InternalGetHandle(), 2048, out ignored);
-                mockLibuv.ReadCallback(socket.InternalGetHandle(), 0, ref ignored);
-                Assert.False(connection.SocketInput.CheckFinOrThrow());
+                Connection connection = null;
+                await context.Thread.PostAsync(_ =>
+                {
+                    var socket = new MockSocket(mockLibuv, Thread.CurrentThread.ManagedThreadId, trace);
+                    connection = new Connection(context, socket);
+                    connection.Start();
+
+                    Libuv.uv_buf_t ignored;
+                    mockLibuv.AllocCallback(socket.InternalGetHandle(), 2048, out ignored);
+                    mockLibuv.ReadCallback(socket.InternalGetHandle(), 0, ref ignored);
+                    Assert.False(connection.SocketInput.CheckFinOrThrow());
+                }, null);
 
                 connection.ConnectionControl.End(ProduceEndType.SocketDisconnect);
             }
@@ -12,6 +12,7 @@
 using Microsoft.AspNetCore.Server.KestrelTests.TestHelpers;
 using Microsoft.AspNetCore.Testing;
 using Microsoft.Extensions.Internal;
+using Moq;
 using Xunit;
 
 namespace Microsoft.AspNetCore.Server.KestrelTests
@@ -724,6 +725,7 @@ public void TakeStartLineCallsConsumingCompleteWithFurthestExamined()
             {
                 var connectionContext = new ConnectionContext()
                 {
+                    ConnectionControl = new Mock<IConnectionControl>().Object,
                     DateHeaderValueManager = new DateHeaderValueManager(),
                     ServerAddress = ServerAddress.FromUrl("http://localhost:5000"),
                     ServerOptions = new KestrelServerOptions(),
@@ -770,6 +772,7 @@ public void TakeStartLineReturnsWhenGivenIncompleteRequestLines(string requestLi
             {
                 var connectionContext = new ConnectionContext()
                 {
+                    ConnectionControl = new Mock<IConnectionControl>().Object,
                     DateHeaderValueManager = new DateHeaderValueManager(),
                     ServerAddress = ServerAddress.FromUrl("http://localhost:5000"),
                     ServerOptions = new KestrelServerOptions(),
@@ -787,6 +790,59 @@ public void TakeStartLineReturnsWhenGivenIncompleteRequestLines(string requestLi
         }
 
         [Fact]
+        public void TakeStartLineDisablesKeepAliveTimeoutOnFirstByteAvailable()
+        {
+            var trace = new KestrelTrace(new TestKestrelTrace());
+            var ltp = new LoggingThreadPool(trace);
+            using (var pool = new MemoryPool())
+            using (var socketInput = new SocketInput(pool, ltp))
+            {
+                var connectionControl = new Mock<IConnectionControl>();
+                var connectionContext = new ConnectionContext()
+                {
+                    ConnectionControl = connectionControl.Object,
+                    DateHeaderValueManager = new DateHeaderValueManager(),
+                    ServerAddress = ServerAddress.FromUrl("http://localhost:5000"),
+                    ServerOptions = new KestrelServerOptions(),
+                    Log = trace
+                };
+                var frame = new Frame<object>(application: null, context: connectionContext);
+                frame.Reset();
+
+                var requestLineBytes = Encoding.ASCII.GetBytes("G");
+                socketInput.IncomingData(requestLineBytes, 0, requestLineBytes.Length);
+
+                frame.TakeStartLine(socketInput);
+                connectionControl.Verify(cc => cc.CancelTimeout());
+            }
+        }
+
+        [Fact]
+        public void TakeStartLineDoesNotDisableKeepAliveTimeoutIfNoDataAvailable()
+        {
+            var trace = new KestrelTrace(new TestKestrelTrace());
+            var ltp = new LoggingThreadPool(trace);
+            using (var pool = new MemoryPool())
+            using (var socketInput = new SocketInput(pool, ltp))
+            {
+                var connectionControl = new Mock<IConnectionControl>();
+                var connectionContext = new ConnectionContext()
+                {
+                    ConnectionControl = connectionControl.Object,
+                    DateHeaderValueManager = new DateHeaderValueManager(),
+                    ServerAddress = ServerAddress.FromUrl("http://localhost:5000"),
+                    ServerOptions = new KestrelServerOptions(),
+                    Log = trace
+                };
+                var frame = new Frame<object>(application: null, context: connectionContext);
+                frame.Reset();
+
+                frame.TakeStartLine(socketInput);
+                connectionControl.Verify(cc => cc.CancelTimeout(), Times.Never);
+            }
+        }
+
+        [Fact]
         public void TakeMessageHeadersCallsConsumingCompleteWithFurthestExamined()
         {
             var trace = new KestrelTrace(new TestKestrelTrace());
@@ -864,5 +920,35 @@ public void TakeMessageHeadersReturnsWhenGivenIncompleteHeaders(string headers)
                 Assert.Equal(false, frame.TakeMessageHeaders(socketInput, (FrameRequestHeaders)frame.RequestHeaders));
             }
         }
+
+        [Fact]
+        public void RequestProcessingAsyncEnablesKeepAliveTimeout()
+        {
+            var trace = new KestrelTrace(new TestKestrelTrace());
+            var ltp = new LoggingThreadPool(trace);
+            using (var pool = new MemoryPool())
+            using (var socketInput = new SocketInput(pool, ltp))
+            {
+                var connectionControl = new Mock<IConnectionControl>();
+                var connectionContext = new ConnectionContext()
+                {
+                    ConnectionControl = connectionControl.Object,
+                    DateHeaderValueManager = new DateHeaderValueManager(),
+                    ServerAddress = ServerAddress.FromUrl("http://localhost:5000"),
+                    ServerOptions = new KestrelServerOptions(),
+                    Log = trace
+                };
+                var frame = new Frame<object>(application: null, context: connectionContext);
+                frame.Reset();
+
+                var requestProcessingTask = frame.RequestProcessingAsync();
+                connectionControl.Verify(cc => cc.SetTimeout((long)connectionContext.ServerOptions.Limits.KeepAliveTimeout.TotalMilliseconds));
+
+                frame.Stop();
+                socketInput.IncomingFin();
+
+                requestProcessingTask.Wait();
+            }
+        }
     }
 }
@@ -161,11 +161,11 @@ public void KeepAliveTimeoutDefault()
         [InlineData(2.1)]
         [InlineData(2.5)]
         [InlineData(2.9)]
-        public void KeepAliveTimeoutIsRoundedToTheNextSecond(double seconds)
+        public void KeepAliveTimeoutValid(double seconds)
         {
             var o = new KestrelServerLimits();
             o.KeepAliveTimeout = TimeSpan.FromSeconds(seconds);
-            Assert.Equal(Math.Ceiling(seconds), o.KeepAliveTimeout.TotalSeconds);
+            Assert.Equal(seconds, o.KeepAliveTimeout.TotalSeconds);
         }
     }
 }
@@ -15,6 +15,7 @@ public class MockConnection : Connection, IDisposable
 
         public MockConnection(KestrelServerOptions options)
         {
+            ConnectionControl = this;
             RequestAbortedSource = new CancellationTokenSource();
             ServerOptions = options;
         }
@@ -129,6 +129,7 @@ unsafe public MockLibuv()
             _uv_timer_init = (loop, handle) => 0;
             _uv_timer_start = (handle, callback, timeout, repeat) => 0;
             _uv_timer_stop = handle => 0;
+            _uv_now = (loop) => DateTime.UtcNow.Ticks / TimeSpan.TicksPerMillisecond;
         }
 
         public Func<UvStreamHandle, int, Action<int>, int> OnWrite { get; set; }
@@ -22,7 +22,12 @@ public TestInput()
         {
             var trace = new KestrelTrace(new TestKestrelTrace());
             var ltp = new LoggingThreadPool(trace);
-            var context = new Frame<object>(null, new ConnectionContext() { ServerAddress = new ServerAddress() })
+            var connectionContext = new ConnectionContext()
+            {
+                ServerAddress = new ServerAddress(),
+                ServerOptions = new KestrelServerOptions()
+            };
+            var context = new Frame<object>(null, connectionContext)
             {
                 DateHeaderValueManager = new DateHeaderValueManager(),
                 ServerAddress = ServerAddress.FromUrl("http://localhost:5000"),
@@ -63,7 +68,11 @@ public void End(ProduceEndType endType)
         {
         }
 
-        public void Stop()
+        public void SetTimeout(long milliseconds)
+        {
+        }
+
+        public void CancelTimeout()
         {
         }