F# 异步堆栈溢出

Question

我对基于异步的程序中的堆栈溢出感到惊讶。我怀疑主要问题在于以下函数，该函数应该组成两个异步计算以并行执行并等待两者完成：

let ( <|> ) (a: Async<unit>) (b: Async<unit>) =
    async {
        let! x = Async.StartChild a
        let! y = Async.StartChild b
        do! x
        do! y
    }

定义了此函数后，我有以下 mapReduce 程序，该程序尝试利用 map 和 reduce 部分中的并行性。通俗地说，其想法是使用共享通道触发 N 个映射器和 N-1 化简器，等待它们完成，然后从通道读取结果。我有自己的 Channel 实现，这里用 ConcurrentBag 替换以缩短代码（该问题影响两者）：

let mapReduce (map    : 'T1 -> Async<'T2>)
              (reduce : 'T2 -> 'T2 -> Async<'T2>)
              (input  : seq<'T1>) : Async<'T2> =
    let bag = System.Collections.Concurrent.ConcurrentBag()

    let rec read () =
        async {
            match bag.TryTake() with
            | true, value -> return value
            | _           -> do! Async.Sleep 100
                             return! read ()
        }

    let write x =
        bag.Add x
        async.Return ()

    let reducer =
        async {
            let! x = read ()
            let! y = read ()
            let! r = reduce x y
            return bag.Add r
        }

    let work =
        input
        |> Seq.map (fun x -> async.Bind(map x, write))
        |> Seq.reduce (fun m1 m2 -> m1 <|> m2 <|> reducer)

    async {
        do! work
        return! read ()
    }

现在，以下基本测试开始在 n=10000 上抛出 StackOverflowException：

let test n  =
    let map x      = async.Return x
    let reduce x y = async.Return (x + y)
    mapReduce map reduce [0..n]
    |> Async.RunSynchronously

编辑： <|> 组合器的另一种实现使测试在 N=10000 上成功：

let ( <|> ) (a: Async<unit>) (b: Async<unit>) =
  Async.FromContinuations(fun (ok, _, _) ->
    let count = ref 0
    let ok () =
        lock count (fun () ->
            match !count with
            | 0 -> incr count
            | _ -> ok ())
    Async.Start <|
        async {
            do! a
            return ok ()
        }
    Async.Start <|
        async {
            do! b
            return ok ()
        })

这对我来说真的很惊讶，因为这是我的假设Async.StartChild 正在做。关于哪种解决方案是最佳的有什么想法吗？

Answer 1

我认为在启动使用 <|> 运算符创建的异步工作流程时会发生堆栈溢出异常。对 Async.StartChild 的调用启动第一个工作流程，该工作流程使用 <|> 进行组合，因此它再次调用 Async.StartChild > 等。

修复此问题的一个简单方法是在计时器的处理程序中安排工作流程（以便它不会添加到当前堆栈中）。类似这样的事情：

let ( <|> ) (a: Async<unit>) (b: Async<unit>) =
    async {
        do! Async.Sleep 1
        let! x = Async.StartChild a
        let! y = Async.StartChild b
        do! x
        do! y }

修复它的更好方法是创建您自己的 Seq.reduce - 当前的实现将其逐一折叠，这样您将获得一棵深度为 10000 的树，其中仅包含右侧是单个工作项，左侧是所有其他工作项。如果您创建了工作项的平衡二叉树，那么它不应该堆栈溢出，因为高度只有 15 左右。

编辑尝试用以下函数替换Seq.reduce：

module Seq = 
  let reduceBallanced f input =
    let arr = input |> Array.ofSeq
    let rec reduce s t =
      if s + 1 >= t then arr.[s]
      else 
        let m = (s + t) / 2
        f (reduce s m) (reduce m t)
    reduce 0 arr.Length

Answer 2

我相信托马斯的直觉是正确的，但在花了相当多的时间弄清楚这一点之后，这是我自己的话和更多细节。

1. 问题在于，由于过度同步，上述代码没有实现预期的mapReduce算法。特别是，a <|> b<|>在 a 和 b 完成之前，c 不会启动 c，因此实际上 <|>< /code> 对于两次以上计算的并行性是没有用的。

2. 第二个问题是 async.Return x 与 Async.FromContinuations(fun (ok, _, _) -> ok x) 同构。该示例实际上是在单线程上按顺序执行的，并且分配的闭包导致堆栈崩溃。

对于好奇的读者，下面是我第二次尝试设计这个算法，它似乎表现得更好一些（n=100000 上大约 1 秒，n=100000 上大约 21 秒code> 具有使用 Async.Sleep 1000 扩展的映射和化简函数，我有 Core i3）。

let mapReduce (map    : 'T1 -> Async<'T2>)
              (reduce : 'T2 -> 'T2 -> Async<'T2>)
              (input  : seq<'T1>) : Async<'T2> =
    let run (a: Async<'T>) (k: 'T -> unit) =
        Async.StartWithContinuations(a, k, ignore, ignore)
    Async.FromContinuations <| fun (ok, _, _) ->
        let k = ref 0
        let agent =
            new MailboxProcessor<_>(fun chan ->
                async {
                    for i in 2 .. k.Value do
                        let! x = chan.Receive()
                        let! y = chan.Receive()
                        return run (reduce x y) chan.Post
                    let! r = chan.Receive()
                    return ok r
                })
        k :=
            (0, input)
            ||> Seq.fold (fun count x ->
                run (map x) agent.Post
                count + 1)
        agent.Start()

Answer 3

非常有趣的讨论！
我对 Async.Parallel 也有类似的问题，

let (<||>) first second = async { let! results = Async.Parallel([|first; second|]) in return   (results.[0], results.[1]) } 

let test = async { do! Async.Sleep 100 } 
(test, [1..10000]) 
||> List.fold (fun state value -> (test <||> state) |> Async.Ignore) 
|> Async.RunSynchronously // stackoverflow

我非常沮丧......所以我通过创建自己的并行组合器来解决它。

let parallel<'T>(computations : Async<'T> []) : Async<'T []> =
  Async.FromContinuations (fun (cont, exnCont, _) ->
    let count = ref computations.Length
    let results : 'T [] = Array.zeroCreate computations.Length
    computations 
        |> Array.iteri (fun i computation ->
            Async.Start <|
                async { 
                    try
                        let! res = computation
                        results.[i] <- res 
                    with ex -> exnCont ex

                    let n = System.Threading.Interlocked.Decrement(count)
                    if n = 0 then 
                        results |> cont 
                }))

最后受到讨论的启发，我实现了以下mapReduce函数

// (|f ,⊗|)

let mapReduce (mapF : 'T -> Async<'R>) (reduceF : 'R -> 'R -> Async<'R>) (input : 'T []) : Async<'R> = 
let rec mapReduce' s e =
    async { 
        if s + 1 >= e then return! mapF input.[s]
        else 
            let m = (s + e) / 2
            let! (left, right) =  mapReduce' s m <||> mapReduce' m e
            return! reduceF left right
    }
mapReduce' 0 input.Length

Answer 4

另一种简单的实现可能类似于：

let mapReduce' (map    : 'T1 -> Async<'T2>)
              (reduce : 'T2 -> 'T2 -> Async<'T2>)
              (input  : seq<'T1>) : Async<'T2> = 
        async {
            let! r = input |> Seq.map map |> Async.Parallel
            return r |> Array.toSeq 
                   |> Seq.reduce (fun a b -> reduce a b |> Async.RunSynchronously)

        }

在这种情况下，映射阶段是并行执行的，然后缩减阶段是顺序执行的，因为它对先前计算的值具有数据依赖性。