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

f#

我最近被告知,在

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块中或传入的 lambda 中包含哪些代码Async.FromContinuations

回答

TL;DR: 不是那样的。在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, then 1, 2, 3, then started,然后等待 3 秒,然后打印4,除此之外的所有4线程 ID 都相同。您可以看到,until 的所有内容都Async.Sleep在同一个线程上同步执行,但是在那之后async执行停止并且主程序继续执行,打印started然后阻塞 on 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参数被称为“继续”——即它们是表示程序在调用or后如何继续的函数。是的,这正是什么意思。fgAsync.FromContinuations

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

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

以连续传递风格对其进行编码将允许此类函数在请求进行中时不会阻塞当前线程calculator.com。你问,阻塞线程有什么问题?我会向您介绍最初提示您问题的原始答案。


其次,当您编写这些async { ... }块时,编译器会给您一些帮助。它采用看起来像一步一步的命令式程序并将其“展开”为一系列继续传递调用的过程。这种展开的“断点”是所有以 bang - 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执行过程比这要复杂一些。在内部,它使用了一种叫做“trampoline”的技术,简单地说,它只是一个循环,在每一轮都运行一个 thunk,但至关重要的是,正在运行的 thunk 也可以“要求”它运行另一个 thunk,如果它这样做了,循环会一直这样做,以此类推,直到下一个 thunk 不再要求运行另一个 thunk,然后整个事情最终停止。
  3. Async.StartImmediate在我的例子中,我专门用来开始计算,因为它Async.StartImmediate会按照它在锡上所说的去做:它会立即开始运行计算,就在那里。这就是为什么一切都与主程序在同一线程上运行的原因。模块中有许多可供选择的启动功能Async。例如,Async.Start将在线程池上启动计算。从log "1"to的行log "3"仍然会同步发生,它们之间没有线程切换,但它将发生在来自log "start"and的不同线程上log "starting"。在这种情况下,线程切换async甚至会在计算开始之前发生,所以它不算数。

以上是如何判断异步计算是否发送到线程池?的全部内容。
THE END
分享
二维码
< <上一篇
下一篇>>