Rx结合CancellationToken使用偶发注册回调执行中断及同Token回调未触发问题

在编写结合CancellationToken的Rx查询逻辑时，观测到偶发的异常现象：

• 同一CancellationToken实例上共注册两个回调：一个是独立于查询逻辑、在查询流程启动前就直接注册的；另一个是嵌入在查询内部、作为终止信号传递链路的一部分存在。
• 设计上这个Token的核心用途是传递查询的终止信号，但实际运行时偶发出现：查询内部的回调执行到中途就被卡住，完全无法走完执行流程，进而导致同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");
    }
}

测试过程中观测到两个稳定的问题关联特征：

• 将代码中的Switch运算符替换为Merge后，问题完全消失
• 把Task.Delay(1000, cts.Token)抛出的取消异常手动捕获吞掉，问题也不会出现

运行输出对比 #

正常流程的预期输出为：

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

从缺失的日志可以判断，问题触发时observer.OnNext(Unit.Default)的调用始终没有执行完成，整个流程卡在了这一步。通常连续运行3-4次即可触发一次异常。

可复现的运行环境 #

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

辅助观测结论 #

如果修改Observable.Create的委托逻辑，不把cts.Token.Register返回的CancellationTokenRegistration直接作为可释放对象返回，而是手动完成注册后返回Disposable.Empty，问题就会消失，修改后的局部代码如下：

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

但这种写法直接丢弃了Token注册返回的句柄，并不属于合理修复——查询正常结束时无法主动注销内部注册的回调，长期运行会造成内存泄漏。 #

待解答问题 #

1. 该偶发卡死问题的根本触发原因是什么？
2. 应该如何调整查询中和CancellationToken相关的逻辑，既能完成传递查询终止信号的设计目标，又不会干扰同一CancellationToken下其他已注册回调的正常执行？

内容的提问来源于stack exchange，提问作者Theodor Zoulias