Feature Request: Add ability to explicitly send EOF for connected Streams

[geoffkizer]

Background and Motivation

Connected Streams -- like NetworkStream, PipeStream, SslStream, and the new QuicStream class -- need the ability to send EOF to indicate that the write side has completed without closing the Stream (and thus closing the read side too).

At the Socket layer, we support this via Socket.Shutdown(SocketShutdown.Send). We don't expose this in NetworkStream, however.

This is most useful for "connected" Streams that can Read and Write but not Seek. It's not clear whether this concept applies to streams like FileStream that can Seek.

Proposed API

Based on comments from this issue: #43101

The basic idea is to add an API like the following:

public abstract class Stream
{
  public virtual bool CanShutdownWrite =>
    false;
  public virtual void ShutdownWrite() =>
    Flush();
  public virtual ValueTask ShutdownWriteAsync(CancellationToken cancellationToken = default) =>
    new ValueTask(FlushAsync(cancellationToken));
}

We'd like naming guidance:

• "ShutdownWrite" corresponds to existing Socket terminology ("shutdown"), but isn't necessarily the most intuitive term for non-Socket scenarios. Suggestions welcome. Unfortunately we cannot use "EndWrite" for obvious reasons. Other alternatives include "FinishWrite" and "CompleteWrite".
• ShutdownWriteAsync will complete once e.g. a shutdown packet has been sent, NOT when the peer acknowledges the shutdown. There is some concern that this isn't obvious from the name.

The hard question is where this API lives. There are three options here:

1. Add a new abstract base class ConnectedStream (or whatever -- naming suggestions welcome) that derives from Stream and that our various existing streams inherit where appropriate (e.g. NetworkStream, SslStream, PipeStream, QuicStream, possibly others).

   • This has a con: other streams like GzipStream that wrap a base stream and would be appropriate to pass-through a shutdown request no longer can.

