F#事件在Async工作流中不起作用

Question

我想对代理人进行Post-Fire-Reply.基本上,代理触发事件然后回复调用者.但是,我要么继续收到超时错误,要么事件无法正确触发.我尝试过Post-Fire,它停止了超时错误,但事件没有触发.

let evt = new Event<int>()
let stream = evt.Publish

type Agent<'T> = MailboxProcessor<'T>
type Fire = Fire of int

let agent = Agent.Start(fun inbox -> 
    let rec loop() = async {
        let! msg = inbox.Receive()
        let (Fire i) = msg
        evt.Trigger i }
    loop())

let on i fn = 
    stream 
    |> Observable.filter (fun x -> x = i) 
    |> Observable.filter (fun x -> x <> 1)
    |> Observable.subscribe (fun x -> fn x) 

let rec collatz n =
    printfn "%d" n
    on n (fun i -> 
        if (i % 2 = 0) then collatz (i/2)
        else collatz (3*n + 1)) |> ignore

    agent.Post (Fire n) // this does not work
    // evt.Trigger n // this does works


collatz 13    

这是一个简单的实验,它重复创建一个函数来查找Collat​​z系列中的下一个数字,然后调用自身返回值,直到达到1.

似乎发生的事情是触发器只触发一次.我尝试了Async.RunSynchronously/Async.Start/StartChild/SynchronizationContext的每个组合,我能想到但没有进展.我发现了一个类似于我正在做的博客,但这对我没有帮助

编辑 感谢Fyodor Soikin指出我的疏忽.最初的问题仍然存在,我希望同时触发事件并回复结果,但是会超时.

let evt = new Event<int>()
let stream = evt.Publish

type Agent<'T> = MailboxProcessor<'T>
type Command = 
    | Fire of int
    | Get of int * AsyncReplyChannel<int>

let agent = Agent.Start(fun inbox -> 
    let rec loop() = async {
        let! msg = inbox.Receive()
        match msg with
        | Fire i -> evt.Trigger i 
        | Get (i,ch) -> 
            evt.Trigger i 
            ch.Reply(i)
        return! loop() }
    loop())

let on i fn = 
    stream 
    |> Observable.filter (fun x -> x = i) 
    |> Observable.filter (fun x -> x <> 1)
    |> Observable.subscribe (fun x -> fn x) 

let rec collatz n =
    printfn "%d" n
    on n (fun i -> 
        if (i % 2 = 0) then collatz (i/2)
        else collatz (3*n + 1)) |> ignore

    agent.PostAndReply (fun ch -> (Get (n, ch))) |> ignore // timeout
    agent.PostAndAsyncReply (fun ch -> (Get (n, ch))) |> Async.Ignore |> Async.Start // works but I need the result
    agent.PostAndAsyncReply (fun ch -> (Get (n, ch))) |> Async.RunSynchronously |> ignore // timeout

collatz 13

Answer 1

你的loop功能不循环.它接收第一条消息,触发事件,然后只是...退出.永远不要尝试接收第二条消息.

您需要使该功能连续工作:处理第一条消息,然后返回接收下一条消息,然后接收下一条消息,依此类推.像这样:

let agent = Agent.Start(fun inbox -> 
    let rec loop() = async {
        let! msg = inbox.Receive()
        let (Fire i) = msg
        evt.Trigger i
        return! loop() }
    loop())

编辑

由于您已达到问题的限制,我将在此处回答您的修改.

您在第二个代码段中获得超时的原因是您的代码中存在死锁.让我们跟踪执行情况以查看它.

1. 线程1:代理启动.
2. 线程2:第一个collatz电话.
3. 线程2:第一个collatz呼叫向代理发布消息.
4. 线程1:代理接收消息.
5. 线程1:代理触发事件.
6. 线程1:由于事件,第二次collatz呼叫发生.
7. 线程1:第二个collatz呼叫向座席发布消息.
8. 线程1:第二个collatz呼叫开始等待代理响应.

这就是执行结束的地方.代理程序此时无法响应(事实上,它甚至无法接收下一条消息!),因为它的指令指针仍在内部evt.Trigger.该evt.Trigger电话还没有恢复,所以loop功能还没有递归,所以该inbox.Receive功能还没有被调用,所以第二个消息在代理的队列仍在等待.

所以你给自己一个经典的死锁:collatz等待代理接收它的消息,但代理正在等待collatz完成处理事件.

对此最简单,最愚蠢的解决方案是异步触发事件:

    async { evt.Trigger i } |> Async.Start

这将确保事件处理程序不是"正确"执行,而是异步执行,可能在不同的线程上执行.这将允许代理在继续执行自己的执行循环之前不等待事件被处理.

通常,在处理多线程和异步时,不应该直接调用未知代码.代理不应该直接调用evt.Trigger,或者它不能控制的任何其他东西,因为代码可能正在等待代理本身(这就是你的情况),从而引入了死锁.