Expose a UseTransportThread property on KestrelServerOptions (#1695)

- This property will force Kestrel to use whatever scheduler the transport
used when write and read callbacks are fired. The default value is false so
all calls to user code including connection adapters, and the application function,
and cancellation token callbacks.
- Transports may expose configuration that changes what the transport thread is.
- Removed InternalKestrelServerOptions.cs
- Added a configurable UseSockets overload (even though there are no options yet)
- Remove RequiresDispatch from the IConnectionInformation

samples/SampleApp/Startup.cs

@@ -44,6 +44,9 @@ namespace SampleApp
             var host = new WebHostBuilder()
                 .UseKestrel(options =>
                 {
+                    // Run callbacks on the transport thread
+                    options.UseTransportThread = true;
+
                     options.Listen(IPAddress.Loopback, 5000, listenOptions =>
                     {
                         // Uncomment the following to enable Nagle's algorithm for this endpoint.
@@ -51,6 +54,7 @@ namespace SampleApp
 
                         listenOptions.UseConnectionLogging();
                     });
+
                     options.Listen(IPAddress.Loopback, 5001, listenOptions =>
                     {
                         listenOptions.UseHttps("testCert.pfx", "testPassword");
@@ -65,14 +69,12 @@ namespace SampleApp
                 .UseLibuv(options =>
                 {
                     // Uncomment the following line to change the default number of libuv threads for all endpoints.
-                    options.ThreadCount = 4;
+                    // options.ThreadCount = 4;
                 })
                 .UseContentRoot(Directory.GetCurrentDirectory())
                 .UseStartup<Startup>()
                 .Build();
 
-            host.ServerFeatures.Get<InternalKestrelServerOptions>().ThreadPoolDispatching = false;
-
             host.Run();
         }
     }

src/Microsoft.AspNetCore.Server.Kestrel.Core/Internal/ConnectionHandler.cs

@@ -27,8 +27,8 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal
 
         public IConnectionContext OnConnection(IConnectionInformation connectionInfo)
         {
-            var inputPipe = connectionInfo.PipeFactory.Create(GetInputPipeOptions(requiresDispatch: connectionInfo.RequiresDispatch, writerScheduler: connectionInfo.InputWriterScheduler));
-            var outputPipe = connectionInfo.PipeFactory.Create(GetOutputPipeOptions(requiresDispatch: connectionInfo.RequiresDispatch, readerScheduler: connectionInfo.OutputReaderScheduler));
+            var inputPipe = connectionInfo.PipeFactory.Create(GetInputPipeOptions(connectionInfo.InputWriterScheduler));
+            var outputPipe = connectionInfo.PipeFactory.Create(GetOutputPipeOptions(connectionInfo.OutputReaderScheduler));
 
             var connectionId = CorrelationIdGenerator.GetNextId();
             var frameConnectionId = Interlocked.Increment(ref _lastFrameConnectionId);
@@ -70,18 +70,18 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal
         }
 
         // Internal for testing
-        internal PipeOptions GetInputPipeOptions(bool requiresDispatch, IScheduler writerScheduler) => new PipeOptions
+        internal PipeOptions GetInputPipeOptions(IScheduler writerScheduler) => new PipeOptions
         {
-            ReaderScheduler = (requiresDispatch ? (IScheduler)_serviceContext.ThreadPool : (IScheduler)InlineScheduler.Default),
+            ReaderScheduler = _serviceContext.ThreadPool,
             WriterScheduler = writerScheduler,
             MaximumSizeHigh = _serviceContext.ServerOptions.Limits.MaxRequestBufferSize ?? 0,
             MaximumSizeLow = _serviceContext.ServerOptions.Limits.MaxRequestBufferSize ?? 0
         };
 
-        internal PipeOptions GetOutputPipeOptions(bool requiresDispatch, IScheduler readerScheduler) => new PipeOptions
+        internal PipeOptions GetOutputPipeOptions(IScheduler readerScheduler) => new PipeOptions
         {
             ReaderScheduler = readerScheduler,
-            WriterScheduler = (requiresDispatch ? (IScheduler)_serviceContext.ThreadPool : (IScheduler)InlineScheduler.Default),
+            WriterScheduler = _serviceContext.ThreadPool,
             MaximumSizeHigh = GetOutputResponseBufferSize(),
             MaximumSizeLow = GetOutputResponseBufferSize()
         };

src/Microsoft.AspNetCore.Server.Kestrel.Core/Internal/InternalKestrelServerOptions.cs

@@ -1,12 +0,0 @@
-// 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.
-
-namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal
-{
-    public class InternalKestrelServerOptions
-    {
-        // This will likely be replace with transport-specific configuration.
-        // https://github.com/aspnet/KestrelHttpServer/issues/828
-        public bool ThreadPoolDispatching { get; set; } = true;
-    }
-}

src/Microsoft.AspNetCore.Server.Kestrel.Core/KestrelServer.cs

@@ -54,21 +54,17 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core
             }
 
             Options = options.Value ?? new KestrelServerOptions();
-            InternalOptions = new InternalKestrelServerOptions();
             _transportFactory = transportFactory;
             _logger = loggerFactory.CreateLogger("Microsoft.AspNetCore.Server.Kestrel");
             Features = new FeatureCollection();
             _serverAddresses = new ServerAddressesFeature();
             Features.Set(_serverAddresses);
-            Features.Set(InternalOptions);
         }
 
         public IFeatureCollection Features { get; }
 
         public KestrelServerOptions Options { get; }
 
-        private InternalKestrelServerOptions InternalOptions { get; }
-
         public async Task StartAsync<TContext>(IHttpApplication<TContext> application, CancellationToken cancellationToken)
         {
             try
@@ -95,13 +91,13 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core
                 _heartbeat = new Heartbeat(new IHeartbeatHandler[] { dateHeaderValueManager, connectionManager }, systemClock, trace);
 
                 IThreadPool threadPool;
-                if (InternalOptions.ThreadPoolDispatching)
+                if (Options.UseTransportThread)
                 {
-                    threadPool = new LoggingThreadPool(trace);
+                    threadPool = new InlineLoggingThreadPool(trace);
                 }
                 else
                 {
-                    threadPool = new InlineLoggingThreadPool(trace);
+                    threadPool = new LoggingThreadPool(trace);
                 }
 
                 var serviceContext = new ServiceContext

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

@@ -28,6 +28,11 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core
         /// </remarks>
         public bool AddServerHeader { get; set; } = true;
 
+        /// <summary>
+        /// Gets or sets a value that determines if Kestrel should use the transport thread thread when executing user code.
+        /// </summary>
+        public bool UseTransportThread { get; set; }
+
         /// <summary>
         /// Enables the Listen options callback to resolve and use services registered by the application during startup.
         /// Typically initialized by UseKestrel()"/>.

src/Microsoft.AspNetCore.Server.Kestrel.Transport.Abstractions/IConnectionInformation.cs

@@ -13,8 +13,6 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Transport.Abstractions
 
         PipeFactory PipeFactory { get; }
 
-        bool RequiresDispatch { get; }
-
         IScheduler InputWriterScheduler { get; }
         IScheduler OutputReaderScheduler { get; }
     }

src/Microsoft.AspNetCore.Server.Kestrel.Transport.Libuv/Internal/LibuvConnectionContext.cs

@@ -24,7 +24,6 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Transport.Libuv.Internal
         public IPEndPoint LocalEndPoint { get; set; }
 
         public PipeFactory PipeFactory => ListenerContext.Thread.PipeFactory;
-        public bool RequiresDispatch => true;
         public IScheduler InputWriterScheduler => ListenerContext.Thread;
         public IScheduler OutputReaderScheduler => ListenerContext.Thread;
     }

