Remove lazily initialization of the pipes (#1850)

- This fixes race condition

src/Microsoft.AspNetCore.Http.Connections/HttpConnectionDispatcher.cs

@@ -379,7 +379,7 @@ namespace Microsoft.AspNetCore.Http.Connections
             context.Response.ContentType = "application/json";
 
             // Establish the connection
-            var connection = _manager.CreateConnection();
+            var connection = CreateConnection(options);
 
             // Set the Connection ID on the logging scope so that logs from now on will have the
             // Connection ID metadata set.
@@ -602,24 +602,9 @@ namespace Microsoft.AspNetCore.Http.Connections
                 return null;
             }
 
-            EnsureConnectionStateInternal(connection, options);
-
             return connection;
         }
 
-        private void EnsureConnectionStateInternal(HttpConnectionContext connection, HttpConnectionOptions options)
-        {
-            // If the connection doesn't have a pipe yet then create one, we lazily create the pipe to save on allocations until the client actually connects
-            if (connection.Transport == null)
-            {
-                var transportPipeOptions = new PipeOptions(pauseWriterThreshold: options.TransportMaxBufferSize, resumeWriterThreshold: options.TransportMaxBufferSize / 2, readerScheduler: PipeScheduler.ThreadPool, useSynchronizationContext: false);
-                var appPipeOptions = new PipeOptions(pauseWriterThreshold: options.ApplicationMaxBufferSize, resumeWriterThreshold: options.ApplicationMaxBufferSize / 2, readerScheduler: PipeScheduler.ThreadPool, useSynchronizationContext: false);
-                var pair = DuplexPipe.CreateConnectionPair(transportPipeOptions, appPipeOptions);
-                connection.Transport = pair.Application;
-                connection.Application = pair.Transport;
-            }
-        }
-
         // This is only used for WebSockets connections, which can connect directly without negotiating
         private async Task<HttpConnectionContext> GetOrCreateConnectionAsync(HttpContext context, HttpConnectionOptions options)
         {
@@ -629,7 +614,7 @@ namespace Microsoft.AspNetCore.Http.Connections
             // There's no connection id so this is a brand new connection
             if (StringValues.IsNullOrEmpty(connectionId))
             {
-                connection = _manager.CreateConnection();
+                connection = CreateConnection(options);
             }
             else if (!_manager.TryGetConnection(connectionId, out connection))
             {
@@ -639,11 +624,17 @@ namespace Microsoft.AspNetCore.Http.Connections
                 return null;
             }
 
-            EnsureConnectionStateInternal(connection, options);
-
             return connection;
         }
 
+        private HttpConnectionContext CreateConnection(HttpConnectionOptions options)
+        {
+            var transportPipeOptions = new PipeOptions(pauseWriterThreshold: options.TransportMaxBufferSize, resumeWriterThreshold: options.TransportMaxBufferSize / 2, readerScheduler: PipeScheduler.ThreadPool, useSynchronizationContext: false);
+            var appPipeOptions = new PipeOptions(pauseWriterThreshold: options.ApplicationMaxBufferSize, resumeWriterThreshold: options.ApplicationMaxBufferSize / 2, readerScheduler: PipeScheduler.ThreadPool, useSynchronizationContext: false);
+
+            return _manager.CreateConnection(transportPipeOptions, appPipeOptions);
+        }
+
         private class EmptyServiceProvider : IServiceProvider
         {
             public static EmptyServiceProvider Instance { get; } = new EmptyServiceProvider();

src/Microsoft.AspNetCore.Http.Connections/HttpConnectionManager.cs

@@ -67,33 +67,30 @@ namespace Microsoft.AspNetCore.Http.Connections
             return false;
         }
 
+        public HttpConnectionContext CreateConnection()
+        {
+            return CreateConnection(PipeOptions.Default, PipeOptions.Default);
+        }
+
         /// <summary>
         /// Creates a connection without Pipes setup to allow saving allocations until Pipes are needed.
         /// </summary>
         /// <returns></returns>
-        public HttpConnectionContext CreateConnection()
+        public HttpConnectionContext CreateConnection(PipeOptions transportPipeOptions, PipeOptions appPipeOptions)
         {
             var id = MakeNewConnectionId();
 
             _logger.CreatedNewConnection(id);
             var connectionTimer = HttpConnectionsEventSource.Log.ConnectionStart(id);
-
             var connection = new HttpConnectionContext(id);
+            var pair = DuplexPipe.CreateConnectionPair(transportPipeOptions, appPipeOptions);
+            connection.Transport = pair.Application;
+            connection.Application = pair.Transport;
 
             _connections.TryAdd(id, (connection, connectionTimer));
             return connection;
         }
 
-        public HttpConnectionContext CreateConnection(PipeOptions transportPipeOptions, PipeOptions appPipeOptions)
-        {
-            var connection = CreateConnection();
-            var pair = DuplexPipe.CreateConnectionPair(transportPipeOptions, appPipeOptions);
-            connection.Application = pair.Transport;
-            connection.Transport = pair.Application;
-
-            return connection;
-        }
-
         public void RemoveConnection(string id)
         {
             if (_connections.TryRemove(id, out var pair))

test/Microsoft.AspNetCore.Http.Connections.Tests/HttpConnectionManagerTests.cs

@@ -24,7 +24,8 @@ namespace Microsoft.AspNetCore.Http.Connections.Tests
             Assert.Null(connection.TransportTask);
             Assert.Null(connection.Cancellation);
             Assert.NotEqual(default, connection.LastSeenUtc);
-            Assert.Null(connection.Transport);
+            Assert.NotNull(connection.Transport);
+            Assert.NotNull(connection.Application);
         }
 
         [Fact]
@@ -197,8 +198,8 @@ namespace Microsoft.AspNetCore.Http.Connections.Tests
             var connection = connectionManager.CreateConnection();
 
             Assert.NotNull(connection.ConnectionId);
-            Assert.Null(connection.Transport);
+            Assert.NotNull(connection.Transport);
-            Assert.Null(connection.Application);
+            Assert.NotNull(connection.Application);
 
             await connection.DisposeAsync();
             Assert.Equal(HttpConnectionContext.ConnectionStatus.Disposed, connection.Status);