Fixed regression caused by transport refactoring (#1573)

* Fixed regression caused by transport refactoring
- Libuv should not close the socket until it has started it. Before this was enforced
by calling ReadStart before starting the frame but the flow has changed. Now that closing the connection
is communicated via the pipe, we need to start consuming writes after calling ReadStart.
- Renamed OnSocketClosed to Close and moved dispose and logging into that method.

Fixes #1571

src/Microsoft.AspNetCore.Server.Kestrel.Transport.Libuv/Internal/Http/Connection.cs

@@ -21,7 +21,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Internal.Http
 
         private static readonly Action<UvStreamHandle, int, object> _readCallback =
             (handle, status, state) => ReadCallback(handle, status, state);
-        
+
         private static readonly Func<UvStreamHandle, int, object, LibuvFunctions.uv_buf_t> _allocCallback =
             (handle, suggestedsize, state) => AllocCallback(handle, suggestedsize, state);
 
@@ -80,6 +80,11 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Internal.Http
                 // Start socket prior to applying the ConnectionAdapter
                 _socket.ReadStart(_allocCallback, _readCallback, this);
                 _lastTimestamp = Thread.Loop.Now();
+
+                // This *must* happen after socket.ReadStart
+                // The socket output consumer is the only thing that can close the connection. If the 
+                // output pipe is already closed by the time we start then it's fine since, it'll close gracefully afterwards.
+                var ignore = Output.StartWrites();
             }
             catch (Exception e)
             {
@@ -100,8 +105,12 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Internal.Http
         }
 
         // Called on Libuv thread
-        public virtual void OnSocketClosed()
+        public virtual void Close()
         {
+            _socket.Dispose();
+
+            Log.ConnectionStop(ConnectionId);
+
             KestrelEventSource.Log.ConnectionStop(this);
 
             Input.Complete(new TaskCanceledException("The request was aborted"));

src/Microsoft.AspNetCore.Server.Kestrel.Transport.Libuv/Internal/Http/SocketOutputConsumer.cs

@@ -37,8 +37,6 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Internal.Http
             _connectionId = connectionId;
             _log = log;
             _writeReqPool = thread.WriteReqPool;
-
-            var ignore = StartWrites();
         }
 
         public async Task StartWrites()
@@ -56,7 +54,6 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Internal.Http
                         var writeResult = await writeReq.WriteAsync(_socket, buffer);
                         _writeReqPool.Return(writeReq);
 
-                        // REVIEW: Locking here, do we need to take the context lock?
                         OnWriteCompleted(writeResult.Status, writeResult.Error);
                     }
 
@@ -80,9 +77,8 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Internal.Http
             // We're done reading
             _pipe.Complete();
 
-            _socket.Dispose();
-            _connection.OnSocketClosed();
-            _log.ConnectionStop(_connectionId);
+            // Close the connection
+            _connection.Close();
         }
 
         private void OnWriteCompleted(int writeStatus, Exception writeError)
@@ -124,13 +120,13 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Internal.Http
             var shutdownReq = new UvShutdownReq(_log);
             shutdownReq.Init(_thread.Loop);
             shutdownReq.Shutdown(_socket, (req, status, state) =>
-                {
-                    req.Dispose();
-                    _log.ConnectionWroteFin(_connectionId, status);
+            {
+                req.Dispose();
+                _log.ConnectionWroteFin(_connectionId, status);
 
-                    tcs.TrySetResult(null);
-                },
-                this);
+                tcs.TrySetResult(null);
+            },
+            this);
 
             return tcs.Task;
         }

test/Microsoft.AspNetCore.Server.KestrelTests/SocketOutputTests.cs

@@ -527,6 +527,7 @@ namespace Microsoft.AspNetCore.Server.KestrelTests
             var socket = new MockSocket(_mockLibuv, _kestrelThread.Loop.ThreadId, new TestKestrelTrace());
             var socketOutput = new SocketOutputProducer(pipe.Writer, frame, "0", trace);
             var consumer = new SocketOutputConsumer(pipe.Reader, _kestrelThread, socket, connection ?? new MockConnection(), "0", trace);
+            var ignore = consumer.StartWrites();
 
             return socketOutput;
         }

test/shared/MockConnection.cs

@@ -27,7 +27,7 @@ namespace Microsoft.AspNetCore.Testing
             return TaskCache.CompletedTask;
         }
 
-        public override void OnSocketClosed()
+        public override void Close()
         {
             _socketClosedTcs.SetResult(null);
         }