Add PipeWriter.UnflushedBytes

[davidfowl]

Background and Motivation

Sometimes its useful to know how many bytes have been buffered and not flushed by the pipe writer. This can be used to optimize how much the other side can see at once (a scenario where the pipe writer wants to write with a minimal chunk size). The JSON serializer does a similar thing today when it writing to the output stream

runtime/src/libraries/System.Text.Json/src/System/Text/Json/Serialization/WriteStack.cs

Line 32 in f7308f0

public int FlushThreshold;

Proposed API

namespace System.IO.Pipelines
{
    public abstract class PipeWriter 
    {
+       pubic virtual long? UnflushedBytes { get; }
    }
}

Usage Examples

async Task CopyWithChunks(Stream source, PipeWriter writer, long flushThreshold)
{
    while (true)
    {
        var memory = writer.GetMemory();
        var read = await stream.ReadAsync(memory);

        if (read == 0)
        {
            break;
        }

        writer.Advance(read);

        // Flush if we've reached the threshold
        if (writer.UnflushedBytes > flushThreshold)
        {
            await writer.FlushAsync();
        }
    }

    if (writer.UnflushedBytes > 0)
    {
        await writer.FlushAsync();
    }
}

Alternatives

namespace System.IO.Pipelines
{
    public abstract class PipeWriter 
    {
+       pubic virtual long? BytesWritten { get; }
    }
}

Risks

The nullable makes it a bit tricky to use but we need to indicate that a derived PipeWriter might not have implemented it.

[Wraith2]

Are you sure it needs to be nullable?

[davidfowl]

Are you sure it needs to be nullable?

I was thinking about it being 0 and what sort of defensive code you'd need to write if it was zero or some sentinal value. Maybe it should default to int.MaxValue 😄

[Wraith2]

Given that the natural range will be 0>max I'd have thought the sentinel would be one of the usual suspects of -1 or MinValue. Being nullable just raises the question of what null means to me, negatives I can apply similar experience to and gloss over but null? not sure what that'd mean. For high perf code if it has no real meaning avoiding unneeded nullables seemed wise.

[davidfowl]

So:

async Task CopyWithChunks(Stream source, PipeWriter writer, long flushThreshold)
{
    while (true)
    {
        var memory = writer.GetMemory();
        var read = await stream.ReadAsync(memory);

        if (read == 0)
        {
            break;
        }

        writer.Advance(read);

        // Flush if we've reached the threshold
        if (writer.UnflushedBytes == -1 || writer.UnflushedBytes > flushThreshold)
        {
            await writer.FlushAsync();
        }
    }

    if (writer.UnflushedBytes == -1 || writer.UnflushedBytes > 0)
    {
        await writer.FlushAsync();
    }
}

If you make it int.MaxValue the code could stay the same 😄 . I'm worried about subtle bugs when people use the field for arithmetic operations on the sentinel value.

[davidfowl]

Forgot this is a dupe of #26818

[HurricanKai]

Would be very helpful if this was added. I am currently faking this behaviour myself by tracking the write count myself.
Another reason why this is very useful is that there is only FlushAsync and not Flush, so when writing high-performance serialization code (not using async) there is no way to flush during that, I usually want to flush in the outer serialization loop, as that's async. There it's helpful to know the amount of written to the pipe.
Also helpful for diagnostics and debugging. (Also yes I'm late 😅)

[bartonjs]

Video

• The discussion favored this being a Try method to better communicate to the caller (and code reviewer) that the implementation might not have an answer; then circled back and split the Nullable into a virtual Can property and a virtual throwing property.

namespace System.IO.Pipelines
{
    partial class PipeWriter 
    {
         public virtual bool CanGetUnflushedBytes => false;
         public virtual long UnflushedBytes => throw new NotImplementedException(some message here);
    }
}

[davidfowl]

@Wraith2 do you want to take a stab at implementing this?