Apakah Parallel.ForEach membatasi jumlah utas aktif?

107

Diberikan kode ini:

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

Akankah 1000 utas muncul hampir secara bersamaan?

Jader Dias
sumber

Jawaban:

149

Tidak, itu tidak akan memulai 1000 utas - ya, itu akan membatasi berapa banyak utas yang digunakan. Ekstensi Paralel menggunakan jumlah inti yang sesuai, berdasarkan berapa banyak yang Anda miliki secara fisik dan berapa banyak yang sudah sibuk. Ini mengalokasikan pekerjaan untuk setiap inti dan kemudian menggunakan teknik yang disebut mencuri pekerjaan untuk membiarkan setiap utas memproses antriannya sendiri secara efisien dan hanya perlu melakukan akses lintas-utas yang mahal saat diperlukan.

Silahkan lihat di Tim PFX Blog untuk banyak informasi tentang bagaimana mengalokasikan pekerjaan dan segala macam topik lainnya.

Perhatikan bahwa dalam beberapa kasus Anda juga dapat menentukan derajat paralelisme yang Anda inginkan.

Jon Skeet
sumber
2
Saya menggunakan Parallel.ForEach (FilePathArray, path => ... untuk membaca sekitar 24.000 file malam ini membuat satu file baru untuk setiap file yang saya baca. Kode yang sangat sederhana. Tampaknya bahkan 6 utas sudah cukup untuk membanjiri disk 7200 RPM Saya membaca dari pada pemanfaatan 100%. Selama beberapa jam saya menyaksikan perpustakaan Paralel memutar lebih dari 8.000 utas. Saya menguji menggunakan MaxDegreeOfParallelism dan cukup yakin 8000+ utas menghilang. Saya telah mengujinya beberapa kali sekarang dengan yang sama Hasilnya.
Jake Drew
Itu bisa memulai 1000 utas untuk beberapa 'DoSomething' yang merosot. (Seperti dalam kasus di mana saya saat ini berurusan dengan masalah dalam kode produksi yang gagal untuk menetapkan batas dan menelurkan 200+ utas sehingga memunculkan kumpulan koneksi SQL .. Saya sarankan mengatur Max DOP untuk pekerjaan apa pun yang tidak dapat dianggap remeh alasannya tentang sebagai terikat secara eksplisit dengan CPU.)
user2864740
28

Pada mesin inti tunggal ... Parallel.ForEach partisi (potongan) koleksi yang dikerjakannya di antara sejumlah utas, tetapi jumlah itu dihitung berdasarkan algoritme yang memperhitungkan dan tampaknya terus memantau pekerjaan yang dilakukan oleh utas yang dialokasikan ke ForEach. Jadi jika bagian tubuh ForEach memanggil fungsi IO-terikat / pemblokiran yang berjalan lama yang akan membuat utas menunggu, algoritme akan menelurkan lebih banyak utas dan mempartisi ulang koleksi di antara mereka . Jika utas selesai dengan cepat dan tidak memblokir utas IO misalnya, seperti hanya menghitung beberapa angka,algoritme akan meningkatkan (atau memang menurunkan) jumlah utas ke titik di mana algoritme menganggap optimal untuk throughput (waktu penyelesaian rata-rata dari setiap iterasi) .

Pada dasarnya kumpulan utas di belakang semua fungsi pustaka Paralel, akan menghasilkan jumlah utas yang optimal untuk digunakan. Jumlah inti prosesor fisik hanya membentuk sebagian dari persamaan. BUKAN ada hubungan satu lawan satu yang sederhana antara jumlah inti dan jumlah utas yang muncul.

Saya tidak menemukan dokumentasi seputar pembatalan dan penanganan utas sinkronisasi sangat membantu. Semoga MS dapat memberikan contoh yang lebih baik di MSDN.

Jangan lupa, kode tubuh harus ditulis untuk dijalankan di banyak utas, bersama dengan semua pertimbangan keamanan utas yang biasa, kerangka kerja belum mengabstraksi faktor itu ... belum.

Pengembang Microsoft
sumber
1
".. jika bagian tubuh ForEach memanggil fungsi pemblokiran yang berjalan lama yang akan membuat utas menunggu, algoritme akan memunculkan lebih banyak utas .." - Dalam kasus yang merosot, ini berarti mungkin ada banyak utas yang dibuat sebanyak yang diizinkan per ThreadPool.
pengguna2864740
2
Anda benar, untuk IO, ini dapat mengalokasikan +100 utas saat saya
men-
5

Ini menghasilkan jumlah utas yang optimal berdasarkan jumlah prosesor / inti. Mereka tidak akan bertelur sekaligus.

Colin Mackay
sumber
4

Pertanyaan bagus. Dalam contoh Anda, tingkat paralelisasi cukup rendah bahkan pada prosesor quad core, tetapi dengan beberapa menunggu, tingkat paralelisasi bisa menjadi cukup tinggi.

// 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());
}

Sekarang lihat apa yang terjadi ketika operasi menunggu ditambahkan untuk mensimulasikan permintaan 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());
}

Saya belum membuat perubahan apa pun dan level konkurensi / paralelisasi telah melonjak secara drastis. Konkurensi dapat ditingkatkan batasnya dengan 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());
}

Saya merekomendasikan pengaturan ParallelOptions.MaxDegreeOfParallelism . Ini tidak serta merta meningkatkan jumlah utas yang digunakan, tetapi ini akan memastikan Anda hanya memulai jumlah utas yang masuk akal, yang tampaknya menjadi perhatian Anda.

Terakhir untuk menjawab pertanyaan Anda, tidak, Anda tidak akan memulai semua utas sekaligus. Gunakan Parallel.Invoke jika Anda ingin memanggil secara paralel dengan sempurna, misalnya menguji kondisi balapan.

// 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)));
}
Timothy Gonzalez
sumber