[Wip] Add timeouts

Author: benaadams

src/Microsoft.AspNet.Server.Kestrel/Http/Frame.cs

@@ -39,6 +39,13 @@ public abstract partial class Frame : FrameContext, IFrameControl
         private static readonly byte[] _bytesDate = Encoding.ASCII.GetBytes("Date: ");
         private static readonly byte[] _bytesEndHeaders = Encoding.ASCII.GetBytes("\r\n\r\n");
 
+        private static readonly TimerCallback _timeoutRequest = (o) => ((Frame)o).TimeoutRequest();
+        protected readonly Timer _timeout;
+        /* TODO: Need to move to config */
+        private static readonly int _keepAliveTimout = 160000;
+        private static readonly int _headerTimeout = 160000;
+        private static readonly int _bodyTimeout = 160000;
+
         private readonly object _onStartingSync = new Object();
         private readonly object _onCompletedSync = new Object();
         protected readonly FrameRequestHeaders _requestHeaders = new FrameRequestHeaders();
@@ -50,8 +57,9 @@ public abstract partial class Frame : FrameContext, IFrameControl
 
         private bool _requestProcessingStarted;
         private Task _requestProcessingTask;
-        protected volatile bool _requestProcessingStopping; // volatile, see: https://msdn.microsoft.com/en-us/library/x13ttww7.aspx
-        protected volatile bool _requestAborted;
+        // volatile, see: https://msdn.microsoft.com/en-us/library/x13ttww7.aspx
+        // enum doesn't work with Interlocked
+        protected volatile int _frameState;
         protected CancellationTokenSource _abortedCts;
         protected CancellationToken? _manuallySetRequestAbortToken;
 
@@ -87,6 +95,9 @@ public Frame(ConnectionContext context,
             _prepareRequest = prepareRequest;
             _pathBase = context.ServerAddress.PathBase;
 
+            _frameState = FrameState.Waiting;
+            _timeout = new Timer(_timeoutRequest, this, _keepAliveTimout, Timeout.Infinite); 
+
             FrameControl = this;
             Reset();
         }
@@ -155,7 +166,7 @@ public CancellationToken RequestAborted
                 var cts = _abortedCts;
                 return
                     cts != null ? cts.Token :
-                    _requestAborted ? new CancellationToken(true) :
+                    _frameState == FrameState.Aborted ? new CancellationToken(true) :
                     RequestAbortedSource.Token;
             }
             set
@@ -173,7 +184,7 @@ private CancellationTokenSource RequestAbortedSource
                 // Get the abort token, lazily-initializing it if necessary.
                 // Make sure it's canceled if an abort request already came in.
                 var cts = LazyInitializer.EnsureInitialized(ref _abortedCts, () => new CancellationTokenSource());
-                if (_requestAborted)
+                if (_frameState == FrameState.Aborted)
                 {
                     cts.Cancel();
                 }
@@ -185,8 +196,91 @@ public bool HasResponseStarted
             get { return _responseStarted; }
         }
 
