Cover

src/System.IO.Pipelines/src/Properties/InternalsVisibleTo.cs

@@ -0,0 +1,9 @@
+// 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.Runtime.CompilerServices;
+
+[assembly: InternalsVisibleTo("System.IO.Pipelines.Tests, PublicKey=002400000480000094000000060200000024000052534131000400000100010015c01ae1f50e8cc09ba9eac9147cf8fd9fce2cfe9f8dce4f7301c4132ca9fb50ce8cbf1df4dc18dd4d210e4345c744ecb3365ed327efdbc52603faa5e21daa11234c8c4a73e51f03bf192544581ebe107adee3a34928e39d04e524a9ce729d5090bfd7dad9d10c722c0def9ccc08ff0a03790e48bcd1f9b6c476063e1966a1c4")]

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

@@ -419,14 +419,14 @@ internal void AdvanceReader(SequencePosition consumed, SequencePosition examined
                     _readerAwaitable.Reset();
                 }
 
-                _readingState.End();
-
                 while (returnStart != null && returnStart != returnEnd)
                 {
                     returnStart.ResetMemory();
                     ReturnSegmentUnsynchronized(returnStart);
                     returnStart = returnStart.NextSegment;
                 }
+
+                _readingState.End();
             }
 
             TrySchedule(_writerScheduler, continuation);
