- Use PipeWriter to write files

- Use Async Disposable for FileStream
- Use PipeWriter in FileResult
- Call StartAsync before using the PipeWriter
- Add SendFileAsync to HttpProtocol and call StartAsync in there to make sure we're not buffering more than we need to.
- Use the array pool as a fallback for large buffers
- Max buffer size of 64K

Author: Kahbazi

src/Hosting/TestHost/src/ResponseFeature.cs

@@ -146,7 +146,7 @@ public void DisableBuffering()
 
         public Task SendFileAsync(string path, long offset, long? count, CancellationToken cancellation)
         {
-            return SendFileFallback.SendFileAsync(Stream, path, offset, count, cancellation);
+            return SendFileFallback.SendFileAsync(Writer, path, offset, count, cancellation);
         }
 
         public Task CompleteAsync()

src/Http/Http.Extensions/src/Microsoft.AspNetCore.Http.Extensions.csproj

@@ -1,4 +1,4 @@
-<Project Sdk="Microsoft.NET.Sdk">
+<Project Sdk="Microsoft.NET.Sdk">
 
   <PropertyGroup>
     <Description>ASP.NET Core common extension methods for HTTP abstractions, HTTP headers, HTTP request/response, and session state.</Description>

src/Http/Http.Extensions/src/SendFileResponseExtensions.cs

@@ -5,7 +5,6 @@
 using System.IO;
 using System.Threading;
 using System.Threading.Tasks;
-using Microsoft.AspNetCore.Http.Extensions;
 using Microsoft.AspNetCore.Http.Features;
 using Microsoft.Extensions.FileProviders;
 
@@ -16,8 +15,6 @@ namespace Microsoft.AspNetCore.Http
     /// </summary>
     public static class SendFileResponseExtensions
     {
-        private const int StreamCopyBufferSize = 64 * 1024;
-
         /// <summary>
         /// Sends the given file using the SendFile extension.
         /// </summary>
@@ -124,13 +121,13 @@ private static async Task SendFileAsyncCore(HttpResponse response, IFileInfo fil
                     {
                         fileContent.Seek(offset, SeekOrigin.Begin);
                     }
-                    await StreamCopyOperation.CopyToAsync(fileContent, response.Body, count, StreamCopyBufferSize, localCancel);
+                    await StreamCopyOperationInternal.CopyToAsync(fileContent, response.BodyWriter, count, localCancel);
                 }
                 catch (OperationCanceledException) when (useRequestAborted) { }
             }
             else
             {
-                await response.SendFileAsync(file.PhysicalPath, offset, count, cancellationToken);
+                await SendFileAsyncCore(response, file.PhysicalPath, offset, count, cancellationToken);
             }
         }
 

src/Http/Http.Extensions/src/StreamCopyOperation.cs

@@ -2,6 +2,7 @@
 // Licensed under the Apache License, Version 2.0. See License.txt in the project root for license information.
 
 using System.IO;
+using System.IO.Pipelines;
 using System.Threading;
 using System.Threading.Tasks;
 
@@ -28,5 +29,14 @@ public static Task CopyToAsync(Stream source, Stream destination, long? count, C
         /// <param name="cancel">The token to monitor for cancellation requests. The default value is <see cref="P:System.Threading.CancellationToken.None" />.</param>
         public static Task CopyToAsync(Stream source, Stream destination, long? count, int bufferSize, CancellationToken cancel)
             => StreamCopyOperationInternal.CopyToAsync(source, destination, count, bufferSize, cancel);
+
+        /// <summary>Asynchronously reads the given number of bytes from the source stream and writes them using pipe writer.</summary>
+        /// <returns>A task that represents the asynchronous copy operation.</returns>
+        /// <param name="source">The stream from which the contents will be copied.</param>
+        /// <param name="writer">The PipeWriter to which the contents of the current stream will be copied.</param>
+        /// <param name="count">The count of bytes to be copied.</param>
+        /// <param name="cancel">The token to monitor for cancellation requests. The default value is <see cref="P:System.Threading.CancellationToken.None" />.</param>
+        public static Task CopyToAsync(Stream source, PipeWriter writer, long? count, CancellationToken cancel)
+            => StreamCopyOperationInternal.CopyToAsync(source, writer, count, cancel);
     }
 }

src/Http/Http/src/SendFileFallback.cs

