Runtime: PipeWriter: no way to wait for reader to catch up

[AArnott]

When I use a Pipe as a way to write to something else, there is no way for the PipeWriter to flush such that it returns a Task that signifies when the PipeReader has caught up with the PipeWriter up to a given point. When I'm writing directly to a stream, I can write several times, then call await Stream.FlushAsync() and know that everything I've written has been transmitted before continuing.

But when using a PipeWriter, there's no way to know when the PipeReader is done up to a point, short of completing the writer, which of course is not a repeatable thing like flushing is.

Even completing the writer is really tedious. You can see a test I wrote that demonstrates that it takes 16 lines of code to do the one-time flush operation of knowing when the reader of the pipe is caught up, as you can see below.

The PipeWriter has a FlushAsync method but from the only docs we have (the blog) it is hinted at that this method doesn't complete when the reader is done, but rather just gives the reader access to what we've written (and I guess serves as a force-slowdown on the writer if the reader is far behind).

Can we please get a proper FlushAsync method that indicates when the reader has caught up with the writer at a given point?

[Fact]
public async Task UsePipeWriter()
{
    byte[] expectedBuffer = GetRandomBuffer(2048);
    var stream = new MemoryStream(expectedBuffer.Length);
    var writer = stream.UsePipeWriter(this.TimeoutToken);
    await writer.WriteAsync(expectedBuffer.AsMemory(0, 1024), this.TimeoutToken);
    await writer.WriteAsync(expectedBuffer.AsMemory(1024, 1024), this.TimeoutToken);

    // As a means of waiting for the async process that copies what we write onto the stream,
    // complete our writer and wait for the reader to complete also.
    writer.Complete();
    var readerDone = new TaskCompletionSource<object>();
    writer.OnReaderCompleted(
        (ex, tcs) =>
        {
            if (ex != null)
            {
                readerDone.SetException(ex);
            }
            else
            {
                readerDone.SetResult(null);
            }
        },
        null);
    await readerDone.Task.WithCancellation(this.TimeoutToken);

    Assert.Equal(expectedBuffer, stream.ToArray());
}

Where UsePipeWriter is defined this way:

public static PipeWriter UsePipeWriter(this Stream stream, CancellationToken cancellationToken = default)
{
    Requires.NotNull(stream, nameof(stream));

    var pipe = new Pipe();
    Task.Run(async delegate
    {
        try
        {
            while (true)
            {
                ReadResult readResult = await pipe.Reader.ReadAsync(cancellationToken);
                if (readResult.Buffer.Length > 0)
                {
                    foreach (ReadOnlyMemory<byte> segment in readResult.Buffer)
                    {
                        await stream.WriteAsync(segment, cancellationToken);
                    }

                    await stream.FlushAsync(cancellationToken).ConfigureAwait(false);
                }

                pipe.Reader.AdvanceTo(readResult.Buffer.End);

                if (readResult.IsCompleted)
                {
                    break;
                }
            }

            pipe.Reader.Complete();
        }
        catch (Exception ex)
        {
            pipe.Reader.Complete(ex);
            throw;
        }
    }).Forget();
    return pipe.Writer;
}

[benaadams]

Use the Pipe constructor taking PipeOptions and set pauseWriterThreshold and resumeWriterThreshold?

/cc @davidfowl

[davidfowl]

I think part of the problem is that the intent of the Pipe specifically is that the reader and writer are completely decoupled. In this scenario it seems as though you're looking for a way to know when the written data was fully consumed by the other side.

What's the scenario?

[AArnott]

@benaadams I'm not looking for thresholds. Those are useful in their own right. I'm looking specifically for a way to know when the reader has done its job.

@davidfowl The scenario is interop across APIs. In my Nerdbank.Streams package I'm including some classes that make a Stream look like a PipeReader/PipeWriter, or vice versa. That way I (and others) can use pipes everywhere they make sense, without fear that I'm cutting off my audience that may very well require a Stream to interact with the so many more APIs that accept Stream. Stream.FlushAsync tells me when the writing is done (actually on the network, for example), but there is nothing like that in PipeWriter.
In one respect, it's nice that pipes will eventually flush without having to worry or block on a flush call, whereas streams may buffer and let 1KB sit in memory forever if you forgot to call flush. But it's nevertheless important when emulating Stream behavior to be able to say when the "reader" has done its work.

I'd like to emulate Stream properly regardless, but the only important scenario I can think of at the moment is: "When can I close the network stream?" (answer: when writing to it and flushing is done).
Given the finality of closing the stream, the OnReaderCompleted callback is probably sufficient. But as I said earlier, that's 16 lines of code. Why is their no ReaderCompletion property on the PipeWriter so I can easily just await it and realize any exceptions it threw like in so many other modern APIs from .NET?
In fact for my own convenience I wrote this helper methods that let me await on this (below). Seriously though, it shouldn't require so much code, IMO.

    private Task WaitForReaderCompletion(PipeWriter writer)
    {
        Requires.NotNull(writer, nameof(writer));

        var readerDone = new TaskCompletionSource<object>();
        writer.OnReaderCompleted(
            (ex, tcs) =>
            {
                if (ex != null)
                {
                    readerDone.SetException(ex);
                }
                else
                {
                    readerDone.SetResult(null);
                }
            },
            null);
        return readerDone.Task.WithCancellation(this.TimeoutToken);
    }

    private Task WaitForWriterCompletion(PipeReader reader)
    {
        Requires.NotNull(reader, nameof(reader));

        var writerDone = new TaskCompletionSource<object>();
        reader.OnWriterCompleted(
            (ex, tcs) =>
            {
                if (ex != null)
                {
                    writerDone.SetException(ex);
                }
                else
                {
                    writerDone.SetResult(null);
                }
            },
            null);
        return writerDone.Task.WithCancellation(this.TimeoutToken);
    }

