在多线程代码中使用F#事件

Question

我正在使用F#中的异步工作流和代理工作很多,而我对事件的深入了解我注意到Event <_>()类型不是线程安全的.在这里,我不是在谈论举办活动的常见问题.我实际上是在谈论订阅和删除/处理事件.出于测试目的,我写了这个简短的程序

let event = Event<int>()
let sub   = event.Publish

[<EntryPoint>]
let main argv =
    let subscribe sub x = async {
        let mutable disposables = []
        for i=0 to x do
            let dis = Observable.subscribe (fun x -> printf "%d" x) sub
            disposables <- dis :: disposables
        for dis in disposables do
            dis.Dispose()
    }

    Async.RunSynchronously(async{
        let! x = Async.StartChild (subscribe sub 1000)
        let! y = Async.StartChild (subscribe sub 1000)
        do! x
        do! y
        event.Trigger 1
        do! Async.Sleep 2000
    })
    0

程序很简单,我创建一个事件和一个函数,为它订阅特定数量的事件,然后处理每个处理程序.我使用另一个异步计算来使用Async.StartChild生成这些函数的两个实例.两个函数完成后,我触发事件以查看是否还有一些处理程序.

但是当event.Trigger(1)被调用时,结果是仍有一些处理程序被重新命名为事件.一些"1"将打印到控制台.这通常意味着订阅和/或处置不是线程安全的.

这就是我没想到的.如果订阅和Disposing不是线程安全的,那么一般情况下如何安全地使用事件？确实事件也可以在线程之外使用,并且触发器不会并行或在不同线程上生成任何函数.但是对我来说,事件在Async,基于代理的代码中使用或者通常与Threads一起使用在某种程度上是正常的.它们通常用作收集Backroundworker线程信息的通信.使用Async.AwaitEvent,可以订阅一个事件.如果订阅和Disposing不是线程安全的,那么在这样的环境中如何使用事件呢？Async.AwaitEvent的目的是什么？考虑到Async工作流做线程希望只使用Async.AwaitEvent基本上是"按设计破坏",如果默认情况下订阅/处置事件不是线程安全的.

我面临的一般问题是.订阅和处置不是线程安全的吗？从我的例子来看,它似乎看起来像,但可能我错过了一些重要的细节.我目前在我的设计中经常使用Event,我通常有MailboxProcessors并使用事件进行通知.所以问题是.如果事件不是线程安全的,那么我目前使用的整个设计根本不是线程安全的.那么这种情况有什么解决方法呢？创建一个全新的线程安全事件实现？他们是否已经存在一些面临这个问题的实现？或者是否有其他选项可以在高度线程化的环境中安全地使用Event？

Answer 1

仅供参考；的实现Event<int>可以在这里找到。

有趣的一点似乎是：

member e.AddHandler(d) =
  x.multicast <- (System.Delegate.Combine(x.multicast, d) :?> Handler<'T>)
member e.RemoveHandler(d) = 
  x.multicast <- (System.Delegate.Remove(x.multicast, d) :?> Handler<'T>)

订阅事件将当前事件处理程序与传递给订阅的事件处理程序结合在一起。这个组合的事件处理程序替换了当前的事件处理程序。

从并发性角度来看，问题在于这里存在一个竞争条件，即并发订阅者可能会使用当前事件处理程序与之结合，而“最后一个”会写回处理程序获胜（在并行性中，最后一个概念很难理解，天，但nvm）。

此处可以做的是引入一个CAS循环，Interlocked.CompareAndExchange但会增加性能开销，从而损害非并发用户。不过，可以通过PR来查看它是否受到F＃社区的好评。

关于您该怎么做的第二个问题，我只能说我会做。我会选择创建一个FSharpEvent支持受保护的订阅/取消订阅的版本。FSharpEvent如果您的公司的FOSS政策允许的话，也许可以以此为依据。如果成功，那么它可以构成将来对F＃核心库的PR。

我不知道您的要求，但也有可能，如果您需要的是协程（即Async）而不是线程，则可以重写程序以仅使用1个线程，因此您不会受到这种竞争状况的影响。