Address PR feedback

Author: Maxwe11

src/System.IO.Pipes/src/System.IO.Pipes.csproj

@@ -131,8 +131,8 @@
     <Compile Include="System\IO\Pipes\ConnectionCompletionSource.cs" />
     <Compile Include="System\IO\Pipes\NamedPipeClientStream.Windows.cs" />
     <Compile Include="System\IO\Pipes\NamedPipeServerStream.Windows.cs" />
+    <Compile Include="System\IO\Pipes\PipeIOCompletionSource.cs" />
     <Compile Include="System\IO\Pipes\PipeStream.Windows.cs" />
-    <Compile Include="System\IO\Pipes\PipeStreamCompletionSource.cs" />
   </ItemGroup>
   <ItemGroup Condition=" '$(TargetsUnix)' == 'true' ">
     <Compile Include="Microsoft\Win32\SafeHandles\SafePipeHandle.Unix.cs" />

src/System.IO.Pipes/src/System/IO/Pipes/PipeStreamCompletionSource.cs renamed to src/System.IO.Pipes/src/System/IO/Pipes/PipeIOCompletionSource.cs

@@ -9,7 +9,7 @@
 
 namespace System.IO.Pipes
 {
-    // TODO: consolidate logic between PipeStreamCompletionSource and ConnectionCompletionSource
+    // TODO: consolidate logic between PipeIOCompletionSource and ConnectionCompletionSource
     internal sealed unsafe class PipeStreamCompletionSource : TaskCompletionSource<int>
     {
         private const int NoResult = 0;
@@ -18,12 +18,12 @@ internal sealed unsafe class PipeStreamCompletionSource : TaskCompletionSource<i
         private const int RegisteringCancellation = 4;
         private const int CompletedCallback = 8;
 
+        private readonly CancellationToken _cancellationToken;
         private readonly bool _isWrite;
         private readonly PipeStream _pipeStream;
         private readonly ThreadPoolBoundHandle _threadPoolBinding;
 
         private CancellationTokenRegistration _cancellationRegistration;
-        private CancellationToken _cancellationToken;
         private int _errorCode;
         private bool _isMessageComplete;
         private int _numBytes; // number of buffer read OR written
@@ -134,7 +134,6 @@ private void AsyncCallback(uint errorCode, uint numBytes)
                     case Interop.mincore.Errors.ERROR_PIPE_NOT_CONNECTED:
                     case Interop.mincore.Errors.ERROR_NO_DATA:
                         errorCode = 0;
-                        numBytes = 0;
                         break;
                 }
             }
