Fix PipeStream leak on Windows when pipe is disposed with a pending operation (#115453)

* Fix PipeStream cleanup on Windows when pipe is disposed with a pending operation

The same pattern occurs in a couple of other places, as well.

* Make dispose cancellation unconditional

Author: stephentoub

src/libraries/System.IO.Pipes/src/System/IO/Pipes/NamedPipeServerStream.Unix.cs

@@ -139,10 +139,7 @@ internal override void DisposeCore(bool disposing)
 
             if (disposing)
             {
-                if (State != PipeState.Closed)
-                {
-                    _internalTokenSource.Cancel();
-                }
+                _internalTokenSource.Cancel();
             }
         }
 

src/libraries/System.IO.Pipes/src/System/IO/Pipes/NamedPipeServerStream.Windows.cs

@@ -62,7 +62,11 @@ internal override void TryToReuse(PipeValueTaskSource source)
             {
                 if (Interlocked.CompareExchange(ref _reusableConnectionValueTaskSource, connectionSource, null) is not null)
                 {
-                    source._preallocatedOverlapped.Dispose();
+                    source.Dispose();
+                }
+                else if (State == PipeState.Closed)
+                {
+                    Interlocked.Exchange(ref _reusableConnectionValueTaskSource, null)?.Dispose();
                 }
             }
         }

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

@@ -255,9 +255,25 @@ internal virtual void TryToReuse(PipeValueTaskSource source)
             if (source is ReadWriteValueTaskSource readWriteSource)
             {
                 ref ReadWriteValueTaskSource? field = ref readWriteSource._isWrite ? ref _reusableWriteValueTaskSource : ref _reusableReadValueTaskSource;
+
+                // Try to return the instance. If there's already something in the field, then there had been concurrent
+                // operations and someone else already returned their instance, so just dispose of this one (the "pool" has
+                // a size of 1). If there's not something there and we're successful in storing our instance, the 99% case is
+                // this is normal operation, there wasn't a concurrent operation, and we just successfully returned the rented
+                // instance. However, it could also be because the stream has been disposed, in which case the disposal process
+                // would have nulled out the field, and since this instance wasn't present when disposal happened, it won't have
+                // been disposed. Further, disposal won't happen again, so just leaving the instance there will result in its
+                // resources being leaked. Instead, _after_ returning the instance we then check whether the stream is now
+                // disposed, and if it is, we then try to re-rent the instance and dispose of it. It's fine if the disposal happens
+                // after we've stored the instance back and before we check the stream's state, as then this code will be racing
+                // with disposal and only one of the two will succeed in exchanging out the instance.
                 if (Interlocked.CompareExchange(ref field, readWriteSource, null) is not null)
                 {
-                    source._preallocatedOverlapped.Dispose();
+                    source.Dispose();
+                }
+                else if (State == PipeState.Closed)
+                {
+                    Interlocked.Exchange(ref field, null)?.Dispose();
                 }
             }
         }

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

@@ -144,6 +144,11 @@ public override Task FlushAsync(CancellationToken cancellationToken)
 
         protected override void Dispose(bool disposing)
         {
+            // Mark the pipe as closed before calling DisposeCore. That way, other threads that might
+            // be synchronizing on shared resources disposed of in DisposeCore will be guaranteed to
+            // see the closed state after that synchronization.
+            _state = PipeState.Closed;
+
             try
             {
                 // Nothing will be done differently based on whether we are
@@ -159,8 +164,6 @@ protected override void Dispose(bool disposing)
             {
                 base.Dispose(disposing);
             }
-
-            _state = PipeState.Closed;
         }
 
         // ********************** Public Properties *********************** //

src/libraries/System.Private.CoreLib/src/Microsoft/Win32/SafeHandles/SafeFileHandle.OverlappedValueTaskSource.Windows.cs

@@ -35,7 +35,11 @@ private void TryToReuse(OverlappedValueTaskSource source)
 
             if (Interlocked.CompareExchange(ref _reusableOverlappedValueTaskSource, source, null) is not null)
             {
-                source._preallocatedOverlapped.Dispose();
+                source.Dispose();
+            }
+            else if (IsClosed)
+            {
+                Interlocked.Exchange(ref _reusableOverlappedValueTaskSource, null)?.Dispose();
             }
         }