@@ -3,6 +3,7 @@
 
 using System;
 using System.IO;
+using System.IO.Pipelines;
 using System.Threading;
 using System.Threading.Tasks;
 
@@ -20,6 +21,31 @@ public static class SendFileFallback
         /// <param name="cancellationToken">A <see cref="CancellationToken"/> used to abort the transmission.</param>
         /// <returns></returns>
         public static async Task SendFileAsync(Stream destination, string filePath, long offset, long? count, CancellationToken cancellationToken)
+        {
+            await using FileStream fileStream = GetFileStream(filePath, offset, count, cancellationToken);
+
+            fileStream.Seek(offset, SeekOrigin.Begin);
+            await StreamCopyOperationInternal.CopyToAsync(fileStream, destination, count, cancellationToken);
+        }
+
+        /// <summary>
+        /// Write the segment of the file using pipe writer.
+        /// </summary>
+        /// <param name="writer"></param>
+        /// <param name="filePath">The full disk path to the file.</param>
+        /// <param name="offset">The offset in the file to start at.</param>
+        /// <param name="count">The number of bytes to send, or null to send the remainder of the file.</param>
+        /// <param name="cancellationToken">A <see cref="CancellationToken"/> used to abort the transmission.</param>
+        /// <returns></returns>
+        public static async Task SendFileAsync(PipeWriter writer, string filePath, long offset, long? count, CancellationToken cancellationToken)
+        {
+            await using FileStream fileStream = GetFileStream(filePath, offset, count, cancellationToken);
+
+            fileStream.Seek(offset, SeekOrigin.Begin);
+            await StreamCopyOperationInternal.CopyToAsync(fileStream, writer, count, cancellationToken);
+        }
+
+        private static FileStream GetFileStream(string filePath, long offset, long? count, CancellationToken cancellationToken)
         {
             var fileInfo = new FileInfo(filePath);
             if (offset < 0 || offset > fileInfo.Length)
@@ -34,21 +60,15 @@ public static async Task SendFileAsync(Stream destination, string filePath, long
 
             cancellationToken.ThrowIfCancellationRequested();
 
-            int bufferSize = 1024 * 16;
+            int bufferSize = 1;
 
-            var fileStream = new FileStream(
+            return new FileStream(
                 filePath,
                 FileMode.Open,
                 FileAccess.Read,
                 FileShare.ReadWrite,
                 bufferSize: bufferSize,
                 options: FileOptions.Asynchronous | FileOptions.SequentialScan);
-
-            using (fileStream)
-            {
-                fileStream.Seek(offset, SeekOrigin.Begin);
-                await StreamCopyOperationInternal.CopyToAsync(fileStream, destination, count, bufferSize, cancellationToken);
-            }
         }
     }
 }

src/Http/Shared/StreamCopyOperationInternal.cs

@@ -5,6 +5,7 @@
 using System.Buffers;
 using System.Diagnostics;
 using System.IO;
+using System.IO.Pipelines;
 using System.Threading;
 using System.Threading.Tasks;
 