@@ -170,12 +169,9 @@ private void AsyncCallback(uint errorCode, uint numBytes)
 
         private void Cancel()
         {
-            // Storing to locals to avoid data races
             SafeHandle handle = _threadPoolBinding.Handle;
             NativeOverlapped* overlapped = Overlapped;
 
-            Debug.Assert(overlapped != null && !Task.IsCompleted, "IO should not have completed yet");
-
             // If the handle is still valid, attempt to cancel the IO
             if (!handle.IsInvalid)
             {

src/System.IO.Pipes/src/System/IO/Pipes/PipeStream.Windows.cs

@@ -13,7 +13,7 @@ namespace System.IO.Pipes
 {
     public abstract partial class PipeStream : Stream
     {
-        private static readonly Task<int> ZeroTask = Task.FromResult(0);
+        private static readonly Task<int> s_zeroTask = Task.FromResult(0);
         internal const bool CheckOperationsRequiresSetHandle = true;
         internal ThreadPoolBoundHandle _threadPoolBinding;
 
@@ -296,7 +296,7 @@ public virtual PipeTransmissionMode ReadMode
         // -----------------------------
 
         [SecurityCritical]
-        private unsafe Task WriteAsyncCorePrivate(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
+        private Task WriteAsyncCorePrivate(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
         {
             Debug.Assert(_handle != null, "_handle is null");
             Debug.Assert(!_handle.IsClosed, "_handle is closed");
@@ -306,8 +306,6 @@ private unsafe Task WriteAsyncCorePrivate(byte[] buffer, int offset, int count,
             Debug.Assert(offset >= 0, "offset is negative");
             Debug.Assert(count >= 0, "count is negative");
 
-            // fixed doesn't work well with zero length arrays. Set the zero-byte flag in case
-            // caller needs to do any cleanup
             if (buffer.Length == 0)
             {
                 return Task.CompletedTask;
@@ -318,7 +316,11 @@ private unsafe Task WriteAsyncCorePrivate(byte[] buffer, int offset, int count,
                 int errorCode = 0;
 
                 // Queue an async WriteFile operation and pass in a packed overlapped
-                int r = WriteFileNative(_handle, buffer, offset, count, completionSource.Overlapped, out errorCode);
+                int r;
+                unsafe
+                {
+                    r = WriteFileNative(_handle, buffer, offset, count, completionSource.Overlapped, out errorCode);
+                }
 
                 // WriteFile, the OS version, will return 0 on failure, but this WriteFileNative 
                 // wrapper returns -1. This will return the following:
@@ -443,7 +445,7 @@ private unsafe int WriteFileNative(SafePipeHandle handle, byte[] buffer, int off
         }
 
         [SecurityCritical]
-        private unsafe Task<int> ReadAsyncCorePrivate(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
+        private Task<int> ReadAsyncCorePrivate(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
         {
             Debug.Assert(_handle != null, "_handle is null");
             Debug.Assert(!_handle.IsClosed, "_handle is closed");
@@ -453,20 +455,22 @@ private unsafe Task<int> ReadAsyncCorePrivate(byte[] buffer, int offset, int cou
             Debug.Assert(offset >= 0, "offset is negative");
             Debug.Assert(count >= 0, "count is negative");
 
-            // handle zero-length buffers separately; fixed keyword ReadFileNative doesn't like
-            // 0-length buffers. Call user callback and we're done
             if (buffer.Length == 0)
             {
                 UpdateMessageCompletion(false);
-                return ZeroTask;
+                return s_zeroTask;
             }
             else
             {
                 var completionSource = new PipeStreamCompletionSource(this, buffer, cancellationToken, isWrite: false);
 
                 // Queue an async ReadFile operation and pass in a packed overlapped
                 int errorCode = 0;
-                int r = ReadFileNative(_handle, buffer, offset, count, completionSource.Overlapped, out errorCode);
+                int r;
+                unsafe
+                {
+                    r = ReadFileNative(_handle, buffer, offset, count, completionSource.Overlapped, out errorCode);
+                }
 
                 // ReadFile, the OS version, will return 0 on failure, but this ReadFileNative wrapper
                 // returns -1. This will return the following:
@@ -487,13 +491,16 @@ private unsafe Task<int> ReadAsyncCorePrivate(byte[] buffer, int offset, int cou
                         case Interop.mincore.Errors.ERROR_PIPE_NOT_CONNECTED:
                             State = PipeState.Broken;
 
-                            // Clear the overlapped status bit for this special case. Failure to do so looks 
-                            // like we are freeing a pending overlapped.
-                            completionSource.Overlapped->InternalLow = IntPtr.Zero;
+                            unsafe
+                            {
+                                // Clear the overlapped status bit for this special case. Failure to do so looks 
+                                // like we are freeing a pending overlapped.
+                                completionSource.Overlapped->InternalLow = IntPtr.Zero;
+                            }
+
                             completionSource.ReleaseResources();
-                            completionSource.SetCompletedSynchronously();
                             UpdateMessageCompletion(true);
-                            return ZeroTask;
+                            return s_zeroTask;
 
                         case Interop.mincore.Errors.ERROR_IO_PENDING:
                             break;
@@ -525,7 +532,7 @@ internal void UpdateMessageCompletion(bool completion)
         {
             // Set message complete to true because the pipe is broken as well.
             // Need this to signal to readers to stop reading.
-            _isMessageComplete = (_state == PipeState.Broken || completion);
+            _isMessageComplete = (completion || _state == PipeState.Broken);
         }
 
         /// <summary>