Hành vi không nhất quán khi hủy các loại Asyncs khác nhau

Question

Tôi gặp sự cố với hành vi có vẻ không nhất quán khi hủy các loại Asyncs khác nhau.

Để tái tạo sự cố, giả sử có một hàm lấy danh sách "công việc" (Async < _> list), chờ cho chúng hoàn tất và in kết quả của chúng. Các chức năng cũng được một mã thông báo hủy để nó có thể bị hủy bỏ:

let processJobs jobs cancel = 
    Async.Start(async { 
    try 
     let! results = jobs |> Async.Parallel 
     printfn "%A" results 
    finally 
     printfn "stopped" 
    }, cancel) 

các hàm được gọi như thế:

let jobs = [job1(); job2(); job3(); job4(); job5()] 
use cancel = new CancellationTokenSource() 

processJobs jobs cancel.Token 

Và phần nào sau đó bị hủy:

Thread.Sleep(1000) 
printfn "cancelling..." 
cancel.Cancel() 

Khi hủy mã thông báo nguồn bị hủy, chức năng sẽ thực thi khối cuối cùng và in "dừng".

Điều đó làm việc tốt cho công việc1, 2 và 3, nhưng không hoạt động khi có công việc4 hoặc job5 trong danh sách.

Job1 chỉ Async.Sleeps:

let job1() = async { 
    do! Async.Sleep 1000000 
    return 10 
} 

Job2 bắt đầu một số Childs async và chờ đợi cho họ:

let job2() = async { 
    let! child1 = Async.StartChild(async { 
    do! Async.Sleep 1000000 
    return 10 
    }) 

    let! child2 = Async.StartChild(async { 
    do! Async.Sleep 1000000 
    return 10 
    }) 

    let! results = [child1; child2] |> Async.Parallel 
    return results |> Seq.sum 
} 

Job3 chờ đợi đối với một số xử lý chờ đợi xấu xí đó là thiết lập bởi một số chuỗi thậm chí còn xấu hơn:

let job3() = async { 
    use doneevent = new ManualResetEvent(false) 

    let thread = Thread(fun() -> Thread.Sleep(1000000); doneevent.Set() |> ignore) 
    thread.Start() 

    do! Async.AwaitWaitHandle(doneevent :> WaitHandle) |> Async.Ignore 

    return 30 
} 

Job4 bài viết đến và chờ đợi một câu trả lời từ một MailboxProcessor:

let job4() = async { 
    let worker = MailboxProcessor.Start(fun inbox -> async { 
    let! (msg:AsyncReplyChannel<int>) = inbox.Receive() 
    do! Async.Sleep 1000000 
    msg.Reply 40 
    }) 

    return! worker.PostAndAsyncReply (fun reply -> reply) // <- cannot cancel this 
} 

Job5 chờ đợi cho một Task (hoặc TaskCompletionSource):

let job5() = async { 
    let tcs = TaskCompletionSource<int>() 

    Async.Start(async { 
    do! Async.Sleep 1000000 
    tcs.SetResult 50 
    }) 

    return! (Async.AwaitTask tcs.Task) // <- cannot cancel this 
} 

Tại sao Job1, 2 và 3 bị hủy bỏ ("dừng lại" được in), trong khi Job4 và 5 làm cho chức năng treo "mãi mãi"?

Cho đến nay tôi luôn dựa vào F # để xử lý việc hủy bỏ hậu trường - miễn là tôi đang sử dụng các khối không đồng bộ và sử dụng! (Let !, do !, return!, ...) mọi thứ phải là tốt .. nhưng đó không phải là trường hợp tất cả các thời gian.

Quote:

Trong F # công việc không đồng bộ, các đối tượng CancellationToken được truyền xung quanh tự động dưới vỏ bọc. Điều này có nghĩa là chúng tôi không phải làm bất kỳ điều gì đặc biệt để hỗ trợ hủy. Khi chạy không đồng bộ quy trình làm việc, chúng tôi có thể cung cấp mã thông báo hủy và mọi thứ sẽ hoạt động tự động .

mã hoàn chỉnh có sẵn ở đây: http://codepad.org/euVO3xgP

EDIT

tôi nhận thấy rằng đường ống một async qua Async.StartAsTask tiếp theo Async.AwaitTask làm cho nó hủy ngang trong mọi trường hợp.

ví dụ cho Job4 đó có nghĩa là thay đổi dòng:

return! worker.PostAndAsyncReply (fun reply -> reply) 

tới:

return! cancelable <| worker.PostAndAsyncReply (fun reply -> reply) 

Với con người hủy ngang:

let cancelable (x:Async<_>) = async { 
    let! cancel = Async.CancellationToken 
    return! Async.StartAsTask(x, cancellationToken = cancel) |> Async.AwaitTask 
} 

Các công trình tương tự cho làm Job5 hủy ngang.

Nhưng .. đó chỉ là giải pháp thay thế và tôi khó có thể đặt xung quanh mỗi cuộc gọi đến một Unync không rõ < _>.

Answer 1

Chỉ có Async. các phương thức xử lý bằng cách sử dụng chính CancellationToken mặc định.

Trong MailboxProcessor dụ của bạn hủy nên đi vào phương pháp Bắt đầu

let! ct= Async.CancellationToken 
use worker := MailboxProcessor.Start(theWork, ct) 

Trong ví dụ TaskCompletionSource, bạn sẽ phải đăng ký một callback để hủy bỏ nó.

let! ct = Async.CancellationToken 
use canceler = ct.Register(fun() -> tcs.TrySetCanceled() |> ignore)