Implementación de subprocesamiento múltiple en C# (revisión de código)

Question

Saludos.

Estoy tratando de implementar algún código multiproceso en una aplicación. El propósito de este código es validar elementos que la base de datos le da. La validación puede llevar bastante tiempo (unos pocos cientos de ms a unos pocos segundos), por lo que este proceso debe dividirse en su propio hilo para cada elemento.

La base de datos puede darle 20 o 30 elementos por segundo al principio, pero eso comienza a disminuir rápidamente, alcanzando finalmente alrededor de 65K elementos durante 24 horas, en cuyo punto se cierra la aplicación.

Me gustaría que cualquier persona más conocedora pudiera echar un vistazo a mi código y ver si hay algún problema obvio. Nadie con quien trabajo sabe multihilo, así que estoy realmente solo, en este.

Aquí está el código. Es un poco largo, pero debería ser bastante claro. Avíseme si tiene algún comentario o consejo. ¡Gracias!

public class ItemValidationService 
{ 
    /// <summary> 
    /// The object to lock on in this class, for multithreading purposes. 
    /// </summary> 
    private static object locker = new object(); 

    /// <summary>Items that have been validated.</summary> 
    private HashSet<int> validatedItems; 

    /// <summary>Items that are currently being validated.</summary> 
    private HashSet<int> validatingItems; 

    /// <summary>Remove an item from the index if its links are bad.</summary> 
    /// <param name="id">The ID of the item.</param> 
    public void ValidateItem(int id) 
    { 
     lock (locker) 
     { 
      if 
      (
       !this.validatedItems.Contains(id) && 
       !this.validatingItems.Contains(id) 
      ){ 
       ThreadPool.QueueUserWorkItem(sender => 
       { 
        this.Validate(id); 
       }); 
      } 
     } 

    } // method 

    private void Validate(int itemId) 
    { 
     lock (locker) 
     { 
      this.validatingItems.Add(itemId); 
     } 

     // ********************************************* 
     // Time-consuming routine to validate an item... 
     // ********************************************* 

     lock (locker) 
     { 
      this.validatingItems.Remove(itemId); 
      this.validatedItems.Add(itemId); 
     } 

    } // method 

} // class 

Answer 1

El grupo de subprocesos es una opción conveniente si usted tiene peso ligero procesamiento esporádica que no es sensible al tiempo. Sin embargo, recuerdo haber leído en MSDN que no es apropiado para el procesamiento a gran escala de esta naturaleza.

Lo usé para algo bastante similar a esto y lo lamento. Tomé un enfoque de hilo de trabajo en aplicaciones posteriores y estoy mucho más feliz con el nivel de control que tengo.

Mi patrón favorito en el modelo de hilo de trabajador es crear un hilo maestro que contenga una cola de elementos de tareas. Luego bifurque a un grupo de trabajadores que saque elementos de esa cola para procesar. Utilizo una cola de bloqueo para que cuando el proceso no tenga elementos, los trabajadores simplemente bloqueen hasta que algo se introduzca en la cola. En este modelo, el hilo maestro produce elementos de trabajo de alguna fuente (db, etc.) y los hilos de trabajo los consumen.

Answer 2

Me preocuparía el rendimiento aquí. Indicó que la base de datos puede darle 20-30 elementos por segundo y un elemento podría demorar unos segundos para ser validado. Eso podría ser una gran cantidad de hilos, usando sus métricas, en el peor de los casos 60-90 hilos. Creo que debes reconsiderar el diseño aquí. Michael mencionó un patrón agradable. El uso de la cola realmente ayuda a mantener las cosas bajo control y organizadas. También se podría utilizar un semáforo para controlar el número de subprocesos creados, es decir, podría tener un número máximo de subprocesos permitidos, pero con cargas más pequeñas, no necesariamente tendría que crear el número máximo si menos terminaban haciendo el trabajo. - es decir, su propio tamaño de grupo puede ser dinámico con un límite.

