Enable cancellation for anonymous pipes and non-async named pipes on Windows

[stephentoub]

Although ReadAsync, WriteAsync, and WaitForConnectionAsync on pipes all accept a CancellationToken, that token is only usable on Windows for canceling an in-flight operation when the pipe is using overlapped I/O. If the pipe was created for non-overlapped I/O, as is the case for anonymous pipes and can be the case for named pipes, the token stops being useful for anything other than an up-front cancellation check.

This change fixes that by using CancelSynchronousIo to cancel the synchronous I/O performed as part of these async operations, which are implemented as async-over-sync (queueing to the thread pool a work item that performs the synchronous I/O).

(The Unix implementation already supports cancellation in these situations.)

Fixes #31390
Closes #63536

While this is implemented differently, I drew inspiration from poizan42's experiments in #31390 (comment). Thanks!

And even though we're now actually registering for cancellation, performance is actually better even when cancellation is never requested:

Method	Toolchain	Cancelable	Named	Mean	Ratio	Allocated	Alloc Ratio
ReadWriteAsync	\main\corerun.exe	False	False	12.272 us	1.00	400 B	1.00
ReadWriteAsync	\pr\corerun.exe	False	False	6.597 us	0.54	84 B	0.21
ReadWriteAsync	\main\corerun.exe	False	True	32.442 us	1.00	400 B	1.00
ReadWriteAsync	\pr\corerun.exe	False	True	29.737 us	0.93	109 B	0.27
ReadWriteAsync	\main\corerun.exe	True	False	12.581 us	1.00	400 B	1.00
ReadWriteAsync	\pr\corerun.exe	True	False	8.036 us	0.62	85 B	0.21
ReadWriteAsync	\main\corerun.exe	True	True	31.889 us	1.00	400 B	1.00
ReadWriteAsync	\pr\corerun.exe	True	True	29.458 us	0.92	115 B	0.29

using BenchmarkDotNet.Attributes;
using BenchmarkDotNet.Running;
using System.IO.Pipes;

[MemoryDiagnoser]
public class Program
{
    static void Main(string[] args) => BenchmarkSwitcher.FromAssembly(typeof(Program).Assembly).Run(args);

    private Stream _server;
    private Stream _client;
    private byte[] _buffer = new byte[1];
    private CancellationTokenSource _cts = new CancellationTokenSource();

    [Params(false, true)]
    public bool Cancelable { get; set; }

    [Params(false, true)]
    public bool Named { get; set; }

    [GlobalSetup]
    public void Setup()
    {
        if (Named)
        {
            string name = Guid.NewGuid().ToString("N");
            var server = new NamedPipeServerStream(name, PipeDirection.Out);
            var client = new NamedPipeClientStream(".", name, PipeDirection.In);
            Task.WaitAll(server.WaitForConnectionAsync(), client.ConnectAsync());
            _server = server;
            _client = client;
        }
        else
        {
            var server = new AnonymousPipeServerStream(PipeDirection.Out);
            var client = new AnonymousPipeClientStream(PipeDirection.In, server.ClientSafePipeHandle);
            _server = server;
            _client = client;
        }
    }

    [GlobalCleanup]
    public void Cleanup()
    {
        _server.Dispose();
        _client.Dispose();
    }

    [Benchmark(OperationsPerInvoke = 1000)]
    public async Task ReadWriteAsync()
    {
        CancellationToken ct = Cancelable ? _cts.Token : default;
        for (int i = 0; i < 1000; i++)
        {
            ValueTask<int> read = _client.ReadAsync(_buffer, ct);
            await _server.WriteAsync(_buffer, ct);
            await read;
        }
    }
}

[ghost]

Tagging subscribers to this area: @dotnet/area-system-io
See info in area-owners.md if you want to be subscribed.

Issue Details

---

Although ReadAsync, WriteAsync, and WaitForConnectionAsync on pipes all accept a CancellationToken, that token is only usable on Windows for canceling an in-flight operation when the pipe is using overlapped I/O. If the pipe was created for non-overlapped I/O, as is the case for anonymous pipes and can be the case for named pipes, the token stops being useful for anything other than an up-front cancellation check.

This change fixes that by using CancelSynchronousIo to cancel the synchronous I/O performed as part of these async operations, which are implemented as async-over-sync (queueing to the thread pool a work item that performs the synchronous I/O).

(The Unix implementation already supports cancellation in these situations.)

Fixes #31390
Closes #63536

While this is implemented differently, I drew inspiration from poizan42's experiments in #31390 (comment). Thanks!

And even though we're now actually registering for cancellation, performance is actually better even when cancellation is never requested:

