Add DuplexStream and implement for NetworkStream and QuicStream

src/libraries/Common/tests/StreamConformanceTests/System/IO/StreamConformanceTests.cs

@@ -2938,6 +2938,72 @@ public override ValueTask<int> ReadAsync(Memory<byte> buffer, CancellationToken
         }
     }
 
+    /// <summary>Base class for a connected stream that can have writes completed.</summary>
+    public abstract class DuplexConnectedStreamConformanceTests : ConnectedStreamConformanceTests
+    {
+        [Theory]
+        [InlineData(true)]
+        [InlineData(false)]
+        public virtual async Task CompleteWrites_EndsReads(bool isAsync)
+        {
+            using StreamPair streams = await CreateConnectedStreamsAsync();
+            byte[] buffer = new byte[8];
+
+            foreach ((Stream writable, Stream readable) in GetReadWritePairs(streams))
+            {
+                await writable.WriteAsync(buffer);
+                await writable.FlushAsync();
+
+                int received = 0;
+                while (received != buffer.Length)
+                {
+                    int len = await readable.ReadAsync(buffer.AsMemory(0, buffer.Length - received));
+                    received += len;
+                }
+
+                ValueTask<int> readTask = readable.ReadAsync(buffer);
+
+                Assert.False(readTask.IsCompleted);
+                Assert.True(writable.CanWrite);
+
+                DuplexStream writableDuplexStream = Assert.IsAssignableFrom<DuplexStream>(writable);
+                if (isAsync) await writableDuplexStream.CompleteWritesAsync();
+                else writableDuplexStream.CompleteWrites();
+
+                Assert.Equal(0, await readTask);
+                Assert.False(writable.CanWrite);
+
+                await Assert.ThrowsAnyAsync<Exception>(async () => await writable.WriteAsync(buffer));
+            }
+        }
+
+        [Theory]
+        [InlineData(true)]
+        [InlineData(false)]
+        public virtual async Task CompleteWrites_CalledTwice_Success(bool isAsync)
+        {
+            using StreamPair streams = await CreateConnectedStreamsAsync();
+            byte[] buffer = new byte[8];
+
+            foreach ((Stream writable, Stream readable) in GetReadWritePairs(streams))
+            {
+                DuplexStream writableDuplexStream = Assert.IsAssignableFrom<DuplexStream>(writable);
+
+                Assert.True(writable.CanWrite);
+
+                if (isAsync) await writableDuplexStream.CompleteWritesAsync();
+                else writableDuplexStream.CompleteWrites();
+
+                Assert.False(writable.CanWrite);
+
+                if (isAsync) await writableDuplexStream.CompleteWritesAsync();
+                else writableDuplexStream.CompleteWrites();
+
+                Assert.False(writable.CanWrite);
+            }
+        }
+    }
+
     /// <summary>Provides a disposable, enumerable tuple of two streams.</summary>
     public class StreamPair : IDisposable, IEnumerable<Stream>
     {

src/libraries/Common/tests/System/IO/ConnectedStreams.cs

@@ -46,13 +46,13 @@ public static (Stream Writer, Stream Reader) CreateUnidirectional(int initialBuf
 
         /// <summary>Creates a pair of streams that are connected for bidirectional communication.</summary>
         /// <remarks>Writing to one stream produces data readable by the either, and vice versa.</remarks>
-        public static (Stream Stream1, Stream Stream2) CreateBidirectional() =>
+        public static (DuplexStream Stream1, DuplexStream Stream2) CreateBidirectional() =>
             CreateBidirectional(StreamBuffer.DefaultInitialBufferSize, StreamBuffer.DefaultMaxBufferSize);
 
         /// <summary>Creates a pair of streams that are connected for bidirectional communication.</summary>
         /// <param name="initialBufferSize">The initial buffer size to use when storing data in the connection.</param>
         /// <remarks>Writing to one stream produces data readable by the either, and vice versa.</remarks>
-        public static (Stream Stream1, Stream Stream2) CreateBidirectional(int initialBufferSize) =>
+        public static (DuplexStream Stream1, DuplexStream Stream2) CreateBidirectional(int initialBufferSize) =>
             CreateBidirectional(initialBufferSize, StreamBuffer.DefaultMaxBufferSize);
 
         /// <summary>Creates a pair of streams that are connected for bidirectional communication.</summary>
@@ -62,7 +62,7 @@ public static (Stream Writer, Stream Reader) CreateUnidirectional(int initialBuf
         /// writes will block until additional space becomes available.
         /// </param>
         /// <remarks>Writing to one stream produces data readable by the either, and vice versa.</remarks>
-        public static (Stream Stream1, Stream Stream2) CreateBidirectional(int initialBufferSize, int maxBufferSize)
+        public static (DuplexStream Stream1, DuplexStream Stream2) CreateBidirectional(int initialBufferSize, int maxBufferSize)
         {
             // Each direction needs a buffer; one stream will use b1 for reading and b2 for writing,
             // and the other stream will do the inverse.
@@ -275,7 +275,7 @@ private void ThrowIfWritingNotSupported()
                 throw new NotSupportedException();
         }
 
-        private sealed class BidirectionalStreamBufferStream : Stream
+        private sealed class BidirectionalStreamBufferStream : DuplexStream
         {
             private readonly StreamBuffer _readBuffer;
             private readonly StreamBuffer _writeBuffer;
@@ -309,7 +309,7 @@ protected override void Dispose(bool disposing)
             }
 
             public override bool CanRead => !_disposed;
-            public override bool CanWrite => !_disposed;
+            public override bool CanWrite => !_disposed && !_writeBuffer.IsComplete;
             public override bool CanSeek => false;
 
             public override void Flush() => ThrowIfDisposed();
@@ -325,6 +325,20 @@ public override Task FlushAsync(CancellationToken cancellationToken)
                 return Task.CompletedTask;
             }
 
+            public override void CompleteWrites()
+            {
+                ThrowIfDisposed();
+                _writeBuffer.EndWrite();
+            }
+
+            public override ValueTask CompleteWritesAsync(CancellationToken cancellationToken = default)
+            {
+                ThrowIfDisposed();
+                if (cancellationToken.IsCancellationRequested) return ValueTask.FromCanceled(cancellationToken);
+                _writeBuffer.EndWrite();
+                return default;
+            }
+
             public override int Read(byte[] buffer, int offset, int count)
             {
                 ValidateBufferArguments(buffer, offset, count);

src/libraries/Common/tests/System/Net/Http/Http3LoopbackStream.cs

@@ -118,7 +118,7 @@ public async Task SendFrameAsync(long frameType, ReadOnlyMemory<byte> framePaylo
 
         public async Task ShutdownSendAsync()
         {
-            _stream.Shutdown();
+            await _stream.CompleteWritesAsync().ConfigureAwait(false);
             await _stream.ShutdownWriteCompleted().ConfigureAwait(false);
         }
 

src/libraries/Common/tests/Tests/System/IO/ConnectedStreamsTests.cs

@@ -14,7 +14,7 @@ public class UnidirectionalConnectedStreamsTests : ConnectedStreamConformanceTes
             Task.FromResult<StreamPair>(ConnectedStreams.CreateUnidirectional());
     }
 
-    public class BidirectionalConnectedStreamsTests : ConnectedStreamConformanceTests
+    public class BidirectionalConnectedStreamsTests : DuplexConnectedStreamConformanceTests
     {
         protected override int BufferedSize => StreamBuffer.DefaultMaxBufferSize;
         protected override bool FlushRequiredToWriteData => false;

src/libraries/System.Net.Http/src/System/Net/Http/SocketsHttpHandler/Http3RequestStream.cs

@@ -152,7 +152,7 @@ public async Task<HttpResponseMessage> SendAsync(CancellationToken cancellationT
                     }
                     else
                     {
-                        _stream.Shutdown();
+                        await _stream.CompleteWritesAsync(cancellationToken).ConfigureAwait(false);
                     }
                 }
 
@@ -372,7 +372,7 @@ private async Task SendContentAsync(HttpContent content, CancellationToken cance
                 _sendBuffer.Discard(_sendBuffer.ActiveLength);
             }
 
-            _stream.Shutdown();
+            await _stream.CompleteWritesAsync(cancellationToken).ConfigureAwait(false);
         }
 
         private async ValueTask WriteRequestContentAsync(ReadOnlyMemory<byte> buffer, CancellationToken cancellationToken)

src/libraries/System.Net.Quic/ref/System.Net.Quic.cs

@@ -78,7 +78,7 @@ public partial class QuicOptions
         public long MaxBidirectionalStreams { get { throw null; } set { } }
         public long MaxUnidirectionalStreams { get { throw null; } set { } }
     }
-    public sealed partial class QuicStream : System.IO.Stream
+    public sealed partial class QuicStream : System.IO.DuplexStream
     {
         internal QuicStream() { }
         public override bool CanRead { get { throw null; } }
@@ -91,6 +91,8 @@ public sealed partial class QuicStream : System.IO.Stream
         public void AbortWrite(long errorCode) { }
         public override System.IAsyncResult BeginRead(byte[] buffer, int offset, int count, System.AsyncCallback? callback, object? state) { throw null; }
         public override System.IAsyncResult BeginWrite(byte[] buffer, int offset, int count, System.AsyncCallback? callback, object? state) { throw null; }
+        public override void CompleteWrites() { }
+        public override System.Threading.Tasks.ValueTask CompleteWritesAsync(System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) { throw null; }
         protected override void Dispose(bool disposing) { }
         public override int EndRead(System.IAsyncResult asyncResult) { throw null; }
         public override void EndWrite(System.IAsyncResult asyncResult) { }
@@ -102,7 +104,6 @@ public sealed partial class QuicStream : System.IO.Stream
         public override System.Threading.Tasks.ValueTask<int> ReadAsync(System.Memory<byte> buffer, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) { throw null; }
         public override long Seek(long offset, System.IO.SeekOrigin origin) { throw null; }
         public override void SetLength(long value) { }
-        public void Shutdown() { }
         public System.Threading.Tasks.ValueTask ShutdownWriteCompleted(System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) { throw null; }
         public override void Write(byte[] buffer, int offset, int count) { }
         public override void Write(System.ReadOnlySpan<byte> buffer) { }

src/libraries/System.Net.Quic/src/System/Net/Quic/Implementations/Mock/MockStream.cs

@@ -178,12 +178,19 @@ internal override ValueTask ShutdownWriteCompleted(CancellationToken cancellatio
             return default;
         }
 
-        internal override void Shutdown()
+        public override void CompleteWrites()
         {
             CheckDisposed();
+            WriteStreamBuffer?.EndWrite();
+        }
+
+        public override ValueTask CompleteWritesAsync(CancellationToken cancellationToken)
+        {
+            CheckDisposed();
+            if (cancellationToken.IsCancellationRequested) return ValueTask.FromCanceled(cancellationToken);
 
-            // This seems to mean shutdown send, in particular, not both.
             WriteStreamBuffer?.EndWrite();
+            return default;
         }
 
         private void CheckDisposed()
@@ -198,7 +205,7 @@ public override void Dispose()
         {
             if (!_disposed)
             {
-                Shutdown();
+                CompleteWrites();
 
                 _disposed = true;
             }
@@ -208,7 +215,7 @@ public override ValueTask DisposeAsync()
         {
             if (!_disposed)
             {
-                Shutdown();
+                CompleteWrites();
 
                 _disposed = true;
             }

src/libraries/System.Net.Quic/src/System/Net/Quic/Implementations/MsQuic/MsQuicStream.cs

@@ -4,7 +4,9 @@
 using System.Buffers;
 using System.Collections.Generic;
 using System.Diagnostics;
+using System.IO;
 using System.Net.Quic.Implementations.MsQuic.Internal;
+using System.Runtime.ExceptionServices;
 using System.Runtime.InteropServices;
 using System.Threading;
 using System.Threading.Tasks;
@@ -396,10 +398,37 @@ internal override async ValueTask ShutdownWriteCompleted(CancellationToken cance
             await _state.ShutdownWriteCompletionSource.Task.ConfigureAwait(false);
         }
 
-        internal override void Shutdown()
+        public override void CompleteWrites()
         {
             ThrowIfDisposed();
-            StartShutdown(QUIC_STREAM_SHUTDOWN_FLAGS.GRACEFUL, errorCode: 0);
+            try
+            {
+                StartShutdown(QUIC_STREAM_SHUTDOWN_FLAGS.GRACEFUL, errorCode: 0);
+            }
+            catch (Exception ex) when (ex is not OutOfMemoryException)
+            {
+                throw new IOException("Unable to complete writes: " + ex.Message, ex);
+            }
+        }
+
+        // We don't wait for QUIC_STREAM_EVENT_SEND_SHUTDOWN_COMPLETE event here,
+        // because it is only sent to us once the peer has acknowledged the shutdown.
+        // Instead, this method acts more like shutdown(SD_SEND) in that it only "queues"
+        // the shutdown packet to be sent without any waiting for completion.
+        public override ValueTask CompleteWritesAsync(CancellationToken cancellationToken)
+        {
+            ThrowIfDisposed();
+            if (cancellationToken.IsCancellationRequested) return ValueTask.FromCanceled(cancellationToken);
+
+            try
+            {
+                StartShutdown(QUIC_STREAM_SHUTDOWN_FLAGS.GRACEFUL, errorCode: 0);
+                return default;
+            }
+            catch (Exception ex) when (ex is not OutOfMemoryException)
+            {
+                return ValueTask.FromException(ExceptionDispatchInfo.SetCurrentStackTrace(new IOException("Unable to complete writes: " + ex.Message, ex)));
+            }
         }
 
         // TODO consider removing sync-over-async with blocking calls.

src/libraries/System.Net.Quic/src/System/Net/Quic/Implementations/QuicStreamProvider.cs

@@ -39,12 +39,14 @@ internal abstract class QuicStreamProvider : IDisposable, IAsyncDisposable
 
         internal abstract ValueTask ShutdownWriteCompleted(CancellationToken cancellationToken = default);
 
-        internal abstract void Shutdown();
-
         internal abstract void Flush();
 
         internal abstract Task FlushAsync(CancellationToken cancellationToken);
 
+        public abstract void CompleteWrites();
+
+        public abstract ValueTask CompleteWritesAsync(CancellationToken cancellationToken);
+
         public abstract void Dispose();
 
         public abstract ValueTask DisposeAsync();

src/libraries/System.Net.Quic/src/System/Net/Quic/QuicStream.cs

@@ -9,7 +9,7 @@
 
 namespace System.Net.Quic
 {
-    public sealed class QuicStream : Stream
+    public sealed class QuicStream : DuplexStream
     {
         private readonly QuicStreamProvider _provider;
 
@@ -101,7 +101,9 @@ public override Task WriteAsync(byte[] buffer, int offset, int count, Cancellati
 
         public ValueTask ShutdownWriteCompleted(CancellationToken cancellationToken = default) => _provider.ShutdownWriteCompleted(cancellationToken);
 
-        public void Shutdown() => _provider.Shutdown();
+        public override void CompleteWrites() => _provider.CompleteWrites();
+
+        public override ValueTask CompleteWritesAsync(CancellationToken cancellationToken = default) => _provider.CompleteWritesAsync(cancellationToken);
 
         protected override void Dispose(bool disposing)
         {

src/libraries/System.Net.Quic/tests/FunctionalTests/MsQuicTests.cs

@@ -125,7 +125,7 @@ public async Task WriteTests(int[][] writes, WriteType writeType)
                         }
                     }
 
-                    stream.Shutdown();
+                    await stream.CompleteWritesAsync();
                     await stream.ShutdownWriteCompleted();
                 },
                 async serverConnection =>
@@ -143,7 +143,7 @@ public async Task WriteTests(int[][] writes, WriteType writeType)
                     int expectedTotalBytes = writes.SelectMany(x => x).Sum();
                     Assert.Equal(expectedTotalBytes, totalBytes);
 
-                    stream.Shutdown();
+                    await stream.CompleteWritesAsync();
                     await stream.ShutdownWriteCompleted();
                 });
         }

src/libraries/System.Net.Quic/tests/FunctionalTests/QuicStreamConnectedStreamConformanceTests.cs

@@ -81,7 +81,7 @@ public sealed class MsQuicQuicStreamConformanceTests : QuicStreamConformanceTest
 
     }
 
-    public abstract class QuicStreamConformanceTests : ConnectedStreamConformanceTests
+    public abstract class QuicStreamConformanceTests : DuplexConnectedStreamConformanceTests
     {
         public SslServerAuthenticationOptions GetSslServerAuthenticationOptions()
         {

src/libraries/System.Net.Quic/tests/FunctionalTests/QuicStreamTests.cs

@@ -450,7 +450,7 @@ public async Task ReadWrite_Random_Success(int readSize, int writeSize)
                         sendBuffer = sendBuffer.Slice(chunk.Length);
                     }
 
-                    clientStream.Shutdown();
+                    await clientStream.CompleteWritesAsync();
                     await clientStream.ShutdownWriteCompleted();
                 },
                 async serverConnection =>

src/libraries/System.Net.Sockets/ref/System.Net.Sockets.cs

@@ -114,7 +114,7 @@ public partial class MulticastOption
         public int InterfaceIndex { get { throw null; } set { } }
         public System.Net.IPAddress? LocalAddress { get { throw null; } set { } }
     }
