[Backport][Pipelines] Noop Writes when Reader completed (#42982)

* [Backport][Pipelines] Noop Writes when Reader completed

* move DisposeTrackingBufferPool to its own file

* assembly version

* fb

* Fb

* packageIndex

Author: BrennanConroy

pkg/Microsoft.Private.PackageBaseline/packageIndex.json

@@ -2232,7 +2232,8 @@
       "InboxOn": {},
       "AssemblyVersionInPackageVersion": {
         "4.0.0.0": "4.5.1",
-        "4.0.0.1": "4.5.2"
+        "4.0.0.1": "4.5.2",
+        "4.0.0.2": "4.5.3"
       }
     },
     "System.IO.Pipes": {

src/System.IO.Pipelines/dir.props

@@ -2,8 +2,8 @@
 <Project ToolsVersion="14.0" DefaultTargets="Build" xmlns="http://schemas.microsoft.com/developer/msbuild/2003">
   <Import Project="..\dir.props" />
   <PropertyGroup>
-    <AssemblyVersion>4.0.0.1</AssemblyVersion>
-    <PackageVersion>4.5.3</PackageVersion>
+    <AssemblyVersion>4.0.0.2</AssemblyVersion>
+    <PackageVersion>4.5.4</PackageVersion>
     <AssemblyKey>Open</AssemblyKey>
   </PropertyGroup>
 </Project>

src/System.IO.Pipelines/src/System/IO/Pipelines/Pipe.cs

@@ -269,6 +269,12 @@ internal void Advance(int bytesWritten)
                 ThrowHelper.ThrowInvalidOperationException_NotWritingNoAlloc();
             }
 
