如何判断异步计算是否发送到线程池

Question

我最近被告知在

async {
return! async { return "hi" } }
|> Async.RunSynchronously
|> printfn "%s"

嵌套的Async<'T>(async { return 1 })不会被发送到线程池进行评估，而在

async {
use ms = new MemoryStream [| 0x68uy; 0x69uy |]
use sr = new StreamReader (ms)
return! sr.ReadToEndAsync () |> Async.AwaitTask }
|> Async.RunSynchronously
|> printfn "%s"

嵌套Async<'T>(sr.ReadToEndAsync () |> Async.AwaitTask)将是。当Async<'T>在异步操作(如let!或return!)中执行时，它决定是否发送到线程池？特别是，您将如何定义发送到线程池的线程？您必须在async块或传递给Async.FromContinuations的 lambda 中包含什么代码？

Answer 1

TL;大卫：不是那样的。async本身不会向线程池"发送"任何内容。它所做的只是运行延续，直到它们停止。如果其中一个延续决定继续使用新线程 - 那么，这就是线程切换发生的时候。

---

让我们设置一个小例子来说明会发生什么：

let log str = printfn $"{str}: thread = {Thread.CurrentThread.ManagedThreadId}"
let f = async {
log "1"
let! x = async { log "2"; return 42 }
log "3"
do! Async.Sleep(TimeSpan.FromSeconds(3.0))
log "4"
}
log "starting"
f |> Async.StartImmediate
log "started"
Console.ReadLine()

如果运行此脚本，它将打印、starting、1、2、3、started，然后等待 3 秒钟，然后打印4，除4之外，所有这些脚本都将具有相同的线程 ID。您可以看到，直到Async.Sleep之前的所有内容都在同一线程上同步执行，但在此之后async执行停止，主程序执行继续，打印started，然后在ReadLine上阻塞。当Async.Sleep醒来并想要继续执行时，原始线程已经在ReadLine上被阻塞，因此异步计算可以继续在新线程上运行。

这是怎么回事？这是如何运作的？

首先，异步计算的结构方式是"延续传递风格"。这是一种技术，其中每个函数都不会将其结果返回给调用方，而是调用另一个函数，将结果作为其参数传递。

让我用一个例子来说明：

// "Normal" style:
let f x = x + 5
let g x = x * 2
printfn "%d" (f (g 3)) // prints 11
// Continuation-passing style:
let f x next = next (x + 5)
let g x next = next (x * 2)
g 3 (fun res1 -> f res1 (fun res2 -> printfn "%d" res2))

这被称为"延续传递"，因为next参数称为"延续" - 即它们是表示程序在调用f或g后如何继续的函数。是的，这正是Async.FromContinuations的意思。

表面上看起来非常愚蠢和迂回，这允许我们做的是让每个函数决定何时、如何甚至是否继续发生。例如，我们上面的f函数可以做一些异步的事情，而不仅仅是简单地返回结果：

let f x next = httpPost "http://calculator.com/add5" x next

以延续传递样式对其进行编码将允许此类函数在对calculator.com的请求进行时不会阻塞当前线程。你问阻塞线程有什么问题？我会把你推荐给最初提示你问题的原始答案。

---

其次，当你编写这些async { ... }块时，编译器会给你一点帮助。它采用看起来像是一步一步的命令式程序，并将其"展开"为一系列持续传递调用。这种展开的"突破"点是所有以爆炸结束的结构——let!、do!、return!。

例如，上面的async块看起来像这样(F#-ish 伪代码)：

let return42 onDone = 
log "2"
onDone 42
let f onDone =
log "1"
return42 (fun x ->
log "3"
Async.Sleep (3 seconds) (fun () ->
log "4"
onDone ()
)
)

在这里，您可以清楚地看到return42函数只是立即调用其延续，从而使从log "1"到log "3"的整个事情完全同步，而Async.Sleep函数不会立即调用其延续，而是安排它稍后(3 秒后)在线程池上运行。这就是线程切换发生的地方。

最后，这就是您问题的答案：为了使async计算跳转线程，传递给Async.FromContinuations的回调应该做任何事情，但立即调用成功延续。

---

进一步调查的几点注意事项

1. 上面示例中的onDone技术在技术上称为"monadic bind"，实际上在实际的 F# 程序中，它由async.Bind方法表示。这个答案也可能有助于理解这个概念。
2. 以上有点过于简单化了。实际上，async执行比这要复杂一些。在内部，它使用一种称为"蹦床"的技术，简单来说，这只是一个循环，在每个转弯处运行一个thunk，但至关重要的是，正在运行的thunk也可以"要求"它运行另一个thunk，如果它这样做，循环将这样做，依此类推，永远，直到下一个thunk不要求运行另一个thunk， 然后整个事情终于停止了。
3. 在我的示例中，我专门使用了Async.StartImmediate来启动计算，因为Async.StartImmediate将按照它在锡上所说的去做：它将立即开始运行计算，就在那里。这就是为什么一切都在与主程序相同的线程上运行的原因。Async模块中有许多替代启动功能。例如，Async.Start将在线程池上启动计算。从log "1"到log "3"的行仍将同步发生，无需在它们之间切换线程，但它将发生在与log "start"和log "starting"不同的线程上。在这种情况下，线程切换将在async计算开始之前发生，因此不计算在内。