Rio Transport on WinThreadpool

Author: benaadams

KestrelHttpServer.sln

@@ -70,6 +70,10 @@ Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Microsoft.AspNetCore.Server
 EndProject
 Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Microsoft.AspNetCore.Server.Kestrel.Tests", "test\Microsoft.AspNetCore.Server.Kestrel.Tests\Microsoft.AspNetCore.Server.Kestrel.Tests.csproj", "{4F1C30F8-CCAA-48D7-9DF6-2A84021F5BCC}"
 EndProject
+Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "TransportChoiceApp", "samples\TransportChoiceApp\TransportChoiceApp.csproj", "{18BE39D5-C307-4F4D-BBB8-082BFAE8AF39}"
+EndProject
+Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Microsoft.AspNetCore.Server.Kestrel.Transport.WindowsRio", "src\Microsoft.AspNetCore.Server.Kestrel.Transport.WindowsRio\Microsoft.AspNetCore.Server.Kestrel.Transport.WindowsRio.csproj", "{DD57C592-6BCC-4A06-B4C7-0D23A286DFB1}"
+EndProject
 Global
 	GlobalSection(SolutionConfigurationPlatforms) = preSolution
 		Debug|Any CPU = Debug|Any CPU
@@ -248,6 +252,30 @@ Global
 		{4F1C30F8-CCAA-48D7-9DF6-2A84021F5BCC}.Release|x64.Build.0 = Release|Any CPU
 		{4F1C30F8-CCAA-48D7-9DF6-2A84021F5BCC}.Release|x86.ActiveCfg = Release|Any CPU
 		{4F1C30F8-CCAA-48D7-9DF6-2A84021F5BCC}.Release|x86.Build.0 = Release|Any CPU
+		{18BE39D5-C307-4F4D-BBB8-082BFAE8AF39}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+		{18BE39D5-C307-4F4D-BBB8-082BFAE8AF39}.Debug|Any CPU.Build.0 = Debug|Any CPU
+		{18BE39D5-C307-4F4D-BBB8-082BFAE8AF39}.Debug|x64.ActiveCfg = Debug|Any CPU
+		{18BE39D5-C307-4F4D-BBB8-082BFAE8AF39}.Debug|x64.Build.0 = Debug|Any CPU
+		{18BE39D5-C307-4F4D-BBB8-082BFAE8AF39}.Debug|x86.ActiveCfg = Debug|Any CPU
+		{18BE39D5-C307-4F4D-BBB8-082BFAE8AF39}.Debug|x86.Build.0 = Debug|Any CPU
+		{18BE39D5-C307-4F4D-BBB8-082BFAE8AF39}.Release|Any CPU.ActiveCfg = Release|Any CPU
+		{18BE39D5-C307-4F4D-BBB8-082BFAE8AF39}.Release|Any CPU.Build.0 = Release|Any CPU
+		{18BE39D5-C307-4F4D-BBB8-082BFAE8AF39}.Release|x64.ActiveCfg = Release|Any CPU
+		{18BE39D5-C307-4F4D-BBB8-082BFAE8AF39}.Release|x64.Build.0 = Release|Any CPU
+		{18BE39D5-C307-4F4D-BBB8-082BFAE8AF39}.Release|x86.ActiveCfg = Release|Any CPU
+		{18BE39D5-C307-4F4D-BBB8-082BFAE8AF39}.Release|x86.Build.0 = Release|Any CPU
+		{DD57C592-6BCC-4A06-B4C7-0D23A286DFB1}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+		{DD57C592-6BCC-4A06-B4C7-0D23A286DFB1}.Debug|Any CPU.Build.0 = Debug|Any CPU
+		{DD57C592-6BCC-4A06-B4C7-0D23A286DFB1}.Debug|x64.ActiveCfg = Debug|Any CPU
+		{DD57C592-6BCC-4A06-B4C7-0D23A286DFB1}.Debug|x64.Build.0 = Debug|Any CPU
+		{DD57C592-6BCC-4A06-B4C7-0D23A286DFB1}.Debug|x86.ActiveCfg = Debug|Any CPU
+		{DD57C592-6BCC-4A06-B4C7-0D23A286DFB1}.Debug|x86.Build.0 = Debug|Any CPU
+		{DD57C592-6BCC-4A06-B4C7-0D23A286DFB1}.Release|Any CPU.ActiveCfg = Release|Any CPU
+		{DD57C592-6BCC-4A06-B4C7-0D23A286DFB1}.Release|Any CPU.Build.0 = Release|Any CPU
+		{DD57C592-6BCC-4A06-B4C7-0D23A286DFB1}.Release|x64.ActiveCfg = Release|Any CPU
+		{DD57C592-6BCC-4A06-B4C7-0D23A286DFB1}.Release|x64.Build.0 = Release|Any CPU
+		{DD57C592-6BCC-4A06-B4C7-0D23A286DFB1}.Release|x86.ActiveCfg = Release|Any CPU
+		{DD57C592-6BCC-4A06-B4C7-0D23A286DFB1}.Release|x86.Build.0 = Release|Any CPU
 	EndGlobalSection
 	GlobalSection(SolutionProperties) = preSolution
 		HideSolutionNode = FALSE