@@ -13,7 +14,7 @@ namespace Microsoft.AspNetCore.Http
     // FYI: In most cases the source will be a FileStream and the destination will be to the network.
     internal static class StreamCopyOperationInternal
     {
-        private const int DefaultBufferSize = 4096;
+        private const int DefaultBufferSize = 4 * 1024;
 
         /// <summary>Asynchronously reads the given number of bytes from the source stream and writes them to another stream.</summary>
         /// <returns>A task that represents the asynchronous copy operation.</returns>
@@ -42,13 +43,13 @@ public static async Task CopyToAsync(Stream source, Stream destination, long? co
             {
                 Debug.Assert(source != null);
                 Debug.Assert(destination != null);
-                Debug.Assert(!bytesRemaining.HasValue || bytesRemaining.GetValueOrDefault() >= 0);
+                Debug.Assert(!bytesRemaining.HasValue || bytesRemaining.Value >= 0);
                 Debug.Assert(buffer != null);
 
                 while (true)
                 {
                     // The natural end of the range.
-                    if (bytesRemaining.HasValue && bytesRemaining.GetValueOrDefault() <= 0)
+                    if (bytesRemaining.HasValue && bytesRemaining.Value <= 0)
                     {
                         return;
                     }
@@ -58,7 +59,7 @@ public static async Task CopyToAsync(Stream source, Stream destination, long? co
                     int readLength = buffer.Length;
                     if (bytesRemaining.HasValue)
                     {
-                        readLength = (int)Math.Min(bytesRemaining.GetValueOrDefault(), (long)readLength);
+                        readLength = (int)Math.Min(bytesRemaining.Value, (long)readLength);
                     }
                     int read = await source.ReadAsync(buffer, 0, readLength, cancel);
 
@@ -83,5 +84,66 @@ public static async Task CopyToAsync(Stream source, Stream destination, long? co
                 ArrayPool<byte>.Shared.Return(buffer);
             }
         }
+
+        /// <summary>Asynchronously reads the given number of bytes from the source stream and writes them using pipe writer.</summary>
+        /// <returns>A task that represents the asynchronous copy operation.</returns>
+        /// <param name="source">The stream from which the contents will be copied.</param>
+        /// <param name="writer">The PipeWriter to which the contents of the current stream will be copied.</param>
+        /// <param name="count">The count of bytes to be copied.</param>
+        /// <param name="cancel">The token to monitor for cancellation requests.</param>
+        public static Task CopyToAsync(Stream source, PipeWriter writer, long? count, CancellationToken cancel)
+        {
+            if (count == null)
+            {
+                // No length, do a copy with the default buffer size (based on whatever the pipe settings are, default is 4K)
+                return source.CopyToAsync(writer, cancel);
+            }
+
+            static async Task CopyToAsync(Stream source, PipeWriter writer, long bytesRemaining, CancellationToken cancel)
+            {
+                // The array pool likes powers of 2
+                const int maxBufferSize = 64 * 1024;
+                const int minBufferSize = 1024;
+
+                // We know exactly how much we're going to copy
+                while (bytesRemaining > 0)
+                {
+                    var bufferSize = (int)Math.Clamp(bytesRemaining, minBufferSize, maxBufferSize);
+
+                    // The natural end of the range.
+                    var memory = writer.GetMemory(bufferSize);
+
+                    if (memory.Length > bytesRemaining)
+                    {
+                        memory = memory.Slice(0, (int)bytesRemaining);
+                    }
+
+                    var read = await source.ReadAsync(memory, cancel);
+
+                    bytesRemaining -= read;
+
+                    // End of the source stream.
+                    if (read == 0)
+                    {
+                        break;
+                    }
+
+                    writer.Advance(read);
+
+                    var result = await writer.FlushAsync(cancel);
+
+                    if (result.IsCompleted)
+                    {
+                        break;
+                    }
+                }
+            }
+
+            Debug.Assert(source != null);
+            Debug.Assert(writer != null);
+            Debug.Assert(count >= 0);
+
+            return CopyToAsync(source, writer, count.Value, cancel);
+        }
     }
 }

src/Middleware/StaticFiles/startvs.cmd

@@ -0,0 +1,3 @@
+@ECHO OFF
+
+%~dp0..\..\..\startvs.cmd %~dp0StaticFiles.slnf

src/Mvc/Mvc.Core/src/Infrastructure/FileResultExecutorBase.cs

@@ -360,19 +360,19 @@ protected static ILogger CreateLogger<T>(ILoggerFactory factory)
 
         protected static async Task WriteFileAsync(HttpContext context, Stream fileStream, RangeItemHeaderValue range, long rangeLength)
         {
-            var outputStream = context.Response.Body;
-            using (fileStream)
+            var outputPipeWriter = context.Response.BodyWriter;
+            await using (fileStream)
             {
                 try
                 {
                     if (range == null)
                     {
-                        await StreamCopyOperation.CopyToAsync(fileStream, outputStream, count: null, bufferSize: BufferSize, cancel: context.RequestAborted);
+                        await StreamCopyOperation.CopyToAsync(fileStream, outputPipeWriter, count: null, cancel: context.RequestAborted);
                     }
                     else
                     {
                         fileStream.Seek(range.From.Value, SeekOrigin.Begin);
-                        await StreamCopyOperation.CopyToAsync(fileStream, outputStream, rangeLength, BufferSize, context.RequestAborted);
+                        await StreamCopyOperation.CopyToAsync(fileStream, outputPipeWriter, rangeLength, context.RequestAborted);
                     }
                 }
                 catch (OperationCanceledException)

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

@@ -70,7 +70,7 @@ internal class Http1OutputProducer : IHttpOutputProducer, IDisposable
         // These should be cleared by the end of the request
         private List<CompletedBuffer> _completedSegments;
         private Memory<byte> _currentSegment;
-        private IMemoryOwner<byte> _currentSegmentOwner;
+        private object _currentSegmentOwner;
         private int _position;
         private bool _startCalled;
 
@@ -436,7 +436,15 @@ public void Dispose()
 
         private void DisposeCurrentSegment()
         {
-            _currentSegmentOwner?.Dispose();
+            if (_currentSegmentOwner is IMemoryOwner<byte> owner)
+            {
+                owner.Dispose();
+            }
+            else if (_currentSegmentOwner is byte[] array)
+            {
+                ArrayPool<byte>.Shared.Return(array);
+            }
+
             _currentSegmentOwner = null;
             _currentSegment = default;
         }
@@ -600,13 +608,13 @@ private Memory<byte> GetChunkedMemory(int sizeHint)
                 // Calculating ChunkWriter.GetBeginChunkByteCount isn't free, so instead, we can add
                 // the max length for the prefix and suffix and add it to the sizeHint.
                 // This still guarantees that the memory passed in will be larger than the sizeHint.
-                sizeHint += MaxBeginChunkLength + EndChunkLength; 
+                sizeHint += MaxBeginChunkLength + EndChunkLength;
                 UpdateCurrentChunkMemory(sizeHint);
             }
             // Check if we need to allocate a new memory.
             else if (_advancedBytesForChunk >= _currentChunkMemory.Length - _currentMemoryPrefixBytes - EndChunkLength - sizeHint && _advancedBytesForChunk > 0)
             {
-                sizeHint += MaxBeginChunkLength + EndChunkLength; 
+                sizeHint += MaxBeginChunkLength + EndChunkLength;
                 var writer = new BufferWriter<PipeWriter>(_pipeWriter);
                 WriteCurrentChunkMemoryToPipeWriter(ref writer);
                 writer.Commit();
@@ -632,7 +640,7 @@ private void UpdateCurrentChunkMemory(int sizeHint)
             }
             _currentChunkMemoryUpdated = true;
         }
-       
+
         private void WriteCurrentChunkMemoryToPipeWriter(ref BufferWriter<PipeWriter> writer)
         {
             Debug.Assert(_advancedBytesForChunk <= _currentChunkMemory.Length);
@@ -714,8 +722,8 @@ private void AddSegment(int sizeHint = 0)
             }
             else
             {
-                _currentSegment = new byte[sizeHint];
-                _currentSegmentOwner = null;
+                _currentSegment = ArrayPool<byte>.Shared.Rent(sizeHint);
+                _currentSegmentOwner = _currentSegment;
             }
 
             _position = 0;
@@ -741,14 +749,14 @@ private static void ThrowSuffixSent()
         /// </summary>
         private readonly struct CompletedBuffer
         {
-            private readonly IMemoryOwner<byte> _memoryOwner;
+            private readonly object _memoryOwner;
 
             public Memory<byte> Buffer { get; }
             public int Length { get; }
 
             public ReadOnlySpan<byte> Span => Buffer.Span.Slice(0, Length);
 
-            public CompletedBuffer(IMemoryOwner<byte> owner, Memory<byte> buffer, int length)
+            public CompletedBuffer(object owner, Memory<byte> buffer, int length)
             {
                 _memoryOwner = owner;
 
@@ -758,9 +766,13 @@ public CompletedBuffer(IMemoryOwner<byte> owner, Memory<byte> buffer, int length
 
             public void Return()
             {
-                if (_memoryOwner != null)
+                if (_memoryOwner is IMemoryOwner<byte> owner)
+                {
+                    owner.Dispose();
+                }
+                else if (_memoryOwner is byte[] array)
                 {
-                    _memoryOwner.Dispose();
+                    ArrayPool<byte>.Shared.Return(array);
                 }
             }
         }

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

@@ -329,7 +329,7 @@ void IHttpResponseBodyFeature.DisableBuffering()
 
         Task IHttpResponseBodyFeature.SendFileAsync(string path, long offset, long? count, CancellationToken cancellation)
         {
-            return SendFileFallback.SendFileAsync(ResponseBody, path, offset, count, cancellation);
+            return SendFileAsync(path, offset, count, cancellation);
         }
 
         Task IHttpResponseBodyFeature.CompleteAsync()