Method	Toolchain	Cancelable	Named	Mean	Ratio	Allocated	Alloc Ratio
ReadWriteAsync	\main\corerun.exe	False	False	12.272 us	1.00	400 B	1.00
ReadWriteAsync	\pr\corerun.exe	False	False	6.597 us	0.54	84 B	0.21
ReadWriteAsync	\main\corerun.exe	False	True	32.442 us	1.00	400 B	1.00
ReadWriteAsync	\pr\corerun.exe	False	True	29.737 us	0.93	109 B	0.27
ReadWriteAsync	\main\corerun.exe	True	False	12.581 us	1.00	400 B	1.00
ReadWriteAsync	\pr\corerun.exe	True	False	8.036 us	0.62	85 B	0.21
ReadWriteAsync	\main\corerun.exe	True	True	31.889 us	1.00	400 B	1.00
ReadWriteAsync	\pr\corerun.exe	True	True	29.458 us	0.92	115 B	0.29

using BenchmarkDotNet.Attributes;
using BenchmarkDotNet.Running;
using System.IO.Pipes;

[MemoryDiagnoser]
public class Program
{
    static void Main(string[] args) => BenchmarkSwitcher.FromAssembly(typeof(Program).Assembly).Run(args);

    private Stream _server;
    private Stream _client;
    private byte[] _buffer = new byte[1];
    private CancellationTokenSource _cts = new CancellationTokenSource();

    [Params(false, true)]
    public bool Cancelable { get; set; }

    [Params(false, true)]
    public bool Named { get; set; }

    [GlobalSetup]
    public void Setup()
    {
        if (Named)
        {
            string name = Guid.NewGuid().ToString("N");
            var server = new NamedPipeServerStream(name, PipeDirection.Out);
            var client = new NamedPipeClientStream(".", name, PipeDirection.In);
            Task.WaitAll(server.WaitForConnectionAsync(), client.ConnectAsync());
            _server = server;
            _client = client;
        }
        else
        {
            var server = new AnonymousPipeServerStream(PipeDirection.Out);
            var client = new AnonymousPipeClientStream(PipeDirection.In, server.ClientSafePipeHandle);
            _server = server;
            _client = client;
        }
    }

    [GlobalCleanup]
    public void Cleanup()
    {
        _server.Dispose();
        _client.Dispose();
    }

    [Benchmark(OperationsPerInvoke = 1000)]
    public async Task ReadWriteAsync()
    {
        CancellationToken ct = Cancelable ? _cts.Token : default;
        for (int i = 0; i < 1000; i++)
        {
            ValueTask<int> read = _client.ReadAsync(_buffer, ct);
            await _server.WriteAsync(_buffer, ct);
            await read;
        }
    }
}

Author:	stephentoub
Assignees:	-
Labels:	area-System.IO, tenet-performance
Milestone:	7.0.0

[danmoseley]

Do we have this covered in the perf benchmarks? If not maybe we could reuse what you have there.

Edit: never mind, it's not a mainline scenario.

[adamsitnik]

It's a really smart way of solving this problem! Thank you for adding all the comments @stephentoub !

We need to figure out a way to reuse the logic for FileStream, RandomAccess and Pipes. And ideally solve #28585 while we are there ;)

[stephentoub]

Do we have this covered in the perf benchmarks

Mostly. The only thing it lacks is passing in a cancellation token.
https://github.com/dotnet/performance/tree/f7ec27a4da2d61b97184f07989e5d9c6e0c21988/src/benchmarks/micro/libraries/System.IO.Pipes

[GSPP]

@stephentoub I've got a question about how this works. What happens in this sequence of events:

• We get directly to before the CancelSynchronousIo call
• The IO starts and completes
• The main thread moves on and starts another IO at some other program location
• Bug: That other IO now gets canceled

I have fought with this kind of IO cancellation before and was then not able to make it work reliably. Curious to learn how you did that.

[poizan42]

@GSPP

1. CancelSynchronousIo is called from the callback for the cancellation token registration: cdc5fa4#diff-f58ee0c0eea63450b3dfbc3640a760f0bfd17cd2747b9ba9b48dc0f88cdf8b6cR357
2. The CancellationTokenRegistration is disposed once the IO has completed:
   • cdc5fa4#diff-582554297ac13e502a90bf95dbe6c446d351cdbcb24ba7905c12427945272cddR217
   • cdc5fa4#diff-f58ee0c0eea63450b3dfbc3640a760f0bfd17cd2747b9ba9b48dc0f88cdf8b6cR227
   • cdc5fa4#diff-f58ee0c0eea63450b3dfbc3640a760f0bfd17cd2747b9ba9b48dc0f88cdf8b6cR262
3. CancellationTokenRegistration.Dispose() blocks until the callback in (1) has completed, hence the main thread cannot move on. (This is unfortunately poorly documented, but see discussion at Allow fire and forget CancellationTokenRegisteration.Dispose #19827)