Add IsFlush to Pipelines ReadResult

[benaadams]

public partial struct ReadResult
{
    public bool IsFlush => (_resultFlags & ResultFlags.Flush) != 0;
}

[Flags]
internal enum ResultFlags : byte
{
    None = 0x0,
    Canceled = 0x1,
    Completed = 0x2,
+   Flush = 0x4
}

And set for calls to FlushAsync()

This allows the reader; if it chooses, to differentiate between a WriteAsync() and a FlushAsync() (if the writer differentiates)

Use case

If WriteAsync and FlushAsync are differentiated; a TextWriter type on a Pipeline does not need to maintain arrays of internal buffered data to have good performance but only needs a char (or two?) to maintain the outstanding surrogate pairs for encoding; the rest can be directly written to the Pipe.

If they don't differentiate then it can't write directly to the Pipe as writing a very small chunk followed by a flush in a loop would have an unacceptable performance degradation; so its back to maintaining internal arrays and then writing in a large chunk 😢

/cc @davidfowl @pakrym @stephentoub

[davidfowl]

Some questions:

• What happens if the consumer ignores it?
• What happens if the consumer is relying on this semantic the producer never flushes is that a bug?
• This method that does the partial flush/write cannot have back pressure because if the reader is using it to buffer, you can end up dead locked if the reader doesn't consume enough to release the threshold

[davidfowl]

Also we're thinking https://github.com/dotnet/corefx/issues/27185 this plus cancellation be better to solve issues with smarter buffering.

/cc @pakrym @halter73

[benaadams]

What happens if the consumer ignores it?

Its a signal; it may be validly ignored as you may do currently with Flush; for example if you don't have enough data to validly do something with. e.g. input socket flushes due to no more data; however isn't currently enough data to create a http request.

What happens if the consumer is relying on this semantic the producer never flushes is that a bug?

CompleteWriter should apply IsFlush; if you complete a read but remain over the threshold of max buffered data, that should throw. Doesn't really change semantics?

This method that does the partial flush/write

WriteAsync should be the method. The virtual implementation should go via Flush as now; so by default it will work; and if you implement it you should be looking deeper.

public virtual PipeAwaiter<FlushResult> WriteAsync(ReadOnlyMemory<byte> source, CancellationToken cancellationToken = default)
{
    this.Write(source.Span);
    return FlushAsync(cancellationToken);
}

cannot have back pressure because if the reader is using it to buffer, you can end up dead locked if the reader doesn't consume enough to release the threshold

That should throw (and it should do currently anyway?)

Also we're thinking dotnet/corefx#27185 this be better to solve issues with smater buffering.

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

You need something to unlock the Read (with given reason) when the amount is smaller; like IsFlush or you may be stuck waiting for a long time.

[davidfowl]

You need something to unlock the Read (with given reason) when the amount is smaller; like IsFlush or you may be stuck waiting for a long time.

That's why you cancel on timer, to yield the read. You a timer anyways, you don't know if a full flush is coming.

[benaadams]

That's why you cancel on timer, to yield the read. You a timer anyways, you don't know if a full flush is coming.

You could use the same argument for TextWriter which generally WriteAsync to internal buffer; FlushAsync when the buffer is full, have an explicit FlushAsync available, and Flush on Dispose. People seem to use them happily without timers?

[davidfowl]

Not really, you can't tell today if a Stream is buffering or not, each stream behaves differently in that regard. You can't even tell if Streams require flushing or not, so the state today is that you always Flush or you write and nothing happens (or you know what the Stream implementation does).

[benaadams]

so the state today is that you always Flush or you write and nothing happens

I still don't completely understand. With Streams you don't always Flush; you do a bunch of Writes and then you Flush; you don't Flush on every Write.

Which would be the same pattern as I'm proposing; e.g. you Write until you have nothing current left to Write and then you Flush; so it wouldn't need timers?

[davidfowl]

Which would be the same pattern as I'm proposing; e.g. you Write until you have nothing current left to Write and then you Flush; so it wouldn't need timers?

I need to understand the scenario you're trying to implement, what it looks like today and what it would look like with the IsFlush flag. I'd also like to understand the decisions that consumers make based on IsFlush. For the nagaling scenario this is much less efficient as you potentially wake up the reader much too often. With dotnet/corefx#27185, the pipe implementation can just avoid the schedule if you're less than the threshold. Of course it's then up to you to either pass a cancellation token or something to implement a timeout if the size limit (MTU for example) hasn't been reached.

[benaadams]

For the nagaling scenario this is much less efficient as you potentially wake up the reader much too often.

With IsFlush the reader decides whether it wants to do anything based on the available size; with ReadAsync(xxxxx) the pipe decides whether to contact the reader based on the available size, they are mostly the same and also could be combined.

Lets iron out some questions in dotnet/corefx#27185 as it could achieve the same ends.

[muratg]

Had a chat with @davidfowl, we're not doing this at this time.

[dotnet-policy-service]

Due to lack of recent activity, this issue has been marked as a candidate for backlog cleanup. It will be closed if no further activity occurs within 14 more days. Any new comment (by anyone, not necessarily the author) will undo this process.

This process is part of our issue cleanup automation.