如何判断是否将异步计算发送到线程池? [英] How to tell if an async computation is sent to the thread pool?

问题描述

我最近被告知

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

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

the nested Async<'T> (async { return 1 }) would not be sent to the thread pool for evaluation, whereas in

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！<)中将其发送到线程池时./code>?特别是，如何定义发送到线程池的线程?您必须在 async 块中或传递给 Async.FromContinuations 的lambda中包括什么代码?

the nested Async<'T> (sr.ReadToEndAsync () |> Async.AwaitTask) would be. What is it about an Async<'T> that decides whether it's sent to the thread pool when it's executed in an asynchronous operation like let! or return!? In particular, how would you define one which is sent to the thread pool? What code do you have to include in the async block, or in the lambda passed into Async.FromContinuations?

推荐答案

TL; DR:不是那样. async 本身不会发送"；线程池中的任何内容.它所做的只是继续运行直到停止.而且，如果这些延续之一决定继续在新线程上-那就是在发生线程切换时.

TL;DR: It's not quite like that. The async itself doesn't "send" anything to the thread pool. All it does is just run continuations until they stop. And if one of those continuations decides to continue on a new thread - well, that's when thread switching happens.

让我们建立一个小例子来说明发生的情况:

Let's set up a small example to illustrate what happens:

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 上被阻塞，因此异步计算可以继续在新线程上运行.

If you run this script, it will print, starting, then 1, 2, 3, then started, then wait 3 seconds, and then print 4, and all of them except 4 will have the same thread ID. You can see that everything until Async.Sleep is executed synchronously on the same thread, but after that async execution stops and the main program execution continues, printing started and then blocking on ReadLine. By the time Async.Sleep wakes up and wants to continue execution, the original thread is already blocked on ReadLine, so the async computation gets to continue running on a new one.

这是怎么回事?这个功能如何?

What's going on here? How does this function?

首先，异步计算的结构方式在"; continuation-passing style" .这是一种技术，其中每个函数都不将其结果返回给调用方，而是调用另一个函数，并将结果作为其参数传递.

First, the way the async computation is structured is in "continuation-passing style". It's a technique where every function doesn't return its result to the caller, but calls another function instead, passing the result as its parameter.

让我举例说明:

// "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 参数称为"continuations"，-即它们是表示调用 f 或 g 后程序继续的函数.是的，这正是 Async.FromContinuations 的意思.

This is called "continuation-passing" because the next parameters are called "continuations" - i.e. they're functions that express how the program continues after calling f or g. And yes, this is exactly what Async.FromContinuations means.

表面上看起来非常愚蠢和回旋，这使我们能够做的是让每个函数确定何时，如何甚至如果继续发生.例如，上面的 f 函数可能正在做异步操作，而不仅仅是返回结果:

Seeming very silly and roundabout on the surface, what this allows us to do is for each function to decide when, how, or even if its continuation happens. For example, our f function from above could be doing something asynchronous instead of just plain returning the result:

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

以连续传递样式对其进行编码将允许该函数在对 calculator.com 的请求进行过程中不阻止当前线程.您问阻塞线程有什么问题吗?我将引导您参考最初提示您问题的原始答案.

Coding it in continuation-passing style would allow such function to not block the current thread while the request to calculator.com is in flight. What's wrong with blocking the thread, you ask? I'll refer you to the original answer that prompted your question in the first place.

第二，当您编写这些 async {...} 块时，编译器会为您提供一些帮助.这需要一个循序渐进的命令式程序，并且需要展开".它进入一系列连续传递的调用.该中断"是指中断".展开的要点是所有以爆炸结尾的结构- let！， do！， return！.

Second, when you write those async { ... } blocks, the compiler gives you a little help. It takes what looks like a step-by-step imperative program and "unrolls" it into a series of continuation-passing calls. The "breaking" points for this unfolding are all the constructs that end with a bang - let!, do!, return!.

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

The above async block, for example, would look somethiing like this (F#-ish pseudocode):

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秒内)运行.那就是线程切换发生的地方.

Here, you can plainly see that the return42 function simply calls its continuation right away, thus making the whole thing from log "1" to log "3" completely synchronous, whereas the Async.Sleep function doesn't call its continuation right away, instead scheduling it to be run later (in 3 seconds) on the thread pool. That's where the thread switching happens.

最后，这是您问题的答案:为了使 async 计算跳转线程，将回调传递给 Async.FromContinuations ，除了立即致电成功延续.

And here, finally, lies the answer to your question: in order to have the async computation jump threads, your callback passed to Async.FromContinuations should do anything but call the success continuation immediately.

一些需要进一步调查的笔记

1. 以上示例中的 onDone 技术在技术上称为单键绑定" ，实际上在实际的F#程序中，它是由 async.Bind 方法表示的.此答案可能也有助于理解该概念.
2. 以上内容有些过分简化.实际上 异步执行要比这复杂一些.在内部，它使用称为蹦床" 的技术，以简单的术语来说，一个在每个转弯上运行一个thunk的循环，但是至关重要的是，运行的thunk也可以询问"一个循环.它运行另一个重击程序，如果执行了，循环将一直执行，依此类推，直到下一个重击程序不再要求运行另一个重击程序，然后整个过程终于停止.
3. 在我的示例中，我专门使用了 Async.StartImmediate 来开始计算，因为 Async.StartImmediate 将按照其提示进行操作:它将开始运行立即进行计算.这就是为什么所有内容都与主程序在同一线程上运行的原因.在 Async中，有许多替代的启动函数模块.例如， Async.Start 将在线程池上开始计算.从 log"1" 到 log"3" 的行仍将全部同步发生，而无需在它们之间进行线程切换，但是它将在与 log开始" 和 log开始" .在这种情况下，线程切换将在 async 计算甚至开始之前进行，因此不计入.

1. The onDone technique in the above example is technically called "monadic bind", and indeed in real F# programs it's represented by the async.Bind method. This answer might also be of help understanding the concept.
2. The above is a bit of an oversimplification. In reality the async execution is a bit more complicated than that. Internally it uses a technique called "trampoline", which in plain terms is just a loop that runs a single thunk on every turn, but crucially, the running thunk can also "ask" it to run another thunk, and if it does, the loop will do so, and so on, forever, until the next thunk doesn't ask to run another thunk, and then the whole thing finally stops.
3. I specifically used Async.StartImmediate to start the computation in my example, because Async.StartImmediate will do just what it says on the tin: it will start running the computation immediately, right there. That's why everything ran on the same thread as the main program. There are many alternative starting functions in the Async module. For example, Async.Start will start the computation on the thread pool. The lines from log "1" to log "3" will still all happen synchronously, without thread switching between them, but it will happen on a different thread from log "start" and log "starting". In this case thread switching will happen before the async computation even starts, so it doesn't count.

这篇关于如何判断是否将异步计算发送到线程池?的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！