From b1e8f0cdea2564ebbeb9dbe28f8c09f601f79ec2 Mon Sep 17 00:00:00 2001 From: Ben Adams Date: Thu, 3 Dec 2015 09:38:30 +0000 Subject: [PATCH] Add IThreadPool interface --- .../Filter/FilteredStreamAdapter.cs | 4 ++-- .../Http/Connection.cs | 8 ++++---- .../Http/SocketInput.cs | 12 +++++------ .../Http/SocketOutput.cs | 10 +++++----- .../Infrastructure/IThreadPool.cs | 15 ++++++++++++++ .../Infrastructure/KestrelThread.cs | 8 ++++---- ...eadPoolActions.cs => LoggingThreadPool.cs} | 5 ++--- .../KestrelServer.cs | 2 +- .../ServiceContext.cs | 4 ++-- .../FrameTests.cs | 4 ++-- .../SocketOutputTests.cs | 20 +++++++++---------- .../TestInput.cs | 4 ++-- .../TestServiceContext.cs | 2 ++ 13 files changed, 57 insertions(+), 41 deletions(-) create mode 100644 src/Microsoft.AspNet.Server.Kestrel/Infrastructure/IThreadPool.cs rename src/Microsoft.AspNet.Server.Kestrel/Infrastructure/{ThreadPoolActions.cs => LoggingThreadPool.cs} (93%) diff --git a/src/Microsoft.AspNet.Server.Kestrel/Filter/FilteredStreamAdapter.cs b/src/Microsoft.AspNet.Server.Kestrel/Filter/FilteredStreamAdapter.cs index 3266d88d64..1409940c86 100644 --- a/src/Microsoft.AspNet.Server.Kestrel/Filter/FilteredStreamAdapter.cs +++ b/src/Microsoft.AspNet.Server.Kestrel/Filter/FilteredStreamAdapter.cs @@ -20,9 +20,9 @@ namespace Microsoft.AspNet.Server.Kestrel.Filter Stream filteredStream, MemoryPool2 memory, IKestrelTrace logger, - ThreadPoolActions threadPoolActions) + IThreadPool threadPool) { - SocketInput = new SocketInput(memory, threadPoolActions); + SocketInput = new SocketInput(memory, threadPool); SocketOutput = new StreamSocketOutput(filteredStream, memory); _log = logger; diff --git a/src/Microsoft.AspNet.Server.Kestrel/Http/Connection.cs b/src/Microsoft.AspNet.Server.Kestrel/Http/Connection.cs index dd7236f523..b49ffd4316 100644 --- a/src/Microsoft.AspNet.Server.Kestrel/Http/Connection.cs +++ b/src/Microsoft.AspNet.Server.Kestrel/Http/Connection.cs @@ -41,8 +41,8 @@ namespace Microsoft.AspNet.Server.Kestrel.Http _connectionId = Interlocked.Increment(ref _lastConnectionId); - _rawSocketInput = new SocketInput(Memory2, ThreadPoolActions); - _rawSocketOutput = new SocketOutput(Thread, _socket, Memory2, this, _connectionId, Log, ThreadPoolActions); + _rawSocketInput = new SocketInput(Memory2, ThreadPool); + _rawSocketOutput = new SocketOutput(Thread, _socket, Memory2, this, _connectionId, Log, ThreadPool); } public void Start() @@ -114,7 +114,7 @@ namespace Microsoft.AspNet.Server.Kestrel.Http { // Frame.Abort calls user code while this method is always // called from a libuv thread. - ThreadPool.QueueUserWorkItem(state => + System.Threading.ThreadPool.QueueUserWorkItem(state => { var connection = (Connection)state; connection._frame.Abort(); @@ -124,7 +124,7 @@ namespace Microsoft.AspNet.Server.Kestrel.Http private void ApplyConnectionFilter() { - var filteredStreamAdapter = new FilteredStreamAdapter(_filterContext.Connection, Memory2, Log, ThreadPoolActions); + var filteredStreamAdapter = new FilteredStreamAdapter(_filterContext.Connection, Memory2, Log, ThreadPool); SocketInput = filteredStreamAdapter.SocketInput; SocketOutput = filteredStreamAdapter.SocketOutput; diff --git a/src/Microsoft.AspNet.Server.Kestrel/Http/SocketInput.cs b/src/Microsoft.AspNet.Server.Kestrel/Http/SocketInput.cs index a85634331d..5f850b5d68 100644 --- a/src/Microsoft.AspNet.Server.Kestrel/Http/SocketInput.cs +++ b/src/Microsoft.AspNet.Server.Kestrel/Http/SocketInput.cs @@ -16,7 +16,7 @@ namespace Microsoft.AspNet.Server.Kestrel.Http private static readonly Action _awaitableIsNotCompleted = () => { }; private readonly MemoryPool2 _memory; - private readonly ThreadPoolActions _threadPoolActions; + private readonly IThreadPool _threadPool; private readonly ManualResetEventSlim _manualResetEvent = new ManualResetEventSlim(false); private Action _awaitableState; @@ -27,10 +27,10 @@ namespace Microsoft.AspNet.Server.Kestrel.Http private MemoryPoolBlock2 _pinned; private readonly object _sync = new Object(); - public SocketInput(MemoryPool2 memory, ThreadPoolActions threadPoolActions) + public SocketInput(MemoryPool2 memory, IThreadPool threadPool) { _memory = memory; - _threadPoolActions = threadPoolActions; + _threadPool = threadPool; _awaitableState = _awaitableIsNotCompleted; } @@ -130,7 +130,7 @@ namespace Microsoft.AspNet.Server.Kestrel.Http if (awaitableState != _awaitableIsCompleted && awaitableState != _awaitableIsNotCompleted) { - _threadPoolActions.Run(awaitableState); + _threadPool.Run(awaitableState); } } @@ -190,7 +190,7 @@ namespace Microsoft.AspNet.Server.Kestrel.Http if (awaitableState != _awaitableIsCompleted && awaitableState != _awaitableIsNotCompleted) { - _threadPoolActions.Run(awaitableState); + _threadPool.Run(awaitableState); } } @@ -212,7 +212,7 @@ namespace Microsoft.AspNet.Server.Kestrel.Http } else if (awaitableState == _awaitableIsCompleted) { - _threadPoolActions.Run(continuation); + _threadPool.Run(continuation); } else { diff --git a/src/Microsoft.AspNet.Server.Kestrel/Http/SocketOutput.cs b/src/Microsoft.AspNet.Server.Kestrel/Http/SocketOutput.cs index ef6b5e4f1f..64c5808efb 100644 --- a/src/Microsoft.AspNet.Server.Kestrel/Http/SocketOutput.cs +++ b/src/Microsoft.AspNet.Server.Kestrel/Http/SocketOutput.cs @@ -25,7 +25,7 @@ namespace Microsoft.AspNet.Server.Kestrel.Http private readonly Connection _connection; private readonly long _connectionId; private readonly IKestrelTrace _log; - private readonly ThreadPoolActions _threadPoolActions; + private readonly IThreadPool _threadPool; // This locks all access to _tail, _isProducing and _returnFromOnProducingComplete. // _head does not require a lock, since it is only used in the ctor and uv thread. @@ -57,14 +57,14 @@ namespace Microsoft.AspNet.Server.Kestrel.Http Connection connection, long connectionId, IKestrelTrace log, - ThreadPoolActions threadPoolActions) + IThreadPool threadPool) { _thread = thread; _socket = socket; _connection = connection; _connectionId = connectionId; _log = log; - _threadPoolActions = threadPoolActions; + _threadPool = threadPool; _tasksPending = new Queue>(_initialTaskQueues); _tasksCompleted = new Queue>(_initialTaskQueues); @@ -322,11 +322,11 @@ namespace Microsoft.AspNet.Server.Kestrel.Http var tcs = _tasksCompleted.Dequeue(); if (_lastWriteError == null) { - _threadPoolActions.Complete(tcs); + _threadPool.Complete(tcs); } else { - _threadPoolActions.Error(tcs, _lastWriteError); + _threadPool.Error(tcs, _lastWriteError); } } diff --git a/src/Microsoft.AspNet.Server.Kestrel/Infrastructure/IThreadPool.cs b/src/Microsoft.AspNet.Server.Kestrel/Infrastructure/IThreadPool.cs new file mode 100644 index 0000000000..fde95f4241 --- /dev/null +++ b/src/Microsoft.AspNet.Server.Kestrel/Infrastructure/IThreadPool.cs @@ -0,0 +1,15 @@ +// 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.Threading.Tasks; + +namespace Microsoft.AspNet.Server.Kestrel.Infrastructure +{ + public interface IThreadPool + { + void Complete(TaskCompletionSource tcs); + void Error(TaskCompletionSource tcs, Exception ex); + void Run(Action action); + } +} \ No newline at end of file diff --git a/src/Microsoft.AspNet.Server.Kestrel/Infrastructure/KestrelThread.cs b/src/Microsoft.AspNet.Server.Kestrel/Infrastructure/KestrelThread.cs index 98c9ef3f4f..64a1c9b81f 100644 --- a/src/Microsoft.AspNet.Server.Kestrel/Infrastructure/KestrelThread.cs +++ b/src/Microsoft.AspNet.Server.Kestrel/Infrastructure/KestrelThread.cs @@ -33,14 +33,14 @@ namespace Microsoft.AspNet.Server.Kestrel private bool _initCompleted = false; private ExceptionDispatchInfo _closeError; private IKestrelTrace _log; - private ThreadPoolActions _threadPoolActions; + private IThreadPool _threadPool; public KestrelThread(KestrelEngine engine) { _engine = engine; _appLifetime = engine.AppLifetime; _log = engine.Log; - _threadPoolActions = engine.ThreadPoolActions; + _threadPool = engine.ThreadPool; _loop = new UvLoopHandle(_log); _post = new UvAsyncHandle(_log); _thread = new Thread(ThreadStart); @@ -268,14 +268,14 @@ namespace Microsoft.AspNet.Server.Kestrel work.CallbackAdapter(work.Callback, work.State); if (work.Completion != null) { - _threadPoolActions.Complete(work.Completion); + _threadPool.Complete(work.Completion); } } catch (Exception ex) { if (work.Completion != null) { - _threadPoolActions.Error(work.Completion, ex); + _threadPool.Error(work.Completion, ex); } else { diff --git a/src/Microsoft.AspNet.Server.Kestrel/Infrastructure/ThreadPoolActions.cs b/src/Microsoft.AspNet.Server.Kestrel/Infrastructure/LoggingThreadPool.cs similarity index 93% rename from src/Microsoft.AspNet.Server.Kestrel/Infrastructure/ThreadPoolActions.cs rename to src/Microsoft.AspNet.Server.Kestrel/Infrastructure/LoggingThreadPool.cs index cbcba6c239..967787dc1f 100644 --- a/src/Microsoft.AspNet.Server.Kestrel/Infrastructure/ThreadPoolActions.cs +++ b/src/Microsoft.AspNet.Server.Kestrel/Infrastructure/LoggingThreadPool.cs @@ -4,18 +4,17 @@ using System; using System.Threading; using System.Threading.Tasks; -using Microsoft.AspNet.Server.Kestrel.Http; namespace Microsoft.AspNet.Server.Kestrel.Infrastructure { - public class ThreadPoolActions + public class LoggingThreadPool : IThreadPool { private readonly IKestrelTrace _log; private readonly WaitCallback _runAction; private readonly WaitCallback _completeTcs; - public ThreadPoolActions(IKestrelTrace log) + public LoggingThreadPool(IKestrelTrace log) { _log = log; diff --git a/src/Microsoft.AspNet.Server.Kestrel/KestrelServer.cs b/src/Microsoft.AspNet.Server.Kestrel/KestrelServer.cs index 00d4605ba2..b8e9c5dd07 100644 --- a/src/Microsoft.AspNet.Server.Kestrel/KestrelServer.cs +++ b/src/Microsoft.AspNet.Server.Kestrel/KestrelServer.cs @@ -64,7 +64,7 @@ namespace Microsoft.AspNet.Server.Kestrel }, AppLifetime = _applicationLifetime, Log = trace, - ThreadPoolActions = new ThreadPoolActions(trace), + ThreadPool = new LoggingThreadPool(trace), DateHeaderValueManager = dateHeaderValueManager, ConnectionFilter = information.ConnectionFilter, NoDelay = information.NoDelay diff --git a/src/Microsoft.AspNet.Server.Kestrel/ServiceContext.cs b/src/Microsoft.AspNet.Server.Kestrel/ServiceContext.cs index 531750f3a3..31eeea898d 100644 --- a/src/Microsoft.AspNet.Server.Kestrel/ServiceContext.cs +++ b/src/Microsoft.AspNet.Server.Kestrel/ServiceContext.cs @@ -21,7 +21,7 @@ namespace Microsoft.AspNet.Server.Kestrel { AppLifetime = context.AppLifetime; Log = context.Log; - ThreadPoolActions = context.ThreadPoolActions; + ThreadPool = context.ThreadPool; FrameFactory = context.FrameFactory; DateHeaderValueManager = context.DateHeaderValueManager; ConnectionFilter = context.ConnectionFilter; @@ -32,7 +32,7 @@ namespace Microsoft.AspNet.Server.Kestrel public IKestrelTrace Log { get; set; } - public ThreadPoolActions ThreadPoolActions { get; set; } + public IThreadPool ThreadPool { get; set; } public Func, Frame> FrameFactory { get; set; } diff --git a/test/Microsoft.AspNet.Server.KestrelTests/FrameTests.cs b/test/Microsoft.AspNet.Server.KestrelTests/FrameTests.cs index b6267e384c..cc9bc26618 100644 --- a/test/Microsoft.AspNet.Server.KestrelTests/FrameTests.cs +++ b/test/Microsoft.AspNet.Server.KestrelTests/FrameTests.cs @@ -50,8 +50,8 @@ namespace Microsoft.AspNet.Server.KestrelTests public void EmptyHeaderValuesCanBeParsed(string rawHeaders, int numHeaders) { var trace = new KestrelTrace(new TestKestrelTrace()); - var tpa = new ThreadPoolActions(trace); - var socketInput = new SocketInput(new MemoryPool2(), tpa); + var ltp = new LoggingThreadPool(trace); + var socketInput = new SocketInput(new MemoryPool2(), ltp); var headerCollection = new FrameRequestHeaders(); var headerArray = Encoding.ASCII.GetBytes(rawHeaders); diff --git a/test/Microsoft.AspNet.Server.KestrelTests/SocketOutputTests.cs b/test/Microsoft.AspNet.Server.KestrelTests/SocketOutputTests.cs index 4028cfff1d..dca77cf881 100644 --- a/test/Microsoft.AspNet.Server.KestrelTests/SocketOutputTests.cs +++ b/test/Microsoft.AspNet.Server.KestrelTests/SocketOutputTests.cs @@ -41,8 +41,8 @@ namespace Microsoft.AspNet.Server.KestrelTests var kestrelThread = kestrelEngine.Threads[0]; var socket = new MockSocket(kestrelThread.Loop.ThreadId, new TestKestrelTrace()); var trace = new KestrelTrace(new TestKestrelTrace()); - var tpa = new ThreadPoolActions(trace); - var socketOutput = new SocketOutput(kestrelThread, socket, memory, null, 0, trace, tpa); + var ltp = new LoggingThreadPool(trace); + var socketOutput = new SocketOutput(kestrelThread, socket, memory, null, 0, trace, ltp); // I doubt _maxBytesPreCompleted will ever be over a MB. If it is, we should change this test. var bufferSize = 1048576; @@ -88,8 +88,8 @@ namespace Microsoft.AspNet.Server.KestrelTests var kestrelThread = kestrelEngine.Threads[0]; var socket = new MockSocket(kestrelThread.Loop.ThreadId, new TestKestrelTrace()); var trace = new KestrelTrace(new TestKestrelTrace()); - var tpa = new ThreadPoolActions(trace); - var socketOutput = new SocketOutput(kestrelThread, socket, memory, null, 0, trace, tpa); + var ltp = new LoggingThreadPool(trace); + var socketOutput = new SocketOutput(kestrelThread, socket, memory, null, 0, trace, ltp); var bufferSize = maxBytesPreCompleted; var buffer = new ArraySegment(new byte[bufferSize], 0, bufferSize); @@ -145,8 +145,8 @@ namespace Microsoft.AspNet.Server.KestrelTests var kestrelThread = kestrelEngine.Threads[0]; var socket = new MockSocket(kestrelThread.Loop.ThreadId, new TestKestrelTrace()); var trace = new KestrelTrace(new TestKestrelTrace()); - var tpa = new ThreadPoolActions(trace); - var socketOutput = new SocketOutput(kestrelThread, socket, memory, null, 0, trace, tpa); + var ltp = new LoggingThreadPool(trace); + var socketOutput = new SocketOutput(kestrelThread, socket, memory, null, 0, trace, ltp); var bufferSize = maxBytesPreCompleted; @@ -226,8 +226,8 @@ namespace Microsoft.AspNet.Server.KestrelTests var kestrelThread = kestrelEngine.Threads[0]; var socket = new MockSocket(kestrelThread.Loop.ThreadId, new TestKestrelTrace()); var trace = new KestrelTrace(new TestKestrelTrace()); - var tpa = new ThreadPoolActions(trace); - var socketOutput = new SocketOutput(kestrelThread, socket, memory, null, 0, trace, tpa); + var ltp = new LoggingThreadPool(trace); + var socketOutput = new SocketOutput(kestrelThread, socket, memory, null, 0, trace, ltp); var bufferSize = maxBytesPreCompleted; var buffer = new ArraySegment(new byte[bufferSize], 0, bufferSize); @@ -303,8 +303,8 @@ namespace Microsoft.AspNet.Server.KestrelTests var kestrelThread = kestrelEngine.Threads[0]; var socket = new MockSocket(kestrelThread.Loop.ThreadId, new TestKestrelTrace()); var trace = new KestrelTrace(new TestKestrelTrace()); - var tpa = new ThreadPoolActions(trace); - var socketOutput = new SocketOutput(kestrelThread, socket, memory, null, 0, trace, tpa); + var ltp = new LoggingThreadPool(trace); + var socketOutput = new SocketOutput(kestrelThread, socket, memory, null, 0, trace, ltp); // block 1 var start = socketOutput.ProducingStart(); diff --git a/test/Microsoft.AspNet.Server.KestrelTests/TestInput.cs b/test/Microsoft.AspNet.Server.KestrelTests/TestInput.cs index 2337510d32..827d47ba84 100644 --- a/test/Microsoft.AspNet.Server.KestrelTests/TestInput.cs +++ b/test/Microsoft.AspNet.Server.KestrelTests/TestInput.cs @@ -15,11 +15,11 @@ namespace Microsoft.AspNet.Server.KestrelTests public TestInput() { var trace = new KestrelTrace(new TestKestrelTrace()); - var tpa = new ThreadPoolActions(trace); + var ltp = new LoggingThreadPool(trace); var memory2 = new MemoryPool2(); FrameContext = new FrameContext { - SocketInput = new SocketInput(memory2, tpa), + SocketInput = new SocketInput(memory2, ltp), ConnectionControl = this, FrameControl = this }; diff --git a/test/Microsoft.AspNet.Server.KestrelTests/TestServiceContext.cs b/test/Microsoft.AspNet.Server.KestrelTests/TestServiceContext.cs index 9686915699..54dfc3c3b3 100644 --- a/test/Microsoft.AspNet.Server.KestrelTests/TestServiceContext.cs +++ b/test/Microsoft.AspNet.Server.KestrelTests/TestServiceContext.cs @@ -4,6 +4,7 @@ using Microsoft.AspNet.Http; using Microsoft.AspNet.Server.Kestrel; using Microsoft.AspNet.Server.Kestrel.Http; +using Microsoft.AspNet.Server.Kestrel.Infrastructure; namespace Microsoft.AspNet.Server.KestrelTests { @@ -15,6 +16,7 @@ namespace Microsoft.AspNet.Server.KestrelTests { AppLifetime = new LifetimeNotImplemented(); Log = new TestKestrelTrace(); + ThreadPool = new LoggingThreadPool(Log); DateHeaderValueManager = new TestDateHeaderValueManager(); }