-    public partial class NetworkStream : System.IO.Stream
+    public partial class NetworkStream : System.IO.DuplexStream
     {
         public NetworkStream(System.Net.Sockets.Socket socket) { }
         public NetworkStream(System.Net.Sockets.Socket socket, bool ownsSocket) { }
@@ -135,6 +135,8 @@ public partial class NetworkStream : System.IO.Stream
         public override System.IAsyncResult BeginRead(byte[] buffer, int offset, int count, System.AsyncCallback? callback, object? state) { throw null; }
         public override System.IAsyncResult BeginWrite(byte[] buffer, int offset, int count, System.AsyncCallback? callback, object? state) { throw null; }
         public void Close(int timeout) { }
+        public override void CompleteWrites() { }
+        public override System.Threading.Tasks.ValueTask CompleteWritesAsync(System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) { throw null; }
         protected override void Dispose(bool disposing) { }
         public override int EndRead(System.IAsyncResult asyncResult) { throw null; }
         public override void EndWrite(System.IAsyncResult asyncResult) { }

src/libraries/System.Net.Sockets/src/Resources/Strings.resx

@@ -330,4 +330,7 @@
   <data name="net_sockets_handle_already_used" xml:space="preserve">
     <value>Handle is already used by another Socket.</value>
   </data>
+  <data name="net_io_shutdownfailure" xml:space="preserve">
+    <value>Unable to complete writes on the transport connection: {0}.</value>
+  </data>
 </root>