+            // If the reader is completed we no-op Advance but leave GetMemory and FlushAsync alone
+            if (_readerCompletion.IsCompleted)
+            {
+                return;
+            }
+
             if (bytesWritten > 0)
             {
                 Debug.Assert(!_writingHead.ReadOnly);

src/System.IO.Pipelines/tests/Infrastructure/DisposeTrackingBufferPool.cs

@@ -0,0 +1,76 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// The .NET Foundation licenses this file to you under the MIT license.
+// See the LICENSE file in the project root for more information.
+
+using System.Buffers;
+using System.Threading.Tasks;
+using Xunit;
+
+namespace System.IO.Pipelines.Tests
+{
+    public class DisposeTrackingBufferPool : TestMemoryPool
+    {
+        public int DisposedBlocks { get; set; }
+        public int CurrentlyRentedBlocks { get; set; }
+
+        public override IMemoryOwner<byte> Rent(int size)
+        {
+            return new DisposeTrackingMemoryManager(new byte[size], this);
+        }
+
+        protected override void Dispose(bool disposing)
+        {
+        }
+
+        private class DisposeTrackingMemoryManager : MemoryManager<byte>
+        {
+            private byte[] _array;
+
+            private readonly DisposeTrackingBufferPool _bufferPool;
+
+            public DisposeTrackingMemoryManager(byte[] array, DisposeTrackingBufferPool bufferPool)
+            {
+                _array = array;
+                _bufferPool = bufferPool;
+                _bufferPool.CurrentlyRentedBlocks++;
+            }
+
+            public override Memory<byte> Memory => CreateMemory(_array.Length);
+
+            public bool IsDisposed => _array == null;
+
+            public override MemoryHandle Pin(int elementIndex = 0)
+            {
+                throw new NotImplementedException();
+            }
+
+            public override void Unpin()
+            {
+                throw new NotImplementedException();
+            }
+
+            protected override bool TryGetArray(out ArraySegment<byte> segment)
+            {
+                if (IsDisposed)
+                    throw new ObjectDisposedException(nameof(DisposeTrackingBufferPool));
+                segment = new ArraySegment<byte>(_array);
+                return true;
+            }
+
+            protected override void Dispose(bool disposing)
+            {
+                _bufferPool.DisposedBlocks++;
+                _bufferPool.CurrentlyRentedBlocks--;
+
+                _array = null;
+            }
+
+            public override Span<byte> GetSpan()
+            {
+                if (IsDisposed)
+                    throw new ObjectDisposedException(nameof(DisposeTrackingBufferPool));
+                return _array;
+            }
+        }
+    }
+}

src/System.IO.Pipelines/tests/PipePoolTests.cs

@@ -10,72 +10,6 @@ namespace System.IO.Pipelines.Tests
 {
     public class PipePoolTests
     {
-        private class DisposeTrackingBufferPool : TestMemoryPool
-        {
-            public int DisposedBlocks { get; set; }
-            public int CurrentlyRentedBlocks { get; set; }
-
-            public override IMemoryOwner<byte> Rent(int size)
-            {
-                return new DisposeTrackingMemoryManager(new byte[size], this);
-            }
-
-            protected override void Dispose(bool disposing)
-            {
-            }
-
-            private class DisposeTrackingMemoryManager : MemoryManager<byte>
-            {
-                private byte[] _array;
-
-                private readonly DisposeTrackingBufferPool _bufferPool;
-
-                public DisposeTrackingMemoryManager(byte[] array, DisposeTrackingBufferPool bufferPool)
-                {
-                    _array = array;
-                    _bufferPool = bufferPool;
-                    _bufferPool.CurrentlyRentedBlocks++;
-                }
-
-                public override Memory<byte> Memory => CreateMemory(_array.Length);
-
-                public bool IsDisposed => _array == null;
-
-                public override MemoryHandle Pin(int elementIndex = 0)
-                {
-                    throw new NotImplementedException();
-                }
-
-                public override void Unpin()
-                {
-                    throw new NotImplementedException();
-                }
-
-                protected override bool TryGetArray(out ArraySegment<byte> segment)
-                {
-                    if (IsDisposed)
-                        throw new ObjectDisposedException(nameof(DisposeTrackingBufferPool));
-                    segment = new ArraySegment<byte>(_array);
-                    return true;
-                }
-
-                protected override void Dispose(bool disposing)
-                {
-                    _bufferPool.DisposedBlocks++;
-                    _bufferPool.CurrentlyRentedBlocks--;
-
-                    _array = null;
-                }
-
-                public override Span<byte> GetSpan()
-                {
-                    if (IsDisposed)
-                        throw new ObjectDisposedException(nameof(DisposeTrackingBufferPool));
-                    return _array;
-                }
-            }
-        }
-
         [Fact]
         public async Task AdvanceToEndReturnsAllBlocks()
         {

src/System.IO.Pipelines/tests/PipeWriterTests.cs

@@ -195,5 +195,40 @@ public void GetMemory_AdjustsToPoolMaxBufferSize()
             var memory = buffer.GetMemory(int.MaxValue);
             Assert.True(memory.Length >= 4096);
         }
+
+        [Fact]
+        public async Task WriteAsyncWithACompletedReaderNoops()
+        {
+            var pool = new DisposeTrackingBufferPool();
+            var pipe = new Pipe(new PipeOptions(pool));
+            pipe.Reader.Complete();
+
+            byte[] writeBuffer = new byte[100];
+            for (var i = 0; i < 10000; i++)
+            {
+                await pipe.Writer.WriteAsync(writeBuffer);
+            }
+
+            Assert.Equal(0, pool.CurrentlyRentedBlocks);
+        }
+
+        [Fact]
+        public async Task GetMemoryFlushWithACompletedReaderNoops()
+        {
+            var pool = new DisposeTrackingBufferPool();
+            var pipe = new Pipe(new PipeOptions(pool));
+            pipe.Reader.Complete();
+
+            for (var i = 0; i < 10000; i++)
+            {
+                var mem = pipe.Writer.GetMemory();
+                pipe.Writer.Advance(mem.Length);
+                await pipe.Writer.FlushAsync(default);
+            }
+
+            Assert.Equal(1, pool.CurrentlyRentedBlocks);
+            pipe.Writer.Complete();
+            Assert.Equal(0, pool.CurrentlyRentedBlocks);
+        }
     }
 }

src/System.IO.Pipelines/tests/System.IO.Pipelines.Tests.csproj

@@ -12,6 +12,7 @@
   <PropertyGroup Condition="'$(Configuration)|$(Platform)' == 'uap-Windows_NT-Debug|AnyCPU'" />
   <PropertyGroup Condition="'$(Configuration)|$(Platform)' == 'uap-Windows_NT-Release|AnyCPU'" />
   <ItemGroup>
+    <Compile Include="Infrastructure\DisposeTrackingBufferPool.cs" />
     <Compile Include="BackpressureTests.cs" />
     <Compile Include="FlushAsyncCancellationTests.cs" />
     <Compile Include="FlushAsyncCompletionTests.cs" />

src/packages.builds

@@ -24,6 +24,9 @@
       <AdditionalProperties>$(AdditionalProperties)</AdditionalProperties>
     </Project>
     <!-- add specific builds / pkgproj's here to include in servicing builds -->
+    <Project Include="$(MSBuildThisFileDirectory)System.IO.Pipelines\pkg\System.IO.Pipelines.pkgproj">
+      <AdditionalProperties>$(AdditionalProperties)</AdditionalProperties>
+    </Project>
   </ItemGroup>
 
   <!-- Need the PackageIndexFile file property from baseline.props -->