F# Akka.NET 代理同步时的性能优化答案

【问题标题】：F# Akka.NET agents performance optimization when synchronizingF# Akka.NET 代理同步时的性能优化
【发布时间】：2015-04-14 23:17:11
【问题描述】：

我正在尝试解决以下问题。我有一些实时运行的代理，具有几毫秒的大心跳，因此它们处理的操作顺序大多是确定性的（因为消息处理不是瓶颈）。

现在，我正在对不再有心跳的系统进行大量模拟（否则将需要几个世纪） - 但我需要确保保留操作顺序。为此，我采用了以下解决方案：模拟器确保每个代理都处理了他的消息队列，方法是发布一个虚拟同步消息并在等待答案时阻塞。这确实适用于我的应用程序，但它所花费的时间并不直观 - 因为单线程实现会快一个数量级（我猜 - x 100 ish - 虽然我没有测试过）。

我已经隔离了一个显示问题的小测试，甚至尝试使用另一个库 akka.net

type Greet = 
| Greet of string
| Hello of AsyncReplyChannel<bool>
| Hello2  

[<EntryPoint>]
let main argv =
    let system = System.create "MySystem" <| Configuration.load()    
    let greeter = spawn system "greeter" <| fun mailbox ->
        let rec loop() = actor {
            let! msg = mailbox.Receive()
            let sender = mailbox.Sender()
            match msg with
                | Greet who -> () // printf "Hello, %s!\n" who
                | Hello2 -> sender.Tell(true)
                | _ -> ()
            return! loop()
            }
        loop()

    let greeterF =
        MailboxProcessor.Start
            (fun inbox ->                
                async {
                    while true do
                        let! msg = inbox.Receive()
                        match msg with
                        | Greet who -> () // printf "Hello, %s!\n" who
                        | Hello reply -> reply.Reply true
                        | _ -> ()
                    }
            )

    let n = 1000000

    let t1 = System.Diagnostics.Stopwatch()
    t1.Start()
    for i = 1 to n do
        let rep = greeterF.PostAndReply(fun reply -> (Hello reply)) |> ignore
        ()

    printfn "elapsed Mailbox:%A" t1.ElapsedMilliseconds

    t1.Restart()

    for i = 1 to n do        
        let res = greeter.Ask (Hello2)
        let rr = res.Result
        ()

    printfn "elapsed Akka:%A" t1.ElapsedMilliseconds
    System.Console.ReadLine () |> ignore

    0

基本上，仅仅 100 万次同步都需要大约 10 秒 - 而不是计算所涉及的任何内容，这……很不幸。

我想知道是否有人遇到过同样的问题，是否有任何方法可以关闭开销，强制所有内容以单线程模式运行……这比停用所有 cpu 更好，但在bios - 或者在没有代理的情况下编写整个系统的克隆。

非常感谢任何帮助。

【问题讨论】：

您确定这种开销在实际计算到位后会很明显吗？话虽如此，我对Hopac 的体验非常好，据称可以节省大量开销。
每个背后的实际计算大多是数字化的，归结为一堆基本操作——最多是这里和那里的几个内存分配。我认为没有太多的方法-锁定/线程切换等..慢了一个数量级。我想我必须投入一些硬件。听起来很傻，但很有效。

标签： performance optimization f# mailboxprocessor akka.net

【解决方案1】：

Akka.NET 版本在这里变慢的原因是你与演员的沟通方式：

main process    Task     FutureActorRef  !!ThreadPool!!   greeter
    Ask ---------------------->
                              Tell-----------> 
                                             MailboxRun ----->
                                 (greeter mailbox is empty)  |                 
                               <--------------------------Tell 
                  <--Complete task
    <----------.Result

对于每次迭代，都会创建一个 TPL 任务
然后，单条消息会发送给迎宾员。
然后主进程在等待响应返回时阻塞。
greeter 回复，从而完成FutureActorRef 中的任务

冲洗并重复.. 这种设计将导致 Akka.NET 为每条消息启动和停止欢迎程序“邮箱运行”，因为邮箱队列在每次迭代中都变空了。这会导致每个传递的单个消息的线程池调度。

这有点像进入您的汽车，将踏板踩到金属上，然后突然停下并下车，然后再次重复该过程。这不是快速旅行的一种非常有效的方式。

@Aaronontheweb 的建议只有在您解决了代码中的上述问题后才会生效。邮箱需要能够不断地从内部队列中挑选消息来批量处理消息以实现全吞吐量。

相反，将生产者与消费者分开。创建一个 Actor 来监听来自欢迎者的响应。一旦该参与者处理了您的 1000000 条消息，让该参与者将 WorkCompleted 消息发送回消费者。

[编辑] 我自己试了一下，我不懂 F#，所以它可能不完全地道:)

open Akka
open Akka.Actor
open Akka.FSharp

type Greet = 
| Greet of string
| Hello of AsyncReplyChannel<bool>
| Hello2 

type Consume =
| Response
| SetSender

