Làm thế nào để giới hạn số lượng các chủ đề được tạo cho một hoạt động Seq.map không đồng bộ trong F #?

Question

Các thiết lập hiện tại đi một cái gì đó như thế này

array 
|> Seq.map (fun item -> async { return f item}) 
|> Async.Parallel 
|> Async.RunSynchronously 

Vấn đề là, điều này có xu hướng tạo ra quá nhiều chủ đề và sụp đổ các ứng dụng theo định kỳ.

Cách giới hạn số lượng chủ đề trong trường hợp này (ví dụ: Environment.ProcessorCount)?

Answer 1

Nếu bạn muốn parallelize tính CPU-chuyên sâu mà phải mất một mảng (hoặc chuỗi có) như một đầu vào, sau đó nó có thể là một ý tưởng tốt hơn để sử dụng PSeq mô-đun từ F# PowerPack (mà chỉ có sẵn trên NET 4.0 mặc dù). Nó cung cấp một phiên bản song song của nhiều chức năng tiêu chuẩn Array.xyz. Để biết thêm thông tin, bạn cũng có thể xem F# translation trong số Parallel Programming with .NET mẫu.

Mã để giải quyết vấn đề của bạn sẽ là một chút đơn giản hơn so với sử dụng quy trình công việc:

array |> PSeq.map f 
     |> PSeq.toArray 

Một số khác biệt giữa hai tùy chọn là:

• PSeq được tạo ra sử dụng Task Song song Library (TPL) từ .NET 4.0, được tối ưu hóa để làm việc với một số lượng lớn các tác vụ đòi hỏi nhiều CPU.
• Async được triển khai trong thư viện F # và hỗ trợ các hoạt động không đồng bộ (không chặn) như I/O trong các hoạt động đồng thời chạy.

Tóm lại, nếu bạn cần các hoạt động không đồng bộ (ví dụ: I/O) thì Async là tùy chọn tốt nhất. Nếu bạn có một số lượng lớn các tác vụ đòi hỏi nhiều CPU, thì PSeq có thể là lựa chọn tốt hơn (trên .NET 4.0)

Answer 2

Có một vài điều bạn có thể làm.

Đầu tiên, vì điều này sử dụng số ThreadPool, bạn có thể sử dụng ThreadPool.SetMaxThreads.

Thứ hai, bạn có thể giới thiệu ga riêng của bạn cùng những dòng này:

let throttle = makeThrottle(8) 
array 
|> Seq.map (fun item -> async { do! throttle.Wait() 
           return f item}) 
|> Async.Parallel 
|> Async.RunSynchronously 

makeThrottle() sẽ không quá khó để viết, nhưng nó sẽ phải chịu một chút đồng bộ trên cao. Nếu bạn đang cố gắng để song song rất nhiều thứ mà bạn đang hết bộ nhớ, chi phí ga có thể là một vấn đề không. (Hãy cho tôi biết nếu bạn cần mẫu cho loại mã này.)

Cuối cùng, nếu điều này thực sự gây ra sự cố, có vẻ như bạn có thể đang làm điều gì sai. Các ThreadPool thường (nhưng không phải luôn luôn) hiện một công việc tốt quản lý chính nó. Nhưng trong các trường hợp khác nhau, thiết kế ga của riêng bạn có thể có giá trị cho ứng dụng của bạn anyway.

Answer 3

Đây là một ví dụ làm việc về cách thực hiện điều này bằng cách sử dụng Semaphore theo tinh thần của đề xuất của Brian:

open System 

let throttle n fs = 
    seq { let n = new Threading.Semaphore(n, n) 
      for f in fs -> 
       async { let! ok = Async.AwaitWaitHandle(n) 
         let! result = Async.Catch f 
         n.Release() |> ignore 
         return match result with 
          | Choice1Of2 rslt -> rslt 
          | Choice2Of2 exn -> raise exn 
        } 
     } 

let f i = async { printfn "start %d" i 
        do! Async.Sleep(2000) 
       } 
let fs = Seq.init 10 f 

fs |> throttle 2 |> Async.Parallel |> Async.RunSynchronously |> ignore