Wrote a test for unaccepted connections

davidfowl · davidfowl · commit c5b27b2cdc8e · 2019-05-26T18:56:45.000-07:00
diff --git a/src/Servers/Kestrel/Transport.Libuv/src/Internal/LibuvConnectionListener.cs b/src/Servers/Kestrel/Transport.Libuv/src/Internal/LibuvConnectionListener.cs
@@ -2,7 +2,6 @@
 // Licensed under the Apache License, Version 2.0. See License.txt in the project root for license information.
 
 using System;
-using System.Collections.Concurrent;
 using System.Collections.Generic;
 using System.Linq;
 using System.Net;
@@ -19,6 +18,8 @@ internal class LibuvConnectionListener : IConnectionListener
     {
         private readonly List<ListenerContext> _listeners = new List<ListenerContext>();
         private IAsyncEnumerator<LibuvConnection> _acceptEnumerator;
+        private bool _stopped;
+        private bool _disposed;
 
         public LibuvConnectionListener(LibuvTransportContext context, EndPoint endPoint)
             : this(new LibuvFunctions(), context, endPoint)
@@ -43,7 +44,54 @@ public LibuvConnectionListener(LibuvFunctions uv, LibuvTransportContext context,
 
         public EndPoint EndPoint { get; set; }
 
-        public async Task StopThreadsAsync()
+        public async ValueTask<ConnectionContext> AcceptAsync(CancellationToken cancellationToken = default)
+        {
+            if (_disposed)
+            {
+                throw new ObjectDisposedException(GetType().FullName);
+            }
+
+            if (await _acceptEnumerator.MoveNextAsync())
+            {
+                return _acceptEnumerator.Current;
+            }
+
+            // null means we're done...
+            return null;
+        }
+
+        public async ValueTask StopAsync(CancellationToken cancellationToken = default)
+        {
+            await UnbindAsync().ConfigureAwait(false);
+        }
+
+        public async ValueTask DisposeAsync()
+        {
+            if (_disposed)
+            {
+                return;
+            }
+
+            _disposed = true;
+
+            await UnbindAsync().ConfigureAwait(false);
+
+            if (_acceptEnumerator != null)
+            {
+                while (await _acceptEnumerator.MoveNextAsync())
+                {
+                    _acceptEnumerator.Current.Abort();
+                }
+
+                await _acceptEnumerator.DisposeAsync();
+            }
+
+            _listeners.Clear();
+
+            await StopThreadsAsync().ConfigureAwait(false);
+        }
+
+        internal async Task StopThreadsAsync()
         {
             try
             {
@@ -71,7 +119,7 @@ await Task.WhenAll(Threads.Select(thread => thread.StopAsync(TimeSpan.FromSecond
 #endif
         }
 
-        public async Task BindAsync()
+        internal async Task BindAsync()
         {
             // TODO: Move thread management to LibuvTransportFactory
             // TODO: Split endpoint management from thread management
@@ -143,7 +191,7 @@ private async IAsyncEnumerator<LibuvConnection> AcceptConnections()
             while (remainingSlots > 0)
             {
                 // Calling GetAwaiter().GetResult() is safe because we know the task is completed
-                (LibuvConnection connection, int slot) = (await Task.WhenAny(slots)).GetAwaiter().GetResult();
+                (var connection, var slot) = (await Task.WhenAny(slots)).GetAwaiter().GetResult();
 
                 // If the connection is null then the listener was closed
                 if (connection == null)
@@ -166,58 +214,26 @@ private async IAsyncEnumerator<LibuvConnection> AcceptConnections()
             }
         }
 
-        public async Task UnbindAsync()
+        internal async Task UnbindAsync()
         {
+            if (_stopped)
+            {
+                return;
+            }
+
+            _stopped = true;
+
             var disposeTasks = _listeners.Select(listener => ((IAsyncDisposable)listener).DisposeAsync()).ToArray();
 
             if (!await WaitAsync(Task.WhenAll(disposeTasks), TimeSpan.FromSeconds(5)).ConfigureAwait(false))
             {
                 Log.LogError(0, null, "Disposing listeners failed");
             }
-
-            _listeners.Clear();
         }
 
         private static async Task<bool> WaitAsync(Task task, TimeSpan timeout)
         {
             return await Task.WhenAny(task, Task.Delay(timeout)).ConfigureAwait(false) == task;
         }
-
-        public async ValueTask<ConnectionContext> AcceptAsync(CancellationToken cancellationToken = default)
-        {
-            if (await _acceptEnumerator.MoveNextAsync())
-            {
-                return _acceptEnumerator.Current;
-            }
-
-            // null means we're done...
-            return null;
-        }
-
-        public async ValueTask StopAsync(CancellationToken cancellationToken)
-        {
-            await UnbindAsync().ConfigureAwait(false);
-        }
-
-        public async ValueTask DisposeAsync()
-        {
-            // TODO: ConfigureAwait
-            await UnbindAsync().ConfigureAwait(false);
-
-            if (_acceptEnumerator != null)
-            {
-                // TODO: Log how many connections were unaccepted
-                if (await _acceptEnumerator.MoveNextAsync().ConfigureAwait(false))
-                {
-                    // Abort the connection
-                    _acceptEnumerator.Current.Abort();
-                }
-
-                // Dispose the enumerator
-                await _acceptEnumerator.DisposeAsync().ConfigureAwait(false);
-            }
-
-            await StopThreadsAsync().ConfigureAwait(false);
-        }
     }
 }
diff --git a/src/Servers/Kestrel/Transport.Libuv/src/Internal/Listener.cs b/src/Servers/Kestrel/Transport.Libuv/src/Internal/Listener.cs
@@ -197,7 +197,6 @@ protected virtual void DispatchConnection(UvStreamHandle socket)
 
         public virtual async Task DisposeAsync()
         {
-            StopAcceptingConnections();
             // 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 LibuvTransport will never be disposed and
@@ -210,6 +209,8 @@ await Thread.PostAsync(listener =>
 
                     listener._closed = true;
 
+                    listener.StopAcceptingConnections();
+
                 }, this).ConfigureAwait(false);
             }
 
diff --git a/src/Servers/Kestrel/Transport.Libuv/src/Internal/ListenerContext.cs b/src/Servers/Kestrel/Transport.Libuv/src/Internal/ListenerContext.cs
@@ -2,6 +2,7 @@
 // Licensed under the Apache License, Version 2.0. See License.txt in the project root for license information.
 
 using System;
+using System.Diagnostics;
 using System.IO.Pipelines;
 using System.Net;
 using System.Net.Sockets;
@@ -89,7 +90,8 @@ protected internal void HandleConnection(UvStreamHandle socket)
                 var connection = new LibuvConnection(socket, TransportContext.Log, Thread, remoteEndPoint, localEndPoint, InputOptions, OutputOptions);
                 connection.Start();
 
-                _acceptQueue.Writer.TryWrite(connection);
+                bool accepted = _acceptQueue.Writer.TryWrite(connection);
+                Debug.Assert(accepted, "The connection was not written to the channel!");
             }
             catch (Exception ex)
             {
diff --git a/src/Servers/Kestrel/Transport.Libuv/src/Internal/ListenerSecondary.cs b/src/Servers/Kestrel/Transport.Libuv/src/Internal/ListenerSecondary.cs
@@ -177,7 +177,6 @@ private void FreeBuffer()
 
         public async Task DisposeAsync()
         {
-            StopAcceptingConnections();
             // 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 LibuvTransport will never be disposed and
@@ -191,11 +190,15 @@ await Thread.PostAsync(listener =>
 
                     listener._closed = true;
 
+                    listener.StopAcceptingConnections();
+
                 }, this).ConfigureAwait(false);
             }
             else
             {
                 FreeBuffer();
+
+                StopAcceptingConnections();
             }
         }
     }