[davidfowl]

@davidfowl The scenario is interop across APIs. In my Nerdbank.Streams package I'm including some classes that make a Stream look like a PipeReader/PipeWriter, or vice versa. That way I (and others) can use pipes everywhere they make sense, without fear that I'm cutting off my audience that may very well require a Stream to interact with the so many more APIs that accept Stream.

We have a work item to address the Stream -> PipeReader/PipeWriter adapter problem. In hindsight we should have probably shipped basic something in the box since everyone will end up writing this. Versions exist in both SignalR and Kestrel and @mgravell has his own in https://www.nuget.org/packages/Pipelines.Sockets.Unofficial/.

Stream.FlushAsync tells me when the writing is done (actually on the network, for example), but there is nothing like that in PipeWriter.

Unfortunately, that's by design when you're using the Pipe. The other way to expose the Stream as a PipeReader/PipeWriter includes implementing a PipeReader and PipeWriter over the Stream directly. That will give you fuller fidelity.

When can I close the network stream?" (answer: when writing to it and flushing is done).
Given the finality of closing the stream, the OnReaderCompleted callback is probably sufficient.

If you're talking about exposing the NetworkStream as a PipeReader/Writer then it should be represented as an IDuplexPipe and it should only close after both sides are done with it. Here's an example how Kestrel does it https://github.com/aspnet/KestrelHttpServer/blob/a44007adfb6f88c652348621bb1466d7278bc926/src/Kestrel.Transport.Sockets/Internal/SocketConnection.cs#L73.

But as I said earlier, that's 16 lines of code. Why is their no ReaderCompletion property on the PipeWriter so I can easily just await it and realize any exceptions it threw like in so many other modern APIs from .NET?

The main reason the APIs are callback based an not Task based is because Tasks's aren't resettable. you can reset the Pipe and reuse it for subsequent operations. In the past that would have required an allocation but we could potentially use a ValueTask here instead (now that we have it). I agree though, it's pretty annoying.

Can you make an official API proposal for exposing the WriterCompletion and ReaderCompletion as ValueTask (non generic)?

[AArnott]

What does Pipe.Reset() do, exactly? Clear all buffers, complete all readers/writers, and start fresh?
When is that useful?
Using ValueTask for a Completion property doesn't make sense in my opinion because ValueTask will still allocate if it's not completed and someone awaits it, I thought, which would be the majority of the time.

Can you make an official API proposal for exposing the WriterCompletion and ReaderCompletion as ValueTask (non generic)?

Sure.

[davidfowl]

What does Pipe.Reset() do, exactly? Clear all buffers, complete all readers/writers, and start fresh?

Yep.

When is that useful?

When you want to pool the Pipe implementation for reuse.

Using ValueTask for a Completion property doesn't make sense in my opinion because ValueTask will still allocate if it's not completed and someone awaits it, I thought, which would be the majority of the time.

ValueTask has new super powers and it works well for this scenario. It's not about async vs sync completion, it's about being able to reuse the backing awaitable object.

[davidfowl]

I’m closing this as it’s by design

[AArnott]

The other way to expose the Stream as a PipeReader/PipeWriter includes implementing a PipeReader and PipeWriter over the Stream directly. That will give you fuller fidelity.

@davidfowl: I'm looking into doing this now, for WebSocket and Stream. But this requires that I understand SequencePosition which appears to be undocumented. Any suggestions?

[davidfowl]

Websocket would be tougher but maybe you can make it work. Here’s an example of an incomplete implementation https://github.com/davidfowl/StringsAreEvil/blob/301d68ab85d17fbf2d3a4f0d01f33b04fac36ec5/StringsAreEvil/Program.cs#L351

[davidfowl]

A few notes:

• this doesn’t handle cancellation
• it doesn’t handle the case where more than 4K needs to be buffered
• it doesn’t correctly yield the read if examined is non zero

I’m sure there are more bugs

[AArnott]

Your sample doesn't show a sample derived type of ReadOnlySequenceSegment, which I'm keen to learn more about how to properly do. Your code slices a ReadOnlySequence<byte> but presumably the arrays backing the consumed bytes are simply discarded rather than returned to a pool for reuse. I'd like to be as efficient as possible.

[davidfowl]

The only real implementation so far is in pipelines. There may be others in corefxlab and in tests. You’re basically trying to reimplement pieces of the pipe class which is non trivial but the code is there.

[AArnott]

That's what I figured. I would simply use Pipe, but that requires that I build an async state machine that constantly reads from a stream/websocket and fills a buffer up to some threshold rather than only reading on demand and leveraging the network buffers already in place for web socket and other streams. I'm not sure whether that's a good or a bad thing, to be honest. But I thought a thinner pipe facade layer over web socket and stream would involve a custom PipeReader that only reads when asked to.

What do you think?

[AArnott]

And anyway, there seems an obvious hold where a Sequence<T> class should be. I'm writing one up -- unless it's on the .NET team's backlog and near the top, already.