[<EntryPoint>]
let main argv =

    let system = System.create "MySystem" <| Configuration.load()    
    let greeter = spawn system "greeter" <| fun mailbox ->
        let rec loop() = actor {
            let! msg = mailbox.Receive()
            let sender = mailbox.Sender()
            match msg with
                | Greet who -> () // printf "Hello, %s!\n" who
                | Hello2 -> sender.Tell(Response)
                | _ -> ()
            return! loop()
            }
        loop()

    let consumer = spawn system "consumer" <| fun mailbox ->
        let rec loop(count,sender : IActorRef) = actor {
            if count = 1000000 then sender.Tell(true)
            let! msg = mailbox.Receive()
            match msg with
            | Response -> return! loop(count+1,sender)
            | SetSender -> return! loop(count,mailbox.Sender())

        }  
        loop(0,null)      

    let n = 1000000

    let t1 = System.Diagnostics.Stopwatch()
    t1.Start()   
    for i = 1 to n do        
        greeter.Tell(Hello2,consumer)

    let workdone = consumer.Ask SetSender
    workdone.Wait()

    printfn "elapsed Akka:%A" t1.ElapsedMilliseconds
    System.Console.ReadLine () |> ignore

    0

我更新了您的代码，为参与者响应使用单独的消费者，然后在处理完所有回复后回复。

通过这样做，您在我的机器上的处理时间现在降至 650 毫秒。

如果您想要更好的吞吐量，您需要让更多的参与者参与进来以实现更多的并行化。

我不确定这对您的特定情况是否有帮助

【讨论】：

对于我的特定应用程序，我认为这是一个可以将同步频率降低到可接受水平的低级成果。我仍然必须同步，因为不同的消息会以某种隐式的时间一致性进行处理。其他建议目前似乎更涉及 - 我想我会在有机会时重新考虑这个问题。我还观察到一些同步“循环”，即代理 1 与 2 与 3 同步等同步。比从主线程进行同步更快

【解决方案2】：

这里是稍加修改的MailboxProcessor版本：

module MBPAsync =
  type Greet = 
   | Greet of string
   | Hello of AsyncReplyChannel<bool>

  let run n =
    let timer = Stopwatch.StartNew ()

    use greeter =
      MailboxProcessor.Start <| fun inbox -> async {
        while true do
          let! msg = inbox.Receive()
          match msg with
           | Greet who -> () // printf "Hello, %s!\n" who
           | Hello reply -> reply.Reply true
      }

    Async.RunSynchronously <| async {
      for i = 1 to n do
        do! Async.Ignore (greeter.PostAndAsyncReply Hello)
    }

    let elapsed = timer.Elapsed
    printfn "%A" elapsed

这里的区别在于这个版本使用PostAndAsyncReply 并将计算保持在异步工作流中。在我的快速测试中，这似乎比使用 PostAndReply 快得多，但是 YMMV。

我从上面的 MBP 版本中得到的时间大致是这样的：

> MBPAsync.run 1000000 ;;
00:00:02.6883486
val it : unit = ()

之前的评论提到了我的 Hopac 库。这是使用 Hopac 的优化版本：

module Hop =
  type Greet = 
   | Greet of string
   | Hello of IVar<bool>

  let run n =
    let timer = Stopwatch.StartNew ()

    let greeterCh = ch ()
    do greeterCh >>= function
          | Greet who -> Job.unit ()
          | Hello reply -> reply <-= true
       |> Job.forever
       |> server

    Job.forUpToIgnore 1 n <| fun _ ->
        let reply = ivar ()
        greeterCh <-- Hello reply >>.
        reply
    |> run

    let elapsed = timer.Elapsed
    printfn "%A" elapsed

我从上面的 Hopac 版本中得到的时间大致是这样的：

> Hop.run 1000000 ;;
00:00:00.1088768
val it : unit = ()

【讨论】：

嗨@user3841986，是否有任何实际的 hopac 版本变体？似乎一些运算符不再存在于代码库中

【解决方案3】：

我不是 F# 开发人员，但我是 Akka.NET 的核心开发人员。针对您的场景的一些想法：

如果您只使用一个演员来完成这项工作，您可以尝试使用PinnedDispatcher - 这样演员一直在自己的专用线程上运行。这将为您节省不必要的上下文切换开销。
您还可以将此PinnedDispatcher 的邮箱吞吐量设置为比正常设置高得多。即，将吞吐量值设置为 10000（或其他值）而不是正常的 25。假设您的邮箱内容大量增长，这应该可以节省邮箱同步开销。

您的调度程序配置可能如下所示：

 my-pinned-dispatcher {
      type = PinnedDispatcher
      throughput = 1000 #your mileage may vary
 }

然后配置一个actor来使用它

C# 流畅界面

var myActor = myActorSystem.ActorOf(Props.Create<FooActor>()
.WithDispatcher("my-pinned-dispatcher");

配置

akka.actor.deployment{
   /greeter{
     dispatcher = my-pinned-dispatcher
   }
}

这两个选项都可以通过 HOCON 在 App.config 或 Web.config 中进行配置，或者您可以使用 Props 类上的流畅界面来执行此操作。另外值得注意的是：there's a bug with pinned dispatchers at the moment，但这应该会在我们下周发布的下一个维护版本 (v1.0.1) 中修复。

您的里程可能会有所不同，但这是我会尝试的 - 基本上它只是旨在帮助减少围绕单个参与者的争用和开销。

【讨论】：