那些遇到过的问题

F#异步挂起

Rei*_*aka 4 f# asynchronous

我有一些相当简单的F#异步代码从Wikipedia上下载了一百篇随机文章(用于研究).

出于某种原因,代码在下载期间的任意时间点挂起.有时它是在50之后,有时是在80之后.

异步代码本身非常简单:

let parseWikiAsync(url:string, count:int ref) =
    async {
            use wc = new WebClientWithTimeout(Timeout = 5000)
            let! html = wc.AsyncDownloadString(Uri(url))
            let ret =
                try html |> parseDoc |> parseArticle
                with | ex -> printfn "%A" ex; None
            lock count (fun () ->
                if !count % 10 = 0 then
                    printfn "%d" !count
                count := !count + 1
            )
            return ret
    }
Run Code Online (Sandbox Code Playgroud)

因为我无法通过fsi弄清楚问题是什么,所以我创建了WebClientWithTimeout,这是一个System.Net.WebClient允许我指定超时的包装器:

type WebClientWithTimeout() =
    inherit WebClient()
    member val Timeout = 60000 with get, set

    override x.GetWebRequest uri =
        let r = base.GetWebRequest(uri)
        r.Timeout <- x.Timeout
        r
Run Code Online (Sandbox Code Playgroud)

然后我使用异步组合器来检索超过一百页,并清除所有返回parseWikiAsync调用的文章None(其中大多数是"消歧页面")直到我有100篇文章:

let en100 =
    let count = ref 0
    seq { for _ in 1..110 -> parseWikiAsync("http://en.wikipedia.org/wiki/Special:Random", count) }
    |> Async.Parallel
    |> Async.RunSynchronously
    |> Seq.choose id
    |> Seq.take 100
Run Code Online (Sandbox Code Playgroud)

当我编译代码并在调试器中运行它时,只有三个线程,其中只有一个运行实际代码 - 异步管道.另外两个对于位置"不可用",而在调用堆栈中没有任何内容.

我认为这意味着它不会卡AsyncDownloadString在parseWikiAsync 中或任何地方.还有什么可能导致这个？

哦,最初,在异步代码实际启动之前需要大约一整分钟.之后它会以相当合理的速度运行,直到它无限期地再次挂起.

这是主线程的调用堆栈:

>   mscorlib.dll!System.Threading.WaitHandle.InternalWaitOne(System.Runtime.InteropServices.SafeHandle waitableSafeHandle, long millisecondsTimeout, bool hasThreadAffinity, bool exitContext) + 0x22 bytes 
    mscorlib.dll!System.Threading.WaitHandle.WaitOne(int millisecondsTimeout, bool exitContext) + 0x28 bytes    
    FSharp.Core.dll!Microsoft.FSharp.Control.AsyncImpl.ResultCell<Microsoft.FSharp.Control.AsyncBuilderImpl.Result<Microsoft.FSharp.Core.FSharpOption<Program.ArticleData>[]>>.TryWaitForResultSynchronously(Microsoft.FSharp.Core.FSharpOption<int> timeout) + 0x36 bytes  
    FSharp.Core.dll!Microsoft.FSharp.Control.CancellationTokenOps.RunSynchronously<Microsoft.FSharp.Core.FSharpOption<Program.ArticleData>[]>(System.Threading.CancellationToken token, Microsoft.FSharp.Control.FSharpAsync<Microsoft.FSharp.Core.FSharpOption<Program.ArticleData>[]> computation, Microsoft.FSharp.Core.FSharpOption<int> timeout) + 0x1ba bytes 
    FSharp.Core.dll!Microsoft.FSharp.Control.FSharpAsync.RunSynchronously<Microsoft.FSharp.Core.FSharpOption<Program.ArticleData>[]>(Microsoft.FSharp.Control.FSharpAsync<Microsoft.FSharp.Core.FSharpOption<Program.ArticleData>[]> computation, Microsoft.FSharp.Core.FSharpOption<int> timeout, Microsoft.FSharp.Core.FSharpOption<System.Threading.CancellationToken> cancellationToken) + 0xb9 bytes   
    WikiSurvey.exe!<StartupCode$WikiSurvey>.$Program.main@() Line 97 + 0x55 bytes   F#