+        private void TimeoutRequest()
+        {
+            if (TransitionToState(FrameState.Timeout) == FrameState.Timeout)
+            {
+                SocketInput?.AbortAwaiting();
+            }
+        }
+        protected int TransitionToState(int state)
+        {
+#pragma warning disable CS0420 // A reference to a volatile field will not be treated as volatile
+            int prevState = Volatile.Read(ref _frameState);
+#pragma warning restore CS0420
+
+            switch (state)
+            {
+                case FrameState.Waiting:
+                    if (prevState == FrameState.Waiting) return FrameState.Waiting;
+                    // can only transition to Waiting from ReadingBody
+                    prevState = Interlocked.CompareExchange(ref _frameState, FrameState.Waiting, FrameState.ReadingBody);
+                    if (prevState == FrameState.ReadingBody)
+                    {
+                        // only reset timer on transition into this state
+                        _timeout.Change(_keepAliveTimout, Timeout.Infinite);
+                        return FrameState.Waiting;
+                    }
+                    return prevState;
+                case FrameState.ReadingHeaders:
+                    if (prevState == FrameState.ReadingHeaders) return FrameState.ReadingHeaders;
+                    // can only transition to ReadingHeaders from Waiting
+                    prevState = Interlocked.CompareExchange(ref _frameState, FrameState.ReadingHeaders, FrameState.Waiting);
+                    if (prevState == FrameState.Waiting)
+                    {
+                        // only reset timer on transition into this state
+                        _timeout.Change(_headerTimeout, Timeout.Infinite);
+                        return FrameState.ReadingHeaders;
+                    }
+                    return prevState;
+                case FrameState.ReadingBody:
+                    /* TODO Need to plumb in SocketInput reads */
+                    if (prevState == FrameState.ReadingBody)
+                    {
+                        // reset timer on each read
+                        _timeout.Change(_bodyTimeout, Timeout.Infinite);
+                        return FrameState.ReadingBody;
+                    }
+                    // can only transition to ReadingBody from ReadingHeaders
+                    prevState = Interlocked.CompareExchange(ref _frameState, FrameState.ReadingBody, FrameState.ReadingHeaders);
+                    if (prevState == FrameState.ReadingHeaders)
+                    {
+                        // only reset timer if state correct
+                        _timeout.Change(_headerTimeout, Timeout.Infinite);
+                        return FrameState.ReadingBody;
+                    }
+                    return prevState;
+                case FrameState.Stopping:
+                    // marker state, can't transition into it.
+                    throw new InvalidOperationException();
+                case FrameState.Timeout:
+                    if (prevState >= FrameState.Timeout) return prevState;
+                    // can transition to Timeout from states below it
+                    do
+                    {
+                        prevState = Interlocked.CompareExchange(ref _frameState, FrameState.Timeout, prevState);
+                    } while (prevState < FrameState.Timeout);
+                    return prevState < FrameState.Timeout ? FrameState.Timeout : prevState;
+                case FrameState.Stopped:
+                    if (prevState >= FrameState.Stopped) return prevState;
+                    // can transition to Stopped from states below it
+                    do
+                    {
+                        prevState = Interlocked.CompareExchange(ref _frameState, FrameState.Stopped, prevState);
+                    } while (prevState < FrameState.Stopped);
+                    return prevState < FrameState.Stopped ? FrameState.Stopped : prevState;
+                case FrameState.Aborted:
+                    // can transition to Aborted from any state
+                    return (_frameState = FrameState.Aborted);
+                default:
+                    throw new InvalidOperationException();
+            }
+        }
+
         public void Reset()
         {
+            TransitionToState(FrameState.Waiting);
+
             _onStarting = null;
             _onCompleted = null;
 
@@ -273,10 +367,8 @@ public void Start()
         /// </summary>
         public Task Stop()
         {
-            if (!_requestProcessingStopping)
-            {
-                _requestProcessingStopping = true;
-            }
+            TransitionToState(FrameState.Stopped);
+
             return _requestProcessingTask ?? TaskUtilities.CompletedTask;
         }
 
@@ -285,8 +377,7 @@ public Task Stop()
         /// </summary>
         public void Abort()
         {
-            _requestProcessingStopping = true;
-            _requestAborted = true;
+            TransitionToState(FrameState.Aborted);
 
             _requestBody?.Abort();
             _responseBody?.Abort();
@@ -573,7 +664,7 @@ protected Task ProduceEnd()
                 if (_responseStarted)
                 {
                     // We can no longer respond with a 500, so we simply close the connection.
-                    _requestProcessingStopping = true;
+                    TransitionToState(FrameState.Stopped);
                     return TaskUtilities.CompletedTask;
                 }
                 else
@@ -589,6 +680,12 @@ protected Task ProduceEnd()
 
             if (!_responseStarted)
             {
+                if (_frameState > FrameState.Stopping && _frameState < FrameState.Stopped)
+                {
+                    // State is status code
+                    StatusCode = _frameState;
+                    ReasonPhrase = null;
+                }
                 return ProduceEndAwaited();
             }
 
@@ -701,6 +798,12 @@ protected bool TakeStartLine(SocketInput input)
                 {
                     return false;
                 }
+
+                if (TransitionToState(FrameState.ReadingHeaders) != FrameState.ReadingHeaders)
+                {
+                    return false;
+                }
+
                 var method = begin.GetAsciiString(scan);
 
                 scan.Take();
@@ -946,5 +1049,20 @@ private enum HttpVersionType
             Http1_0 = 0,
             Http1_1 = 1
         }
+
+        // enum doesn't work with Interlocked
+        protected class FrameState
+        {
+            public const int Waiting = 0;
+            public const int ReadingHeaders = 1;
+            public const int ReadingBody = 2;
+            // Do not change order of these with out changing comparision tests
+            public const int Stopping = 99;
+            // States are status codes
+            public const int Timeout = 408;
+            // Other final states
+            public const int Stopped = 1000;
+            public const int Aborted = 1001;
+        }
     }
 }

