Rào cản không đồng bộ trong F #

Question

Tôi đã viết một chương trình trong F # liệt kê tất cả các thư mục trên đĩa không đồng bộ. Một tác vụ async liệt kê tất cả các tệp trong một thư mục đã cho và tạo các nhiệm vụ không đồng bộ riêng biệt (các trình tiện ích: Tôi bắt đầu chúng bằng cách sử dụng Async.Start) để liệt kê các thư mục con. Tất cả chúng đều truyền đạt kết quả tới trung tâm MailboxProcessor.

Vấn đề của tôi là, làm thế nào để phát hiện ra rằng tất cả các nhiệm vụ daemon đã kết thúc và sẽ không có tệp nào khác đến. Về cơ bản, tôi cần một rào cản cho tất cả các nhiệm vụ là trẻ em trực tiếp và gián tiếp của nhiệm vụ hàng đầu của tôi. Tôi không thể tìm thấy bất cứ điều gì như thế trong mô hình không đồng bộ của F #.

Điều tôi đã làm thay vào đó là tạo một Trình xử lý hộp thư riêng biệt, nơi tôi đăng ký bắt đầu và kết thúc của mỗi tác vụ. Khi số đếm hoạt động chuyển sang không, tôi đã hoàn thành. Nhưng tôi không hài lòng với giải pháp đó. Bất cứ một đề nghị nào khác?

Answer 1

Bạn đã thử sử dụng Async.Parallel? Tức là, thay vì Async.Start mỗi thư mục con, chỉ cần kết hợp các tác vụ thư mục con vào một async đơn qua Async.Parallel. Sau đó, bạn kết thúc với một (nested) ngã ba tham gia nhiệm vụ mà bạn có thể RunSynchronously và chờ kết quả cuối cùng.

EDIT

Dưới đây là một số mã gần đúng, cho thấy các ý chính, nếu không phải là đầy đủ chi tiết:

open System.IO 

let agent = MailboxProcessor.Start(fun mbox -> 
    async { 
     while true do 
      let! msg = mbox.Receive() 
      printfn "%s" msg 
    }) 

let rec traverse dir = 
    async { 
     agent.Post(dir) 
     let subDirs = Directory.EnumerateDirectories(dir) 
     return! [for d in subDirs do yield traverse d] 
       |> Async.Parallel |> Async.Ignore 
    } 

traverse "d:\\" |> Async.RunSynchronously 
// now all will be traversed, 
// though Post-ed messages to agent may still be in flight 

EDIT 2

Dưới đây là phiên bản chờ đợi sử dụng trả lời:

open System.IO 

let agent = MailboxProcessor.Start(fun mbox -> 
    async { 
     while true do 
      let! dir, (replyChannel:AsyncReplyChannel<unit>) = mbox.Receive() 
      printfn "%s" dir 
      replyChannel.Reply() 
    }) 

let rec traverse dir = 
    async { 
     let r = agent.PostAndAsyncReply(fun replyChannel -> dir, replyChannel) 
     let subDirs = Directory.EnumerateDirectories(dir) 
     do! [for d in subDirs do yield traverse d] 
       |> Async.Parallel |> Async.Ignore 
     do! r // wait for Post to finish 
    } 

traverse "c:\\Projects\\" |> Async.RunSynchronously 
// now all will be traversed to completion 

Answer 2

Bạn chỉ có thể sử dụng Interlocked để tăng và giảm khi bạn bắt đầu/kết thúc nhiệm vụ và hoàn thành tất cả khi đến 0. Tôi đã sử dụng chiến lược này trong mã tương tự với MailboxProcessors.

Answer 3

Có thể bạn nên sử dụng Task.Factory.StartNew() và Task.WaitAll().

Answer 4

Đây có lẽ là một bài tập học tập, nhưng có vẻ như bạn sẽ hài lòng với một danh sách lười biếng của tất cả các tệp. Ăn cắp từ câu trả lời của Brian ở trên ... (và tôi nghĩ rằng một cái gì đó như thế này là trong tất cả các F # sách, mà tôi không có với tôi ở nhà)

open System.IO 

let rec traverse dir = 
seq { 
    let subDirs = Directory.EnumerateDirectories(dir) 
    yield dir 
    for d in subDirs do 
     yield! traverse d 

} 

Đối với những gì nó có giá trị, tôi đã tìm thấy các công việc Async trong F # rất hữu ích cho các vấn đề song song "dễ dàng xấu hổ", mặc dù tôi đã không cố gắng đa nhiệm đa nhiệm nói chung.

Answer 5

Chỉ cần làm rõ: Tôi nghĩ có thể có một giải pháp tốt hơn tương tự như những gì người ta có thể làm trong Chapel. Ở đó bạn có một tuyên bố "đồng bộ", một rào cản chờ đợi cho tất cả các nhiệm vụ sinh ra trong một tuyên bố để kết thúc. Dưới đây là một ví dụ từ cuốn hướng dẫn Chapel:

def concurrentUpdate(tree: Tree) { 
    if requiresUpdate(tree) then 
     begin update(tree); 
    if !tree.isLeaf { 
     concurrentUpdate(tree.left); 
     concurrentUpdate(tree.right); 
    } 
} 
sync concurrentUpdate(tree); 

Các "bắt đầu" tuyên bố tạo ra một nhiệm vụ được chạy song song, phần nào tương tự như F # "async" khối với Async.Start.