2. Just add a new virtual method to Stream itself, ideally with a reasonable default implementation, e.g. just calling Dispose (if there's no good default behavior, then potentially also a Can/Is capability test property).

   • This has a con: this feature isn't really applicable for many uses of Stream.
   • The "CanShutdownWrite" property is really only needed if we add this to Stream. We can probably remove it otherwise. Without it, any code depending on correct shutdown behavior would break in a difficult to diagnose way.
   • Not specifically API but there is debate on if the Shutdown methods should default throw or flush. The intended behavior is that a successful shutdown also implies a flush.

3. Add a mix-in via an interface like IConnectedStream that can be tested for, e.g. if (stream is IConnectedStream cs) cs.ShutdownWrites();

   • This has the same con as a new class: it breaks composition with transforming streams.

We'd like naming guidance: Shutdown()

Another issue to consider is whether to provide an overload for Write/WriteAsync that allows you to pass a flag indicating that the Stream should send EOF after the specified buffer, e.g. something like this:

public abstract class Stream
{
  public virtual Task WriteAsync(ReadOnlyMemory<byte> buffer, CancellationToken cancellationToken, bool shutdownWrite);
}

This is important because it enables the data and EOF to be sent in a single packet, which can improve performance, specifically for QUIC (and HTTP3 over QUIC) as well as HTTP2 request streams.

If we add a WriteAsync overload like this, then we could opt not to do ShutdownWrite at all and instead achieve this by passing a 0-length buffer and shutdownWrite=true to the WriteAsync overload.

cc @scalablecory @stephentoub

[Dotnet-GitSync-Bot]

I couldn't figure out the best area label to add to this issue. If you have write-permissions please help me learn by adding exactly one area label.

[geoffkizer]

@scalablecory Is this required by HTTP3? I.e. do we need to be able to shutdown writes on the QUIC stream as part of sending the HTTP3 request?

[scalablecory]

@geoffkizer yes QUIC needs both a ShutdownWrites() and a Write(..., bool shutdownWrites).

The 2nd one is important to QUIC because it causes the data and the FIN bit to be sent in a single packet.

(Note: QUIC needs a whole lot more too, so it will need its own abstraction anyway. I think we could live without the 2nd on Stream)

[geoffkizer]

The 2nd one is important to QUIC because it causes the data and the FIN bit to be sent in a single packet.

Good point -- I added some text at the end about this in the proposal.

[scalablecory]

SslStream should implement this too, and send a close_notify alert. This requires I/O -- we could probably hide it behind the scenes, but should consider making this shutdown method async. @wfurt

Things like lengthless HTTP/1.0 requests can't be implemented over TLS without this.

[scalablecory]

Note: "ShutdownWrite" corresponds to existing Socket terminology ("shutdown"), but isn't necessarily the most intuitive term for non-Socket scenarios. Suggestions welcome. Unfortunately we cannot use "EndWrite" for obvious reasons. Other alternatives include "FinishWrite" and "CompleteWrite".

Our other stream type PipeWriter uses the language "Complete", which seems perfectly fine here. Unless objections, we should go with that if only for consistency.

[geoffkizer]

"Complete" always strikes me as awkward because it's not clear what you are completing... it seems to imply that previous writes are not "complete" somehow until I call this.

Another option here would be "CloseWrite".

[scalablecory]

One quirk I've ran into while prototyping this is how it behaves when shutdown is not supported.

Say I'm taking a generic Stream and depend e.g. HTTP/1.0 behavior where you need to shutdown writes to trigger a response.

If we make the default implementation do nothing, my code might block indefinitely waiting for a response that will never come.

Similarly, if we make the default implementation throw, I won't discover things will break until I'm at the end of a potentially very large write operation.

Here is a pattern that does work:

• Introduce (I hate these) a virtual bool CanShutdownWrites => false member.
• Wrapper streams will implement as override bool CanShutdownWrites => baseStream.CanShutdownWrites
• HTTP/1.0 client can now check very early if the Stream is incompatible.

One thing I've tried that does not work is to introduce a DuplexStream base type: this makes generic wrapper streams (think e.g. GzipStream) impossible to implement in a way that works based on the base stream.

[scalablecory]

@stephentoub I'd love to see this in .NET 6 -- are you okay bringing to API review?

[stephentoub]

What is the proposed API at this point? I see open questions and varying ideas without a solidified answer.

[scalablecory]

I've done some prototyping here and would propose the API as:

enum FlushMode
{
    None,
    FlushWrites,
    FlushAndShutdownWrites
}

class Stream
{
    public virtual bool CanShutdownWrites { get; }
    public virtual void Flush(FlushMode flushMode);
    public virtual ValueTask FlushAsync(FlushMode flushMode, CancellationToken cancellationToken = default);

    public virtual void Write(ReadOnlySpan<byte> buffer, FlushMode flushMode);
    public virtual ValueTask WriteAsync(ReadOnlyMemory<byte> buffer, FlushMode flushMode, CancellationToken cancellationToken);
}

The CanShutdownWrites would be so that any use of the Stream can verify up front that their network-oriented/etc. code will work if it depends on shutdowns.

I think Shutdown()'s behavior would also be to flush any buffers, so combining it with Flush() feels natural. Bonus: we can have a ValueTask-returning FlushAsync.

The Write overloads are stretch goals here. They'd be useful for a couple optimizations things:

• Combining Write()+Flush() can alter how buffering is or isn't done, reducing copies.
• MsQuic and others have APIs have a Write(bool flush) feature that this would map to very nicely.
• Some protocols (e.g. QUIC) communicate EOF as part of their data framing, and this would allow us to send EOF without sending two frames or even two separate packets.

Default implementation would look like:

    public virtual bool CanShutdownWrites { get; } => false;

    public virtual void Flush(FlushMode flushMode)
    {
        if(flushMode == FlushMode.FlushWrites) Flush();
        else if(flushMode == FlushMode.FlushAndShutdownWrites) throw new IOException("stream does not support shutdown.");
    }

    public virtual ValueTask FlushAsync(FlushMode flushMode, CancellationToken cancellationToken = default) =>
        flushMode == FlushMode.FlushWrites => new ValueTask(FlushAsync(cancellationToken)) :
        flushMode == FlushMode.FlushAndShutdownWrites => ValueTask.FromException(new IOException("stream does not support shutdown.")) :
        cancellationToken.IsCancellationRequested => ValueTask.FromCanceled(cancellationToken) :
        default;

    public virtual void Write(ReadOnlySpan<byte> buffer, FlushMode flushMode)
    {
        Write(buffer);
        Flush(flushMode);
    }

    public virtual ValueTask WriteAsync(ReadOnlyMemory<byte> buffer, FlushMode flushMode, CancellationToken cancellationToken)
    {
        ValueTask writeTask = WriteAsync(buffer, cancellationToken);

        return flushMode == FlushMode.None
            ? writeTask
            : FinishWriteAndFlushAsync(writeTask, this, flushMode, cancellationToken);

        static async ValueTask FinishWriteAndFlushAsync(ValueTask writeTask, Stream stream, FlushMode flushMode, CancellationToken cancellationToken)
        {
            await writeTask.ConfigureAwait(false);
            await stream.FlushAsync(flushMode, cancellationToken).ConfigureAwait(false);
        }
    }

Wrapper streams without buffering like GzipStream would be able to pass through support:

class MyCustomStream : Stream
{
    readonly Stream _baseStream;

    public override bool CanShutdownWrites =>
        _baseStream.CanShutdownWrites;

    public override void Flush(FlushMode flushMode) =>
        _baseStream.Flush(flushMode);

    public override ValueTask FlushAsync(FlushMode flushMode, CancellationToken cancellationToken) =>
        _baseStream.FlushAsync(flushMode, cancellationToken);
}

With buffering would look like:

class MyCustomStream : Stream
{
    readonly Stream _baseStream;
    readonly Buffer _buffer;

    public override bool CanShutdownWrites =>
        _baseStream.CanShutdownWrites;

    public override void Flush(FlushMode flushMode)
    {
        if(flushMode == FlushMode.None)
        {
            return;
        }
        if(_buffer.ActiveLength != 0)
        {
            _baseStream.Write(_buffer.ActiveSpan, flushMode);
        }
        else
        {
            _baseStream.Flush(flushMode);
        }
    }

    public override ValueTask FlushAsync(FlushMode flushMode, CancellationToken cancellationToken) =>
        _buffer.ActiveLength != 0 ? _baseStream.WriteAsync(_buffer.ActiveMemory, flushMode, cancellationToken) :
        _baseStream.FlushAsync(flushMode, cancellationToken);
}

[tmds]

@scalablecory maybe instead of adding flags specific for flushing, we can add a more generic WriteFlags that can be used for some of the other Stream related API proposals you and @geoffkizer have.

[Flags]
enum WriteFlags
{
}
class Stream
{
    public virtual void Write(ReadOnlySpan<byte> buffer, WriteFlags writeFlags);
    public virtual ValueTask WriteAsync(ReadOnlyMemory<byte> buffer, WriteFlags writeFlags, CancellationToken cancellationToken);
}

[scalablecory]

@tmds Yeah, I thought of that too.

If we can find any additional flags we might want, lets consider them. I can't think of any off the top of my head. Adding it as a future extensibility point is dubious, as any future flags would not have to break anyone who implemented it before the new flag was added.

[scalablecory]

Can see a prototype being used of my proposal here:

https://github.com/dotnet/runtimelab/blob/e654eb59970f86a1978246a9c2a94f0383db6279/src/libraries/System.Net.Http.Primitives/src/System/Net/Http/Primitives/Http1Connection.cs#L725

[geoffkizer]

I am not sure we are adding any value with the Flush overloads.

Flush(None) doesn't make any sense.
Flush(FlushWrites) is just the same as existing Flush.
Flush(FlushAndShutdownWrites) can be accomplished by calling the new Write overload with an empty buffer.

[stephentoub]

I'm very hesitant to add new Write overloads that also implicitly flush. Now every time we add new write APIs (e.g. vector I/O), are we going to overload for flushing, or have additional defaulted flush arguments, or some such things?

[geoffkizer]

I'm very hesitant to add new Write overloads that also implicitly flush.

To be clear -- are you objecting specifically to passing FlushWrites to Write, or also FlushAndShutdownWrites too?

[stephentoub]

My comment was specific to adding new Write overloads that take a flush enum, in particular because then it seems we basically need to add new WriteXx overloads that take the enum for existing write methods and future write methods.

[geoffkizer]

One alternative (as described in the original post here) is to have this be a bool that has nothing to do with flush specifically:

public virtual Task WriteAsync(ReadOnlyMemory buffer, CancellationToken cancellationToken, bool shutdownWrite);

But I assume your objection applies to this alternative as well, for the same reason.

[scalablecory]

I'm very hesitant to add new Write overloads that also implicitly flush. Now every time we add new write APIs (e.g. vector I/O), are we going to overload for flushing, or have additional defaulted flush arguments, or some such things?

I'm defaulting the flush argument for the vector Write and it's working great.

But yes, we could get by without Write overloads if needed. It just won't be as efficient.

[scalablecory]

I am not sure we are adding any value with the Flush overloads.

It's a more intuitive API. Shutdown implies flush, and this makes that clear. It gets rid of the question (by both implementers and users) "I called shutdown, should I have called flush too?"

A ValueTask returning FlushAsync is also a good thing to have.

Flush(None) doesn't make any sense.

Agreed; None is only there to support the Write overload. Might make sense to split the enum into two, but I've found value (see the prototype code linked) in having them both use the same enum.

Flush(FlushAndShutdownWrites) can be accomplished by calling the new Write overload with an empty buffer.

I would expect this to work for any implementation, but it feels very bad to suggest that be THE way to shutdown.

[geoffkizer]

Seems like we should also discuss this issue at the same time: #44782

[geoffkizer]

I would expect this to work for any implementation, but it feels very bad to suggest that be THE way to shutdown.

I agree with this, but I'd suggest exposing a simple ShutdownWrites() call instead of exposing it through Flush.

We want auto-flush streams like NetworkStream and SslStream to support sending EOF. Currently when I use those streams, I ignore Flush completely. It seems weird to me that we'd now say, use Flush to send EOF.

[scalablecory]

I'd suggest exposing a simple ShutdownWrites() call instead of exposing it through Flush.

We want auto-flush streams like NetworkStream and SslStream to support sending EOF. Currently when I use those streams, I ignore Flush completely. It seems weird to me that we'd now say, use Flush to send EOF.

I'm having trouble seeing the downside here, only upside.

I guess your argument is that Shutdown should not imply Flush. What do you think behavior should be on a Stream if you call Shutdown without having a flushed?

[stephentoub]

All built-in types that use a Stream will be updated to call ShutdownWrite() when they would have called Flush()+Dispose() inside of e.g. a Dispose().

You mean call ShutdownWrite+Dispose? If you mean just ShutdownWrite and not Dispose, we can't do that.

CanShutdownWrite

Other Can methods that return false mean that the corresponding operations fail when used. This one apparently doesn't. That's not a deal breaker, but it's an unfortunate inconsistency.

What code will use this property? What will a typical decision based on it look like?

[scalablecory]

All built-in types that use a Stream will be updated to call ShutdownWrite() when they would have called Flush()+Dispose() inside of e.g. a Dispose().

You mean call ShutdownWrite+Dispose? If you mean just ShutdownWrite and not Dispose, we can't do that.

Yes.

CanShutdownWrite

Other Can methods that return false mean that the corresponding operations fail when used. This one apparently doesn't. That's not a deal breaker, but it's an unfortunate inconsistency.

That's a good point. I don't think that'd be bad behavior, honestly.

What code will use this property? What will a typical decision based on it look like?

void MakeHttp10Request(Stream stream)
{
   if(!stream.CanShutdownWrite) throw new Exception("HTTP/1.0 requires the ability to shutdown writes on a Stream");
}

void MakeHttp30Request(Stream stream)
{
   if(!stream.CanShutdownWrite) throw new Exception("HTTP/3.0 requires the ability to shutdown writes on a Stream");
}

[stephentoub]

I don't think that'd be bad behavior, honestly

Throwing from ShutdownWrite in the base implementation? What would all of those call sites you wanted to update do then? If they'd all be required to make the virtual Can call and then fallback themselves to Flush, that seems similarly less than ideal.

if(!stream.CanShutdownWrite) throw new Exception("HTTP/1.0 requires the ability to shutdown writes on a Stream");

What would PipeStream for example return from CanShutdownWrite? We wouldn't let it be used in SocketsHttpHandler just because it doesn't support shutting down just one direction?

[scalablecory]

Throwing from ShutdownWrite in the base implementation? What would all of those call sites you wanted to update do then? If they'd all be required to make the virtual Can call and then fallback themselves to Flush, that seems similarly less than ideal.

Yeah, that's a good point. I'm worried that if you need proper shutdown semantics and you forget to check CanShutdownWrite, you'd get silent failures.

What would PipeStream for example return from CanShutdownWrite? We wouldn't let it be used in SocketsHttpHandler just because it doesn't support shutting down just one direction?

It just means we wouldn't let it be used with HTTP/1.0 with a lengthless request, as there is no chunked encoding to denote EOF and without EOF the other side won't know when request content ends, to start sending the response.

[bartonjs]

Video

We felt that the concept really warranted introducing a new middle type to the hierarchy:

namespace System.IO
{
    public abstract class DuplexStream : Stream
    {
        // when disposed, this write-only stream will call CompleteWrites().
        // this allows compat with e.g. StreamWriter that knows nothing about shutdown.
        public Stream GetWriteOnlyStream() => throw null;
    
        public abstract void CompleteWrites();
        public abstract ValueTask CompleteWritesAsync(CancellationToken cancellationToken = default);
    
        override all the things;
    }
}

partial class NetworkStream : DuplexStream 
{
}

and other streams as needed.

(I edited so that NetworkStream derives from DuplexStream, not BidirectionalStream)

[davidfowl]

cc @Tratcher @halter73

[campersau]

The name DuplexStream exists in several libraries already would this clash?
Like aspnet https://github.com/dotnet/aspnetcore/blob/6868d7fb664facb03373d85f222280b248ed105a/src/Servers/IIS/IIS/src/Core/DuplexStream.cs
and others https://github.com/search?l=C%23&q=%22class+duplexstream%22&type=Code

[scalablecory]

@campersau looking here (thanks @stephentoub for showing me this wonderful tool): https://grep.app/search?q=class%20DuplexStream%20&filter[lang][0]=C%23

I think we're okay.

There appear to be very few, and all of them are internal types. Almost all of them are ASP.NET or YARP (CC @Tratcher) that we can get fixed easily enough. What's left is @jstedfast's MailKit and a couple unmaintained and/or Framework-only repos.