@@ -467,7 +467,7 @@ internal void OnWriterCompleted(Action<Exception, object> callback, object state
         {
             if (callback == null)
             {
-                ThrowHelper.ThrowArgumentOutOfRangeException(ExceptionArgument.callback);
+                ThrowHelper.ThrowArgumentNullException(ExceptionArgument.callback);
             }
 
             PipeCompletionCallbacks completionCallbacks;

src/System.IO.Pipelines/src/System/IO/Pipelines/PipeAwaitable.cs

@@ -2,11 +2,13 @@
 // 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.Diagnostics;
 using System.Runtime.CompilerServices;
 using System.Threading;
 
 namespace System.IO.Pipelines
 {
+    [DebuggerDisplay("CancelledState: {_canceledState}, IsCompleted: {IsCompleted}")]
     internal struct PipeAwaitable
     {
         private static readonly Action _awaitableIsCompleted = () => { };
@@ -134,11 +136,6 @@ public bool ObserveCancelation()
             return false;
         }
 
-        public override string ToString()
-        {
-            return $"CancelledState: {_canceledState}, {nameof(IsCompleted)}: {IsCompleted}";
-        }
-
         private enum CanceledState
         {
             NotCanceled = 0,

src/System.IO.Pipelines/src/System/IO/Pipelines/PipeCompletion.cs

@@ -9,6 +9,7 @@
 
 namespace System.IO.Pipelines
 {
+    [DebuggerDisplay("IsCompleted: {" + nameof(IsCompleted) + "}")]
     internal struct PipeCompletion
     {
         private static readonly ArrayPool<PipeCompletionCallback> CompletionCallbackPool = ArrayPool<PipeCompletionCallback>.Shared;

src/System.IO.Pipelines/src/System/IO/Pipelines/PipeReaderState.cs

@@ -1,10 +1,12 @@
 // Copyright (c) Microsoft. All rights reserved.
 // Licensed under the MIT license. See LICENSE file in the project root for full license information.
 
+using System.Diagnostics;
 using System.Runtime.CompilerServices;
 
 namespace System.IO.Pipelines
 {
+    [DebuggerDisplay("State: {_state}")]
     internal struct PipeReaderState
     {
         private State _state;
@@ -38,18 +40,13 @@ public void End()
         {
             if (_state == State.Inactive)
             {
-                ThrowHelper.CreateInvalidOperationException_NoReadToComplete();
+                ThrowHelper.ThrowInvalidOperationException_NoReadToComplete();
             }
 
             _state = State.Inactive;
         }
 
         public bool IsActive => _state == State.Active;
-
-        public override string ToString()
-        {
-            return $"State: {_state}";
-        }
     }
 
     internal enum State: byte

src/System.IO.Pipelines/src/System/IO/Pipelines/ThrowHelper.cs

@@ -14,7 +14,7 @@ internal class ThrowHelper
 
         internal static void ThrowArgumentNullException(ExceptionArgument argument) { throw CreateArgumentNullException(argument); }
         [MethodImpl(MethodImplOptions.NoInlining)]
-        private static Exception CreateArgumentNullException(ExceptionArgument argument) { return new ArgumentOutOfRangeException(argument.ToString()); }
+        private static Exception CreateArgumentNullException(ExceptionArgument argument) { return new ArgumentNullException(argument.ToString()); }
 
         public static void ThrowInvalidOperationException_NotWritingNoAlloc()
         {
@@ -157,7 +157,7 @@ public static void ThrowInvalidOperationException_ResetIncompleteReaderWriter()
         [MethodImpl(MethodImplOptions.NoInlining)]
         public static Exception CreateInvalidOperationException_ResetIncompleteReaderWriter()
         {
-            return new InvalidOperationException(SR.AdvanceToInvalidCursor);
+            return new InvalidOperationException(SR.ReaderAndWriterHasToBeCompleted);
         }
     }
 

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

@@ -1,102 +1,102 @@
-//// 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 Xunit;
-
-//namespace System.IO.Pipelines.Tests
-//{
-//    public class PipeLengthTests : IDisposable
-//    {
-//        public PipeLengthTests()
-//        {
-//            _pool = new TestMemoryPool();
-//            _pipe = new Pipe(new PipeOptions(_pool));
-//        }
-
-//        public void Dispose()
-//        {
-//            _pipe.Writer.Complete();
-//            _pipe.Reader.Complete();
-//            _pool?.Dispose();
-//        }
-
-//        private readonly TestMemoryPool _pool;
-
-//        private readonly Pipe _pipe;
-
-//        [Fact]
-//        public void ByteByByteTest()
-//        {
-//            for (var i = 1; i <= 1024 * 1024; i++)
-//            {
-//                _pipe.Writer.GetMemory(100);
-//                _pipe.Writer.Advance(1);
-//                _pipe.Writer.Commit();
-
-//                Assert.Equal(i, _pipe.Length);
-//            }
-
-//            _pipe.Writer.FlushAsync();
-
-//            for (int i = 1024 * 1024 - 1; i >= 0; i--)
-//            {
-//                ReadResult result = _pipe.Reader.ReadAsync().GetResult();
-//                SequencePosition consumed = result.Buffer.Slice(1).Start;
-
-//                Assert.Equal(i + 1, result.Buffer.Length);
-
-//                _pipe.Reader.AdvanceTo(consumed, consumed);
-
-//                Assert.Equal(i, _pipe.Length);
-//            }
-//        }
-
-//        [Fact]
-//        public void LengthCorrectAfterAlloc0AdvanceCommit()
-//        {
-//            _pipe.Writer.GetMemory(0);
-//            _pipe.Writer.WriteEmpty(10);
-//            _pipe.Writer.Commit();
-
-//            Assert.Equal(10, _pipe.Length);
-//        }
-
-//        [Fact]
-//        public void LengthCorrectAfterAllocAdvanceCommit()
-//        {
-//            PipeWriter writableBuffer = _pipe.Writer.WriteEmpty(10);
-//            writableBuffer.Commit();
-
-//            Assert.Equal(10, _pipe.Length);
-//        }
-
-//        [Fact]
-//        public void LengthDecreasedAfterReadAdvanceConsume()
-//        {
-//            _pipe.Writer.GetMemory(100);
-//            _pipe.Writer.Advance(10);
-//            _pipe.Writer.Commit();
-//            _pipe.Writer.FlushAsync();
-
-//            ReadResult result = _pipe.Reader.ReadAsync().GetResult();
-//            SequencePosition consumed = result.Buffer.Slice(5).Start;
-//            _pipe.Reader.AdvanceTo(consumed, consumed);
-
-//            Assert.Equal(5, _pipe.Length);
-//        }
-
-//        [Fact]
-//        public void LengthNotChangeAfterReadAdvanceExamine()
-//        {
-//            PipeWriter writableBuffer = _pipe.Writer.WriteEmpty(10);
-//            writableBuffer.Commit();
-//            writableBuffer.FlushAsync();
-
-//            ReadResult result = _pipe.Reader.ReadAsync().GetResult();
-//            _pipe.Reader.AdvanceTo(result.Buffer.Start, result.Buffer.End);
-
-//            Assert.Equal(10, _pipe.Length);
-//        }
-//    }
-//}
+// 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 Xunit;
+
+namespace System.IO.Pipelines.Tests
+{
+    public class PipeLengthTests : IDisposable
+    {
+        public PipeLengthTests()
+        {
+            _pool = new TestMemoryPool();
+            _pipe = new Pipe(new PipeOptions(_pool));
+        }
+
+        public void Dispose()
+        {
+            _pipe.Writer.Complete();
+            _pipe.Reader.Complete();
+            _pool?.Dispose();
+        }
+
+        private readonly TestMemoryPool _pool;
+
+        private readonly Pipe _pipe;
+
+        [Fact]
+        public void ByteByByteTest()
+        {
+            for (var i = 1; i <= 1024 * 1024; i++)
+            {
+                _pipe.Writer.GetMemory(100);
+                _pipe.Writer.Advance(1);
+                _pipe.Writer.Commit();
+
+                Assert.Equal(i, _pipe.Length);
+            }
+
+            _pipe.Writer.FlushAsync();
+
+            for (int i = 1024 * 1024 - 1; i >= 0; i--)
+            {
+                ReadResult result = _pipe.Reader.ReadAsync().GetResult();
+                SequencePosition consumed = result.Buffer.Slice(1).Start;
+
+                Assert.Equal(i + 1, result.Buffer.Length);
+
+                _pipe.Reader.AdvanceTo(consumed, consumed);
+
+                Assert.Equal(i, _pipe.Length);
+            }
+        }
+
+        [Fact]
+        public void LengthCorrectAfterAlloc0AdvanceCommit()
+        {
+            _pipe.Writer.GetMemory(0);
+            _pipe.Writer.WriteEmpty(10);
+            _pipe.Writer.Commit();
+
+            Assert.Equal(10, _pipe.Length);
+        }
+
+        [Fact]
+        public void LengthCorrectAfterAllocAdvanceCommit()
+        {
+            PipeWriter writableBuffer = _pipe.Writer.WriteEmpty(10);
+            writableBuffer.Commit();
+
+            Assert.Equal(10, _pipe.Length);
+        }
+
+        [Fact]
+        public void LengthDecreasedAfterReadAdvanceConsume()
+        {
+            _pipe.Writer.GetMemory(100);
+            _pipe.Writer.Advance(10);
+            _pipe.Writer.Commit();
+            _pipe.Writer.FlushAsync();
+
+            ReadResult result = _pipe.Reader.ReadAsync().GetResult();
+            SequencePosition consumed = result.Buffer.Slice(5).Start;
+            _pipe.Reader.AdvanceTo(consumed, consumed);
+
+            Assert.Equal(5, _pipe.Length);
+        }
+
+        [Fact]
+        public void LengthNotChangeAfterReadAdvanceExamine()
+        {
+            PipeWriter writableBuffer = _pipe.Writer.WriteEmpty(10);
+            writableBuffer.Commit();
+            writableBuffer.FlushAsync();
+
+            ReadResult result = _pipe.Reader.ReadAsync().GetResult();
+            _pipe.Reader.AdvanceTo(result.Buffer.Start, result.Buffer.End);
+
+            Assert.Equal(10, _pipe.Length);
+        }
+    }
+}

src/System.IO.Pipelines/tests/PipelineReaderWriterFacts.cs → src/System.IO.Pipelines/tests/PipeReaderWriterFacts.cs

@@ -557,5 +557,56 @@ public async Task WritingDataMakesDataReadableViaPipeline()
 
             _pipe.Reader.AdvanceTo(buffer.Start, buffer.Start);
         }
+
+        [Fact]
+        public async Task DoubleReadThrows()
+        {
+            await _pipe.Writer.WriteAsync(new byte[1]);
+            PipeAwaiter<ReadResult> awaiter = _pipe.Reader.ReadAsync();
+            ReadResult result = awaiter.GetAwaiter().GetResult();
+
+            Assert.Throws<InvalidOperationException>(() => awaiter.GetAwaiter().GetResult());
+
+            _pipe.Reader.AdvanceTo(result.Buffer.Start, result.Buffer.Start);
+        }
+
+        [Fact]
+        public void GetResultBeforeCompletedThrows()
+        {
+            PipeAwaiter<ReadResult> awaiter = _pipe.Reader.ReadAsync();
+
+            Assert.Throws<InvalidOperationException>(() => awaiter.GetAwaiter().GetResult());
+        }
+
+        [Fact]
+        public void CompleteAfterAdvanceThrows()
+        {
+            _pipe.Writer.WriteEmpty(4);
+
+            Assert.Throws<InvalidOperationException>(() => _pipe.Writer.Complete());
+
+            _pipe.Commit();
+        }
+
+        [Fact]
+        public async Task AdvanceWithoutReadThrows()
+        {
+            await _pipe.Writer.WriteAsync(new byte[3]);
+            ReadResult readResult = await _pipe.Reader.ReadAsync();
+            _pipe.Reader.AdvanceTo(readResult.Buffer.Start);
+
+            InvalidOperationException exception = Assert.Throws<InvalidOperationException>(() => _pipe.Reader.AdvanceTo(readResult.Buffer.End));
+            Assert.Equal("No reading operation to complete.", exception.Message);
+        }
+
+        [Fact]
+        public async Task TryReadAfterReadAsyncThrows()
+        {
+            await _pipe.Writer.WriteAsync(new byte[3]);
+            ReadResult readResult = await _pipe.Reader.ReadAsync();
+
+            Assert.Throws<InvalidOperationException>(() => _pipe.Reader.TryRead(out _));
+            _pipe.Reader.AdvanceTo(readResult.Buffer.Start);
+        }
     }
 }

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

@@ -39,7 +39,7 @@ public async Task LengthIsReseted()
 
             _pipe.Reset();
 
-            //Assert.Equal(0, _pipe.Length);
+            Assert.Equal(0, _pipe.Length);
         }
 
         [Fact]