Revert "Revert the output pipe in the DuplexStreamPipeAdapter (#11601)"

This reverts commit 6de357e.

# Conflicts:
#	src/Servers/Kestrel/Core/src/Internal/Http2/Http2FrameWriter.cs
#	src/Servers/Kestrel/Core/src/Middleware/Internal/DuplexPipeStreamAdapter.cs

Author: halter73

src/Servers/Kestrel/Core/src/Middleware/HttpsConnectionMiddleware.cs

@@ -110,10 +110,7 @@ private async Task InnerOnConnectionAsync(ConnectionContext context)
 
             if (_options.ClientCertificateMode == ClientCertificateMode.NoCertificate)
             {
-                sslDuplexPipe = new SslDuplexPipe(context.Transport, inputPipeOptions, outputPipeOptions)
-                {
-                    Log = _logger
-                };
+                sslDuplexPipe = new SslDuplexPipe(context.Transport, inputPipeOptions, outputPipeOptions);
                 certificateRequired = false;
             }
             else
@@ -150,10 +147,7 @@ private async Task InnerOnConnectionAsync(ConnectionContext context)
                         }
 
                         return true;
-                    }))
-                {
-                    Log = _logger
-                };
+                    }));
 
                 certificateRequired = true;
             }

src/Servers/Kestrel/Core/src/Middleware/Internal/DuplexPipeStreamAdapter.cs

@@ -5,7 +5,6 @@
 using System.IO;
 using System.IO.Pipelines;
 using System.Threading.Tasks;
-using Microsoft.Extensions.Logging;
 
 namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal
 {
@@ -15,127 +14,36 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal
     /// <typeparam name="TStream"></typeparam>
     internal class DuplexPipeStreamAdapter<TStream> : DuplexPipeStream, IDuplexPipe where TStream : Stream
     {
-        private readonly Pipe _output;
-        private Task _outputTask;
-        private bool _disposed;
-        private readonly object _disposeLock = new object();
-
         public DuplexPipeStreamAdapter(IDuplexPipe duplexPipe, Func<Stream, TStream> createStream) :
             this(duplexPipe, new StreamPipeReaderOptions(leaveOpen: true), new StreamPipeWriterOptions(leaveOpen: true), createStream)
         {
         }
 
-        public DuplexPipeStreamAdapter(IDuplexPipe duplexPipe, StreamPipeReaderOptions readerOptions, StreamPipeWriterOptions writerOptions, Func<Stream, TStream> createStream) :
-            base(duplexPipe.Input, duplexPipe.Output)
+        public DuplexPipeStreamAdapter(IDuplexPipe duplexPipe, StreamPipeReaderOptions readerOptions, StreamPipeWriterOptions writerOptions, Func<Stream, TStream> createStream) : base(duplexPipe.Input, duplexPipe.Output)
         {
             Stream = createStream(this);
-
-            var outputOptions = new PipeOptions(pool: writerOptions.Pool,
-                                                readerScheduler: PipeScheduler.Inline,
-                                                writerScheduler: PipeScheduler.Inline,
-                                                pauseWriterThreshold: 1,
-                                                resumeWriterThreshold: 1,
-                                                minimumSegmentSize: writerOptions.MinimumBufferSize,
-                                                useSynchronizationContext: false);
-
             Input = PipeReader.Create(Stream, readerOptions);
-
-            // We're using a pipe here because the HTTP/2 stack in Kestrel currently makes assumptions
-            // about when it is ok to write to the PipeWriter. This should be reverted back to PipeWriter.Create once
-            // those patterns are fixed.
-            _output = new Pipe(outputOptions);
+            Output = PipeWriter.Create(Stream, writerOptions);
         }
 
-        public ILogger Log { get; set; }
-
         public TStream Stream { get; }
 
         public PipeReader Input { get; }
 
-        public PipeWriter Output
-        {
-            get
-            {
-                if (_outputTask == null)
-                {
-                    _outputTask = WriteOutputAsync();
-                }
-
-                return _output.Writer;
-            }
-        }
+        public PipeWriter Output { get; }
 
-        public override ValueTask DisposeAsync()
+        protected override void Dispose(bool disposing)
         {
-            lock (_disposeLock)
-            {
-                if (_disposed)
-                {
-                    return default;
-                }
-                _disposed = true;
-            }
-
             Input.Complete();
-            _output.Writer.Complete();
-
-            if (_outputTask == null || _outputTask.IsCompletedSuccessfully)
-            {
-                // Wait for the output task to complete, this ensures that we've copied
-                // the application data to the underlying stream
-                return default;
-            }
-
-            return new ValueTask(_outputTask);
+            Output.Complete();
+            base.Dispose(disposing);
         }
 
-        private async Task WriteOutputAsync()
+        public override ValueTask DisposeAsync()
         {
-            try
-            {
-                while (true)
-                {
-                    var result = await _output.Reader.ReadAsync();
-                    var buffer = result.Buffer;
-
-                    try
-                    {
-                        if (buffer.IsEmpty)
-                        {
-                            if (result.IsCompleted)
-                            {
-                                break;
-                            }
-
-                            await Stream.FlushAsync();
-                        }
-                        else if (buffer.IsSingleSegment)
-                        {
-                            await Stream.WriteAsync(buffer.First);
-                        }
-                        else
-                        {
-                            foreach (var memory in buffer)
-                            {
-                                await Stream.WriteAsync(memory);
-                            }
-                        }
-                    }
-                    finally
-                    {
-                        _output.Reader.AdvanceTo(buffer.End);
-                    }
-                }
-            }
-            catch (Exception ex)
-            {
-                Log?.LogCritical(0, ex, $"{GetType().Name}.{nameof(WriteOutputAsync)}");
-            }
-            finally
-            {
-                _output.Reader.Complete();
-            }
+            Input.Complete();
+            Output.Complete();
+            return base.DisposeAsync();
         }
     }
 }
-

src/Servers/Kestrel/Core/src/Middleware/LoggingConnectionMiddleware.cs

@@ -44,7 +44,6 @@ private class LoggingDuplexPipe : DuplexPipeStreamAdapter<LoggingStream>
             public LoggingDuplexPipe(IDuplexPipe transport, ILogger logger) :
                 base(transport, stream => new LoggingStream(stream, logger))
             {
-                Log = logger;
             }
         }
     }

src/Servers/Kestrel/test/InMemory.FunctionalTests/Http2/Http2TestBase.cs

@@ -444,22 +444,7 @@ protected async Task InitializeConnectionAsync(RequestDelegate application, int
                 CreateConnection();
             }
 
-            var connectionTask = _connection.ProcessRequestsAsync(new DummyApplication(application));
-
-            async Task CompletePipeOnTaskCompletion()
-            {
-                try
-                {
-                    await connectionTask;
-                }
-                finally
-                {
-                    _pair.Transport.Input.Complete();
-                    _pair.Transport.Output.Complete();
-                }
-            }
-
-            _connectionTask = CompletePipeOnTaskCompletion();
+            _connectionTask = _connection.ProcessRequestsAsync(new DummyApplication(application));
 
             await SendPreambleAsync().ConfigureAwait(false);
             await SendSettingsAsync();