src/Microsoft.AspNetCore.Server.Kestrel.Transport.Libuv/WebHostBuilderLibuvExtensions.cs

@@ -33,17 +33,17 @@ namespace Microsoft.AspNetCore.Hosting
         /// <param name="hostBuilder">
         /// The Microsoft.AspNetCore.Hosting.IWebHostBuilder to configure.
         /// </param>
-        /// <param name="options">
+        /// <param name="configureOptions">
         /// A callback to configure Libuv options.
         /// </param>
         /// <returns>
         /// The Microsoft.AspNetCore.Hosting.IWebHostBuilder.
         /// </returns>
-        public static IWebHostBuilder UseLibuv(this IWebHostBuilder hostBuilder, Action<LibuvTransportOptions> options)
+        public static IWebHostBuilder UseLibuv(this IWebHostBuilder hostBuilder, Action<LibuvTransportOptions> configureOptions)
         {
             return hostBuilder.UseLibuv().ConfigureServices(services =>
             {
-                services.Configure(options);
+                services.Configure(configureOptions);
             });
         }
     }

src/Microsoft.AspNetCore.Server.Kestrel.Transport.Sockets/SocketConnection.cs

@@ -215,10 +215,8 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Transport.Sockets
 
         public PipeFactory PipeFactory => _transport.TransportFactory.PipeFactory;
 
