Threadpool/WaitHandle pérdida de recursos/crash

Question

Creo que puede tener que volver a pensar en mi diseño.Estoy teniendo un tiempo difícil de reducir a un error que está causando mi equipo completo se bloquea, a veces lanzando un HRESULT 0x8007000E de VS 2010.

Tengo una aplicación de consola (que lo haré más adelante convertir a un servicio) que se encarga de la transferencia de archivos basado en una base de datos de la cola.

Yo soy de limitación de los hilos permite transferir.Esto es debido a que algunos de los sistemas que se conectan a sólo puede contener un cierto número de conexiones de ciertas cuentas.

Por ejemplo, el Sistema sólo puede aceptar 3 conexiones simultáneas (lo que significa que 3 hilos separados).Cada uno de estos temas tiene su propio objeto de conexión, así que no debe ejecutar en problemas de sincronización, ya que no son para compartir una conexión.

Queremos procesar los archivos de los sistemas en ciclos.Así, por ejemplo, nos permitirá 3 conexiones que se pueden transferir hasta 100 archivos por conexión.Esto significa que, para mover 1000 archivos de Sistema, sólo podemos procesar 300 archivos por ciclo, desde el 3 de subprocesos pueden con 100 archivos de cada uno.Por lo tanto, durante la vigencia de esta transferencia, vamos a tener 10 hilos.Sólo podemos ejecutar 3 a la vez.Por lo tanto, habrá 3 ciclos, y en el último ciclo se use solo 1 hilo de la transferencia de los últimos 100 archivos.(3 hilos x 100 archivos = 300 archivos por ciclo)

La arquitectura actual por ejemplo es:

