Remove over engineered callback writing and just write chunks directly. (#6901)

src/Servers/Kestrel/Core/src/Internal/Http/Http1OutputProducer.cs

@@ -71,7 +71,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http
             return WriteAsync(Constants.EmptyData, cancellationToken);
         }
 
-        public Task WriteAsync<T>(Func<PipeWriter, T, long> callback, T state, CancellationToken cancellationToken)
+        public Task WriteChunkAsync(ReadOnlySpan<byte> buffer, CancellationToken cancellationToken)
         {
             lock (_contextLock)
             {
@@ -80,9 +80,17 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http
                     return Task.CompletedTask;
                 }
 
-                var buffer = _pipeWriter;
+                if (buffer.Length > 0)
-                var bytesCommitted = callback(buffer, state);
+                {
-                _unflushedBytes += bytesCommitted;
+                    var writer = new BufferWriter<PipeWriter>(_pipeWriter);
+
+                    writer.WriteBeginChunkBytes(buffer.Length);
+                    writer.Write(buffer);
+                    writer.WriteEndChunkBytes();
+                    writer.Commit();
+
+                    _unflushedBytes += writer.BytesCommitted;
+                }
             }
 
             return FlushAsync(cancellationToken);

src/Servers/Kestrel/Core/src/Internal/Http/HttpProtocol.cs

@@ -30,7 +30,6 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http
         private static readonly byte[] _bytesConnectionKeepAlive = Encoding.ASCII.GetBytes("\r\nConnection: keep-alive");
         private static readonly byte[] _bytesTransferEncodingChunked = Encoding.ASCII.GetBytes("\r\nTransfer-Encoding: chunked");
         private static readonly byte[] _bytesServer = Encoding.ASCII.GetBytes("\r\nServer: " + Constants.ServerName);
-        private static readonly Func<PipeWriter, ReadOnlyMemory<byte>, long> _writeChunk = WriteChunk;
 
         private readonly object _onStartingSync = new Object();
         private readonly object _onCompletedSync = new Object();
@@ -820,7 +819,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http
                     {
                         return !firstWrite ? Task.CompletedTask : FlushAsync(cancellationToken);
                     }
-                    return WriteChunkedAsync(data, cancellationToken);
+                    return WriteChunkedAsync(data.Span, cancellationToken);
                 }
                 else
                 {
@@ -851,7 +850,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http
                         return;
                     }
 
-                    await WriteChunkedAsync(data, cancellationToken);
+                    await WriteChunkedAsync(data.Span, cancellationToken);
                 }
                 else
                 {
@@ -936,27 +935,9 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http
             }
         }
 
-        private Task WriteChunkedAsync(ReadOnlyMemory<byte> data, CancellationToken cancellationToken)
+        private Task WriteChunkedAsync(ReadOnlySpan<byte> data, CancellationToken cancellationToken)
         {
-            return Output.WriteAsync(_writeChunk, data, cancellationToken);
+            return Output.WriteChunkAsync(data, cancellationToken);
-        }
-
-        private static long WriteChunk(PipeWriter writableBuffer, ReadOnlyMemory<byte> buffer)
-        {
-            var bytesWritten = 0L;
-            if (buffer.Length > 0)
-            {
-                var writer = new BufferWriter<PipeWriter>(writableBuffer);
-
-                writer.WriteBeginChunkBytes(buffer.Length);
-                writer.Write(buffer.Span);
-                writer.WriteEndChunkBytes();
-                writer.Commit();
-
-                bytesWritten = writer.BytesCommitted;
-            }
-
-            return bytesWritten;
         }
 
         public void ProduceContinue()

src/Servers/Kestrel/Core/src/Internal/Http/IHttpOutputProducer.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.IO.Pipelines;
 using System.Threading;
 using System.Threading.Tasks;
 