-        public bool RequiresDispatch => _transport.TransportFactory.ForceDispatch;
-
         public IScheduler InputWriterScheduler => InlineScheduler.Default;
 
-        public IScheduler OutputReaderScheduler => InlineScheduler.Default;
+        public IScheduler OutputReaderScheduler => TaskRunScheduler.Default;
     }
 }

src/Microsoft.AspNetCore.Server.Kestrel.Transport.Sockets/SocketTransportFactory.cs

@@ -4,17 +4,16 @@
 using System;
 using Microsoft.AspNetCore.Server.Kestrel.Internal.System.IO.Pipelines;
 using Microsoft.AspNetCore.Server.Kestrel.Transport.Abstractions;
+using Microsoft.Extensions.Options;
 
 namespace Microsoft.AspNetCore.Server.Kestrel.Transport.Sockets
 {
     public sealed class SocketTransportFactory : ITransportFactory
     {
         private readonly PipeFactory _pipeFactory = new PipeFactory();
-        private readonly bool _forceDispatch;
 
-        public SocketTransportFactory(bool forceDispatch = false)
+        public SocketTransportFactory(IOptions<SocketTransportOptions> options)
         {
-            _forceDispatch = forceDispatch;
         }
 
         public ITransport Create(IEndPointInformation endPointInformation, IConnectionHandler handler)
@@ -38,7 +37,5 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Transport.Sockets
         }
 
         internal PipeFactory PipeFactory => _pipeFactory;
-
-        internal bool ForceDispatch => _forceDispatch;
     }
 }

src/Microsoft.AspNetCore.Server.Kestrel.Transport.Sockets/SocketTransportOptions.cs

@@ -0,0 +1,12 @@
+using System;
+using System.Collections.Generic;
+using System.Text;
+
+namespace Microsoft.AspNetCore.Server.Kestrel.Transport.Sockets
+{
+    // TODO: Come up with some options
+    public class SocketTransportOptions
+    {
+        
+    }
+}

src/Microsoft.AspNetCore.Server.Kestrel.Transport.Sockets/WebHostBuilderSocketExtensions.cs

@@ -1,6 +1,7 @@
 // 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 Microsoft.AspNetCore.Server.Kestrel.Transport.Abstractions;
 using Microsoft.AspNetCore.Server.Kestrel.Transport.Sockets;
 using Microsoft.Extensions.DependencyInjection;
@@ -25,5 +26,25 @@ namespace Microsoft.AspNetCore.Hosting
                 services.AddSingleton<ITransportFactory, SocketTransportFactory>();
             });
         }
