F#的异步如何真正起作用？

Question

我想学习如何async和let!工作在F#.我读过的所有文档都让人感到困惑.使用Async.RunSynchron运行异步块有什么意义？这是异步还是同步？看起来像是一个矛盾.

文档说Async.StartImmediate在当前线程中运行.如果它在同一个线程中运行,它对我来说看起来并不是异步......或者asyncs更像是协程而不是线程.如果是这样的话,他们什么时候退回？

引用MS文档:

使用let的代码行!开始计算,然后暂停线程直到结果可用,此时执行继续.

如果线程等待结果,我为什么要使用它？看起来像普通的旧函数调用.

Async.Parallel有什么作用？它接收一系列Async <'T>.为什么不能并行执行一系列普通函数？

我想我在这里缺少一些非常基本的东西.我想在我理解之后,所有文档和样本都将开始有意义.

Answer 1

一些东西.

一,区别

let resp = req.GetResponse()

和

let! resp = req.AsyncGetReponse()

对于Web请求"在海上"的大概数百毫秒(CPU的永恒),前者使用一个线程(在I/O上阻塞),而后者使用零线程.这是异步最常见的"胜利":您可以编写非阻塞I/O,不会浪费任何线程等待硬盘转动或网络请求返回.(与大多数其他语言不同,您不会被迫进行控制反转并将事物纳入回调中.)

其次,Async.StartImmediate将在当前线程上启动异步.一个典型的用途是使用GUI,你有一些GUI应用程序想要更新UI(例如说"加载......"某处),然后做一些后台工作(从磁盘或其他任何东西加载),然后返回到前台UI线程以在完成时更新UI("完成!"). StartImmediate使异步能够在操作开始时更新UI并捕获,SynchronizationContext以便在操作结束时可以返回GUI以进行UI的最终更新.

接下来,Async.RunSynchronously很少使用(一篇论文是你在任何应用程序中最多调用一次).在限制中,如果您将整个程序编写为异步,则在"main"方法中,您将调用RunSynchronously运行程序并等待结果(例如,在控制台应用程序中打印出结果).这会阻塞一个线程,因此它通常只在程序的异步部分的"顶部"有用,在边界上有同步的东西.(更高级的用户可能更喜欢StartWithContinuations- RunSynchronously从异步回到同步有点"容易入侵".)

最后,Async.Parallelfork-join并行性.您可以编写一个类似于函数而不是asyncs的函数(就像TPL中的东西一样),但F#中的典型最佳位置是并行I/O绑定计算,它们已经是异步对象,所以这是最常用的签名.(对于CPU绑定的并行性,您可以使用asyncs,但您也可以使用TPL.)

Answer 2

异步的用法是保存使用中的线程数.

请参阅以下示例:

let fetchUrlSync url = 
    let req = WebRequest.Create(Uri url)
    use resp = req.GetResponse()
    use stream = resp.GetResponseStream()
    use reader = new StreamReader(stream)
    let contents = reader.ReadToEnd()
    contents 

let sites = ["http://www.bing.com";
             "http://www.google.com";
             "http://www.yahoo.com";
             "http://www.search.com"]

// execute the fetchUrlSync function in parallel 
let pagesSync = sites |> PSeq.map fetchUrlSync  |> PSeq.toList

上面的代码是你想要做的:定义一个函数并并行执行.那么为什么我们需要异步呢？

让我们考虑一下大事.例如,如果网站的数量不是4,而是说10,000!然后需要10,000个线程并行运行它们,这是一个巨大的资源成本.

在异步时:

let fetchUrlAsync url =
    async { let req =  WebRequest.Create(Uri url)
            use! resp = req.AsyncGetResponse()
            use stream = resp.GetResponseStream()
            use reader = new StreamReader(stream)
            let contents = reader.ReadToEnd()
            return contents }
let pagesAsync = sites |> Seq.map fetchUrlAsync |> Async.Parallel |> Async.RunSynchronously

当代码进入时use! resp = req.AsyncGetResponse(),当前线程被放弃,其资源可用于其他目的.如果响应在1秒后回复,那么你的线程可以使用这1秒来处理其他东西.否则线程被阻塞,浪费线程资源1秒钟.

因此,即使您以异步方式并行下载10000个网页,线程数也仅限于少量.

我认为你不是.Net/C#程序员.异步教程通常假定人们知道.Net以及如何在C#中编写异步IO(很多代码).F#中Async构造的神奇之处不在于并行.因为简单并行可以通过其他结构实现,例如.Net并行扩展中的ParallelFor.但是,异步IO更复杂,因为您看到线程放弃执行,当IO完成时,IO需要唤醒其父线程.这是异步魔术用于的地方:在几行简洁代码中,您可以进行非常复杂的控制.

Answer 3

这里有很多好的答案,但我认为我对这个问题采取了不同的角度:F#的异步是如何工作的？

与async/awaitC#F#不同,开发人员实际上可以实现自己的版本Async.这可以是学习如何Async工作的好方法.

(感兴趣的源代码Async可以在这里找到:https://github.com/Microsoft/visualfsharp/blob/fsharp4/src/fsharp/FSharp.Core/control.fs)

作为我们DIY工作流程的基本构建块,我们定义:

type DIY<'T> = ('T->unit)->unit