Al utilizar el grupo de hilos, también me resulta más difícil controlar la ejecución de los hilos del conjunto en el momento de realizar el trabajo. Entonces, a menos que sea fuego y olvide, estoy a favor de una ejecución más controlada. Sé que mencionó que su aplicación se cierra después de que se completen los 65K elementos. ¿Cómo supervisa los hilos para determinar si han completado su trabajo? Es decir, todos los trabajadores en cola han terminado. ¿Está monitoreando el estado de todos los artículos en los HashSets? Creo que al poner tus artículos en cola y que tus propios hilos de trabajo consuman esa cola, puedes obtener más control. Sin embargo, esto puede tener el costo de más sobrecarga en términos de señalización entre hilos para indicar cuándo todos los artículos han sido puestos en cola, lo que les permite salir.

Answer 3

Tenga cuidado, QueueUserWorkItem might fail

Answer 4

Secundo la idea de utilizar una cola de bloqueo y subprocesos de trabajo.Aquí hay una implementación de cola de bloqueo que he usado en el pasado con buenos resultados: http://www.codeproject.com/KB/recipes/boundedblockingqueue.aspx

¿Qué implica su lógica de validación? Si está principalmente vinculado a la CPU, entonces crearía no más de 1 hilo de trabajo por procesador/núcleo en la caja. Esto le dirá la cantidad de procesadores: Environment.ProcessorCount

Si su validación implica E/S como Acceso a archivos o acceso a la base de datos, puede usar algunos hilos más que la cantidad de procesadores.

Answer 5

Existe un posible error de lógica en el código publicado con la pregunta, dependiendo de dónde provenga la identificación del artículo en ValidateItem(int id). ¿Por qué? Debido a que, aunque bloquea correctamente sus validatingItems y validatedItems queues antes de realizar una búsqueda en un elemento de trabajo, no agrega el elemento a la cola validatingItems hasta que el nuevo hilo se inicie. Eso significa que podría haber un intervalo de tiempo en el que otro hilo llame al ValidateItem(id) con la misma identificación (a menos que esto se ejecute en un solo hilo principal).

Agregaría el artículo a la cola validatingItems justo antes de poner en cola el artículo, dentro del candado.

Editar: también QueueUserWorkItem() devuelve un bool, por lo que debe utilizar el valor de retorno para asegurarse de que el elemento se puso en cola y, a continuación, agregarlo a la cola validatingItems.

Answer 6

ThreadPool puede no ser óptimo para atascarse tanto en él. Es posible que desee investigar los límites superiores de sus capacidades y/o hacer las suyas propias.

Además, existe una condición de carrera que existe en su código, si no espera validaciones duplicadas. La llamada a

this.validatingItems.Add(itemId); 

tiene que ocurrir en el hilo principal (ValidateItem), no en el hilo grupo de subprocesos (método Validar). Esta llamada debe aparecer en una línea antes de la puesta en cola del elemento de trabajo en el grupo.

Se encuentra un error peor al no comprobar el retorno de QueueUserWorkItem. Hacer cola puede fallar, y por qué no lanza una excepción es un misterio para todos nosotros. Si devuelve falso, debe eliminar el elemento que se agregó a la lista validatingItems y manejar el error (throw exeception probablemente).

Answer 7

También podría intentar usar el Runtime CCR - Concurrency and Coordination. Está enterrado dentro de Microsoft Robotics Studio, pero proporciona una API excelente para hacer este tipo de cosas.

Solo necesita crear un "Puerto" (esencialmente una cola), conectar un receptor (método que recibe una llamada cuando algo se publica en él) y luego publicar elementos de trabajo en él. El CCR maneja la cola y el hilo del trabajador para ejecutarlo.

Here's a video on Channel9 about the CCR.

es muy alto rendimiento e incluso se está utilizando para no Robótica cosas (Myspace.com utiliza detrás de la scenese para su red de entrega de contenido).