+
+        /// <summary>
+        /// Specify Sockets as the transport to be used by Kestrel.
+        /// </summary>
+        /// <param name="hostBuilder">
+        /// The Microsoft.AspNetCore.Hosting.IWebHostBuilder to configure.
+        /// </param>
+        /// <param name="configureOptions">
+        /// A callback to configure Libuv options.
+        /// </param>
+        /// <returns>
+        /// The Microsoft.AspNetCore.Hosting.IWebHostBuilder.
+        /// </returns>
+        public static IWebHostBuilder UseSockets(this IWebHostBuilder hostBuilder, Action<SocketTransportOptions> configureOptions)
+        {
+            return hostBuilder.UseSockets().ConfigureServices(services =>
+            {
+                services.Configure(configureOptions);
+            });
+        }
     }
 }

test/Microsoft.AspNetCore.Server.Kestrel.Core.Tests/PipeOptionsTests.cs

@@ -24,7 +24,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Tests
 
             var connectionHandler = new ConnectionHandler<object>(listenOptions: null, serviceContext: serviceContext, application: null);
             var mockScheduler = Mock.Of<IScheduler>();
-            var outputPipeOptions = connectionHandler.GetOutputPipeOptions(requiresDispatch: true, readerScheduler: mockScheduler);
+            var outputPipeOptions = connectionHandler.GetOutputPipeOptions(readerScheduler: mockScheduler);
 
             Assert.Equal(expectedMaximumSizeLow, outputPipeOptions.MaximumSizeLow);
             Assert.Equal(expectedMaximumSizeHigh, outputPipeOptions.MaximumSizeHigh);
@@ -43,7 +43,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Tests
 
             var connectionHandler = new ConnectionHandler<object>(listenOptions: null, serviceContext: serviceContext, application: null);
             var mockScheduler = Mock.Of<IScheduler>();
-            var inputPipeOptions = connectionHandler.GetInputPipeOptions(requiresDispatch: true, writerScheduler: mockScheduler);
+            var inputPipeOptions = connectionHandler.GetInputPipeOptions(writerScheduler: mockScheduler);
 
             Assert.Equal(expectedMaximumSizeLow, inputPipeOptions.MaximumSizeLow);
             Assert.Equal(expectedMaximumSizeHigh, inputPipeOptions.MaximumSizeHigh);

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

@@ -49,7 +49,11 @@ namespace Microsoft.AspNetCore.Server.Kestrel.FunctionalTests
 
             Context = context;
 
-            _transport = s_transportFactory.Create(listenOptions, new ConnectionHandler<HttpContext>(listenOptions, context, new DummyApplication(app, httpContextFactory)));
+            // Switch this to test on socket transport
+            var transportFactory = CreateLibuvTransportFactory(context);
+            // var transportFactory = CreateSocketTransportFactory(context);
+
+            _transport = transportFactory.Create(listenOptions, new ConnectionHandler<HttpContext>(listenOptions, context, new DummyApplication(app, httpContextFactory)));
 
             try
             {
@@ -67,11 +71,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.FunctionalTests
             }
         }
 
-        // Switch this to test on socket transport
-        private static readonly ITransportFactory s_transportFactory = CreateLibuvTransportFactory();
-//        private static readonly ITransportFactory s_transportFactory = CreateSocketTransportFactory();
-
-        private static ITransportFactory CreateLibuvTransportFactory()
+        private static ITransportFactory CreateLibuvTransportFactory(TestServiceContext context)
         {
             var transportOptions = new LibuvTransportOptions()
             {
@@ -87,12 +87,11 @@ namespace Microsoft.AspNetCore.Server.Kestrel.FunctionalTests
             return transportFactory;
         }
 
-        private static ITransportFactory CreateSocketTransportFactory()
+        private static ITransportFactory CreateSocketTransportFactory(TestServiceContext context)
         {
-            // For now, force the socket transport to do threadpool dispatch for tests.
-            // There are a handful of tests that deadlock due to test issues if we don't do dispatch.
-            // We should clean these up, but for now, make them work by forcing dispatch.
-            return new SocketTransportFactory(true);
+            var options = new SocketTransportOptions();
+
+            return new SocketTransportFactory(Options.Create(options));
         }
         
         public IPEndPoint EndPoint => _listenOptions.IPEndPoint;