如何使用 Observable FromEventPattern 异步例程避免死锁？

Question

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

下面的代码会死锁。

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);

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

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

Answer 1

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

我创建了一个名为 WaitableEventHelper 的静态助手 class，它包括以下功能。

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 中的先前代码，任务只是任务类型的列表。

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 FromEventPattern 异步例程避免死锁？

How to avoid deadlock with Observable FromEventPattern Async routines?

c#

system.reactive

observable