@@ -269,5 +297,7 @@ Global
 		{2E9CB89D-EC8F-4DD9-A72B-08D5BABF752D} = {2D5D5227-4DBD-499A-96B1-76A36B03B750}
 		{D95A7EC3-48AC-4D03-B2E2-0DA3E13BD3A4} = {D3273454-EA07-41D2-BF0B-FCC3675C2483}
 		{4F1C30F8-CCAA-48D7-9DF6-2A84021F5BCC} = {D3273454-EA07-41D2-BF0B-FCC3675C2483}
+		{DD57C592-6BCC-4A06-B4C7-0D23A286DFB1} = {2D5D5227-4DBD-499A-96B1-76A36B03B750}
+		{18BE39D5-C307-4F4D-BBB8-082BFAE8AF39} = {8A3D00B8-1CCF-4BE6-A060-11104CE2D9CE}
 	EndGlobalSection
 EndGlobal

samples/TransportChoiceApp/Startup.cs

@@ -0,0 +1,78 @@
+// 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.IO;
+using System.Net;
+using System.Text;
+using System.Threading.Tasks;
+using Microsoft.AspNetCore.Builder;
+using Microsoft.AspNetCore.Hosting;
+using Microsoft.AspNetCore.Server.Kestrel.Core.Internal;
+
+namespace TransportChoiceApp
+{
+    public class Startup
+    {
+        private static readonly byte[] _helloWorldPayload = Encoding.UTF8.GetBytes("Hello, World!");
+
+        public void Configure(IApplicationBuilder app)
+        {
+            app.Run((httpContext) =>
+            {
+                var response = httpContext.Response;
+                var payloadLength = _helloWorldPayload.Length;
+                response.StatusCode = 200;
+                response.ContentType = "text/plain";
+                response.ContentLength = payloadLength;
+                return response.Body.WriteAsync(_helloWorldPayload, 0, payloadLength);
+            });
+        }
+
+        public static void Main(string[] args)
+        {
+            var hostBuilder = new WebHostBuilder()
+                .UseKestrel(options =>
+                {
+                    options.Listen(IPAddress.Loopback, 5001);
+                    
+                });
+
+            Console.WriteLine(@"
+Choose a Transport
+ 1. Libuv
+ 2. Sockets
+ 3. Rio
+");
+
+            switch (Console.ReadKey().KeyChar)
+            {
+                case '1':
+                    hostBuilder.UseLibuv(options =>
+                    {
+                        //options.ThreadCount = 4;
+                    });
+                    break;
+                case '2':
+                    hostBuilder.UseSockets();
+                    break;
+                case '3':
+                    hostBuilder.UseWindowsRio();
+                    break;
+                default:
+                    Console.WriteLine("Invalid option");
+                    return;
+            }
+
+            Console.WriteLine();
+
+            hostBuilder
+                .UseContentRoot(Directory.GetCurrentDirectory())
+                .UseStartup<Startup>();
+
+            var host = hostBuilder.Build();
+
+            host.Run();
+        }
+    }
+}

samples/TransportChoiceApp/TransportChoiceApp.csproj

@@ -0,0 +1,16 @@
+<Project Sdk="Microsoft.NET.Sdk.Web">
+
+  <Import Project="..\..\build\common.props" />
+
+  <PropertyGroup>
+    <TargetFramework>netcoreapp2.0</TargetFramework>
+    <IsPackable>false</IsPackable>
+    <OutputType>exe</OutputType>
+  </PropertyGroup>
+
+  <ItemGroup>
+    <ProjectReference Include="..\..\src\Microsoft.AspNetCore.Server.Kestrel.Transport.WindowsRio\Microsoft.AspNetCore.Server.Kestrel.Transport.WindowsRio.csproj" />
+    <ProjectReference Include="..\..\src\Microsoft.AspNetCore.Server.Kestrel\Microsoft.AspNetCore.Server.Kestrel.csproj" />
+  </ItemGroup>
+
+</Project>

src/Microsoft.AspNetCore.Server.Kestrel.Transport.WindowsRio/Internal/AutoResetGateOfT.cs

@@ -0,0 +1,98 @@
+// Copyright (c) .NET Foundation. All rights reserved.
+// Licensed under the MIT license. See LICENSE file in the project root for full license information.
+
+using System;
+using System.Runtime.CompilerServices;
+using System.Threading;
+using System.Threading.Tasks;
+
+namespace Microsoft.AspNetCore.Server.Kestrel.Transport.WindowsRio.Internal
+{
+    /// <summary>
+    /// Simple awaitable gate - intended to synchronize a single producer with a single consumer to ensure the producer doesn't
+    /// produce until the consumer is ready. Similar to a <see cref="TaskCompletionSource{TResult}"/> but reusable so we don't have
+    /// to keep allocating new ones every time.
+    /// </summary>
+    /// <remarks>
+    /// The gate can be in one of two states: "Open", indicating that an await will immediately return and "Closed", meaning that an await
+    /// will block until the gate is opened. The gate is initially "Closed" and can be opened by a call to <see cref="Open"/>. Upon the completion
+    /// of an await, it will automatically return to the "Closed" state (this is done in the <see cref="GetResult"/> call that is injected by the
+    /// compiler's async/await logic).
+    /// </remarks>
+    public class AutoResetGate<T> : ICriticalNotifyCompletion
+    {
+        private static readonly Action _gateIsOpen = () => { };
+
+        private Action _gateState;
+        private T _value;
+
+        /// <summary>
+        /// Returns a boolean indicating if the gate is "open"
+        /// </summary>
+        public bool IsCompleted => ReferenceEquals(_gateState, _gateIsOpen);
+
+        public void UnsafeOnCompleted(Action continuation) => OnCompleted(continuation);
+
+        public void OnCompleted(Action continuation)
+        {
+            // If we're already completed, call the continuation immediately
+            if (IsCompleted)
+            {
+                continuation();
+            }
+            else
+            {
+                // Otherwise, if the current continuation is null, atomically store the new continuation in the field and return the old value
+                var previous = Interlocked.CompareExchange(ref _gateState, continuation, null);
+                if (ReferenceEquals(previous, _gateIsOpen))
+                {
+                    // It got completed in the time between the previous the method and the cmpexch.
+                    // So call the continuation (the value of _continuation will remain _completed because cmpexch is atomic,
+                    // so we didn't accidentally replace it).
+                    continuation();
+                }
+            }
+        }
+
+        /// <summary>
+        /// Resets the gate to continue blocking the waiter. This is called immediately after awaiting the signal.
+        /// </summary>
+        public T GetResult()
+        {
+            var value = _value;
+            _value = default(T);
+
+            // Clear the active continuation to "reset" the state of this event
+            _gateState = null;
+
+            return value;
+        }
+
+        /// <summary>
+        /// Set the gate to allow the waiter to continue.
+        /// </summary>
+        public void Open(T value)
+        {
+            // Set the stored continuation value to a sentinel that indicates the state is completed, then call the previous value.
+            _value = value;
+            var completion = Interlocked.Exchange(ref _gateState, _gateIsOpen);
+            if (!ReferenceEquals(completion, _gateIsOpen))
+            {
+                completion?.Invoke();
+            }
+        }
+
+        private void Close()
+        {
+            // Clear the active continuation to "reset" the state of this event
+            _gateState = null;
+        }
+
+        public AutoResetGate<T> GetAwaiter() => this;
+
+        private void ThrowMultiConsumerUseError()
+        {
+            throw new InvalidOperationException();
+        }
+    }
+}

src/Microsoft.AspNetCore.Server.Kestrel.Transport.WindowsRio/Internal/Networking/BufferMapper.cs

@@ -0,0 +1,135 @@
+// 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.Internal.System.Buffers;
+using Microsoft.AspNetCore.Server.Kestrel.Internal.System.IO.Pipelines;
+
+namespace Microsoft.AspNetCore.Server.Kestrel.Transport.WindowsRio.Internal
+{
+    public class BufferMapper
+    {
+        private readonly object _mappingSync = new object();
+        private BufferMapping[] _bufferIdMappings = new BufferMapping[0];
+
+        public BufferMapper(MemoryPool memoryPool)
+        {
+            memoryPool.RegisterSlabAllocationCallback((slab) => OnSlabAllocated(slab));
+            memoryPool.RegisterSlabDeallocationCallback((slab) => OnSlabDeallocated(slab));
+        }
+
+        internal RioRegisteredBuffer GetRegisteredBuffer(IntPtr address, out long startAddress)
+        {
+            var buffer = default(RioRegisteredBuffer);
+            startAddress = 0;
+
+            // Take local copy to avoid modifications
+            var bufferIdMappings = _bufferIdMappings;
+            var addressLong = address.ToInt64();
+
+            // Can binary search if it's too slow
+            for (var i = 0; i < bufferIdMappings.Length; i++)
+            {
+                var mapping = bufferIdMappings[i];
+                if (addressLong >= mapping.Start && addressLong <= mapping.End)
+                {
+                    buffer = mapping.Buffer;
+                    startAddress = mapping.Start;
+                    break;
+                }
+            }
+
+            return buffer;
+        }
+
+        internal unsafe RioBufferSegment GetSegmentFromBuffer(Buffer<byte> buffer)
+        {
+            // It's ok to unpin the handle here because the memory is from the pool
+            // we created, which is already pinned.
+            var pin = buffer.Pin();
+            var spanPtr = (IntPtr)pin.PinnedPointer;
+            pin.Free();
+
+            long startAddress;
+            long spanAddress = spanPtr.ToInt64();
+            var bufferId = GetRegisteredBuffer(spanPtr, out startAddress);
+
+            checked
+            {
+                var offset = unchecked((uint)(spanAddress - startAddress));
+                return new RioBufferSegment(bufferId, offset, (uint)buffer.Length);
+            }
+        }
+
+        private void OnSlabAllocated(MemoryPoolSlab slab)
+        {
+            var memoryPtr = slab.NativePointer;
+            var buffer = RioRegisteredBuffer.Create(memoryPtr, (uint)slab.Length);
+            var addressLong = memoryPtr.ToInt64();
+
+            // Read, write and swap the mappings under lock
+            lock (_mappingSync)
+            {
+                var currentMappings = _bufferIdMappings;
+                var newMappings = new BufferMapping[currentMappings.Length + 1];
+
+                for (var i = 0; i < currentMappings.Length; i++)
+                {
+                    newMappings[i] = currentMappings[i];
+                }
+
+                newMappings[currentMappings.Length] = new BufferMapping
+                {
+                    Buffer = buffer,
+                    Start = addressLong,
+                    End = addressLong + slab.Length
+                };
+
+                _bufferIdMappings = newMappings;
+            }
+        }
+
+        private void OnSlabDeallocated(MemoryPoolSlab slab)
+        {
+            var memoryPtr = slab.NativePointer;
+            var addressLong = memoryPtr.ToInt64();
+
+            // Read, write and swap the mappings under lock
+            lock (_mappingSync)
+            {
+                var currentMappings = _bufferIdMappings;
+                var newMappings = new BufferMapping[currentMappings.Length - 1];
+
+                for (int i = 0, n = 0; i < currentMappings.Length; i++)
+                {
+                    var bufferMapping = currentMappings[i];
+                    if (addressLong != bufferMapping.Start)
+                    {
+                        // Not being removed, add it to new mappings
+                        newMappings[n] = currentMappings[i];
+                        n++;
+                    }
+                    else
+                    {
+                        // Being removed, dispose, don't add
+                        bufferMapping.Buffer.Dispose();
+                    }
+                }
+
+                _bufferIdMappings = newMappings;
+            }
+        }
+
+        private struct BufferMapping
+        {
+            public RioRegisteredBuffer Buffer;
+            public long Start;
+            public long End;
+
+            public override string ToString()
+            {
+                return $"{Buffer} ({Start}) - ({End})";
+            }
+        }
+    }
+}