@@ -10,7 +9,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http
 {
     public interface IHttpOutputProducer
     {
-        Task WriteAsync<T>(Func<PipeWriter, T, long> callback, T state, CancellationToken cancellationToken);
+        Task WriteChunkAsync(ReadOnlySpan<byte> data, CancellationToken cancellationToken);
         Task FlushAsync(CancellationToken cancellationToken);
         Task Write100ContinueAsync();
         void WriteResponseHeaders(int statusCode, string ReasonPhrase, HttpResponseHeaders responseHeaders);

src/Servers/Kestrel/Core/src/Internal/Http2/Http2OutputProducer.cs

@@ -83,7 +83,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Internal.Http2
             Dispose();
         }
 
-        public Task WriteAsync<T>(Func<PipeWriter, T, long> callback, T state, CancellationToken cancellationToken)
+        public Task WriteChunkAsync(ReadOnlySpan<byte> span, CancellationToken cancellationToken)
         {
             throw new NotImplementedException();
         }

src/Servers/Kestrel/Core/test/OutputProducerTests.cs

@@ -1,4 +1,4 @@
-// Copyright (c) .NET Foundation. All rights reserved.
+// 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;
@@ -46,17 +46,11 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Tests
                 // Close
                 socketOutput.Dispose();
 
-                var called = false;
+                await socketOutput.WriteDataAsync(new byte[] { 1, 2, 3, 4 }, default);
 
-                await socketOutput.WriteAsync((buffer, state) =>
+                Assert.True(socketOutput.Pipe.Reader.TryRead(out var result));
-                {
+                Assert.True(result.IsCompleted);
-                    called = true;
+                Assert.True(result.Buffer.IsEmpty);
-                    return 0;
-                },
-                0,
-                default);
-
-                Assert.False(called);
             }
         }
 
@@ -80,7 +74,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Tests
             mockConnectionContext.Verify(f => f.Abort(null), Times.Once());
         }
 
-        private Http1OutputProducer CreateOutputProducer(
+        private TestHttpOutputProducer CreateOutputProducer(
             PipeOptions pipeOptions = null,
             ConnectionContext connectionContext = null)
         {
@@ -89,8 +83,8 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Tests
 
             var pipe = new Pipe(pipeOptions);
             var serviceContext = new TestServiceContext();
-            var socketOutput = new Http1OutputProducer(
+            var socketOutput = new TestHttpOutputProducer(
-                pipe.Writer,
+                pipe,
                 "0",
                 connectionContext,
                 serviceContext.Log,
@@ -99,5 +93,15 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Core.Tests
 
             return socketOutput;
         }
+
+        private class TestHttpOutputProducer : Http1OutputProducer
+        {
+            public TestHttpOutputProducer(Pipe pipe, string connectionId, ConnectionContext connectionContext, IKestrelTrace log, ITimeoutControl timeoutControl, IHttpMinResponseDataRateFeature minResponseDataRateFeature) : base(pipe.Writer, connectionId, connectionContext, log, timeoutControl, minResponseDataRateFeature)
+            {
+                Pipe = pipe;
+            }
+
+            public Pipe Pipe { get; }
+        }
     }
 }

src/Servers/Kestrel/Transport.Libuv/test/LibuvOutputConsumerTests.cs

@@ -304,13 +304,7 @@ namespace Microsoft.AspNetCore.Server.Kestrel.Transport.Libuv.Tests
                     Assert.NotEmpty(completeQueue);
 
                     // Add more bytes to the write-behind buffer to prevent the next write from
-                    _ = outputProducer.WriteAsync((writableBuffer, state) =>
+                    _ = outputProducer.WriteDataAsync(halfWriteBehindBuffer, default);
-                    {
-                        writableBuffer.Write(state);
-                        return state.Count;
-                    },
-                    halfWriteBehindBuffer,
-                    default);
 
                     // Act
                     var writeTask2 = outputProducer.WriteDataAsync(halfWriteBehindBuffer);