f# - 与 MailboxProcessor 和 Task 的交互永远挂起

Question

我想按顺序处理一系列作业，但我想将这些作业并行排队。

这是我的代码:

open System.Threading.Tasks

let performWork (work : int) =
  task {
    do! Task.Delay 1000

    if work = 7 then
      failwith "Oh no"
    else
      printfn $"Work {work}"
  }

async {
  let w = MailboxProcessor.Start (fun inbox -> async {
    while true do
      let! message = inbox.Receive()

      let (ch : AsyncReplyChannel<_>), work = message

      do!
        performWork work
        |> Async.AwaitTask

      ch.Reply()
  })

  w.Error.Add(fun exn -> raise exn)

  let! completed =
    seq {
      for i = 1 to 10 do
        async {
          do! Async.Sleep 100
          do! w.PostAndAsyncReply(fun ch -> ch, i)

          return i
        }
    }
    |> fun jobs -> Async.Parallel(jobs, maxDegreeOfParallelism = 4)

  printfn $"Completed {Seq.length completed} job(s)."
}
|> Async.RunSynchronously

我希望这段代码在到达工作项 7 时崩溃。

但是，它永远挂起:

$ dotnet fsi ./Test.fsx
Work 3
Work 1
Work 2
Work 4
Work 5
Work 6

我认为 w.Error 事件没有正确触发。

我应该如何捕捉并重新抛出这个错误？

如果我的工作是async，那么它会按预期崩溃:

let performWork (work : int) =
  async {
    do! Async.Sleep 1000

    if work = 7 then
      failwith "Oh no"
    else
      printfn $"Work {work}"
  }

但我不明白为什么这很重要。

---

利用 Result 也可以，但同样，我不知道为什么需要这样做。

async {
  let w = MailboxProcessor.Start (fun inbox -> async {
    while true do
      let! message = inbox.Receive()

      let (ch : AsyncReplyChannel<_>), work = message

      try
        do!
          performWork work
          |> Async.AwaitTask

        ch.Reply(Ok ())
      with exn ->
        ch.Reply(Error exn)
  })

  let performWorkOnWorker (work : int) =
    async {
      let! outcome = w.PostAndAsyncReply(fun ch -> ch, work)

      match outcome with
      | Ok () ->
        return ()
      | Error exn ->
        return raise exn
    }

  let! completed =
    seq {
      for i = 1 to 10 do
        async {
          do! Async.Sleep 100
          do! performWorkOnWorker i

          return i
        }
    }
    |> fun jobs -> Async.Parallel(jobs, maxDegreeOfParallelism = 4)

  printfn $"Completed {Seq.length completed} job(s)."
}
|> Async.RunSynchronously

Answer 1

我认为问题在于您的错误处理:

w.Error.Add(fun exn -> raise exn)

您没有处理异常，而是试图再次引发它，我认为这会导致无限循环。

您可以将其更改为打印异常:

w.Error.Add(printfn "%A")

结果是:

Work 4
Work 2
Work 1
Work 3
Work 5
Work 6
System.AggregateException: One or more errors occurred. (Oh no)
 ---> System.Exception: Oh no
   at Program.performWork@4.MoveNext() in C:\Users\Brian Berns\Source\Repos\FsharpConsole\FsharpConsole\Program.fs:line 8
   --- End of inner exception stack trace ---