diff --git a/src/Servers/Kestrel/Transport.Libuv/test/LibuvTransportTests.cs b/src/Servers/Kestrel/Transport.Libuv/test/LibuvTransportTests.cs
@@ -1,18 +1,17 @@
 // 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.Buffers;
 using System.Collections.Generic;
 using System.Linq;
 using System.Net;
 using System.Net.Http;
 using System.Net.Sockets;
 using System.Text;
-using System.Threading;
 using System.Threading.Tasks;
 using Microsoft.AspNetCore.Http;
 using Microsoft.AspNetCore.Server.Kestrel.Core;
 using Microsoft.AspNetCore.Server.Kestrel.Core.Internal;
-using Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Infrastructure;
 using Microsoft.AspNetCore.Server.Kestrel.Transport.Libuv.Internal;
 using Microsoft.AspNetCore.Server.Kestrel.Transport.Libuv.Tests.TestHelpers;
 using Microsoft.AspNetCore.Testing;
@@ -23,15 +22,6 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Transport.Libuv.Tests
 {
     public class LibuvTransportTests
     {
-        public static TheoryData<ListenOptions> ConnectionAdapterData => new TheoryData<ListenOptions>
-        {
-            new ListenOptions(new IPEndPoint(IPAddress.Loopback, 0)),
-            new ListenOptions(new IPEndPoint(IPAddress.Loopback, 0))
-            {
-                ConnectionAdapters = { new PassThroughConnectionAdapter() }
-            }
-        };
-
         public static IEnumerable<object[]> OneToTen => Enumerable.Range(1, 10).Select(i => new object[] { i });
 
         [Fact]
@@ -42,7 +32,7 @@ public async Task TransportCanBindAndStop()
 
             // The transport can no longer start threads without binding to an endpoint.
             await transport.BindAsync();
-            await transport.StopThreadsAsync();
+            await transport.DisposeAsync();
         }
 
         [Fact]
@@ -52,42 +42,92 @@ public async Task TransportCanBindUnbindAndStop()
             var transport = new LibuvConnectionListener(transportContext, new IPEndPoint(IPAddress.Loopback, 0));
 
             await transport.BindAsync();
-            await transport.UnbindAsync();
-            await transport.StopThreadsAsync();
+            await transport.StopAsync();
+            await transport.DisposeAsync();
         }
 
