Manera más eficiente de procesar una cola con hilos

Question

Tengo una cola en la que se colocan las solicitudes de transformación de Fourier pendientes (operaciones comparativamente lentas), podríamos obtener miles de solicitudes de transformación por segundo en algunos casos, por lo que debe ser rápido.

Estoy actualizando el código anterior para usar .net 4, así como la migración a TPL. Me pregunto cómo es la forma más eficiente (rendimiento más rápido) de manejar esta cola. Me gustaría usar todos los núcleos disponibles.

Actualmente estoy experimentando con un BlockingCollection. Creo una clase de manejador de colas que genera 4 tareas, que bloquean en BlockingCollection y esperan el trabajo entrante. Luego procesan esa transformación pendiente. Código:

public class IncomingPacketQueue : IDisposable 
    { 
     BlockingCollection<IncomingPacket> _packetQ = new BlockingCollection<IncomingPacket>(); 

     public IncomingPacketQueue(int workerCount) 
     { 
      for (int i = 0; i < workerCount; i++) 
      { 
       Task.Factory.StartNew(Consume); 
      } 
     } 

     public void EnqueueSweep(IncomingPacket incoming) 
     { 
      _packetQ.Add(incoming); 
     } 

     private void Consume() 
     { 
      foreach (var sweep in _packetQ.GetConsumingEnumerable()) 
      { 
       //do stuff 
       var worker = new IfftWorker(); 
       Trace.WriteLine(" Thread {0} picking up a pending ifft".With(Thread.CurrentThread.ManagedThreadId)); 
       worker.DoIfft(sweep);     

      } 
     } 

     public int QueueCount 
     { 
      get 
      { 
       return _packetQ.Count; 
      } 
     } 

    #region IDisposable Members 

    public void Dispose() 
    { 
     _packetQ.CompleteAdding(); 
    } 

    #endregion 
    } 

¿Esto parece una buena solución? Parece que se agotan todos los núcleos, aunque actualmente no estoy seguro de cuántos trabajadores debería generar en mi constructor.

Answer 1

Eso parece razonable. He encontrado que BlockingCollection es bastante rápido. Lo uso para procesar decenas de miles de solicitudes por segundo.

Si su aplicación está vinculada al procesador, entonces probablemente no desee crear más trabajadores que sus núcleos. Ciertamente, no desea crear muchos más trabajadores que núcleos. En una máquina de cuatro núcleos, si espera pasar la mayor parte del tiempo haciendo las FFT, entonces cuatro trabajadores se comerán toda la CPU. Más trabajadores solo significan más de los que tienes que manejar. El TPL generalmente equilibrará eso para usted, pero no hay razón para crear, digamos, 100 trabajadores cuando no puede manejar más de un puñado.

Le sugiero que realice pruebas con 3, 4, 5, 6, 7 y 8 trabajadores. Vea cuál le ofrece el mejor rendimiento.

Answer 2

Haga que la cantidad de trabajadores se pueda configurar. También hay demasiados trabajadores y se volverá más lento (como lo indica otro afiche), por lo que debe encontrar el punto óptimo. Un valor configurable permitiría ejecutar pruebas para encontrar el valor óptimo o permitiría que su programa sea adaptable para diferentes tipos de hardware. Sin duda podría colocar este valor en App.Config y leerlo al inicio.

Answer 3

Estoy de acuerdo con Jim. Tu enfoque se ve muy bien. No vas a ser mucho mejor esto. No soy un experto en FFT, pero supongo que estas operaciones están casi 100% vinculadas a la CPU. Si ese es realmente el caso, una buena primera aproximación al número de trabajadores sería una correlación directa de 1 a 1 con la cantidad de núcleos en la máquina. Puede usar Environment.ProcessorCount para obtener este valor. Podría experimentar con un multiplicador de decir 2x o 4x, pero nuevamente, si estas operaciones están vinculadas a la CPU, cualquier cantidad superior a 1x podría causar más sobrecarga. Usar Environment.ProcessorCount haría su código más portátil.

Otra sugerencia ... haga saber al TPL que se trata de hilos dedicados. Puede hacerlo especificando la opción LongRunning.

public IncomingPacketQueue() 
{ 
    for (int i = 0; i < Environment.ProcessorCount; i++) 
    { 
     Task.Factory.StartNew(Consume, TaskCreationOptions.LongRunning); 
    } 
} 

Answer 4

También podría tratar de usar PLINQ para paralelizar el proceso para ver cómo se compara con el enfoque que está utilizando actualmente. Tiene algunos trucos bajo la manga que pueden hacerlo muy eficiente bajo ciertas circunstancias.

_packetQ.GetConsumingEnumerable().AsParallel().ForAll(
    sweep => new IfftWorker().DoIfft(sweep)); 

Answer 5

¿Por qué no usar Parallel.ForEach y dejar que TPL maneje la cantidad de hilos creados?

 Parallel.ForEach(BlockingCollectionExtensions.GetConsumingPartitioneenter(_packetQ), 
         sweep => { 
          //do stuff 
          var worker = new IfftWorker(); 
          Trace.WriteLine(" Thread {0} picking up a pending ifft".With(Thread.CurrentThread.ManagedThreadId)); 
          worker.DoIfft(sweep);     

         }); 

(el GetConsumingPartitioner es parte de la ParallelExtensionsExtras)