如何避免 Observable FromEventPattern Async 例程的死锁？答案

【问题标题】：How to avoid deadlock with Observable FromEventPattern Async routines?如何避免 Observable FromEventPattern Async 例程的死锁？
【发布时间】：2019-01-05 01:37:21
【问题描述】：

我正在使用可观察/反应式扩展来消除某些事件，例如按钮单击或在文本框中输入文本。但是，在关机或关闭的情况下，我需要等待任何未决事件，以便保存操作可以完成，等等。

下面的代码会死锁。

Button b1 = new Button();

var scheduler = new EventLoopScheduler(ts => new Thread(ts)
{
    IsBackground = false
});

var awaiter = Observable.FromEventPattern(h => b1.Click += h, h => b1.Click -= h, scheduler)                
     .Throttle(TimeSpan.FromMilliseconds(5000), scheduler)
     .FirstOrDefaultAsync();

someTaskList.add(awaiter.ToTask());

awaiter.Subscribe
(
    x =>
    {
        //do some work in response to click event
    }
);

//program continues...

然后，在应用程序的其他地方

private async Task CloseApplicationSafely()
{
    await AwaitPendingEvents();
}

private async Task AwaitPendingEvents()
{
    if(someTaskList.Count > 0)
    {
        await Task.WhenAll(someTaskList);
    }
}

然后程序将死锁，如果按钮点击从未发生，则永远等待。这是另一个示例，但带有一个文本框。

var completedTask = Observable.FromEventPattern(h => t1.TextChanged += h, h => t1.TextChanged -= h, scheduler)
    .Select(x => ((TextBox)x.Sender).Text)
    .DistinctUntilChanged()
    .Throttle(TimeSpan.FromMilliseconds(5000), scheduler)
    .ForEachAsync(txt =>
    {
        //do some work, save the text
    });

someTaskList.Add(completedTask);

在这种情况下，文本是否被更改并不重要。如果您等待，变量 completedTask 将永远死锁。 ForEachAsync() 返回一个似乎永远不会被激活的任务。

我做错了什么？希望我的预期功能很明确。我正在消除事件。但我需要等待任何正在处理的未决事件以确保它们完成。如果没有待处理的事件，则无需等待即可继续。谢谢。

【问题讨论】：

您能说明什么是待处理事件吗？待处理事件的来源在哪里？
我不确定我是否理解您的问题。见源。 Observable 绑定到像 Click 或 TextChanged 这样的控制事件。然后使用 Throttle() 消除这些事件。因此，您可以多次快速触发单击该按钮，但它会等到您最后一次单击后的 5 秒，然后再通过订阅触发单个事件。这就是我所说的待处理事件。有人点击了，但订阅尚未执行。
这不是死锁。死锁是一种循环依赖。 A 在等待 B，B 在等待 A，所以他们永远无法完成，因为他们都不会让对方完成。你只是在等待永远不会完成的事情，但永远不会完成的事情不依赖于其他任何东西，它只是永远不会完成的事情。这是一个重要的区别，因为它们的解决方式不同。对于死锁，您删除循环依赖，对于永远不会完成的任务，您要么需要强制它完成，要么在等待之前检查它是否会完成。
当你希望程序结束时，你需要有可观察的结束。尝试使用.Take 或.TakeUntil 来做。

标签： c# observable system.reactive

【解决方案1】：

@Servy 和@Enigmativity 的评论帮助我确定了这一点。对于那些感兴趣的人，这是我想出的解决方案。如果对我的方法有任何建议，请告诉我。

我创建了一个名为 WaitableEventHelper 的静态助手类，其中包含以下函数。

