API PipeWriter.FlushAsync(bool isMoreData, ...

[benaadams]

Background and Motivation

Currently after calling .FlushAsync() on aPipeWriter provided by Pipe marks writing as operations complete; which makes the writing block eligible for collection when the PipeReader advances to the position of the writer (releasing the memory for an idle pipe).

However, the writer may have more to write and is only flushing so too much backlog and latency doesn't build up.

As the write flag is released; if the writer immediately wants to write more, the reader and writer move into a data race and lock contention. So the writer getting more memory with .GetMemory() needs to move through a slower lock protected path.

This additional option would tell the .FlushAsync not to release the write flag; and then calling GetMemory() after the flush would not need to go via contended locks and potentially re-rent memory from the MemoryPool that's just been released to the pool because the PipeReader won the data race.

In fact the PipeWriter would not need to call GetMemory() at all if it still have enough memory in its current block to write to as it would not have been invalidated by the .FlushAsync (by releasing the write flag it also invalidates the current memory so the full process needs to be gone through to require some memory to write to)

Additionally for non-Pipe backed PipeWriters; such as a socket-base one this flag could flow through to the socket Send passing MSG_MORE on linux send(2) flag; which acts as TCP_CORK for that send only pushing out full packets (but without the extra syscall that TCP_CORK needs)

MSG_MORE (Since Linux 2.4.4)
The caller has more data to send. This flag is used with TCP sockets to obtain the same effect as the TCP_CORK socket option (see tcp(7)), with the difference that this flag can be set on a per-call basis.

Or RIO_MSG_DEFER flag on Windows for RioSend

RIO_MSG_DEFER
The request does not need to be executed immediately. This will insert the request into the request queue, but it may or may not trigger the execution of the request.

Proposed API

public abstract partial class PipeWriter
{
    /// <summary>
    /// Makes bytes written available to <see cref="PipeReader"/> and runs <see cref="PipeReader.ReadAsync"/> continuation.
    /// </summary>
    /// <param name="isMoreData"><see cref="bool"/> indicating that more data is to be written imminently.</param>
    /// <param name="cancellationToken">The token to monitor for cancellation requests. The default value is <see cref="CancellationToken.None"/>.</param>
    public
    virtual ValueTask<FlushResult> FlushAsync(bool isMoreData,
                                              CancellationToken cancellationToken = default)
        => FlushAsync(cancellationToken);
}

Usage Examples

Current

input.Advance(bytesReceived);

var flushTask = input.FlushAsync();
if (!_waitForData)
{
    // Previous buffer has been invalidated by the FlushAsync
    // we need to ask for completely new one which may contend
    // with the reawoken Reader from the FlushAsync
    buffer = input.GetMemory(MinAllocBufferSize);
}

var result = await flushTask;

After

input.Advance(bytesReceived);

var flushTask = input.FlushAsync(isMoreData: !_waitForData);
if (!_waitForData)
{
    if (buffer.Length - bytesReceived >= MinAllocBufferSize)
    {
        // We have enough memory, only need to slice the existing one
        // by the same amount as the input.Advance
        buffer = buffer.Slice(bytesReceived);
    }
    else
    {
        buffer = input.GetMemory(MinAllocBufferSize);
    }
}

var result = await flushTask;

Alternative Designs

A PipeOptions level switch to say never release the block even if writing is off; this has the disadvantage of not being able to release the block when the pipe is truly reached an idle point.

Also it doesn't enable other PipeWriters such as a socket based one to use MSG_MORE; RIO_MSG_DEFER.

/cc @davidfowl @halter73

FYI @stephentoub @adamsitnik as discussed at meeting.

[davidfowl]

This leaks the implementation of Pipe into the PipeWriter abstraction. This doesn’t make sense on the other implementations (glad we have more than one now) and it’s obscure enough that it makes me unhappy. I’d much rather prefer to waste a block if memory than make the caller decide how to set the boolean per call

[benaadams]

I think its a good ability for the other PipeWriter implementations?

For a socket implementation of the PipeWriter ; it can be used to pass MSG_MORE as stated above.

For a filesystem implementation of the PipeWriter; it can be used as a hint to wait for more if there isn't enough data and then perform a scatter write (or larger chunk).

For a TLS based pipe it can use it to close of any 16Kb frames (or whatever the threshold is) and send those; while keeping its current incomplete frame open.

Its allowing the implementation of PipeWriter hints that it can use to buffer based on its internal implementation and what the best choices are; which are unlikely to be known outside the implementation (so you couldn't just add a buffering Pipe over the top, as how will that know the packing of TCP packets as MSG_MORE does?)

[benaadams]

For example for non-platform ASP.NET Core it could have all FlushAsync go to FlushAsync(isMoreData: true) instead, which could use the MSG_MORE flag to send any full packets and hold back an incomplete packet; when the it returns to the request loop sees there is no further input data, it could then call FlushAsync(isMoreData: false) to send the final incomplete packet before waiting, which would also allow that block to be released as the pipeline would have become idle.

[davidfowl]

I think the way the issue is proposed doesn’t help much. The proposal should be an overload to FlushAsync that describes some abstract behavior and then go deeper into describing implementation specifics. As it stands it’s an obscure implementation detail to remove locks in a very obscure case when benchmarks have a flag turned on in kestrel.

I prefer the other approach where specific implementations keep the block of memory around because that’s the trade off being made here (which matches what most socket based programs do today anyways).

[omariom]

Alternative approach would be adding mayFlush flag to Advance.
Then the users of the writer would not need to think how often to Flush to not buildup underlying buffers. The implementations could do whatever is better for them. For example, if the buffers are full, it would flush. Or it could RIO_MSG_DEFER knowing that there will be Flush soon on which it will RIO_MSG_COMMIT_ONLY.

[benaadams]

Alternative approach would be adding mayFlush flag to Advance.

Advance isn't async so it can't provide backpressure that isn't blocking; don't want to discourage people from calling FlushAsync

[davidfowl]

This feels very much like this API https://docs.microsoft.com/en-us/dotnet/api/system.net.sockets.socketflags?view=netcore-3.1

[benaadams]

Yes so the SocketFlags that get passed per Send call as per send(2) flag

MSG_MORE (Since Linux 2.4.4)

The caller has more data to send. This flag is used with TCP sockets to obtain the same effect as the TCP_CORK socket option (see tcp(7)), with the difference that this flag can be set on a per-call basis.

Where TCP_CORK in tcp(7) is

TCP_CORK (since Linux 2.2)

If set, don't send out partial frames. All queued partial frames are sent when the option is cleared again. This is useful for prepending headers before calling sendfile(2), or for throughput optimization. As currently implemented, there is a 200 millisecond ceiling on the time for which output is corked by TCP_CORK. If this ceiling is reached, then queued data is automatically transmitted. This option can be combined with TCP_NODELAY only since Linux 2.5.71. This option should not be used in code intended to be portable.

[benaadams]

Worked out some redesigns with @davidfowl; will open new api over weekend