Multiple threads processing connections

src/Kestrel/IKestrelServerInformation.cs

@@ -0,0 +1,10 @@
+// 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 Kestrel
+{
+    public interface IKestrelServerInformation
+    {
+        int ThreadCount { get; set; }
+    }
+}

src/Kestrel/ServerFactory.cs

@@ -38,7 +38,7 @@ public IDisposable Start(IServerInformation serverInformation, Func<IFeatureColl
             var disposables = new List<IDisposable>();
             var information = (ServerInformation)serverInformation;
             var engine = new KestrelEngine(_libraryManager, _appShutdownService);
-            engine.Start(1);
+            engine.Start(information.ThreadCount == 0 ? 1 : information.ThreadCount);
             foreach (var address in information.Addresses)
             {
                 disposables.Add(engine.CreateServer(

src/Kestrel/ServerInformation.cs

@@ -8,7 +8,7 @@
 
 namespace Kestrel
 {
-    public class ServerInformation : IServerInformation
+    public class ServerInformation : IServerInformation, IKestrelServerInformation
     {
         public ServerInformation()
         {
@@ -25,7 +25,7 @@ public void Initialize(IConfiguration configuration)
             foreach (var url in urls.Split(new[] { ';' }, StringSplitOptions.RemoveEmptyEntries))
             {
                 var address = ServerAddress.FromUrl(url);
-                if(address != null)
+                if (address != null)
                 {
                     Addresses.Add(address);
                 }
@@ -41,5 +41,7 @@ public string Name
         }
 
         public IList<ServerAddress> Addresses { get; private set; }
+
+        public int ThreadCount { get; set; }
     }
 }

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

@@ -24,7 +24,7 @@ private static void ReadCallback(UvStreamHandle handle, int nread, Exception err
 
         private readonly UvStreamHandle _socket;
         private Frame _frame;
-        long _connectionId;
+        long _connectionId = 0;
 
         public Connection(ListenerContext context, UvStreamHandle socket) : base(context)
         {

src/Microsoft.AspNet.Server.Kestrel/Http/Listener.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 Microsoft.AspNet.Server.Kestrel.Infrastructure;
 using Microsoft.AspNet.Server.Kestrel.Networking;
 using System;
 using System.Diagnostics;
@@ -53,7 +54,7 @@ public Task StartAsync(
                     ListenSocket = new UvTcpHandle();
                     ListenSocket.Init(Thread.Loop, Thread.QueueCloseHandle);
                     ListenSocket.Bind(new IPEndPoint(IPAddress.Any, port));
-                    ListenSocket.Listen(10, _connectionCallback, this);
+                    ListenSocket.Listen(Constants.ListenBacklog, _connectionCallback, this);
                     tcs.SetResult(0);
                 }
                 catch (Exception ex)
@@ -70,7 +71,12 @@ private void OnConnection(UvStreamHandle listenSocket, int status)
             acceptSocket.Init(Thread.Loop, Thread.QueueCloseHandle);
             listenSocket.Accept(acceptSocket);
 
-            var connection = new Connection(this, acceptSocket);
+            DispatchConnection(acceptSocket);
+        }
+
+        protected virtual void DispatchConnection(UvTcpHandle socket)
+        {
+            var connection = new Connection(this, socket);
             connection.Start();
         }
 

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

@@ -0,0 +1,89 @@
+// 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.Infrastructure;
+using Microsoft.AspNet.Server.Kestrel.Networking;
+using System;
+using System.Collections.Generic;
+using System.Threading.Tasks;
+
+namespace Microsoft.AspNet.Server.Kestrel.Http
+{
+    public class ListenerPrimary : Listener
+    {
+        UvPipeHandle ListenPipe { get; set; }
+
+        List<UvPipeHandle> _dispatchPipes = new List<UvPipeHandle>();
+        int _dispatchIndex;
+        ArraySegment<ArraySegment<byte>> _1234 = new ArraySegment<ArraySegment<byte>>(new[] { new ArraySegment<byte>(new byte[] { 1, 2, 3, 4 }) });
+
+        public ListenerPrimary(IMemoryPool memory) : base(memory)
+        {
+        }
+
+        public async Task StartAsync(
+            string pipeName,
+            string scheme,
+            string host,
+            int port,
+            KestrelThread thread,
+            Func<Frame, Task> application)
+        {
+            await StartAsync(scheme, host, port, thread, application);
+
+            await Thread.PostAsync(_ =>
+            {
+                ListenPipe = new UvPipeHandle();
+                ListenPipe.Init(Thread.Loop, false);
+                ListenPipe.Bind(pipeName);
+                ListenPipe.Listen(Constants.ListenBacklog, OnListenPipe, null);
+            }, null);
+        }
+
+        private void OnListenPipe(UvStreamHandle pipe, int status, Exception error, object state)
+        {
+            if (status < 0)
+            {
+                return;
+            }
+
+            var dispatchPipe = new UvPipeHandle();
+            dispatchPipe.Init(Thread.Loop, true);
+            try
+            {
+                pipe.Accept(dispatchPipe);
+            }
+            catch (Exception)
+            {
+                dispatchPipe.Dispose();
+                return;
+            }
+            _dispatchPipes.Add(dispatchPipe);
+        }
+
+        protected override void DispatchConnection(UvTcpHandle socket)
+        {
+            var index = _dispatchIndex++ % (_dispatchPipes.Count + 1);
+            if (index == _dispatchPipes.Count)
+            {
+                base.DispatchConnection(socket);
+            }
+            else
+            {
+                var dispatchPipe = _dispatchPipes[index];
+                var write = new UvWriteReq();
+                write.Init(Thread.Loop);
+                write.Write2(
+                    dispatchPipe,
+                    _1234,
+                    socket,
+                    (write2, status, error, state) => 
+                    {
+                        write2.Dispose();
+                        ((UvTcpHandle)state).Dispose();
+                    },
+                    socket);
+            }
+        }
+    }
+}

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

@@ -0,0 +1,117 @@
+// 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.Networking;
+using System;
+using System.Diagnostics;
+using System.Runtime.InteropServices;
+using System.Threading.Tasks;
+
+namespace Microsoft.AspNet.Server.Kestrel.Http
+{
+    public class ListenerSecondary : ListenerContext, IDisposable
+    {
+        UvPipeHandle DispatchPipe { get; set; }
+
+        public ListenerSecondary(IMemoryPool memory)
+        {
+            Memory = memory;
+        }
+
+        public Task StartAsync(
+            string pipeName,
+            KestrelThread thread,
+            Func<Frame, Task> application)
+        {
+            Thread = thread;
+            Application = application;
+
+            DispatchPipe = new UvPipeHandle();
+
+            var tcs = new TaskCompletionSource<int>();
+            Thread.Post(_ =>
+            {
+                try
+                {
+                    DispatchPipe.Init(Thread.Loop, true);
+                    var connect = new UvConnectRequest();
+                    connect.Init(Thread.Loop);
+                    connect.Connect(
+                        DispatchPipe,
+                        pipeName,
+                        (connect2, status, error, state) =>
+                        {
+                            connect.Dispose();
+                            if (error != null)
+                            {
+                                tcs.SetException(error);
+                                return;
+                            }
+
+                            try
+                            {
+                                var ptr = Marshal.AllocHGlobal(16);
+                                var buf = Thread.Loop.Libuv.buf_init(ptr, 16);
+
+                                DispatchPipe.ReadStart(
+                                    (_1, _2, _3) => buf,
+                                    (_1, status2, error2, state2) =>
+                                    {
+                                        if (status2 == 0)
+                                        {
+                                            DispatchPipe.Dispose();
+                                            Marshal.FreeHGlobal(ptr);
+                                            return;
+                                        }
+
+                                        var acceptSocket = new UvTcpHandle();
+                                        acceptSocket.Init(Thread.Loop, Thread.QueueCloseHandle);
+
+                                        try
+                                        {
+                                            DispatchPipe.Accept(acceptSocket);
+                                        }
+                                        catch (Exception ex)
+                                        {
+                                            Trace.WriteLine("DispatchPipe.Accept " + ex.Message);
+                                            acceptSocket.Dispose();
+                                            return;
+                                        }
+
+                                        var connection = new Connection(this, acceptSocket);
+                                        connection.Start();
+                                    },
+                                    null);
+
+                                tcs.SetResult(0);
+                            }
+                            catch (Exception ex)
+                            {
+                                DispatchPipe.Dispose();
+                                tcs.SetException(ex);
+                            }
+                        },
+                        null);
+                }
+                catch (Exception ex)
+                {
+                    DispatchPipe.Dispose();
+                    tcs.SetException(ex);
+                }
+            }, null);
+            return tcs.Task;
+        }
+
+        public void Dispose()
+        {
+            // Ensure the event loop is still running.
+            // If the event loop isn't running and we try to wait on this Post
+            // to complete, then KestrelEngine will never be disposed and
+            // the exception that stopped the event loop will never be surfaced.
+            if (Thread.FatalError == null)
+            {
+                Thread.Send(_ => DispatchPipe.Dispose(), null);
+            }
+        }
+    }
+}

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

@@ -4,7 +4,6 @@
 using Microsoft.AspNet.Server.Kestrel.Networking;
 using System;
 using System.Threading;
-using System.Threading;
 
 namespace Microsoft.AspNet.Server.Kestrel.Http
 {

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

@@ -0,0 +1,12 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Threading.Tasks;
+
+namespace Microsoft.AspNet.Server.Kestrel.Infrastructure
+{
+    internal class Constants
+    {
+        public const int ListenBacklog = 128;
+    }
+}

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

@@ -118,6 +118,18 @@ public Task PostAsync(Action<object> callback, object state)
             return tcs.Task;
         }
 
+        public void Send(Action<object> callback, object state)
+        {
+            if (_loop.ThreadId == Thread.CurrentThread.ManagedThreadId)
+            {
+                callback.Invoke(state);
+            }
+            else
+            {
+                PostAsync(callback, state).Wait();
+            }
+        }
+
         private void PostCloseHandle(Action<IntPtr> callback, IntPtr handle)
         {
             lock (_workSync)

src/Microsoft.AspNet.Server.Kestrel/KestrelEngine.cs

@@ -38,10 +38,10 @@ public KestrelEngine(ILibraryManager libraryManager, IApplicationShutdown appShu
                         : "amd64";
 
                     libraryPath = Path.Combine(
-                        libraryPath, 
+                        libraryPath,
                         "native",
                         "windows",
-                        architecture, 
+                        architecture,
                         "libuv.dll");
                 }
                 else if (Libuv.IsDarwin)
@@ -91,16 +91,37 @@ public void Dispose()
 
         public IDisposable CreateServer(string scheme, string host, int port, Func<Frame, Task> application)
         {
-            var listeners = new List<Listener>();
+            var listeners = new List<IDisposable>();
 
             try
             {
+                var pipeName = (Libuv.IsWindows ? @"\\.\pipe\kestrel_" : "/tmp/kestrel_") + Guid.NewGuid().ToString("n");
+
+                var single = Threads.Count == 1;
+                var first = true;
+
                 foreach (var thread in Threads)
                 {
-                    var listener = new Listener(Memory);
+                    if (single)
+                    {
+                        var listener = new Listener(Memory);
+                        listeners.Add(listener);
+                        listener.StartAsync(scheme, host, port, thread, application).Wait();
+                    }
+                    else if (first)
+                    {
+                        var listener = new ListenerPrimary(Memory);
+                        listeners.Add(listener);
+                        listener.StartAsync(pipeName, scheme, host, port, thread, application).Wait();
+                    }
+                    else
+                    {
+                        var listener = new ListenerSecondary(Memory);
+                        listeners.Add(listener);
+                        listener.StartAsync(pipeName, thread, application).Wait();
+                    }
 
-                    listeners.Add(listener);
-                    listener.StartAsync(scheme, host, port, thread, application).Wait();
+                    first = false;
                 }
                 return new Disposable(() =>
                 {