API Proposal: Add PipeReader.ReadAsync overloads to allow waiting for a specific amount of data

[davidfowl]

Often a protocol knows how much data it is waiting for (TLS frames being one example, size headered binary protocols being another).

Instead of ReadAsync() then you check the header and you see you don't have enough, so you ReadAsync() check if you have enough and continue doing loops until you have a complete frame. If you could do a "ReadAsync(xxxxx)" and it wouldn't return unless there was an error/complete or at least that amount of data was available it would solve extra looping in every downstream protocol

Proposed API

public abstract partial class PipeReader
{
+    public virtual ValueTask<ReadResult> ReadAsync(int minimumBytes, CancellationToken cancellationToken = default);
}

/cc @Drawaes

[benaadams]

I like it

"ReadAsync(xxxxx)" and it wouldn't return unless there was an error/complete or at least that amount of data was available it

Should it be a suggestion/hint of size?

What happens if there will be a large delay before the next bit of data? (e.g. TLS may want to close the frame early and send anyway).

Often there are variable sized protocols, which have a preferred frame size (e.g. TCP at packet sizes, TLS at frame size), which can close the frame early and send to next pipe stage/output; however are more efficient doing the send on Frame boundaries.

[davidfowl]

What happens if there will be a large delay before the next bit of data? (e.g. TLS may want to close the frame early and send anyway).

Then you cancel the read to yield it.

[benaadams]

Then you cancel the read to yield it.

So the pattern would be?

public async Task Write15LinesAsync()
{
    PipeWriter buffer = Pipe.Writer;
    var length = 'z' - 'a' + 3;
    for (var i = 0; i < 15; i++)
    {
        var written = WriteAtoZ(buffer.GetMemory(length));
        buffer.Advance(written);
        await buffer.FlushAsync(); // Write data
    }
    await Pipe.Reader.CancelPendingRead(); // Clear any buffered data
}

public int WriteAtoZ(Memory<byte> memory)
{
    Span<byte> span = memory.Span;
    var i = 0;
    for (var c = 'a'; c <= 'z'; c++)
    {
        span[i] = (byte)c;
        i++;
    }
    span[i + 1] = (byte)'\r';
    span[i + 2] = (byte)'\n';

    return i + 2;
}

[benaadams]

Or would cancel/wake up happen with Flush but not Write so it would be:

public async Task Write15LinesAsync()
{
    PipeWriter buffer = Pipe.Writer;
    var length = 'z' - 'a' + 3;
    for (var i = 0; i < 15; i++)
    {
        var written = WriteAtoZ(buffer.GetMemory(length));
        buffer.Advance(written);
        await buffer.WriteAsync(); // Write data
    }
    await buffer.FlushAsync(); // Clear any buffered data
}

e.g. how do you

Then you cancel the read to yield it.

[davidfowl]