public static Task WaitableDebouncer(
    this Control c, 
    Action<EventHandler> addHandler, 
    Action<EventHandler> removeHandler, 
    IScheduler scheduler,
    CancellationToken cancelToken,
    TimeSpan limit,
    Func<Task> func)
{
    var mycts = new CancellationTokenSource();

    bool activated = false;
    bool active = false;

    Func<Task> pending = null;

    var awaiter = Observable.FromEventPattern(addHandler, removeHandler, scheduler)
        .TakeUntil(x => { return cancelToken.IsCancellationRequested; })
        .Do((x) => { activated = true; })
        .Do((x) =>
        {
            //sets pending task to last in sequence
            pending = func;
        })
        .Throttle(limit, scheduler)
        .Do((x) => { active = true; })    //done with throttle
        .ForEachAsync(async (x) =>
        {
            //get func
            var f = pending;

            //remove from list
            pending = null;

            //execute it
            await f();

            //have we been cancelled?
            if (cancelToken.IsCancellationRequested)
            {
                mycts.Cancel();
            }

            //not active
            active = false;

        }, mycts.Token);

    //if cancelled 
    cancelToken.Register(() => 
    {
        //never activated, force cancel
        if (!activated)
        {
            mycts.Cancel();
        }

        //activated in the past but not currently active
        if (activated && !active)
        {
            mycts.Cancel();
        }
    });

    //return new awaiter based on conditions
    return Task.Run(async () =>
    {
        try
        {
            //until awaiter finishes or is cancelled, this will block
            await awaiter;
        }
        catch (Exception)
        {
            //cancelled, don't care
        }

        //if pending isn't null, that means we terminated before ForEachAsync reached it
        //execute it
        if (pending != null)
        {
            await pending();
        }
    });
}

然后我像这样使用它。这是一个单击按钮的示例， b1 是 System.Windows.Forms.Button 对象。这可以是任何东西。对于我的测试应用程序，我正在更改主窗体上某些面板的颜色。根据 OP 中的先前代码，tasks 只是一个 Task 类型的 List。

var awaiter1 = b1.WaitableDebouncer(h => b1.Click += h, h => b1.Click -= h, 
    scheduler, 
    canceller.Token, 
    TimeSpan.FromMilliseconds(5000), 
    async () =>
    {
        Invoke(new Action(() =>
        {
            if (p1.BackColor == Color.Red)
            {
                p1.BackColor = Color.Orange;
            }
            else if (p1.BackColor == Color.Orange)
            {
                p1.BackColor = Color.Yellow;
            }
            else if (p1.BackColor == Color.Yellow)
            {
                p1.BackColor = Color.HotPink;
            }
            else
            {
                p1.BackColor = Color.Red;
            }
        }));
    });

tasks.Add(awaiter1);

另一个用于文本框上的 TextChanged。 t1 是 System.Windows.Forms.TextBox。同样，这可以是任何东西，我只是设置一个静态 someValue 字符串变量并更新 UI 上的标签。

var awaiter2 = t1.WaitableDebouncer(h => t1.TextChanged += h, h => t1.TextChanged -= h, 
    scheduler, 
    canceller.Token, 
    TimeSpan.FromMilliseconds(5000), 
    async () =>
    {
        savedValue = t1.Text;

        Invoke(new Action(() => l1.Text = savedValue));
    });

tasks.Add(awaiter2);

这就是终止或关闭的样子。这可能是应用程序关闭或文件关闭。只是一些我们需要取消绑定这些事件的事件，但在这样做之前保存用户已启动的任何待处理工作。想象一下用户在文本框中输入内容，然后快速点击 X 关闭应用程序。 5 秒还没用完。

private async Task AwaitPendingEvents()
{
    if (tasks.Count > 0)
    {
        await Task.WhenAll(tasks);
    }            
}

我们有一个应用范围内的等待程序。接近尾声。

//main cancel signal
canceller.Cancel();

await AwaitPendingEvents();

到目前为止，我的测试似乎有效。如果从未生成过任何事件，它将取消。如果已生成事件，我们将查看是否有任何未完成的工作尚未通过节流阀。如果是这样，我们取消 observable 并自己执行待处理的工作，这样我们就不必等待计时器。如果有待处理的工作并且我们已经通过了限制，那么我们只需等待并让可观察订阅完成执行它。如果已请求取消，则订阅会在执行后自行取消。

【讨论】：