Run Code Online (Sandbox Code Playgroud)

Jac*_* P. 8

维基百科在这里不应该受到责备,这是Async.Parallel内部工作的结果.类型签名Async.Parallel是seq<Async<'T>> -> Async<'T[]>.它返回一个包含序列中所有结果的Async值 - 因此它不会返回,直到返回中的所有计算seq<Async<'T>>.

为了说明,我修改了您的代码,以便跟踪未完成请求的数量,即已发送到服务器但尚未接收/解析响应的请求.

open Microsoft.FSharp.Control
open Microsoft.FSharp.Control.WebExtensions
open System
open System.Net
open System.Threading

type WebClientWithTimeout() =
    inherit WebClient()

    let mutable timeout = -1
    member __.Timeout
        with get () = timeout
        and set value = timeout <- value

    override x.GetWebRequest uri =
        let r = base.GetWebRequest(uri)
        r.Timeout <- x.Timeout
        r

type ParsedDoc = ParsedDoc
type ParsedArticle = ParsedArticle

let parseDoc (str : string) = ParsedDoc
let parseArticle (doc : ParsedDoc) = Some ParsedArticle

/// A synchronized wrapper around Console.Out so we don't
/// get garbled console output.
let synchedOut =
    System.Console.Out
    |> System.IO.TextWriter.Synchronized

let parseWikiAsync(url : string, outstandingRequestCount : int ref) =
    async {
    use wc = new WebClientWithTimeout(Timeout = 5000)
    wc.Headers.Add ("User-Agent", "Friendly Bot 1.0 (FriendlyBot@friendlybot.com)")

    // Increment the outstanding request count just before we send the request.
    do
        // NOTE : The message must be created THEN passed to synchedOut.WriteLine --
        // piping it (|>) into synchedOut.WriteLine or using fprintfn causes a closure
        // to be created which somehow defeats the synchronization and garbles the output.
        let msg =
            Interlocked.Increment outstandingRequestCount
            |> sprintf "Outstanding requests: %i"
        synchedOut.WriteLine msg

    let! html = wc.AsyncDownloadString(Uri(url))
    let ret =
        try html |> parseDoc |> parseArticle
        with ex ->
            let msg = sprintf "%A" ex
            synchedOut.WriteLine msg
            None

    // Decrement the outstanding request count now that we've
    // received a reponse and parsed it.
    do
        let msg =
            Interlocked.Decrement outstandingRequestCount
            |> sprintf "Outstanding requests: %i"
        synchedOut.WriteLine msg

    return ret
    }

/// Writes a message to the console, passing a value through
/// so it can be used within a function pipeline.
let inline passThruWithMessage (msg : string) value =
    Console.WriteLine msg
    value

let en100 =
    let outstandingRequestCount = ref 0
    seq { for _ in 1..120 ->
            parseWikiAsync("http://en.wikipedia.org/wiki/Special:Random", outstandingRequestCount) }
    |> Async.Parallel
    |> Async.RunSynchronously
    |> passThruWithMessage "Finished running all of the requests."
    |> Seq.choose id
    |> Seq.take 100
Run Code Online (Sandbox Code Playgroud)

如果编译并运行该代码,您将看到如下输出:

Outstanding requests: 4
Outstanding requests: 2
Outstanding requests: 1
Outstanding requests: 3
Outstanding requests: 5
Outstanding requests: 6
Outstanding requests: 7
Outstanding requests: 8
Outstanding requests: 9
Outstanding requests: 10
Outstanding requests: 12
Outstanding requests: 14
Outstanding requests: 15
Outstanding requests: 16
Outstanding requests: 17
Outstanding requests: 18
Outstanding requests: 13
Outstanding requests: 19
Outstanding requests: 20
Outstanding requests: 24
Outstanding requests: 22
Outstanding requests: 26
Outstanding requests: 27
Outstanding requests: 28
Outstanding requests: 29
Outstanding requests: 30
Outstanding requests: 25
Outstanding requests: 21
Outstanding requests: 23
Outstanding requests: 11
Outstanding requests: 29
Outstanding requests: 28
Outstanding requests: 27
Outstanding requests: 26
Outstanding requests: 25
Outstanding requests: 24
Outstanding requests: 23
Outstanding requests: 22
Outstanding requests: 21
Outstanding requests: 20
Outstanding requests: 19
Outstanding requests: 18
Outstanding requests: 17
Outstanding requests: 16
Outstanding requests: 15
Outstanding requests: 14
Outstanding requests: 13
Outstanding requests: 12
Outstanding requests: 11
Outstanding requests: 10
Outstanding requests: 9
Outstanding requests: 8
Outstanding requests: 7
Outstanding requests: 6
Outstanding requests: 5
Outstanding requests: 4
Outstanding requests: 3
Outstanding requests: 2
Outstanding requests: 1
Outstanding requests: 0
Finished running all of the requests.
Run Code Online (Sandbox Code Playgroud)

正如您所看到的,所有请求都是在解析任何请求之前进行的 - 因此,如果您使用较慢的连接,或者您尝试检索大量文档,则服务器可能会丢弃连接,因为它可能假设您没有检索它尝试发送的响应.代码的另一个问题是您需要明确指定要在seq其中生成的元素数量,这使得代码不再可重用.