这是一个接受另一个函数(称为continuation)的函数,该函数在类型的结果'T准备好时调用.这允许DIY<'T>在不阻塞调用线程的情况下启动后台任务.当结果准备就绪时,将调用continuation以允许继续计算.

F#Async构建块有点复杂,因为它还包括取消和异常延续,但基本上就是这样.

为了支持F#工作流语法,我们需要定义一个计算表达式(https://msdn.microsoft.com/en-us/library/dd233182.aspx).虽然这是一个相当先进的F#功能,但它也是F#最神奇的功能之一.定义两个最重要的操作是return:bind这是由F#中使用我们的组合DIY<_>积木成聚集DIY<_>积木.

adaptTask用于适应Task<'T>a DIY<'T>. startChild允许启动几个simulatenous DIY<'T>,请注意它不会启动新线程,但重用调用线程.

这里没有任何进一步的例子是示例程序:

open System
open System.Diagnostics
open System.Threading
open System.Threading.Tasks

// Our Do It Yourself Async workflow is a function accepting a continuation ('T->unit).
// The continuation is called when the result of the workflow is ready. 
// This may happen immediately or after awhile, the important thing is that 
//  we don't block the calling thread which may then continue executing useful code.
type DIY<'T> = ('T->unit)->unit

// In order to support let!, do! and so on we implement a computation expression.
// The two most important operations are returnValue/bind but delay is also generally 
//  good to implement.
module DIY =

    // returnValue is called when devs uses return x in a workflow.
    // returnValue passed v immediately to the continuation.
    let returnValue (v : 'T) : DIY<'T> =
        fun a ->
            a v

    // bind is called when devs uses let!/do! x in a workflow
    // bind binds two DIY workflows together
    let bind (t : DIY<'T>) (fu : 'T->DIY<'U>) : DIY<'U> =
        fun a ->
            let aa tv =
                let u = fu tv
                u a
            t aa

    let delay (ft : unit->DIY<'T>) : DIY<'T> =
        fun a ->
            let t = ft ()
            t a

    // starts a DIY workflow as a subflow
    // The way it works is that the workflow is executed 
    //  which may be a delayed operation. But startChild
    //  should always complete immediately so in order to
    //  have something to return it returns a DIY workflow
    // postProcess checks if the child has computed a value 
    //  ie rv has some value and if we have computation ready
    //  to receive the value (rca has some value).
    //  If this is true invoke ca with v
    let startChild (t : DIY<'T>) : DIY<DIY<'T>> =
        fun a ->
            let l   = obj()
            let rv  = ref None
            let rca = ref None

            let postProcess () =
                match !rv, !rca with
                | Some v, Some ca ->
                    ca v
                    rv  := None
                    rca := None
                | _ , _ -> ()

            let receiver v =
                lock l <| fun () ->
                    rv := Some v
                    postProcess ()

            t receiver

            let child : DIY<'T> =
                fun ca ->
                    lock l <| fun () ->
                        rca := Some ca
                        postProcess ()

            a child

    let runWithContinuation (t : DIY<'T>) (f : 'T -> unit) : unit =
        t f

    // Adapts a task as a DIY workflow
    let adaptTask (t : Task<'T>) : DIY<'T> =
        fun a ->
            let action = Action<Task<'T>> (fun t -> a t.Result)
            ignore <| t.ContinueWith action

    // Because C# generics doesn't allow Task<void> we need to have
    //  a special overload of for the unit Task.
    let adaptUnitTask (t : Task) : DIY<unit> =
        fun a ->
            let action = Action<Task> (fun t -> a ())
            ignore <| t.ContinueWith action

    type DIYBuilder() =
        member x.Return(v)  = returnValue v
        member x.Bind(t,fu) = bind t fu
        member x.Delay(ft)  = delay ft

let diy = DIY.DIYBuilder()

open DIY

[<EntryPoint>]
let main argv = 

    let delay (ms : int) = adaptUnitTask <| Task.Delay ms

    let delayedValue ms v =
        diy {
            do! delay ms
            return v
        }

    let complete = 
        diy {
            let sw = Stopwatch ()
            sw.Start ()

            // Since we are executing these tasks concurrently 
            //  the time this takes should be roughly 700ms
            let! cd1 = startChild <| delayedValue 100 1
            let! cd2 = startChild <| delayedValue 300 2
            let! cd3 = startChild <| delayedValue 700 3

            let! d1 = cd1
            let! d2 = cd2
            let! d3 = cd3

            sw.Stop ()

            return sw.ElapsedMilliseconds,d1,d2,d3
        }

    printfn "Starting workflow"

    runWithContinuation complete (printfn "Result is: %A")

    printfn "Waiting for key"

    ignore <| Console.ReadKey ()

    0

程序的输出应该是这样的:

Starting workflow
Waiting for key
Result is: (706L, 1, 2, 3)

当运行由于Waiting for key未阻止启动工作流程的控制台线程而立即打印的程序注释时.大约700ms后打印结果.

我希望这对一些F#开发者来说很有意思