1. De Un Sistema.Las operaciones de roscado.Temporizador comprueba la cola cada 5 segundos por algo que hacer llamando GetScheduledTask()
2. Si no hay nada que, GetScheduledTask() simplemente no hace nada
3. Si no hay trabajo, crear un grupo de subprocesos subproceso proceso de la obra [de Trabajo Subproceso]
4. El trabajo de Un Hilo se ve que hay 1000 archivos para transferir
5. El trabajo de Un Hilo ve que sólo puede tener 3 hilos de ejecución para el sistema es conseguir los archivos de
6. Trabajo Subproceso se inicia tres nuevos subprocesos de trabajo [B,C,D] y de transferencias de
7. El trabajo de Hilo de espera para B,C,D [WaitHandle.WaitAll(transfersArray)]
8. El trabajo de Un Hilo se ve que todavía hay más archivos en la cola (debe ser de 700 ahora)
9. El trabajo de Un Hilo crea una nueva matriz a esperar en [transfersArray = new TransferArray[3] que es el máximo para el Sistema a, pero podría variar en el sistema
10. Trabajo Subproceso se inicia tres nuevos subprocesos de trabajo [B,C,D] y que los espera [WaitHandle.WaitAll(transfersArray)]
11. El proceso se repite hasta que no haya más archivos a mover.
12. Trabajo Subproceso de las señales de que está hecho

Estoy usando ManualResetEvent para manejar la señalización.

Mis preguntas son:

1. Es allí cualquier evidente circunstancia que puede causar una pérdida de recursos o problema que estoy experimentando?
2. Debo bucle a través de la matriz después de cada WaitHandle.WaitAll(array) y llame a array[index].Dispose()?
3. El número de Identificador bajo el Administrador de Tareas para este proceso lentamente se arrastra para arriba
4. Estoy llamando a la creación inicial de Un Subproceso de trabajo de un Sistema.Las operaciones de roscado.El temporizador.No va a ser ningún problema con esto?El código para que el temporizador está:

(Alguna clase de código de programación)

private ManualResetEvent _ResetEvent;

private void Start()
{
    _IsAlive = true;
    ManualResetEvent transferResetEvent = new ManualResetEvent(false);
    //Set the scheduler timer to 5 second intervals
    _ScheduledTasks = new Timer(new TimerCallback(ScheduledTasks_Tick), transferResetEvent, 200, 5000);
}

private void ScheduledTasks_Tick(object state)
{
    ManualResetEvent resetEvent = null;
    try
    {
        resetEvent = (ManualResetEvent)state;
        //Block timer until GetScheduledTasks() finishes
        _ScheduledTasks.Change(Timeout.Infinite, Timeout.Infinite);
        GetScheduledTasks();
    }
    finally
    {
        _ScheduledTasks.Change(5000, 5000);
        Console.WriteLine("{0} [Main] GetScheduledTasks() finished", DateTime.Now.ToString("MMddyy HH:mm:ss:fff"));
        resetEvent.Set();
    }
}


private void GetScheduledTask()
{
    try 
    { 
        //Check to see if the database connection is still up
        if (!_IsAlive)
        {
            //Handle
            _ConnectionLostNotification = true;
            return;
        }

        //Get scheduled records from the database
        ISchedulerTask task = null;

        using (DataTable dt = FastSql.ExecuteDataTable(
                _ConnectionString, "hidden for security", System.Data.CommandType.StoredProcedure,
                new List<FastSqlParam>() { new FastSqlParam(ParameterDirection.Input, SqlDbType.VarChar, "@ProcessMachineName", Environment.MachineName) })) //call to static class
        {
            if (dt != null)
            {
                if (dt.Rows.Count == 1)
                {  //Only 1 row is allowed
                    DataRow dr = dt.Rows[0];

                    //Get task information
                    TransferParam.TaskType taskType = (TransferParam.TaskType)Enum.Parse(typeof(TransferParam.TaskType), dr["TaskTypeId"].ToString());
                    task = ScheduledTaskFactory.CreateScheduledTask(taskType);

                    task.Description = dr["Description"].ToString();
                    task.IsEnabled = (bool)dr["IsEnabled"];
                    task.IsProcessing = (bool)dr["IsProcessing"];
                    task.IsManualLaunch = (bool)dr["IsManualLaunch"];
                    task.ProcessMachineName = dr["ProcessMachineName"].ToString();
                    task.NextRun = (DateTime)dr["NextRun"];
                    task.PostProcessNotification = (bool)dr["NotifyPostProcess"];
                    task.PreProcessNotification = (bool)dr["NotifyPreProcess"];
                    task.Priority = (TransferParam.Priority)Enum.Parse(typeof(TransferParam.SystemType), dr["PriorityId"].ToString());
                    task.SleepMinutes = (int)dr["SleepMinutes"];
                    task.ScheduleId = (int)dr["ScheduleId"];
                    task.CurrentRuns = (int)dr["CurrentRuns"];
                    task.TotalRuns = (int)dr["TotalRuns"];

                    SchedulerTask scheduledTask = new SchedulerTask(new ManualResetEvent(false), task);
                    //Queue up task to worker thread and start
                    ThreadPool.QueueUserWorkItem(new WaitCallback(this.ThreadProc), scheduledTask);     
                }
            }
        }

    }
    catch (Exception ex)
    {
        //Handle
    }
}

private void ThreadProc(object taskObject)
{
    SchedulerTask task = (SchedulerTask)taskObject;
    ScheduledTaskEngine engine = null;
    try
    {
        engine = SchedulerTaskEngineFactory.CreateTaskEngine(task.Task, _ConnectionString);
        engine.StartTask(task.Task);    
    }
    catch (Exception ex)
    {
        //Handle
    }
    finally
    {
        task.TaskResetEvent.Set();
        task.TaskResetEvent.Dispose();
    }
}

Answer 1

Resulta que el origen de este extraño problema no estaba relacionado con la arquitectura, sino más bien a causa de la conversión de la solución de 3.5 a 4.0.Me re-crea la solución, que no realizan cambios en el código, y el problema nunca ocurrió de nuevo.

Answer 2

0x8007000E es un error de falta de memoria.Que y el número de identificador parecen apuntar a una pérdida de recursos.Asegurarse de la eliminación de cada objeto que implementa IDisposable.Esto incluye las matrices de ManualResetEvents que usted está utilizando.

Si usted tiene el tiempo, usted también puede querer convertir a la utilización de la .NET 4.0 Task clase;fue diseñado para manejar escenarios complejos como este mucho más limpiamente.Mediante la definición de niño Task los objetos, usted puede reducir su total número de hilos (threads son bastante caros, no sólo porque de programación, sino también a causa de su espacio en la pila).

Answer 3

Estoy buscando respuestas a un problema similar (Handles número creciente a lo largo del tiempo).

Eché un vistazo a la arquitectura de la aplicación y desea sugerir algo que podría ayudarte:

Ha escuchado acerca de la IOCP (de Entrada y Salida de Puertos de Finalización).

No estoy seguro de la dificultad para implementar esta usando C#, pero en C/C++ es un pedazo de la torta.Mediante el uso de este se crea un único hilo de la piscina (El número de subprocesos en que la piscina es, en general, se define como 2 x el número de procesadores o procesadores de núcleos en el PC o en el servidor) Asociar este grupo a una IOCP de Manejar, y la piscina hace el trabajo.Consulte la ayuda de estas funciones:CreateIoCompletionPort();PostQueuedCompletionStatus();GetQueuedCompletionStatus();

En General, la creación y la salida de los hilos sobre la marcha puede consumir mucho tiempo y conduce a reducciones del rendimiento y la fragmentación de la memoria.Hay miles de literatura acerca de la IOCP en MSDN y en google.

Answer 4

Creo que deberías reconsiderar su arquitectura por completo.El hecho de que sólo se puede tener 3 simultáneamente conexiones es casi pidiendo uso 1 hilo para generar la lista de archivos y 3 hilos para su proceso.Su productor hilo insertar todos los archivos en cola y el 3 de consumidores hilos de cola, y continuar con el proceso al llegar los elementos en la cola.Una cola de bloqueo puede simplificar considerablemente el código.Si usted está utilizando .NET 4.0, a continuación, usted puede tomar ventaja de la BlockingCollection clase.

public class Example
{
    private BlockingCollection<string> m_Queue = new BlockingCollection<string>();

    public void Start()
    {
        var threads = new Thread[] 
            { 
                new Thread(Producer), 
                new Thread(Consumer), 
                new Thread(Consumer), 
                new Thread(Consumer) 
            };
        foreach (Thread thread in threads)
        {
            thread.Start();
        }
    }

    private void Producer()
    {
        while (true)
        {
            Thread.Sleep(TimeSpan.FromSeconds(5));
            ScheduledTask task = GetScheduledTask();
            if (task != null)
            {
                foreach (string file in task.Files)
                {
                    m_Queue.Add(task);
                }
            }
        }
    }

    private void Consumer()
    {
        // Make a connection to the resource that is assigned to this thread only.
        while (true)
        {
            string file = m_Queue.Take();
            // Process the file.
        }
    }
}

Definitivamente he simplificada cosas en el ejemplo anterior, pero espero que la idea general.Observe cómo esto es mucho más sencillo ya que no hay mucho en el camino de la sincronización de subprocesos (la mayoría serán incorporados en el bloqueo de cola) y, por supuesto, no hay ningún uso de WaitHandle objetos.Obviamente habría que añadir en los mecanismos correctos para cerrar los hilos con gracia, pero que debe ser bastante fácil.