The question is really about where the knowledge is (reader or writer). In this simple example, you can just flush at the end and not Write for each Advance (so it's a bit contrived).

[benaadams]

The question is really about where the knowledge is (reader or writer).

Both sizes have knowledge; the Writer knows when it has no more pending data; the Reader knows either its ideal chunk sizes; or whether it has enough data to parse.

In this simple example, you can just flush at the end and not Write for each Advance

Is simplified for ease of concepts; the point is you can't Advance forever or you'll run out of memory; and you get no backpressure feedback.

Ok. a more complex example:

public async Task ExecuteAsync()
{
    try
    {
        var bytesWritten = 0u;
        while (true)
        {
            if (!_pipe.Reader.TryRead(out var result))
            {
                // No waiting input data
                if (bytesWritten > 0)
                {
                    // Flush any written data before waiting for more input
                    var flushResult = _pipe.Writer.FlushAsync();
                    if (!flushResult.IsCompleted)
                    {
                        await flushResult;
                    }
                    bytesWritten = 0u;
                }
                // Await for more input
                result = await _pipe.Reader.ReadAsync();
            }

            var inputBuffer = result.Buffer;
            var consumed = inputBuffer.Start;
            var examined = inputBuffer.End;

            try
            {
                if (inputBuffer.IsEmpty && result.IsCompleted)
                {
                    break;
                }

                ParseRequest(inputBuffer, out consumed, out examined);

                if (_state != State.Body && result.IsCompleted)
                {
                    // Bad request, finish request processing
                    break;
                }

                if (_state == State.Body)
                {
                    bytesWritten += await ProcessRequestAsync();

                    _state = State.StartLine;
                }
            }
            finally
            {
                _pipe.Reader.AdvanceTo(consumed, examined);
            }
        }

        // Flush any pending data
        await _pipe.Writer.FlushAsync();
        _pipe.Reader.Complete();
    }
    catch (Exception ex)
    {
        _pipe.Reader.Complete(ex);
    }
    finally
    {
        _pipe.Writer.Complete();
    }
}

private async ValueTask<uint> ProcessRequestAsync()
{
    // Processing a request Writes, but doesn't Flush, 
    // unless it needs to trigger immediate sending
    // like after a websocket message, server sent event or
    // at end of </head> to get early browser read of css, fonts and preload, prefetch links
    await _pipe.Writer.WriteAsync(_plainTextBytes);
    return (uint)_plainTextBytes.Length;
}

So the question is how do you differentiate between a forced early wake up, and an enough data wakeup on the Writer and Reader ends.

I'm suggesting using WriteAsync and FlushAsync for the Writer side as they are well understood api tropes from Streams; so new users of Pipelines will naturally start using them correctly in that way as its already what they understand, there are no new concepts to use.

[benaadams]

Also that you want to default the first param of WriteAsync

public virtual PipeAwaiter<FlushResult> WriteAsync(ReadOnlyMemory<byte> source = default, CancellationToken cancellationToken = default)

So you can do

await WriteAsync();

When you are only writing using Span and Advance

[davidfowl]

Is simplified for ease of concepts; the point is you can't Advance forever or you'll run out of memory; and you get no backpressure feedback.

Then don't FlushAsync, use a reasonable threshold that won't run out of memory? What's the scenario where you will need to buffer so much that you could run out?

I'm suggesting using WriteAsync and FlushAsync for the Writer side as they are well understood api tropes from Streams; so new users of Pipelines will naturally start using them correctly in that way as its already what they understand, there are no new concepts to use.

WriteAsync without a buffer is not an easily understood concept that carries over to normal usage.

So the question is how do you differentiate between a forced early wake up, and an enough data wakeup on the Writer and Reader ends.

You need to describe how the other side of the writer wants to handle consuming the data written during request processing.

[benaadams]

Second scenario, TextWriter without arrays as buffers (as generally the encoding happens in Flush so it needs to hold conversion buffers) ; and encoding directly to Pipe without intermediary copies.

Holding onto these arrays causes issues https://github.com/dotnet/coreclr/issues/16389 as well as contention getting the from the pool.

System.OutOfMemoryException:
   at System.Buffers.TlsOverPerCoreLockedStacksArrayPool`1.Rent
   at Microsoft.AspNetCore.WebUtilities.HttpResponseStreamWriter..ctor
   at Microsoft.AspNetCore.Mvc.Internal.MemoryPoolHttpResponseStreamWriterFactory.CreateWriter
   at Microsoft.AspNetCore.Mvc.Formatters.JsonOutputFormatter+<WriteResponseBodyAsync>d__11.MoveNext

With Pipes the arrays just aren't needed as the Pipe brings its own memory.

Stream that wraps Pipe

public class DuplexPipeStream : Stream, IDuplexPipe
{
    private Pipe _pipe;
    PipeReader IDuplexPipe.Input => _pipe.Reader;
    PipeWriter IDuplexPipe.Output => _pipe.Writer;

    public override async Task FlushAsync(CancellationToken cancellationToken)
        => await _pipe.Writer.FlushAsync(cancellationToken);

    public override Task WriteAsync(ReadOnlyMemory<byte> source, CancellationToken cancellationToken = default)
        => _pipe.Writer.WriteAsync(source, cancellationToken);

    public override async ValueTask<int> ReadAsync(Memory<byte> destination, CancellationToken cancellationToken = default)
    {
        while (true)
        {
            var result = await _pipe.Reader.ReadAsync();
            var readableBuffer = result.Buffer;
            try
            {
                if (!readableBuffer.IsEmpty)
                {
                    // buffer.Count is int
                    var count = (int)Math.Min(readableBuffer.Length, destination.Count);
                    readableBuffer = readableBuffer.Slice(0, count);
                    readableBuffer.CopyTo(destination);
                    return count;
                }
                else if (result.IsCompleted)
                {
                    return 0;
                }
            }
            finally
            {
                _pipe.Reader.AdvanceTo(readableBuffer.End, readableBuffer.End);
            }
        }
    }

    public override Task WriteAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
        => WriteAsync(new Memory<byte>(buffer, offset, count), cancellationToken);
    public override Task<int> ReadAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
        => ReadAsync(new Memory<byte>(buffer, offset, count), cancellationToken).AsTask();

    // Other stream overloads
}

Then TextWriter that works with that Stream; encoding on the go in WriteAsync rather than waiting for FlushAsync and having no internal; but it would rely on WriteAsync only occasionally acting as a Flush (as it does in StreamTextWriter currently) to maintain performance.

public class HttpResponseStreamWriter : TextWriter
{
    internal const int DefaultBufferSize = 16 * 1024;

    private IDuplexPipe _pipe;
    private StreamWriter _streamWriter;
    private readonly Encoder _encoder;

    public PipeStreamWriter(Stream stream, Encoding encoding)
        : this(stream, encoding, DefaultBufferSize) {}

    public PipeStreamWriter(Stream stream, Encoding encoding, int bufferSize)
    {
        Encoding = encoding;
        if (stream is IDuplexPipe pipe)
        {
            _pipe = pipe;
        }
        else
        {
            _streamWriter = new StreamWriter(stream, encoding, bufferSize);
        }
    }

    public override Encoding Encoding { get; }

    public unsafe override void Write(char value)
    {
        if (_pipe != null)
        {
            Span<byte> bytes = _pipe.Output.GetSpan(Encoding.GetMaxByteCount(1));
            var encoded = Encoding.GetBytes(new ReadOnlySpan<byte>(&value, sizeof(char)), bytes);
            _pipe.Output.Advance(encoded);
        }
        else
        {
            _streamWriter.Write(value);
        }
        }
        else
        {
            _streamWriter.Write(value);
        }
    }


    public override async Task WriteAsync(ReadOnlyMemory<char> source)
    {
        if (_pipe != null)
        {
            Write(source, default(CancellationToken));
            await _pipe.Output.WriteAsync();
        }
        else
        {
            await _streamWriter.WriteAsync(source);
        }
    }

    private void Write(ReadOnlyMemory<char> source, CancellationToken token)
    {
        ReadOnlySpan<char> input = source.Span;
        int minBytes = Encoding.GetMaxCharCount(1);
        while (input.Length > 0)
        {
            Span<byte> bytes = _pipe.Output.GetSpan(minBytes);
            int totalEncoded = 0;
            while (bytes.Length > 0)
            {
                int toEncode = Math.Min(Encoding.GetMaxCharCount(bytes.Length), input.Length);
                var encoded = Encoding.GetBytes(input.Slice(0, toEncode), bytes);
                input = input.Slice(toEncode);
                bytes = bytes.Slice(encoded);
                totalEncoded += encoded;
                if (bytes.Length < minBytes)
                {
                    break;
                }
            }
            _pipe.Output.Advance(totalEncoded);
        }
    }

    public override Task WriteAsync(char[] values, int index, int count)
        => WriteAsync(new ReadOnlyMemory<char>(values, index, count));

    public override Task WriteAsync(string value)
        => WriteAsync(value.AsReadOnlyMemory());

    public override Task FlushAsync()
    {
        if (_pipe != null)
        {
            await _pipe.Output.FlushAsync();
        }
        else
        {
            await _streamWriter.FlushAsync();
        }
    }

    // Other TextWriter overloads
}

[benaadams]

So the question is if WriteAsync called FlushAsync as now; then TextWriter.FlushAsync wouldn't call FlushAsync; instead it would need to cancel the reader of the next stage in the pipeline

Then you cancel the read to yield it.

So it would additionally need with the wrapping Stream the next stage in the pipeline so it could call the void method CancelPendingRead?

    public override Task FlushAsync()
    {
        if (_pipe != null)
        {
           // Where does _nextPipe come from?
           // Is it a problem CancelPendingRead is acting as a non-awaitable Flush?
            _nextPipe.Input.CancelPendingRead();
        }
        else
        {
            await _streamWriter.FlushAsync();
        }
    }

[benaadams]

btw I think "ReadAsync(xxxxx)" is a good api; I just want more detail on how to wake it up early; and how the Reader tells it was woken up early (other than first checking how much data there is and seeing if it agrees with the amount that was requested; then changing behaviour based on that).

[davidfowl]

You still haven't shown the consuming side. What's on the other end of the pipe that's flushing? I'd like to see the transport code basically and how it uses these 2 features. What I hear you asking for is a force flush from the writer side even though the reader is looking for a specific buffer size.

[benaadams]

What's on the other end of the pipe that's flushing? I'd like to see the transport code basically and how it uses these 2 features.

Something along the lines of:

const int PreferredFrameSize = 1024; // Varies based on Reader's framing size

public async Task WriteFramesAsync(CancellationToken cancellationToken)
{
    try
    {
        await WriteFramesAsync(_pipe.Reader, cancellationToken).ConfigureAwait(false);
    }
    catch (Exception ex)
    {
        _pipe.Reader.Complete(ex);
    }
    finally
    {
        _pipe.Writer.Complete();
    }
}

private async Task WriteFramesAsync(PipeReader reader, CancellationToken cancellationToken)
{
    while (true)
    {
        // Read specifying preferred readsize hint
        ReadResult result = await reader.ReadAsync(PreferredFrameSize, cancellationToken);

        ReadOnlySequence<byte> inputBuffer = result.Buffer;
        SequencePosition consumed = inputBuffer.Start;
        SequencePosition examined = inputBuffer.End;

        bool shouldConsumeAll = (result.IsFlush || result.IsCompleted);

        try
        {
            while (true)
            {
                cancellationToken.ThrowIfCancellationRequested();
                // Output frame, passing on Flush if last frame of current data
                await OutputFrameAsync(
                    inputBuffer, 
                    flush: (inputBuffer.Length <= PreferredFrameSize ? shouldConsumeAll : false), 
                    out consumed, 
                    out examined, 
                    cancellationToken);

                inputBuffer = inputBuffer.Slice(consumed);
                if (inputBuffer.IsEmpty)
                {
                    // Frame completed and no more data, await more data 
                    break;
                }

                if (!result.IsCompleted && inputBuffer.Length < PreferredFrameSize)
                {
                    // Under preferred size, check if more input was made 
                    // available while writing Frames, Completed won't have more data
                    reader.AdvanceTo(consumed, examined);
                    if (reader.TryRead(out ReadResult newResult))
                    {
                        result = newResult;
                        inputBuffer = result.Buffer;
                        consumed = inputBuffer.Start;
                        examined = inputBuffer.End;

                        // Apply inital Flush if originally requested
                        shouldConsumeAll = (shouldConsumeAll || result.IsFlush || result.IsCompleted);
                    }
                }

                if (!shouldConsumeAll && inputBuffer.Length < PreferredFrameSize)
                {
                    // Not enough left to fill a frame and not Flush or Complete, await more data
                    break;
                }
            }

            if (inputBuffer.IsEmpty && result.IsCompleted)
            {
                // Finished, exit
                break;
            }
        }
        finally
        {
            reader.AdvanceTo(consumed, examined);
        }
    }

    reader.Complete();
}

[davidfowl]

I think the problem here is that you need to guarantee that the writer absolutely is going to flush or you can end up in a situation where you don't have enough data for a frame and you haven't consumed enough to release back pressure. Thats' fundamentally what I think is broken about the reader using IsFlush to make decisions like this.

That's why I think you always need this timeout on the reader side that tries to consume data as a failsafe.

[benaadams]

If you are in back pressure you have ReadAsync immediately complete with IsFlush automatically set.

If the reader stops making progress while in back pressure; then it should throw
ThrowHelper.ThrowInvalidOperationException_BackpressureDeadlock();
on Advance or perhaps better next ReadAsync (as its going to return the same data)?

Then it would tear everything down?