¿Parallel.ForEach limita el número de subprocesos activos?

Question

Teniendo en cuenta este código:

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

1000 ¿Todos los hilos desovar casi al mismo tiempo?

Answer 1

No, no comenzará 1000 hilos - sí, limitará cuántos hilos se utilizan. Paralelo Extensiones utiliza una cantidad adecuada de núcleos, en función de cuántos físicamente tiene y cuántos ya están ocupados. Se asigna el trabajo para cada núcleo y luego utiliza una técnica llamada robo de trabajo para permitir que cada subproceso procese su propia cola de manera eficiente y solo tenga que realizar un costoso acceso cruzado cuando realmente lo necesite.

Eche un vistazo a PFX Team Blog para cargas de información acerca de cómo asigna el trabajo y todo tipo de otros temas.

Tenga en cuenta que en algunos casos también puede especificar el grado de paralelismo que desea.

Answer 2

Funciona una cantidad óptima de subprocesos según la cantidad de procesadores/núcleos. No todos engendrarán a la vez.

Answer 3

Consulte Does Parallel.For use one Task per iteration? para obtener una idea de un "modelo mental" para usar. Sin embargo, el autor afirma que "Al final del día, es importante recordar que los detalles de implementación pueden cambiar en cualquier momento".

Answer 4

En una máquina de núcleo único ... Parallel.ForEach particiones (fragmentos) de la colección que está trabajando entre una serie de subprocesos, pero ese número se calcula en función de un algoritmo que tiene en cuenta y parece supervisar continuamente el trabajo realizado por los hilos que está asignando a ForEach. Así que si la parte del cuerpo de ForEach llama a funciones de larga duración IO-bound/blocking que dejarían el hilo en espera, el algoritmo generará más hilos y reparticionará la colección entre ellos. Si los hilos se completan rápidamente y no se bloquean en hilos IO, por ejemplo, simplemente calculando algunos números, el algoritmo incrementará (o disminuirá) el número de hilos hasta un punto donde el algoritmo considere óptimo para el rendimiento (promedio tiempo de finalización de cada iteración).

Básicamente, el grupo de subprocesos detrás de todas las diversas funciones de la biblioteca Paralelo, resolverá un número óptimo de subprocesos para usar. La cantidad de núcleos de procesador físico forma solo parte de la ecuación. NO hay una relación simple de uno a uno entre la cantidad de núcleos y la cantidad de hilos engendrados.

No me resulta muy útil la documentación sobre la cancelación y el manejo de los hilos de sincronización. Es de esperar que MS pueda proporcionar mejores ejemplos en MSDN.

No olvide que el código del cuerpo debe escribirse para ejecutarse en varios hilos, junto con todas las consideraciones habituales de seguridad del hilo, el marco no resume ese factor ... todavía.

Answer 5

Una gran pregunta. En su ejemplo, el nivel de paralelización es bastante bajo incluso en un procesador de cuatro núcleos, pero con un poco de espera, el nivel de paralelización puede ser bastante 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()); 
} 

Ahora mire lo que sucede cuando se agrega una operación de espera para simular una solicitud 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()); 
} 

Todavía no he realizado ningún cambio y el nivel de simultaneidad/paralelización ha aumentado drásticamente. La concurrencia puede tener su límite aumentado 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()); 
} 

Recomendar configuración ParallelOptions.MaxDegreeOfParallelism. No necesariamente aumentará el número de subprocesos en uso, pero garantizará que solo comience una cantidad de subprocesos, lo cual parece ser su preocupación.

Por último, para responder a su pregunta, no obtendrá todos los hilos para comenzar de una vez. Use Parallel.Invoke si desea invocar en paralelo a la perfección, p. probando las condiciones de carrera.

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