奇怪的Rx+CancellationToken问题：有时注册的回调未完成

我观察到一个奇怪的现象，有时会在我编写的 Rx 查询中发生，其中涉及CancellationToken。两个回调注册到同一个CancellationToken，一个在查询之外，一个是查询的一部分。的意图CancellationToken是发出查询终止的信号。发生的情况是，有时第二个回调卡在执行中间，永远不会完成，从而阻止调用第一个回调。

下面是重现该问题的最小示例。虽然不是很小，但我不能再减少了。例如更换Switch运算符与Merge使问题消失。如果抛出异常，也会发生同样的情况Task.Delay(1000, cts.Token)被吞掉了。

public class Program
{
    public static void Main()
    {
        var cts = new CancellationTokenSource(500);
        cts.Token.Register(() => Console.WriteLine("### Token Canceled! ###"));
        try
        {
            Observable
                .Timer(TimeSpan.Zero, TimeSpan.FromMilliseconds(1000))
                .TakeUntil(Observable.Create<Unit>(observer =>
                    cts.Token.Register(() =>
                    {
                        Console.WriteLine("Before observer.OnNext");
                        observer.OnNext(Unit.Default);
                        Console.WriteLine("After observer.OnNext");
                    })))
                .Select(_ =>
                {
                    return Observable.StartAsync(async () =>
                    {
                        Console.WriteLine("Action starting");
                        await Task.Delay(1000, cts.Token);
                        return 1;
                    });
                })
                .Switch()
                .Wait();
        }
        catch (Exception ex) { Console.WriteLine("Failed: {0}", ex.Message); }
        Thread.Sleep(500);
        Console.WriteLine("Finished");
    }
}

预期输出：

Action starting
Before observer.OnNext
After observer.OnNext
### Token Canceled! ###
Failed: A task was canceled.
Finished

实际输出（有时）：

Action starting
Before observer.OnNext
Failed: A task was canceled.
Finished

在小提琴上尝试一下。您可能需要运行该程序 3-4 次才能出现问题。请注意两个缺失的日志条目。看来是调用observer.OnNext(Unit.Default);永远不会完成。

我的问题是：有谁知道导致这个问题的原因是什么？另外，我该如何修改CancellationToken- 查询的相关部分，以便它执行其预期目的（终止查询），而不会干扰相同的其他注册回调CancellationToken?

.NET 5.0.1 和 .NET Framework 4.8、System.Reactive 5.0.0、C# 9

Update:还有 .NET 6.0 和 System.Reactive 5.0.0 (截屏拍摄于 2022 年 6 月 4 日）

还有一项观察：如果我修改以下内容，问题就会停止出现Observable.Create委托，以便它返回一个Disposable.Empty代替CancellationTokenRegistration， 像这样：

.TakeUntil(Observable.Create<Unit>(observer =>
{
    cts.Token.Register(() =>
    {
        Console.WriteLine("Before observer.OnNext");
        observer.OnNext(default);
        Console.WriteLine("After observer.OnNext");
    });
    return Disposable.Empty;
}))

但我不认为忽略由返回的注册cts.Token.Register是一个修复。

一年后，我无法重现它。你还可以吗？我已经在 .NET Framework 4.8 和 .NET 6 上尝试过。

然而，在我的相关问题，我刚刚处理了导致死锁的竞争条件，这might与您所描述的问题相关。

CancellationTokenRegistration.Dispose本质上是一个阻塞调用，它等待所有回调完成，并且这些回调在调用的线程上执行CancellationTokenSource.Cancel（例如，随机计时器回调线程），这可能与调用的线程不同DisposeRx 订阅的数量。

我还没有深入挖掘，但我的案例的解决方法是不调用令牌注册Dispose如果令牌已经被取消。这应该没问题，因为在发出取消信号后令牌注册将被丢弃，并且当 Rx 调用时我们不再对取消回调感兴趣Dispose:

    .TakeUntil(Observable.Create<Unit>(observer => 
    {
        var token = cts.Token;
        var rego = token.Register(() =>
        {
            Console.WriteLine("Before observer.OnNext");
            observer.OnNext(Unit.Default);
            Console.WriteLine("After observer.OnNext");
        });
        return Disposable.Create(() => 
        {
            if (!token.IsCancellationRequested)
            {
                // this 
                rego.Dispose();
            }
        });
    }))       

在 .NET 6 中，另一possible解决方法可能是使用CancellationTokenRegistration.DisposeAsync（尽管我还没有验证它是否可以解决我的情况下的僵局）：

        return Disposable.Create(() => 
        {
            DisposeAsync();
            async void DisposeAsync() => 
              await rego.DisposeAsync().ConfigureAwait(false);
        });

Updated，正如 Theodor 在评论中提到的，使用CancellationTokenRegistration.Unregister代替Dispose是另一种解决方法（非阻塞），这对我来说感觉更干净。