Parallel.ForEach có giới hạn số lượng chủ đề đang hoạt động không?

Question

Với mã này:

var arrayStrings = new string[1000]; 
Parallel.ForEach<string>(arrayStrings, someString => 
{ 
    DoSomething(someString); 
}); 

tất cả 1000 chủ đề sẽ đẻ trứng gần như cùng một lúc?

Answer 1

Không, nó sẽ không bắt đầu 1000 luồng - có, nó sẽ giới hạn số lượng chuỗi được sử dụng. Tiện ích mở rộng song song sử dụng số lượng lõi phù hợp, dựa trên số lượng thực tế bạn có và số lượng đã bận. Nó phân bổ công việc cho mỗi lõi và sau đó sử dụng một kỹ thuật được gọi là đánh cắp công việc để cho mỗi luồng xử lý hàng đợi của riêng nó một cách hiệu quả và chỉ cần thực hiện bất kỳ truy cập chéo luồng đắt tiền nào khi thực sự cần.

Hãy xem PFX Team Blog cho tải thông tin về cách phân bổ công việc và tất cả các loại chủ đề khác.

Lưu ý rằng trong một số trường hợp, bạn cũng có thể chỉ định mức độ song song mà bạn muốn.

Answer 2

Nó hoạt động với số lượng chủ đề tối ưu dựa trên số lượng bộ xử lý/lõi. Họ sẽ không phải tất cả đẻ trứng cùng một lúc.

Answer 3

Xem Does Parallel.For use one Task per iteration? để biết ý tưởng về "mô hình tinh thần" để sử dụng. Tuy nhiên tác giả nói rằng "Vào cuối ngày, điều quan trọng cần nhớ là chi tiết triển khai có thể thay đổi bất kỳ lúc nào".

Answer 4

Trên một máy lõi đơn ... Parallel.ForEach phân vùng (khối) của bộ sưu tập nó làm việc trên một số chủ đề, nhưng con số đó được tính toán dựa trên một thuật toán có tính đến và xuất hiện liên tục theo dõi công việc được thực hiện bởi các chủ đề nó được phân bổ cho ForEach. Vì vậy, nếu phần cơ thể của ForEach gọi ra để chạy IO-bound/chặn chức năng dài sẽ rời khỏi thread chờ đợi xung quanh, thuật toán sẽ sinh ra nhiều chủ đề hơn và phân vùng lại bộ sưu tập giữa chúng. Nếu các chủ đề hoàn thành nhanh chóng và không chặn trên các chuỗi IO ví dụ, chẳng hạn như chỉ cần tính toán một số, thuật toán sẽ tăng lên (hoặc thực sự xuống) số lượng chủ đề đến một điểm mà thuật toán xem xét tối ưu cho thông lượng (trung bình thời gian hoàn thành của mỗi lần lặp).

Về cơ bản, luồng chủ đề đằng sau tất cả các chức năng thư viện song song khác nhau, sẽ tính ra số lượng chủ đề tối ưu để sử dụng. Số lõi xử lý vật lý chỉ là một phần của phương trình. KHÔNG có mối quan hệ một - một đơn giản giữa số lõi và số lượng các luồng được sinh ra.

Tôi không tìm thấy tài liệu xung quanh việc hủy và xử lý các chuỗi đồng bộ hóa rất hữu ích. Hy vọng rằng MS có thể cung cấp các ví dụ tốt hơn trong MSDN.

Đừng quên, mã cơ thể phải được viết để chạy trên nhiều luồng, cùng với tất cả các cân nhắc an toàn chủ đề thông thường, khung công tác không trừu tượng yếu tố đó ... được nêu ra.

Answer 5

Câu hỏi hay. Trong ví dụ của bạn, mức độ song song là khá thấp ngay cả trên một bộ xử lý lõi tứ, nhưng với một số chờ đợi mức độ song song có thể nhận được khá cao.

// Max concurrency: 5 
[Test] 
public void Memory_Operations() 
{ 
    ConcurrentBag<int> monitor = new ConcurrentBag<int>(); 
    ConcurrentBag<int> monitorOut = new ConcurrentBag<int>(); 
    var arrayStrings = new string[1000]; 
    Parallel.ForEach<string>(arrayStrings, someString => 
    { 
     monitor.Add(monitor.Count); 
     monitor.TryTake(out int result); 
     monitorOut.Add(result); 
    }); 

    Console.WriteLine("Max concurrency: " + monitorOut.OrderByDescending(x => x).First()); 
} 

