diff --git a/src/Microsoft.AspNet.Server.Kestrel/Http/Connection.cs b/src/Microsoft.AspNet.Server.Kestrel/Http/Connection.cs index 906faa986..b206d1fad 100644 --- a/src/Microsoft.AspNet.Server.Kestrel/Http/Connection.cs +++ b/src/Microsoft.AspNet.Server.Kestrel/Http/Connection.cs @@ -12,8 +12,8 @@ namespace Microsoft.AspNet.Server.Kestrel.Http { public class Connection : ConnectionContext, IConnectionControl { - private static readonly Action _readCallback = ReadCallback; - private static readonly Func _allocCallback = AllocCallback; + private static readonly Action _readCallback = (handle, status, state) => ReadCallback(handle, status, state); + private static readonly Func _allocCallback = (handle, size, state) => AllocCallback(handle, size, state); private static long _lastConnectionId; @@ -72,12 +72,12 @@ public void Start() if (task.IsFaulted) { connection.Log.LogError("ConnectionFilter.OnConnection", task.Exception); - ConnectionControl.End(ProduceEndType.SocketDisconnect); + connection.ConnectionControl.End(ProduceEndType.SocketDisconnect); } else if (task.IsCanceled) { connection.Log.LogError("ConnectionFilter.OnConnection Canceled"); - ConnectionControl.End(ProduceEndType.SocketDisconnect); + connection.ConnectionControl.End(ProduceEndType.SocketDisconnect); } else { diff --git a/src/Microsoft.AspNet.Server.Kestrel/Http/Frame.cs b/src/Microsoft.AspNet.Server.Kestrel/Http/Frame.cs index 79552ea94..13546271e 100644 --- a/src/Microsoft.AspNet.Server.Kestrel/Http/Frame.cs +++ b/src/Microsoft.AspNet.Server.Kestrel/Http/Frame.cs @@ -157,7 +157,7 @@ public async Task RequestProcessingAsync() } } - while (!terminated && !_requestProcessingStopping && !TakeMessageHeaders(SocketInput)) + while (!terminated && !_requestProcessingStopping && !TakeMessageHeaders(SocketInput, _requestHeaders)) { terminated = SocketInput.RemoteIntakeFin; if (!terminated) @@ -196,6 +196,9 @@ public async Task RequestProcessingAsync() await FireOnCompleted(); await ProduceEnd(); + + // Finish reading the request body in case the app did not. + await RequestBody.CopyToAsync(Stream.Null); } terminated = !_keepAlive; @@ -654,7 +657,7 @@ static string GetString(ArraySegment range, int startIndex, int endIndex) return Encoding.UTF8.GetString(range.Array, range.Offset + startIndex, endIndex - startIndex); } - private bool TakeMessageHeaders(SocketInput input) + public static bool TakeMessageHeaders(SocketInput input, FrameRequestHeaders requestHeaders) { var scan = input.ConsumingStart(); var consumed = scan; @@ -692,6 +695,22 @@ private bool TakeMessageHeaders(SocketInput input) chSecond == '\r' || chSecond == '\n') { + if (chSecond == '\r') + { + var scanAhead = scan; + var chAhead = scanAhead.Take(); + if (chAhead == '\n') + { + chAhead = scanAhead.Take(); + // If the "\r\n" isn't part of "linear whitespace", + // then this header has no value. + if (chAhead != ' ' && chAhead != '\t') + { + break; + } + } + } + beginValue = scan; chSecond = scan.Take(); } @@ -729,9 +748,6 @@ private bool TakeMessageHeaders(SocketInput input) } var name = beginName.GetArraySegment(endName); -#if DEBUG - var nameString = beginName.GetString(endName); -#endif var value = beginValue.GetString(endValue); if (wrapping) { @@ -739,7 +755,7 @@ private bool TakeMessageHeaders(SocketInput input) } consumed = scan; - _requestHeaders.Append(name.Array, name.Offset, name.Count, value); + requestHeaders.Append(name.Array, name.Offset, name.Count, value); break; } } diff --git a/src/Microsoft.AspNet.Server.Kestrel/Http/Listener.cs b/src/Microsoft.AspNet.Server.Kestrel/Http/Listener.cs index 8cfba60cd..f9a876271 100644 --- a/src/Microsoft.AspNet.Server.Kestrel/Http/Listener.cs +++ b/src/Microsoft.AspNet.Server.Kestrel/Http/Listener.cs @@ -3,6 +3,7 @@ using System; using System.Threading.Tasks; +using Microsoft.AspNet.Server.Kestrel.Infrastructure; using Microsoft.AspNet.Server.Kestrel.Networking; using Microsoft.Extensions.Logging; @@ -20,7 +21,7 @@ protected Listener(ServiceContext serviceContext) protected UvStreamHandle ListenSocket { get; private set; } - public Task StartAsync( + public StateCompletionSource StartAsync( ServerAddress address, KestrelThread thread, Func application) @@ -29,20 +30,20 @@ public Task StartAsync( Thread = thread; Application = application; - var tcs = new TaskCompletionSource(); - Thread.Post(_ => + var scs = new StateCompletionSource(this); + Thread.Post(scs2 => { try { - ListenSocket = CreateListenSocket(); - tcs.SetResult(0); + scs2.State.ListenSocket = scs2.State.CreateListenSocket(); + scs2.SetResult(0); } catch (Exception ex) { - tcs.SetException(ex); + scs2.SetException(ex); } - }, null); - return tcs.Task; + }, scs); + return scs; } /// @@ -84,24 +85,24 @@ public void Dispose() // the exception that stopped the event loop will never be surfaced. if (Thread.FatalError == null && ListenSocket != null) { - var tcs = new TaskCompletionSource(); + var scs = new StateCompletionSource(ListenSocket); Thread.Post( - _ => + scs2 => { try { - ListenSocket.Dispose(); - tcs.SetResult(0); + scs2.State.Dispose(); + scs2.SetResult(0); } catch (Exception ex) { - tcs.SetException(ex); + scs2.SetException(ex); } }, - null); + scs); // REVIEW: Should we add a timeout here to be safe? - tcs.Task.Wait(); + scs.Wait(); } ListenSocket = null; diff --git a/src/Microsoft.AspNet.Server.Kestrel/Http/ListenerPrimary.cs b/src/Microsoft.AspNet.Server.Kestrel/Http/ListenerPrimary.cs index 41aeeb3bf..6c045244b 100644 --- a/src/Microsoft.AspNet.Server.Kestrel/Http/ListenerPrimary.cs +++ b/src/Microsoft.AspNet.Server.Kestrel/Http/ListenerPrimary.cs @@ -35,15 +35,21 @@ public async Task StartAsync( KestrelThread thread, Func application) { - await StartAsync(address, thread, application).ConfigureAwait(false); + await StartAsync(address, thread, application); - await Thread.PostAsync(_ => + await Thread.PostAsync(listener => { - ListenPipe = new UvPipeHandle(Log); - ListenPipe.Init(Thread.Loop, false); - ListenPipe.Bind(pipeName); - ListenPipe.Listen(Constants.ListenBacklog, OnListenPipe, null); - }, null).ConfigureAwait(false); + listener.ListenPipe = new UvPipeHandle(listener.Log); + listener.ListenPipe.Init(listener.Thread.Loop, false); + listener.ListenPipe.Bind(pipeName); + listener.ListenPipe.Listen( + Constants.ListenBacklog, + (pipe, status, error, state) => + { + ((ListenerPrimary)state).OnListenPipe(pipe, status, error, state); + }, + listener); + }, this).ConfigureAwait(false); } private void OnListenPipe(UvStreamHandle pipe, int status, Exception error, object state) diff --git a/src/Microsoft.AspNet.Server.Kestrel/Http/ListenerSecondary.cs b/src/Microsoft.AspNet.Server.Kestrel/Http/ListenerSecondary.cs index a34897073..2ee2fa0fb 100644 --- a/src/Microsoft.AspNet.Server.Kestrel/Http/ListenerSecondary.cs +++ b/src/Microsoft.AspNet.Server.Kestrel/Http/ListenerSecondary.cs @@ -22,7 +22,7 @@ protected ListenerSecondary(ServiceContext serviceContext) : base(serviceContext UvPipeHandle DispatchPipe { get; set; } - public Task StartAsync( + public async Task StartAsync( string pipeName, ServerAddress address, KestrelThread thread, @@ -34,89 +34,98 @@ public Task StartAsync( DispatchPipe = new UvPipeHandle(Log); - var tcs = new TaskCompletionSource(); - Thread.Post(_ => + var qscs = new QuadStateCompletionSource(this, pipeName); + Thread.Post(qscs2 => { try { - DispatchPipe.Init(Thread.Loop, true); - var connect = new UvConnectRequest(Log); - connect.Init(Thread.Loop); + var listener = qscs2.State1; + listener.DispatchPipe.Init(listener.Thread.Loop, true); + var connect = new UvConnectRequest(listener.Log); + connect.Init(listener.Thread.Loop); connect.Connect( - DispatchPipe, - pipeName, + listener.DispatchPipe, + qscs2.State2, (connect2, status, error, state) => { - connect.Dispose(); + var qscs3 = (QuadStateCompletionSource)state; + var listener2 = qscs3.State1; + connect2.Dispose(); if (error != null) { - tcs.SetException(error); + qscs3.SetException(error); return; } try { var ptr = Marshal.AllocHGlobal(4); - var buf = Thread.Loop.Libuv.buf_init(ptr, 4); + var buf = listener2.Thread.Loop.Libuv.buf_init(ptr, 4); - DispatchPipe.ReadStart( - (_1, _2, _3) => buf, - (_1, status2, state2) => + qscs3.State3 = ptr; + qscs3.State4 = buf; + + listener2.DispatchPipe.ReadStart( + (handle, status2, state2) => ((QuadStateCompletionSource)state2).State4, + (handle, status2, state2) => { + var qscs4 = (QuadStateCompletionSource)state2; + var listener3 = qscs4.State1; if (status2 < 0) { if (status2 != Constants.EOF) { Exception ex; - Thread.Loop.Libuv.Check(status2, out ex); - Log.LogError("DispatchPipe.ReadStart", ex); + listener3.Thread.Loop.Libuv.Check(status2, out ex); + listener3.Log.LogError("DispatchPipe.ReadStart", ex); } - DispatchPipe.Dispose(); - Marshal.FreeHGlobal(ptr); + listener3.DispatchPipe.Dispose(); + Marshal.FreeHGlobal(qscs4.State3); return; } - if (DispatchPipe.PendingCount() == 0) + if (listener3.DispatchPipe.PendingCount() == 0) { return; } - var acceptSocket = CreateAcceptSocket(); + var acceptSocket = listener3.CreateAcceptSocket(); try { - DispatchPipe.Accept(acceptSocket); + listener3.DispatchPipe.Accept(acceptSocket); } catch (UvException ex) { - Log.LogError("DispatchPipe.Accept", ex); + listener3.Log.LogError("DispatchPipe.Accept", ex); acceptSocket.Dispose(); return; } - var connection = new Connection(this, acceptSocket); + var connection = new Connection(listener3, acceptSocket); connection.Start(); }, - null); + qscs3); - tcs.SetResult(0); + qscs3.SetResult(0); } catch (Exception ex) { - DispatchPipe.Dispose(); - tcs.SetException(ex); + listener2.DispatchPipe.Dispose(); + qscs3.SetException(ex); } }, - null); + qscs2); } catch (Exception ex) { - DispatchPipe.Dispose(); - tcs.SetException(ex); + qscs2.State1.DispatchPipe.Dispose(); + qscs2.SetException(ex); } - }, null); - return tcs.Task; + }, qscs); + await qscs; + return; } /// @@ -132,7 +141,7 @@ public void Dispose() // the exception that stopped the event loop will never be surfaced. if (Thread.FatalError == null) { - Thread.Send(_ => DispatchPipe.Dispose(), null); + Thread.Send(listener => ((ListenerSecondary)listener).DispatchPipe.Dispose(), this); } } } diff --git a/src/Microsoft.AspNet.Server.Kestrel/Http/PipeListener.cs b/src/Microsoft.AspNet.Server.Kestrel/Http/PipeListener.cs index 4c3e35f79..e43627323 100644 --- a/src/Microsoft.AspNet.Server.Kestrel/Http/PipeListener.cs +++ b/src/Microsoft.AspNet.Server.Kestrel/Http/PipeListener.cs @@ -25,7 +25,13 @@ protected override UvStreamHandle CreateListenSocket() var socket = new UvPipeHandle(Log); socket.Init(Thread.Loop, false); socket.Bind(ServerAddress.UnixPipePath); - socket.Listen(Constants.ListenBacklog, ConnectionCallback, this); + socket.Listen( + Constants.ListenBacklog, + (stream, status, error, state) => + { + ConnectionCallback(stream, status, error, state); + }, + this); return socket; } diff --git a/src/Microsoft.AspNet.Server.Kestrel/Http/PipeListenerPrimary.cs b/src/Microsoft.AspNet.Server.Kestrel/Http/PipeListenerPrimary.cs index 1f7eb285e..a67bff81c 100644 --- a/src/Microsoft.AspNet.Server.Kestrel/Http/PipeListenerPrimary.cs +++ b/src/Microsoft.AspNet.Server.Kestrel/Http/PipeListenerPrimary.cs @@ -25,7 +25,13 @@ protected override UvStreamHandle CreateListenSocket() var socket = new UvPipeHandle(Log); socket.Init(Thread.Loop, false); socket.Bind(ServerAddress.UnixPipePath); - socket.Listen(Constants.ListenBacklog, ConnectionCallback, this); + socket.Listen( + Constants.ListenBacklog, + (stream, status, error, state) => + { + ConnectionCallback(stream, status, error, state); + }, + this); return socket; } diff --git a/src/Microsoft.AspNet.Server.Kestrel/Http/SocketOutput.cs b/src/Microsoft.AspNet.Server.Kestrel/Http/SocketOutput.cs index d0d218ac5..4e3fd048a 100644 --- a/src/Microsoft.AspNet.Server.Kestrel/Http/SocketOutput.cs +++ b/src/Microsoft.AspNet.Server.Kestrel/Http/SocketOutput.cs @@ -241,13 +241,14 @@ void ISocketOutput.Write(ArraySegment buffer, bool immediate) buffer, (error, state, calledInline) => { + var tcs2 = (TaskCompletionSource)state; if (error != null) { - tcs.SetException(error); + tcs2.SetException(error); } else { - tcs.SetResult(0); + tcs2.SetResult(0); } }, tcs, @@ -259,7 +260,7 @@ void ISocketOutput.Write(ArraySegment buffer, bool immediate) } } - Task ISocketOutput.WriteAsync(ArraySegment buffer, bool immediate, CancellationToken cancellationToken) + async Task ISocketOutput.WriteAsync(ArraySegment buffer, bool immediate, CancellationToken cancellationToken) { if (!immediate) { @@ -270,48 +271,31 @@ Task ISocketOutput.WriteAsync(ArraySegment buffer, bool immediate, Cancell (error, state, calledInline) => { }, null, immediate: false); - return TaskUtilities.CompletedTask; + return; } // TODO: Optimize task being used, and remove callback model from the underlying Write - var tcs = new TaskCompletionSource(); + var cs = new CompletionSource(); Write( buffer, (error, state, calledInline) => { - if (!calledInline) + var cs2 = (CompletionSource)state; + if (error != null) { - ThreadPool.QueueUserWorkItem(state2 => - { - var tcs2 = (TaskCompletionSource)state2; - if (error != null) - { - tcs2.SetException(error); - } - else - { - tcs2.SetResult(0); - } - }, state); + cs2.SetException(error); } else { - var tcs2 = (TaskCompletionSource)state; - if (error != null) - { - tcs2.SetException(error); - } - else - { - tcs2.SetResult(0); - } + cs2.SetResult(0); } }, - tcs, + cs, immediate: true); - return tcs.Task; + await cs; + return; } private class CallbackContext @@ -377,7 +361,7 @@ public void DoWriteIfNeeded() var _this = (WriteContext)state; _this.WriteStatus = status; _this.WriteError = error; - DoShutdownIfNeeded(); + _this.DoShutdownIfNeeded(); }, this); } @@ -400,9 +384,9 @@ public void DoShutdownIfNeeded() var _this = (WriteContext)state; _this.ShutdownSendStatus = status; - Self._log.ConnectionWroteFin(Self._connectionId, status); + _this.Self._log.ConnectionWroteFin(_this.Self._connectionId, status); - DoDisconnectIfNeeded(); + _this.DoDisconnectIfNeeded(); }, this); } diff --git a/src/Microsoft.AspNet.Server.Kestrel/Http/TcpListener.cs b/src/Microsoft.AspNet.Server.Kestrel/Http/TcpListener.cs index 86feca412..ef483a034 100644 --- a/src/Microsoft.AspNet.Server.Kestrel/Http/TcpListener.cs +++ b/src/Microsoft.AspNet.Server.Kestrel/Http/TcpListener.cs @@ -25,7 +25,13 @@ protected override UvStreamHandle CreateListenSocket() socket.Init(Thread.Loop, Thread.QueueCloseHandle); socket.NoDelay(NoDelay); socket.Bind(ServerAddress); - socket.Listen(Constants.ListenBacklog, ConnectionCallback, this); + socket.Listen( + Constants.ListenBacklog, + (stream, status, error, state) => + { + ConnectionCallback(stream, status, error, state); + }, + this); return socket; } diff --git a/src/Microsoft.AspNet.Server.Kestrel/Http/TcpListenerPrimary.cs b/src/Microsoft.AspNet.Server.Kestrel/Http/TcpListenerPrimary.cs index 4f1ff0a48..53ed9cbab 100644 --- a/src/Microsoft.AspNet.Server.Kestrel/Http/TcpListenerPrimary.cs +++ b/src/Microsoft.AspNet.Server.Kestrel/Http/TcpListenerPrimary.cs @@ -27,7 +27,13 @@ protected override UvStreamHandle CreateListenSocket() socket.Init(Thread.Loop, Thread.QueueCloseHandle); socket.NoDelay(NoDelay); socket.Bind(ServerAddress); - socket.Listen(Constants.ListenBacklog, ConnectionCallback, this); + socket.Listen( + Constants.ListenBacklog, + (stream, status, error, state) => + { + ConnectionCallback(stream, status, error, state); + }, + this); return socket; } diff --git a/src/Microsoft.AspNet.Server.Kestrel/Infrastructure/CompletionSource.cs b/src/Microsoft.AspNet.Server.Kestrel/Infrastructure/CompletionSource.cs new file mode 100644 index 000000000..c6da9aa4f --- /dev/null +++ b/src/Microsoft.AspNet.Server.Kestrel/Infrastructure/CompletionSource.cs @@ -0,0 +1,116 @@ +// 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; +using System.Runtime.CompilerServices; +using System.Threading; + +namespace Microsoft.AspNet.Server.Kestrel.Infrastructure +{ + internal sealed class CompletionSource : INotifyCompletion, ICriticalNotifyCompletion + { + private readonly static Action CALLBACK_RAN = () => { }; + private bool _isCompleted; + private Action _continuation; + private SemaphoreSlim _semaphore; + + private Exception _error; + + private TResult _result; + + public CompletionSource() + { + } + + public bool IsCompleted { get { return _isCompleted; } } + + public void SetResult(TResult result) + { + _result = result; + _isCompleted = true; + + Complete(); + + Interlocked.MemoryBarrier(); + + if (_semaphore != null) + { + _semaphore.Release(); + } + } + + public void SetException(Exception ex) + { + _error = ex; + _isCompleted = true; + + Complete(); + + Interlocked.MemoryBarrier(); + + if (_semaphore != null) + { + _semaphore.Release(); + } + } + + private void Complete() + { + Action continuation = _continuation ?? Interlocked.CompareExchange(ref _continuation, CALLBACK_RAN, null); + if (continuation != null) + { + CompleteCallback(continuation); + } + } + + private static void CompleteCallback(Action continuation) + { + ThreadPool.QueueUserWorkItem((obj) => RunCallback(obj), continuation); + } + + public CompletionSource GetAwaiter() { return this; } + + private static void RunCallback(object callback) + { + ((Action)callback)(); + } + + void INotifyCompletion.OnCompleted(Action continuation) + { + throw new NotImplementedException(); + } + + [System.Security.SecurityCritical] + void ICriticalNotifyCompletion.UnsafeOnCompleted(Action continuation) + { + if (_continuation == CALLBACK_RAN || + Interlocked.CompareExchange( + ref _continuation, continuation, null) == CALLBACK_RAN) + { + CompleteCallback(continuation); + } + } + public TResult GetResult() + { + if (_error != null) + { + throw _error; + } + return _result; + } + + public void Wait() + { + _semaphore = new SemaphoreSlim(0); + + Interlocked.MemoryBarrier(); + + if (_isCompleted) + { + return; + } + + _semaphore.Wait(); + } + } +} diff --git a/src/Microsoft.AspNet.Server.Kestrel/Infrastructure/DualStateCompletionSource.cs b/src/Microsoft.AspNet.Server.Kestrel/Infrastructure/DualStateCompletionSource.cs new file mode 100644 index 000000000..ab8d64cbb --- /dev/null +++ b/src/Microsoft.AspNet.Server.Kestrel/Infrastructure/DualStateCompletionSource.cs @@ -0,0 +1,120 @@ +// 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; +using System.Runtime.CompilerServices; +using System.Threading; + +namespace Microsoft.AspNet.Server.Kestrel.Infrastructure +{ + internal sealed class DualStateCompletionSource : INotifyCompletion, ICriticalNotifyCompletion + { + private readonly static Action CALLBACK_RAN = () => { }; + private bool _isCompleted; + private Action _continuation; + private SemaphoreSlim _semaphore; + + private Exception _error; + private TResult _result; + + public DualStateCompletionSource(TState1 state1, TState2 state2) + { + State1 = state1; + State2 = state2; + } + public TState1 State1 { get; set; } + public TState2 State2 { get; set; } + + public bool IsCompleted { get { return _isCompleted; } } + + public void SetResult(TResult result) + { + _result = result; + _isCompleted = true; + + Complete(); + + Interlocked.MemoryBarrier(); + + if (_semaphore != null) + { + _semaphore.Release(); + } + } + + public void SetException(Exception ex) + { + _error = ex; + _isCompleted = true; + + Complete(); + + Interlocked.MemoryBarrier(); + + if (_semaphore != null) + { + _semaphore.Release(); + } + } + + private void Complete() + { + Action continuation = _continuation ?? Interlocked.CompareExchange(ref _continuation, CALLBACK_RAN, null); + if (continuation != null) + { + CompleteCallback(continuation); + } + } + + private static void CompleteCallback(Action continuation) + { + ThreadPool.QueueUserWorkItem((obj) => RunCallback(obj), continuation); + } + + public DualStateCompletionSource GetAwaiter() { return this; } + + + private static void RunCallback(object callback) + { + ((Action)callback)(); + } + + void INotifyCompletion.OnCompleted(Action continuation) + { + throw new NotImplementedException(); + } + + [System.Security.SecurityCritical] + void ICriticalNotifyCompletion.UnsafeOnCompleted(Action continuation) + { + if (_continuation == CALLBACK_RAN || + Interlocked.CompareExchange( + ref _continuation, continuation, null) == CALLBACK_RAN) + { + CompleteCallback(continuation); + } + } + public TResult GetResult() + { + if (_error != null) + { + throw _error; + } + return _result; + } + + public void Wait() + { + _semaphore = new SemaphoreSlim(0); + + Interlocked.MemoryBarrier(); + + if (_isCompleted) + { + return; + } + + _semaphore.Wait(); + } + } +} diff --git a/src/Microsoft.AspNet.Server.Kestrel/Infrastructure/KestrelThread.cs b/src/Microsoft.AspNet.Server.Kestrel/Infrastructure/KestrelThread.cs index 6c26ec71b..afaf866d4 100644 --- a/src/Microsoft.AspNet.Server.Kestrel/Infrastructure/KestrelThread.cs +++ b/src/Microsoft.AspNet.Server.Kestrel/Infrastructure/KestrelThread.cs @@ -58,13 +58,13 @@ public Task StartAsync() public void Stop(TimeSpan timeout) { - Post(OnStop, null); + Post((thread) => thread.OnStop(), this); if (!_thread.Join((int)timeout.TotalMilliseconds)) { - Post(OnStopRude, null); + Post((thread) => thread.OnStopRude(), this); if (!_thread.Join((int)timeout.TotalMilliseconds)) { - Post(OnStopImmediate, null); + Post((thread) => thread.OnStopImmediate(), this); if (!_thread.Join((int)timeout.TotalMilliseconds)) { #if NET451 @@ -79,12 +79,12 @@ public void Stop(TimeSpan timeout) } } - private void OnStop(object obj) + private void OnStop() { _post.Unreference(); } - private void OnStopRude(object obj) + private void OnStopRude() { _engine.Libuv.walk( _loop, @@ -99,7 +99,7 @@ private void OnStopRude(object obj) IntPtr.Zero); } - private void OnStopImmediate(object obj) + private void OnStopImmediate() { _stopImmediate = true; _loop.Stop(); @@ -120,7 +120,7 @@ public void Post(Action callback, T state) { _workAdding.Enqueue(new Work { - CallbackAdapter = (callback2, state2) => ((Action)callback2).Invoke((T)state2), + CallbackAdapter = CallbackAdapter, Callback = callback, State = state }); @@ -128,6 +128,7 @@ public void Post(Action callback, T state) _post.Send(); } + public Task PostAsync(Action callback, object state) { var tcs = new TaskCompletionSource(); @@ -152,7 +153,7 @@ public Task PostAsync(Action callback, T state) { _workAdding.Enqueue(new Work { - CallbackAdapter = (state1, state2) => ((Action)state1).Invoke((T)state2), + CallbackAdapter = CallbackAdapter, Callback = callback, State = state, Completion = tcs @@ -162,6 +163,11 @@ public Task PostAsync(Action callback, T state) return tcs.Task; } + private static void CallbackAdapter(object callback, object state) + { + ((Action)callback).Invoke((T)state); + } + public void Send(Action callback, object state) { if (_loop.ThreadId == Thread.CurrentThread.ManagedThreadId) @@ -271,7 +277,12 @@ private void DoPostWork() { if (work.Completion != null) { - ThreadPool.QueueUserWorkItem(_ => work.Completion.SetException(ex), null); + ThreadPool.QueueUserWorkItem( + tcs => + { + ((TaskCompletionSource)tcs).SetException(ex); + }, + work.Completion); } else { diff --git a/src/Microsoft.AspNet.Server.Kestrel/Infrastructure/QuadStateCompletionSource.cs b/src/Microsoft.AspNet.Server.Kestrel/Infrastructure/QuadStateCompletionSource.cs new file mode 100644 index 000000000..9fe50a24a --- /dev/null +++ b/src/Microsoft.AspNet.Server.Kestrel/Infrastructure/QuadStateCompletionSource.cs @@ -0,0 +1,123 @@ +// 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; +using System.Runtime.CompilerServices; +using System.Threading; + +namespace Microsoft.AspNet.Server.Kestrel.Infrastructure +{ + internal sealed class QuadStateCompletionSource : INotifyCompletion, ICriticalNotifyCompletion + { + private readonly static Action CALLBACK_RAN = () => { }; + private bool _isCompleted; + private Action _continuation; + private SemaphoreSlim _semaphore; + + private Exception _error; + private TResult _result; + + public QuadStateCompletionSource(TState1 state1, TState2 state2) + { + State1 = state1; + State2 = state2; + } + + public TState1 State1 { get; set; } + public TState2 State2 { get; set; } + public TState3 State3 { get; set; } + public TState4 State4 { get; set; } + + public bool IsCompleted { get { return _isCompleted; } } + + public void SetResult(TResult result) + { + _result = result; + _isCompleted = true; + + Complete(); + + Interlocked.MemoryBarrier(); + + if (_semaphore != null) + { + _semaphore.Release(); + } + } + + public void SetException(Exception ex) + { + _error = ex; + _isCompleted = true; + + Complete(); + + Interlocked.MemoryBarrier(); + + if (_semaphore != null) + { + _semaphore.Release(); + } + } + + private void Complete() + { + Action continuation = _continuation ?? Interlocked.CompareExchange(ref _continuation, CALLBACK_RAN, null); + if (continuation != null) + { + CompleteCallback(continuation); + } + } + + private static void CompleteCallback(Action continuation) + { + ThreadPool.QueueUserWorkItem((obj) => RunCallback(obj), continuation); + } + + public QuadStateCompletionSource GetAwaiter() { return this; } + + + private static void RunCallback(object callback) + { + ((Action)callback)(); + } + + void INotifyCompletion.OnCompleted(Action continuation) + { + throw new NotImplementedException(); + } + + [System.Security.SecurityCritical] + void ICriticalNotifyCompletion.UnsafeOnCompleted(Action continuation) + { + if (_continuation == CALLBACK_RAN || + Interlocked.CompareExchange( + ref _continuation, continuation, null) == CALLBACK_RAN) + { + CompleteCallback(continuation); + } + } + public TResult GetResult() + { + if (_error != null) + { + throw _error; + } + return _result; + } + + public void Wait() + { + _semaphore = new SemaphoreSlim(0); + + Interlocked.MemoryBarrier(); + + if (_isCompleted) + { + return; + } + + _semaphore.Wait(); + } + } +} diff --git a/src/Microsoft.AspNet.Server.Kestrel/Infrastructure/StateCompletionSource.cs b/src/Microsoft.AspNet.Server.Kestrel/Infrastructure/StateCompletionSource.cs new file mode 100644 index 000000000..8fe208fdd --- /dev/null +++ b/src/Microsoft.AspNet.Server.Kestrel/Infrastructure/StateCompletionSource.cs @@ -0,0 +1,120 @@ +// 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; +using System.Runtime.CompilerServices; +using System.Threading; + +namespace Microsoft.AspNet.Server.Kestrel.Infrastructure +{ + public sealed class StateCompletionSource : INotifyCompletion, ICriticalNotifyCompletion + { + private readonly static Action CALLBACK_RAN = () => { }; + private bool _isCompleted; + private Action _continuation; + private SemaphoreSlim _semaphore; + + private Exception _error; + + private TResult _result; + + public StateCompletionSource(TState state) + { + State = state; + } + + public TState State { get; set; } + + public bool IsCompleted { get { return _isCompleted; } } + + public void SetResult(TResult result) + { + _result = result; + _isCompleted = true; + + Complete(); + + Interlocked.MemoryBarrier(); + + if (_semaphore != null) + { + _semaphore.Release(); + } + } + + public void SetException(Exception ex) + { + _error = ex; + _isCompleted = true; + + Complete(); + + Interlocked.MemoryBarrier(); + + if (_semaphore != null) + { + _semaphore.Release(); + } + } + + private void Complete() + { + Action continuation = _continuation ?? Interlocked.CompareExchange(ref _continuation, CALLBACK_RAN, null); + if (continuation != null) + { + CompleteCallback(continuation); + } + } + + private static void CompleteCallback(Action continuation) + { + ThreadPool.QueueUserWorkItem((obj) => RunCallback(obj), continuation); + } + + public StateCompletionSource GetAwaiter() { return this; } + + + private static void RunCallback(object callback) + { + ((Action)callback)(); + } + + void INotifyCompletion.OnCompleted(Action continuation) + { + throw new NotImplementedException(); + } + + [System.Security.SecurityCritical] + void ICriticalNotifyCompletion.UnsafeOnCompleted(Action continuation) + { + if (_continuation == CALLBACK_RAN || + Interlocked.CompareExchange( + ref _continuation, continuation, null) == CALLBACK_RAN) + { + CompleteCallback(continuation); + } + } + public TResult GetResult() + { + if (_error != null) + { + throw _error; + } + return _result; + } + + public void Wait() + { + _semaphore = new SemaphoreSlim(0); + + Interlocked.MemoryBarrier(); + + if (_isCompleted) + { + return; + } + + _semaphore.Wait(); + } + } +} diff --git a/src/Microsoft.AspNet.Server.Kestrel/KestrelEngine.cs b/src/Microsoft.AspNet.Server.Kestrel/KestrelEngine.cs index 9a68cd9df..bd0410652 100644 --- a/src/Microsoft.AspNet.Server.Kestrel/KestrelEngine.cs +++ b/src/Microsoft.AspNet.Server.Kestrel/KestrelEngine.cs @@ -7,10 +7,8 @@ using System.Threading.Tasks; using Microsoft.AspNet.Http.Features; using Microsoft.AspNet.Server.Kestrel.Http; -using Microsoft.AspNet.Server.Kestrel.Infrastructure; using Microsoft.AspNet.Server.Kestrel.Networking; using Microsoft.Extensions.PlatformAbstractions; -using Microsoft.Extensions.Logging; namespace Microsoft.AspNet.Server.Kestrel { diff --git a/src/Microsoft.AspNet.Server.Kestrel/Networking/UvAsyncHandle.cs b/src/Microsoft.AspNet.Server.Kestrel/Networking/UvAsyncHandle.cs index 04c2ecce3..9b4ecb60f 100644 --- a/src/Microsoft.AspNet.Server.Kestrel/Networking/UvAsyncHandle.cs +++ b/src/Microsoft.AspNet.Server.Kestrel/Networking/UvAsyncHandle.cs @@ -8,7 +8,7 @@ namespace Microsoft.AspNet.Server.Kestrel.Networking { public class UvAsyncHandle : UvHandle { - private static Libuv.uv_async_cb _uv_async_cb = AsyncCb; + private static Libuv.uv_async_cb _uv_async_cb = (handle) => AsyncCb(handle); private Action _callback; public UvAsyncHandle(IKestrelTrace logger) : base(logger) diff --git a/src/Microsoft.AspNet.Server.Kestrel/Networking/UvConnectRequest.cs b/src/Microsoft.AspNet.Server.Kestrel/Networking/UvConnectRequest.cs index 0c0ce04b4..a9af611e5 100644 --- a/src/Microsoft.AspNet.Server.Kestrel/Networking/UvConnectRequest.cs +++ b/src/Microsoft.AspNet.Server.Kestrel/Networking/UvConnectRequest.cs @@ -12,7 +12,7 @@ namespace Microsoft.AspNet.Server.Kestrel.Networking /// public class UvConnectRequest : UvRequest { - private readonly static Libuv.uv_connect_cb _uv_connect_cb = UvConnectCb; + private readonly static Libuv.uv_connect_cb _uv_connect_cb = (ptr, status) => UvConnectCb(ptr, status); private Action _callback; private object _state; diff --git a/src/Microsoft.AspNet.Server.Kestrel/Networking/UvHandle.cs b/src/Microsoft.AspNet.Server.Kestrel/Networking/UvHandle.cs index 4b90ecfd5..f0c51549a 100644 --- a/src/Microsoft.AspNet.Server.Kestrel/Networking/UvHandle.cs +++ b/src/Microsoft.AspNet.Server.Kestrel/Networking/UvHandle.cs @@ -9,7 +9,7 @@ namespace Microsoft.AspNet.Server.Kestrel.Networking { public abstract class UvHandle : UvMemory { - private static Libuv.uv_close_cb _destroyMemory = DestroyMemory; + private static Libuv.uv_close_cb _destroyMemory = (memory) => DestroyMemory(memory); private Action, IntPtr> _queueCloseHandle; protected UvHandle(IKestrelTrace logger) : base (logger) diff --git a/src/Microsoft.AspNet.Server.Kestrel/Networking/UvShutdownReq.cs b/src/Microsoft.AspNet.Server.Kestrel/Networking/UvShutdownReq.cs index d6aa2c65a..3d10d99f1 100644 --- a/src/Microsoft.AspNet.Server.Kestrel/Networking/UvShutdownReq.cs +++ b/src/Microsoft.AspNet.Server.Kestrel/Networking/UvShutdownReq.cs @@ -11,7 +11,7 @@ namespace Microsoft.AspNet.Server.Kestrel.Networking /// public class UvShutdownReq : UvRequest { - private readonly static Libuv.uv_shutdown_cb _uv_shutdown_cb = UvShutdownCb; + private readonly static Libuv.uv_shutdown_cb _uv_shutdown_cb = (ptr, status) => UvShutdownCb(ptr, status); private Action _callback; private object _state; diff --git a/src/Microsoft.AspNet.Server.Kestrel/Networking/UvStreamHandle.cs b/src/Microsoft.AspNet.Server.Kestrel/Networking/UvStreamHandle.cs index e6ffafe24..164d72169 100644 --- a/src/Microsoft.AspNet.Server.Kestrel/Networking/UvStreamHandle.cs +++ b/src/Microsoft.AspNet.Server.Kestrel/Networking/UvStreamHandle.cs @@ -10,7 +10,7 @@ namespace Microsoft.AspNet.Server.Kestrel.Networking { public abstract class UvStreamHandle : UvHandle { - private readonly static Libuv.uv_connection_cb _uv_connection_cb = UvConnectionCb; + private readonly static Libuv.uv_connection_cb _uv_connection_cb = (ptr, status) => UvConnectionCb(ptr, status); private readonly static Libuv.uv_alloc_cb _uv_alloc_cb = UvAllocCb; private readonly static Libuv.uv_read_cb _uv_read_cb = UvReadCb; diff --git a/src/Microsoft.AspNet.Server.Kestrel/Networking/UvWriteReq.cs b/src/Microsoft.AspNet.Server.Kestrel/Networking/UvWriteReq.cs index 103ddfc6e..ea4b308c6 100644 --- a/src/Microsoft.AspNet.Server.Kestrel/Networking/UvWriteReq.cs +++ b/src/Microsoft.AspNet.Server.Kestrel/Networking/UvWriteReq.cs @@ -14,7 +14,7 @@ namespace Microsoft.AspNet.Server.Kestrel.Networking /// public class UvWriteReq : UvRequest { - private readonly static Libuv.uv_write_cb _uv_write_cb = UvWriteCb; + private readonly static Libuv.uv_write_cb _uv_write_cb = (ptr, staus) => UvWriteCb(ptr, staus); private IntPtr _bufs; diff --git a/test/Microsoft.AspNet.Server.KestrelTests/EngineTests.cs b/test/Microsoft.AspNet.Server.KestrelTests/EngineTests.cs index 633c131c5..c2d2ad06b 100644 --- a/test/Microsoft.AspNet.Server.KestrelTests/EngineTests.cs +++ b/test/Microsoft.AspNet.Server.KestrelTests/EngineTests.cs @@ -876,6 +876,52 @@ await connection.ReceiveEnd( } } + [Theory] + [MemberData(nameof(ConnectionFilterData))] + public async Task RequestBodyIsConsumedAutomaticallyIfAppDoesntConsumeItFully(ServiceContext testContext) + { + using (var server = new TestServer(async frame => + { + var response = frame.Get(); + var request = frame.Get(); + + Assert.Equal("POST", request.Method); + + response.Headers.Clear(); + response.Headers["Content-Length"] = new[] { "11" }; + + await response.Body.WriteAsync(Encoding.ASCII.GetBytes("Hello World"), 0, 11); + }, testContext)) + { + using (var connection = new TestConnection()) + { + await connection.SendEnd( + "POST / HTTP/1.1", + "Content-Length: 5", + "", + "HelloPOST / HTTP/1.1", + "Transfer-Encoding: chunked", + "", + "C", "HelloChunked", "0", + "POST / HTTP/1.1", + "Content-Length: 7", + "", + "Goodbye"); + await connection.ReceiveEnd( + "HTTP/1.1 200 OK", + "Content-Length: 11", + "", + "Hello WorldHTTP/1.1 200 OK", + "Content-Length: 11", + "", + "Hello WorldHTTP/1.1 200 OK", + "Content-Length: 11", + "", + "Hello World"); + } + } + } + private class TestApplicationErrorLogger : ILogger { public int ApplicationErrorsLogged { get; set; } diff --git a/test/Microsoft.AspNet.Server.KestrelTests/FrameTests.cs b/test/Microsoft.AspNet.Server.KestrelTests/FrameTests.cs index ec73fcd3b..8266a1c2e 100644 --- a/test/Microsoft.AspNet.Server.KestrelTests/FrameTests.cs +++ b/test/Microsoft.AspNet.Server.KestrelTests/FrameTests.cs @@ -1,9 +1,11 @@ // 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 Microsoft.AspNet.Server.Kestrel.Http; +using System; using System.Linq; using System.Text; +using Microsoft.AspNet.Server.Kestrel.Http; +using Microsoft.AspNet.Server.Kestrel.Infrastructure; using Xunit; namespace Microsoft.AspNet.Server.KestrelTests @@ -34,5 +36,34 @@ public void ChunkedPrefixMustBeHexCrLfWithoutLeadingZeros(int dataCount, string Assert.Equal(Encoding.ASCII.GetBytes(expected), beginChunkBytes.ToArray()); } + + [Theory] + [InlineData("Cookie: \r\n\r\n", 1)] + [InlineData("Cookie:\r\n\r\n", 1)] + [InlineData("Cookie:\r\n value\r\n\r\n", 1)] + [InlineData("Cookie\r\n", 0)] + [InlineData("Cookie: \r\nConnection: close\r\n\r\n", 2)] + [InlineData("Connection: close\r\nCookie: \r\n\r\n", 2)] + [InlineData("Connection: close\r\nCookie \r\n", 1)] + [InlineData("Connection:\r\n \r\nCookie \r\n", 1)] + public void EmptyHeaderValuesCanBeParsed(string rawHeaders, int numHeaders) + { + var socketInput = new SocketInput(new MemoryPool2()); + var headerCollection = new FrameRequestHeaders(); + + var headerArray = Encoding.ASCII.GetBytes(rawHeaders); + var inputBuffer = socketInput.IncomingStart(headerArray.Length); + Buffer.BlockCopy(headerArray, 0, inputBuffer.Data.Array, inputBuffer.Data.Offset, headerArray.Length); + socketInput.IncomingComplete(headerArray.Length, null); + + var success = Frame.TakeMessageHeaders(socketInput, headerCollection); + + Assert.True(success); + Assert.Equal(numHeaders, headerCollection.Count()); + + // Assert TakeMessageHeaders consumed all the input + var scan = socketInput.ConsumingStart(); + Assert.True(scan.IsEnd); + } } }