Parallel.ForEach limita il numero di thread attivi?

Question

Dato questo codice:

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

saranno tutte le discussioni 1000 deporre le uova quasi contemporaneamente?

Answer 1

No, non inizierà 1000 thread - sì, limiterà il numero di thread utilizzati. Parallel Extensions utilizza un numero appropriato di core, in base al numero di persone che hanno fisicamente e quante ne sono già occupate. Alloca il lavoro per ogni core e quindi utilizza una tecnica denominata per il furto del lavoro per consentire a ogni thread di elaborare la propria coda in modo efficiente e deve solo eseguire un costoso accesso cross-thread quando è necessario.

Dai un'occhiata allo PFX Team Blog per carichi di informazioni su come esso assegna il lavoro e tutti i tipi di altri argomenti.

Si noti che in alcuni casi è possibile specificare anche il grado di parallelismo desiderato.

Answer 2

Elabora un numero ottimale di thread in base al numero di processori/core. Non si genereranno tutti in una volta.

Answer 3

Vedere Does Parallel.For use one Task per iteration? per un'idea di un "modello mentale" da utilizzare. Tuttavia, l'autore afferma che "Alla fine della giornata, è importante ricordare che i dettagli di implementazione possono cambiare in qualsiasi momento."

Answer 4

Su un computer single core ... Parallel.ForOgni partizione (blocchi) della raccolta su cui sta lavorando tra un numero di thread, ma tale numero viene calcolato in base a un algoritmo che tiene conto e sembra monitorare continuamente il lavoro svolto dai thread che sta allocando a ForEach. Quindi, se la parte del corpo di ForEach richiama le funzioni di blocco/interdipendenza di lunga durata che lascerebbero il thread in attesa, l'algoritmo genererà più thread e ripartirà la raccolta tra loro. Ad esempio, se i thread vengono completati rapidamente e non si bloccano i thread IO, ad esempio semplicemente il calcolo di alcuni numeri, , l'algoritmo aumenterà (o addirittura diminuirà) il numero di thread in un punto in cui l'algoritmo considera ottimale il throughput (media tempo di completamento di ogni iterazione).

Fondamentalmente il pool di thread dietro tutte le varie funzioni della libreria parallela, determinerà un numero ottimale di thread da utilizzare. Il numero di core del processore fisico costituisce solo una parte dell'equazione. NON c'è una semplice relazione uno a uno tra il numero di core e il numero di thread generati.

Non trovo molto utile la documentazione relativa alla cancellazione e alla gestione dei thread di sincronizzazione. Speriamo che la MS possa fornire esempi migliori in MSDN.

Non dimenticare, il codice corpo deve essere scritto per essere eseguito su più thread, insieme con tutte le solite considerazioni sulla sicurezza dei thread, il framework non astragga quel fattore ... ancora.

Answer 5

Ottima domanda. Nel tuo esempio, il livello di parallelizzazione è piuttosto basso anche su un processore quad core, ma con un po 'di attesa il livello di parallelizzazione può diventare piuttosto alto.

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

Ora guarda cosa succede quando viene aggiunta un'operazione di attesa per simulare una richiesta 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()); 
} 

Non ho ancora apportato alcuna modifica e il livello di concorrenza/parallelizzazione è aumentato drammaticamente. La concorrenza può aumentare il suo limite con 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()); 
} 

Si consiglia di impostare ParallelOptions.MaxDegreeOfParallelism. Non aumenterà necessariamente il numero di thread in uso, ma ti garantirà solo l'avvio di un numero ragionevole di thread, che sembra essere la tua preoccupazione.

Infine, per rispondere alla domanda, non è possibile avviare tutte le discussioni contemporaneamente. Usa Parallel.Invoke se stai cercando di invocare in parallelo perfettamente ad es. testare le condizioni della gara.

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