Bây giờ hãy xem điều gì sẽ xảy ra khi hoạt động chờ được thêm vào để mô phỏng yêu cầu HTTP.

// Max concurrency: 34 
[Test] 
public void Waiting_Operations() 
{ 
    ConcurrentBag<int> monitor = new ConcurrentBag<int>(); 
    ConcurrentBag<int> monitorOut = new ConcurrentBag<int>(); 
    var arrayStrings = new string[1000]; 
    Parallel.ForEach<string>(arrayStrings, someString => 
    { 
     monitor.Add(monitor.Count); 

     System.Threading.Thread.Sleep(1000); 

     monitor.TryTake(out int result); 
     monitorOut.Add(result); 
    }); 

    Console.WriteLine("Max concurrency: " + monitorOut.OrderByDescending(x => x).First()); 
} 

Tôi chưa thực hiện bất kỳ thay đổi nào và mức độ đồng thời/song song đã tăng đáng kể. Đồng thời có thể tăng giới hạn của nó với ParallelOptions.MaxDegreeOfParallelism.

// Max concurrency: 43 
[Test] 
public void Test() 
{ 
    ConcurrentBag<int> monitor = new ConcurrentBag<int>(); 
    ConcurrentBag<int> monitorOut = new ConcurrentBag<int>(); 
    var arrayStrings = new string[1000]; 
    var options = new ParallelOptions {MaxDegreeOfParallelism = int.MaxValue}; 
    Parallel.ForEach<string>(arrayStrings, options, someString => 
    { 
     monitor.Add(monitor.Count); 

     System.Threading.Thread.Sleep(1000); 

     monitor.TryTake(out int result); 
     monitorOut.Add(result); 
    }); 

    Console.WriteLine("Max concurrency: " + monitorOut.OrderByDescending(x => x).First()); 
} 

// Max concurrency: 391 
[Test] 
public void Test() 
{ 
    ConcurrentBag<int> monitor = new ConcurrentBag<int>(); 
    ConcurrentBag<int> monitorOut = new ConcurrentBag<int>(); 
    var arrayStrings = new string[1000]; 
    var options = new ParallelOptions {MaxDegreeOfParallelism = int.MaxValue}; 
    Parallel.ForEach<string>(arrayStrings, options, someString => 
    { 
     monitor.Add(monitor.Count); 

     System.Threading.Thread.Sleep(100000); 

     monitor.TryTake(out int result); 
     monitorOut.Add(result); 
    }); 

    Console.WriteLine("Max concurrency: " + monitorOut.OrderByDescending(x => x).First()); 
} 

tôi reccommend thiết ParallelOptions.MaxDegreeOfParallelism. Nó sẽ không nhất thiết phải tăng số lượng các chủ đề được sử dụng, nhưng nó sẽ đảm bảo bạn chỉ bắt đầu một số sane của chủ đề, mà có vẻ là mối quan tâm của bạn.

Cuối cùng để trả lời câu hỏi của bạn, không có bạn sẽ không nhận được tất cả các chủ đề để bắt đầu cùng một lúc. Sử dụng Parallel.Invoke nếu bạn đang tìm kiếm để gọi song song hoàn hảo ví dụ: điều kiện cuộc đua thử nghiệm.

// 636462943623363344 
// 636462943623363344 
// 636462943623363344 
// 636462943623363344 
// 636462943623363344 
// 636462943623368346 
// 636462943623368346 
// 636462943623373351 
// 636462943623393364 
// 636462943623393364 
[Test] 
public void Test() 
{ 
    ConcurrentBag<string> monitor = new ConcurrentBag<string>(); 
    ConcurrentBag<string> monitorOut = new ConcurrentBag<string>(); 
    var arrayStrings = new string[1000]; 
    var options = new ParallelOptions {MaxDegreeOfParallelism = int.MaxValue}; 
    Parallel.ForEach<string>(arrayStrings, options, someString => 
    { 
     monitor.Add(DateTime.UtcNow.Ticks.ToString()); 
     monitor.TryTake(out string result); 
     monitorOut.Add(result); 
    }); 

    var startTimes = monitorOut.OrderBy(x => x.ToString()).ToList(); 
    Console.WriteLine(string.Join(Environment.NewLine, startTimes.Take(10))); 
}