-        [Theory]
-        [MemberData(nameof(ConnectionAdapterData))]
-        public async Task ConnectionCanReadAndWrite(ListenOptions listenOptions)
+        [Fact]
+        public async Task ConnectionCanReadAndWrite()
         {
-            var serviceContext = new TestServiceContext();
-            listenOptions.UseHttpServer(listenOptions.ConnectionAdapters, serviceContext, new DummyApplication(TestApp.EchoApp), HttpProtocols.Http1);
+            var endpoint = new IPEndPoint(IPAddress.Loopback, 0);
 
             var transportContext = new TestLibuvTransportContext();
-            var transport = new LibuvConnectionListener(transportContext, listenOptions.EndPoint);
+            var transport = new LibuvConnectionListener(transportContext, endpoint);
 
             await transport.BindAsync();
-            listenOptions.EndPoint = transport.EndPoint;
+            endpoint = (IPEndPoint)transport.EndPoint;
 
-            var dispatcher = new ConnectionDispatcher(serviceContext, listenOptions.Build());
-            _ = dispatcher.StartAcceptingConnections(transport);
+            async Task EchoServerAsync()
+            {
+                await using var connection = await transport.AcceptAsync();
 
-            using (var socket = TestConnection.CreateConnectedLoopbackSocket(listenOptions.IPEndPoint.Port))
+                while (true)
+                {
+                    var result = await connection.Transport.Input.ReadAsync();
+
+                    if (result.IsCompleted)
+                    {
+                        break;
+                    }
+                    await connection.Transport.Output.WriteAsync(result.Buffer.ToArray());
+
+                    connection.Transport.Input.AdvanceTo(result.Buffer.End);
+                }
+            }
+
+            var serverTask = EchoServerAsync();
+
+            using (var socket = TestConnection.CreateConnectedLoopbackSocket(endpoint.Port))
             {
-                var data = "Hello World";
-                socket.Send(Encoding.ASCII.GetBytes($"POST / HTTP/1.0\r\nContent-Length: 11\r\n\r\n{data}"));
+                var data = Encoding.ASCII.GetBytes("Hello World");
+                socket.Send(data);
+
                 var buffer = new byte[data.Length];
                 var read = 0;
                 while (read < data.Length)
                 {
                     read += socket.Receive(buffer, read, buffer.Length - read, SocketFlags.None);
                 }
+
+                Assert.Equal(data, buffer);
             }
 
-            
-            Assert.True(await serviceContext.ConnectionManager.CloseAllConnectionsAsync(new CancellationTokenSource(TestConstants.DefaultTimeout).Token));
-            await transport.UnbindAsync();
-            await transport.StopThreadsAsync();
+            await serverTask.DefaultTimeout();
+
+            await transport.StopAsync();
+            await transport.DisposeAsync();
+        }
+
+        [Fact]
+        public async Task UnacceptedConnectionsAreAborted()
+        {
+            var endpoint = new IPEndPoint(IPAddress.Loopback, 0);
+
+            var transportContext = new TestLibuvTransportContext();
+            var transport = new LibuvConnectionListener(transportContext, endpoint);
+
+            await transport.BindAsync();
+            endpoint = (IPEndPoint)transport.EndPoint;
+
+            async Task ConnectAsync()
+            {
+                using (var socket = TestConnection.CreateConnectedLoopbackSocket(endpoint.Port))
+                {
+                    var read = await socket.ReceiveAsync(new byte[10], SocketFlags.None);
+                    Assert.Equal(0, read);
+                }
+            }
+
+            var connectTask = ConnectAsync();
+
+            await transport.StopAsync();
+            await transport.DisposeAsync();
+
+            // The connection was accepted because libuv eagerly accepts connections
+            // they sit in a queue in each listener, we want to make sure that resources
+            // are cleaned up if they are never accepted by the caller
+
+            await connectTask.DefaultTimeout();
         }
 
         [ConditionalTheory]