src/Microsoft.AspNet.Server.Kestrel/Http/FrameOfT.cs

@@ -40,9 +40,9 @@ public override async Task RequestProcessingAsync()
         {
             try
             {
-                while (!_requestProcessingStopping)
+                while (_frameState == FrameState.Waiting)
                 {
-                    while (!_requestProcessingStopping && !TakeStartLine(SocketInput))
+                    while (_frameState < FrameState.Stopping && !TakeStartLine(SocketInput))
                     {
                         if (SocketInput.RemoteIntakeFin)
                         {
@@ -51,7 +51,7 @@ public override async Task RequestProcessingAsync()
                         await SocketInput;
                     }
 
-                    while (!_requestProcessingStopping && !TakeMessageHeaders(SocketInput, _requestHeaders))
+                    while (_frameState < FrameState.Stopping && !TakeMessageHeaders(SocketInput, _requestHeaders))
                     {
                         if (SocketInput.RemoteIntakeFin)
                         {
@@ -60,7 +60,7 @@ public override async Task RequestProcessingAsync()
                         await SocketInput;
                     }
 
-                    if (!_requestProcessingStopping)
+                    if (TransitionToState(FrameState.ReadingBody) == FrameState.ReadingBody)
                     {
                         var messageBody = MessageBody.For(HttpVersion, _requestHeaders, this);
                         _keepAlive = messageBody.RequestKeepAlive;
@@ -100,8 +100,8 @@ public override async Task RequestProcessingAsync()
 
                             _application.DisposeContext(context, _applicationException);
 
-                            // If _requestAbort is set, the connection has already been closed.
-                            if (!_requestAborted)
+                            // If Aborted, the connection has already been closed.
+                            if (_frameState != FrameState.Aborted)
                             {
                                 await ProduceEnd();
 
@@ -133,10 +133,11 @@ public override async Task RequestProcessingAsync()
             {
                 try
                 {
+                    _timeout.Dispose();
                     _abortedCts = null;
 
-                    // If _requestAborted is set, the connection has already been closed.
-                    if (!_requestAborted)
+                    // If Aborted, the connection has already been closed.
+                    if (_frameState != FrameState.Aborted)
                     {
                         // Inform client no more data will ever arrive
                         ConnectionControl.End(ProduceEndType.SocketShutdownSend);

src/Microsoft.AspNet.Server.Kestrel/Infrastructure/KestrelThread.cs

@@ -245,6 +245,11 @@ private void ThreadStart(object parameter)
             }
             catch (Exception ex)
             {
+#if DEBUG
+                // Output error in debug for Tests, which do not support application lifetime
+                Console.WriteLine(nameof(KestrelThread) + " Exception");
+                Console.WriteLine(ex);
+#endif
                 _closeError = ExceptionDispatchInfo.Capture(ex);
                 // Request shutdown so we can rethrow this exception
                 // in Stop which should be observable.