如何在 F# 中实现异步而不是并行

Question

（坚持使用异步获取许多网页的常见示例）

我如何异步分拆多个（数百个）网页请求，然后等待所有请求完成，然后再进行下一步？ Async.AsParallel 一次处理几个请求，由 CPU 上的核心数量控制。抓取网页不是 CPU 密集型操作。我对 Async.AsParallel 的加速不满意，正在寻找替代方案。

我尝试将 Async.StartAsTask 和 Task[].WaitAll 之间的点联系起来。本能地，我编写了以下代码，但它无法编译。

let processItemsConcurrently (items : int seq) = 
  let tasks = items |> Seq.map (fun item -> Async.StartAsTask(fetchAsync item))
  Tasks.Task.WaitAll(tasks) 

你会如何处理这个问题？

Answer 1

Async.Parallel 几乎肯定就在这里。不确定你对什么不满意； F# 异步的优势更多地在于异步计算，而不是任务并行的 CPU 绑定内容（后者更适合 Task 和 .NET 4.0 TPL）。这是一个完整的示例：

open System.Diagnostics 
open System.IO
open System.Net
open Microsoft.FSharp.Control.WebExtensions 

let sites = [|
    "http://bing.com"
    "http://google.com"
    "http://cnn.com"
    "http://stackoverflow.com"
    "http://yahoo.com"
    "http://msdn.com"
    "http://microsoft.com"
    "http://apple.com"
    "http://nfl.com"
    "http://amazon.com"
    "http://ebay.com"
    "http://expedia.com"
    "http://twitter.com"
    "http://reddit.com"
    "http://hulu.com"
    "http://youtube.com"
    "http://wikipedia.org"
    "http://live.com"
    "http://msn.com"
    "http://wordpress.com"
    |]

let print s = 
    // careful, don't create a synchronization bottleneck by printing
    //printf "%s" s
    ()

let printSummary info fullTimeMs =
    Array.sortInPlaceBy (fun (i,_,_) -> i) info
//  for i, size, time in info do
//      printfn "%2d  %7d  %5d" i size time
    let longest = info |> Array.map (fun (_,_,time) -> time) |> Array.max
    printfn "longest request took %dms" longest
    let bytes = info |> Array.sumBy (fun (_,size,_) -> float size)
    let seconds = float fullTimeMs / 1000.
    printfn "sucked down %7.2f KB/s" (bytes / 1024.0 / seconds)

let FetchAllSync() =
    let allsw = Stopwatch.StartNew()
    let info = sites |> Array.mapi (fun i url ->
        let sw = Stopwatch.StartNew()
        print "S"
        let req = WebRequest.Create(url) 
        use resp = req.GetResponse()
        use stream = resp.GetResponseStream()
        use reader = new StreamReader(stream,
                            System.Text.Encoding.UTF8, true, 4096) 
        print "-"
        let contents = reader.ReadToEnd()
        print "r"
        i, contents.Length, sw.ElapsedMilliseconds)
    let time = allsw.ElapsedMilliseconds 
    printSummary info time
    time, info |> Array.sumBy (fun (_,size,_) -> size)

let FetchAllAsync() =
    let allsw = Stopwatch.StartNew()
    let info = sites |> Array.mapi (fun i url -> async {
        let sw = Stopwatch.StartNew()
        print "S"
        let req = WebRequest.Create(url) 
        use! resp = req.AsyncGetResponse()
        use stream = resp.GetResponseStream()
        use reader = new AsyncStreamReader(stream, // F# PowerPack
                           System.Text.Encoding.UTF8, true, 4096) 
        print "-"
        let! contents = reader.ReadToEnd()  // in F# PowerPack
        print "r"
        return i, contents.Length, sw.ElapsedMilliseconds })
                    |> Async.Parallel 
                    |> Async.RunSynchronously 
    let time = allsw.ElapsedMilliseconds 
    printSummary info time
    time, info |> Array.sumBy (fun (_,size,_) -> size)

// By default, I think .NET limits you to 2 open connections at once
ServicePointManager.DefaultConnectionLimit <- sites.Length 

