Revert "Delay socket receive/send until first read/flush (#34458)" (#35284)

This reverts commit e2acbb9.

Author: davidfowl

src/Servers/Kestrel/Transport.Sockets/src/Client/SocketConnectionFactory.cs

@@ -81,6 +81,7 @@ public async ValueTask<ConnectionContext> ConnectAsync(EndPoint endpoint, Cancel
                 _outputOptions,
                 _options.WaitForDataBeforeAllocatingBuffer);
 
+            socketConnection.Start();
             return socketConnection;
         }
 

src/Servers/Kestrel/Transport.Sockets/src/Internal/SocketConnection.DuplexPipe.cs

@@ -33,7 +33,6 @@ internal sealed partial class SocketConnection : TransportConnection
         private readonly TaskCompletionSource _waitForConnectionClosedTcs = new TaskCompletionSource();
         private bool _connectionClosed;
         private readonly bool _waitForData;
-        private int _connectionStarted;
 
         internal SocketConnection(Socket socket,
                                   MemoryPool<byte> memoryPool,
@@ -68,32 +67,31 @@ internal SocketConnection(Socket socket,
 
             var pair = DuplexPipe.CreateConnectionPair(inputOptions, outputOptions);
 
-            _originalTransport = pair.Transport;
+            // Set the transport and connection id
+            Transport = _originalTransport = pair.Transport;
             Application = pair.Application;
 
-            Transport = new SocketDuplexPipe(this);
-
             InitializeFeatures();
         }
 
-        public IDuplexPipe InnerTransport => _originalTransport;
-
         public PipeWriter Input => Application.Output;
 
         public PipeReader Output => Application.Input;
 
         public override MemoryPool<byte> MemoryPool { get; }
 
-        private void EnsureStarted()
+        public void Start()
         {
-            if (_connectionStarted == 1 || Interlocked.CompareExchange(ref _connectionStarted, 1, 0) == 1)
+            try
             {
-                return;
+                // Spawn send and receive logic
+                _receivingTask = DoReceive();
+                _sendingTask = DoSend();
+            }
+            catch (Exception ex)
+            {
+                _trace.LogError(0, ex, $"Unexpected exception in {nameof(SocketConnection)}.{nameof(Start)}.");
             }
-
-            // Offload these to avoid potentially blocking the first read/write/flush
-            _receivingTask = Task.Run(DoReceive);
-            _sendingTask = Task.Run(DoSend);
         }
 
         public override void Abort(ConnectionAbortedException abortReason)
@@ -108,9 +106,6 @@ public override void Abort(ConnectionAbortedException abortReason)
         // Only called after connection middleware is complete which means the ConnectionClosed token has fired.
         public override async ValueTask DisposeAsync()
         {
-            // Just in case we haven't started the connection, start it here so we can clean up properly.
-            EnsureStarted();
-
             _originalTransport.Input.Complete();
             _originalTransport.Output.Complete();
 
@@ -130,7 +125,7 @@ public override async ValueTask DisposeAsync()
             }
             catch (Exception ex)
             {
-                _trace.LogError(0, ex, $"Unexpected exception in {nameof(SocketConnection)}.");
+                _trace.LogError(0, ex, $"Unexpected exception in {nameof(SocketConnection)}.{nameof(Start)}.");
             }
             finally
             {

src/Servers/Kestrel/Transport.Sockets/src/Internal/SocketConnection.PipeReader.cs

@@ -136,6 +136,8 @@ internal void Bind()
                         setting.OutputOptions,
                         waitForData: _options.WaitForDataBeforeAllocatingBuffer);
 
+                    connection.Start();
+
                     _settingsIndex = (_settingsIndex + 1) % _settingsCount;
 
                     return connection;