@@ -132,7 +172,7 @@ public async Task OneToTenThreads(int threadCount)
                 }
             }
 
-            await transport.UnbindAsync().ConfigureAwait(false);
+            await transport.StopAsync().ConfigureAwait(false);
 
             await acceptTask.ConfigureAwait(false);
 
@@ -141,7 +181,7 @@ public async Task OneToTenThreads(int threadCount)
                 await serviceContext.ConnectionManager.AbortAllConnectionsAsync().ConfigureAwait(false);
             }
 
-            await transport.StopThreadsAsync().ConfigureAwait(false);
+            await transport.DisposeAsync().ConfigureAwait(false);
         }
     }
 }

Original file line number	Diff line number	Diff line change
`@@ -197,7 +197,6 @@ protected virtual void DispatchConnection(UvStreamHandle socket)`
`197`	`197`
`198`	`198`	`public virtual async Task DisposeAsync()`
`199`	`199`	`{`
`200`		`- StopAcceptingConnections();`
`201`	`200`	`// Ensure the event loop is still running.`
`202`	`201`	`// If the event loop isn't running and we try to wait on this Post`
`203`	`202`	`// to complete, then LibuvTransport will never be disposed and`
`@@ -210,6 +209,8 @@ await Thread.PostAsync(listener =>`
`210`	`209`
`211`	`210`	`listener._closed = true;`
`212`	`211`
	`212`	`+ listener.StopAcceptingConnections();`
	`213`	`+`
`213`	`214`	`}, this).ConfigureAwait(false);`
`214`	`215`	`}`
`215`	`216`
Original file line number	Diff line number	Diff line change
`@@ -177,7 +177,6 @@ private void FreeBuffer()`
`177`	`177`
`178`	`178`	`public async Task DisposeAsync()`
`179`	`179`	`{`
`180`		`- StopAcceptingConnections();`
`181`	`180`	`// Ensure the event loop is still running.`
`182`	`181`	`// If the event loop isn't running and we try to wait on this Post`
`183`	`182`	`// to complete, then LibuvTransport will never be disposed and`
`@@ -191,11 +190,15 @@ await Thread.PostAsync(listener =>`
`191`	`190`
`192`	`191`	`listener._closed = true;`
`193`	`192`
	`193`	`+ listener.StopAcceptingConnections();`
	`194`	`+`
`194`	`195`	`}, this).ConfigureAwait(false);`
`195`	`196`	`}`
`196`	`197`	`else`
`197`	`198`	`{`
`198`	`199`	`FreeBuffer();`
	`200`	`+`
	`201`	`+ StopAcceptingConnections();`
`199`	`202`	`}`
`200`	`203`	`}`
`201`	`204`	`}`