for i in 1..3 do // to warmup and show variance
    let time1,r1 = FetchAllSync()
    printfn "Sync took %dms, result was %d" time1 r1
    let time2,r2 = FetchAllAsync()
    printfn "Async took %dms, result was %d  (speedup=%2.2f)" 
        time2 r2 (float time1/ float time2)
    printfn ""

在我的 4 核机器上，这始终提供近 4 倍的加速。

编辑

为了回复您的评论，我更新了代码。你是对的，我添加了更多站点，但没有看到预期的加速（仍然稳定在 4 倍左右）。我已经开始在上面添加一些调试输出，将继续调查以查看是否有其他东西限制了连接...

编辑

再次编辑了代码。好吧，我发现了可能的瓶颈。下面是 PowerPack 中 AsyncReadToEnd 的实现：

type System.IO.StreamReader with
   member s.AsyncReadToEnd () = 
       FileExtensions.UnblockViaNewThread (fun () -> s.ReadToEnd())

换句话说，它只是阻塞线程池线程并同步读取。啊啊！！！让我看看是否可以解决这个问题。

编辑

好的，PowerPack 中的 AsyncStreamReader 做了正确的事情，我现在正在使用它。

然而，关键问题似乎是方差。

当您点击 cnn.com 时，很多时候结果会在 500 毫秒内返回。但每隔一段时间，您就会收到一个需要 4 秒的请求，这当然可能会杀死明显的异步性能，因为总时间是最不幸的请求的时间。

运行上面的程序，我在家里的 2 核机器上看到速度从大约 2.5 倍到 9 倍。不过，它的变化非常大。程序中仍然可能存在一些我错过的瓶颈，但我认为网络的方差可能解释了我此时所看到的所有内容。

Answer 2

使用 .NET 的响应式扩展与 F# 相结合，您可以编写一个非常优雅的解决方案 - 查看示例 http://blog.paulbetts.org/index.php/2010/11/16/making-async-io-work- for-you-reactive-style/ （这里使用C#，但使用F#也很容易；关键是使用Begin/End方法而不是sync方法，即使你可以让它编译，它也会不必要地阻塞 n 个 ThreadPool 线程，而不是 Threadpool 只是在完成例程进入时拾取它们）

Answer 3

我敢打赌，您所经历的加速对于您的口味来说不够显着，因为您要么使用 WebRequest 的子类型，要么使用依赖它的类（例如 WebClient）。
如果是这种情况，您需要设置 MaxConnection ConnectionManagementElement 上的 （我建议您仅在需要时才将其设置，否则它将成为一个非常耗时的操作）为一个较高的值，具体取决于您想要从应用程序启动的同时连接的数量。

Answer 4

我不是 F# 人员，但从纯粹的 .NET 角度来看，您正在寻找的是 TaskFactory::FromAsync，其中您将包装在任务中的异步调用类似于 HttpRequest::BeginGetResponse。您还可以使用 TaskCompletionSource 封装 WebClient 公开的 EAP 模型。有关这两个主题的更多信息，请参阅 MSDN 上的此处。

希望有了这些知识，您可以找到最接近的原生 F# 方法来完成您想要做的事情。

Answer 5

下面是一些避免未知因素的代码，例如网络访问延迟。我的 CPU 利用率低于 5%，同步和异步代码路径的效率约为 60-80%。

open System.Diagnostics

let numWorkers = 200
let asyncDelay = 50

let main =
   let codeBlocks = [for i in 1..numWorkers -> 
                        async { do! Async.Sleep asyncDelay } ]

   while true do
      printfn "Concurrent started..."
      let sw = new Stopwatch()
      sw.Start()
      codeBlocks |> Async.Parallel |> Async.RunSynchronously |> ignore
      sw.Stop()
      printfn "Concurrent in %d millisec" sw.ElapsedMilliseconds
      printfn "efficiency: %d%%" (int64 (asyncDelay * 100) / sw.ElapsedMilliseconds)

      printfn "Synchronous started..."
      let sw = new Stopwatch()
      sw.Start()
      for codeBlock in codeBlocks do codeBlock |> Async.RunSynchronously |> ignore
      sw.Stop()
      printfn "Synchronous in %d millisec" sw.ElapsedMilliseconds
      printfn "efficiency: %d%%" (int64 (asyncDelay * numWorkers * 100) / sw.ElapsedMilliseconds)

main