更好的解决方案是在某些消费代码需要时检索和解析页面.(如果你考虑一下,那正是F#seq的优点.)我们首先创建一个带有Uri的函数并生成一个seq<Async<'T>>- 即,它产生一个无限的Async<'T>值序列,每个值都将检索来自Uri的内容,解析它,并返回结果.

/// Given a Uri, creates an infinite sequence of whose elements are retrieved
/// from the Uri.
let createDocumentSeq (uri : System.Uri) =
    #if DEBUG
    let outstandingRequestCount = ref 0
    #endif

    Seq.initInfinite <| fun _ ->
        async {
        use wc = new WebClientWithTimeout(Timeout = 5000)
        wc.Headers.Add ("User-Agent", "Friendly Bot 1.0 (FriendlyBot@friendlybot.com)")

        #if DEBUG
        // Increment the outstanding request count just before we send the request.
        do
            // NOTE : The message must be created THEN passed to synchedOut.WriteLine --
            // piping it (|>) into synchedOut.WriteLine or using fprintfn causes a closure
            // to be created which somehow defeats the synchronization and garbles the output.
            let msg =
                Interlocked.Increment outstandingRequestCount
                |> sprintf "Outstanding requests: %i"
            synchedOut.WriteLine msg
        #endif

        let! html = wc.AsyncDownloadString uri
        let ret =
            try Some html
            with ex ->
                let msg = sprintf "%A" ex
                synchedOut.WriteLine msg
                None

        #if DEBUG
        // Decrement the outstanding request count now that we've
        // received a reponse and parsed it.
        do
            let msg =
                Interlocked.Decrement outstandingRequestCount
                |> sprintf "Outstanding requests: %i"
            synchedOut.WriteLine msg
        #endif

        return ret
        }
Run Code Online (Sandbox Code Playgroud)

现在我们使用此函数将页面检索为流:

//
let en100_Streaming =
    #if DEBUG
    let documentCount = ref 0
    #endif

    Uri ("http://en.wikipedia.org/wiki/Special:Random")
    |> createDocumentSeq
    |> Seq.choose (fun asyncDoc ->
        Async.RunSynchronously asyncDoc
        |> Option.bind (parseDoc >> parseArticle))
    #if DEBUG
    |> Seq.map (fun x ->
        let msg =
            Interlocked.Increment documentCount
            |> sprintf "Parsed documents: %i"
        synchedOut.WriteLine msg
        x)
    #endif
    |> Seq.take 50
    // None of the computations actually take place until
    // this point, because Seq.toArray forces evaluation of the sequence.
    |> Seq.toArray
Run Code Online (Sandbox Code Playgroud)

如果您运行该代码,您将看到它从服务器一次提取一个结果,并且不会留下未完成的请求.此外,更改要检索的结果数量非常容易 - 您只需更改传递的值即可Seq.take.

现在虽然流代码工作得很好,但它不会并行执行请求,因此对于大量文档来说可能会很慢.这是一个很容易解决的问题,尽管解决方案可能有点不直观.而不是尝试并行执行整个请求序列 - 这是原始代码中的问题 - 让我们创建一个用于并行Async.Parallel执行小批量请求的函数,然后用于Seq.collect将结果组合回一个平坦的序列.

/// Given a sequence of Async<'T>, creates a new sequence whose elements
/// are computed in batches of a specified size.
let parallelBatch batchSize (sequence : seq<Async<'T>>) =
    sequence
    |> Seq.windowed batchSize
    |> Seq.collect (fun batch ->
        batch
        |> Async.Parallel
        |> Async.RunSynchronously)
Run Code Online (Sandbox Code Playgroud)

要使用此功能,我们只需要对流版本的代码进行一些小的调整:

let en100_Batched =
    let batchSize = 10
    #if DEBUG
    let documentCount = ref 0
    #endif

    Uri ("http://en.wikipedia.org/wiki/Special:Random")
    |> createDocumentSeq
    // Execute batches in parallel
    |> parallelBatch batchSize
    |> Seq.choose (Option.bind (parseDoc >> parseArticle))
    #if DEBUG
    |> Seq.map (fun x ->
        let msg =
            Interlocked.Increment documentCount
            |> sprintf "Parsed documents: %i"
        synchedOut.WriteLine msg
        x)
    #endif
    |> Seq.take 50
    // None of the computations actually take place until
    // this point, because Seq.toArray forces evaluation of the sequence.
    |> Seq.toArray
Run Code Online (Sandbox Code Playgroud)

同样,您可以轻松更改要检索的文档数量,并且可以轻松修改批量大小(同样,我建议您将其保持相当小).如果你愿意,你可以对'流'和'批处'代码做一些调整,这样你就可以在运行时切换它们.

最后一件事 - 我的代码请求不应该超时,所以你可以摆脱这个WebClientWithTimeout类,直接使用WebClient.

归档时间：

查看次数：

1029 次

最近记录：

13 年，4 月 前
如何在F#Async.Parallel中处理HttpWebRequest超时 8
F#中的"限制"异步下载 7
更多相关链接
相关归档
如何取消长时间运行的数据库操作？ 26
超时使用TaskCompletionSource实现的异步方法 19
为什么"\ u1FFF:foo".StartsWith(":")返回true？ 8
如何通过NuGet分发F#类型提供程序？ 6
在 Chrome ajax 异步回发端不起作用 5
异步/等待的呼叫链...等待等待或返回等待？ 5
自定义展开返回累加器 4
为什么这个异步/等待代码不会导致死锁？ 4
在函数方法中解决继承依赖性 2
尽管等待 API 响应，但使用 asyncio 没有加速 2
难疑归档
如何在JavaScript中获取查询字符串值？ 2701
如何使用CSS垂直居中文本？ 2190
使用Git下载特定标签 1892
在SQL表中查找重复值 1789
静态类和单例模式之间的区别？ 1708
使用JavaScript/jQuery滚动到页面顶部？ 1511
如何在正则表达式中使用变量？ 1250
LF将被git中的CRLF取代 - 这是什么,它是否重要？ 1146
在React JSX中循环 1131
.gitignore for Visual Studio项目和解决方案 1115