Threadpool/WaitHandle ressource fuite/accident

Question

Je pense que je devrais repenser à mon design. J'ai du mal à réduire un bug qui bloque complètement mon ordinateur, en lançant parfois un HRESULT 0x8007000E de VS 2010.

J'ai une application de console (que je convertirai plus tard en un service) qui gère transférer des fichiers basés sur une file d'attente de base de données.

Je suis en train d'étrangler les threads autorisés à transférer. En effet, certains systèmes auxquels nous nous connectons ne peuvent contenir qu'un certain nombre de connexions de certains comptes. Par exemple, le système A ne peut accepter que 3 connexions simultanées (ce qui signifie 3 threads séparés). Chacun de ces threads possède son propre objet de connexion unique, donc nous ne devrions pas nous lancer dans des problèmes de synchronisation car ils ne partagent pas une connexion.

Nous voulons traiter les fichiers de ces systèmes par cycles. Ainsi, par exemple, nous autoriserons 3 connexions pouvant transférer jusqu'à 100 fichiers par connexion. Cela signifie que pour déplacer 1000 fichiers du système A, nous ne pouvons traiter que 300 fichiers par cycle, puisque 3 threads sont autorisés avec 100 fichiers chacun. Par conséquent, pendant la durée de ce transfert, nous aurons 10 threads. Nous pouvons seulement courir 3 à la fois. Donc, il y aura 3 cycles, et le dernier cycle utilisera seulement 1 thread pour transférer les 100 derniers fichiers. (3 fils x 100 fichiers = 300 fichiers par cycle)

L'architecture actuelle par exemple:

1. Un System.Threading.Timer vérifie la file d'attente toutes les 5 secondes pour quelque chose à faire en appelant GetScheduledTask()
2. S'il n'y a rien à, GetScheduledTask() ne fonctionne tout simplement S'il n'y a rien
3. travail, créer un fil ThreadPool pour traiter le travail [travail discussion a]
4. travail discussion a voit qu'il ya 1000 fichiers à transférer
5. travail Discussion A voit qu'il ne peut avoir 3 fils en cours d'exécution sur le système, il récupèrera les bons fichiers
6. travail Discussion A commence trois nouveaux fils de travail [B, C, D] et les transferts
7. travail thread A attend B, C, D [WaitHandle.WaitAll(transfersArray)]
8. fil de travail a voit qu'il ya encore plus de fichiers dans la file d'attente (doit être 700 maintenant)
9. travail discussion a crée un nouveau tableau d'attendre le [transfersArray = new TransferArray[3] qui est le maximum pour le système a, mais peut varier sur le système
10. Le thread de travail A démarre trois nouveaux threads de travail [B, C, D] et les attend [WaitHandle.WaitAll(transfersArray)]
11. Le processus se répète jusqu'à ce qu'il n'y ait plus de fichiers à déplacer.
12. travail de la discussion des signaux A qu'il est fait

J'utilise ManualResetEvent pour gérer la signalisation.

Mes questions sont les suivantes:

1. Y at-il des circonstances criantes qui provoquerait une fuite de ressources ou un problème que je ressentais?
2. Si je boucle à travers le tableau après chaque WaitHandle.WaitAll(array) et appelez array[index].Dispose()?
3. Le nombre de handles dans le Gestionnaire des tâches de ce processus se glisse lentement
4. J'appelle la création initiale de travailleurs Discussion A partir d'un System.Threading. Minuteur. Y aura-t-il des problèmes avec ça? Le code de cette temporisation est le suivant:

(Certains code de classe pour la programmation)

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

Il s'avère que la source de ce problème étrange n'était pas liée à l'architecture mais plutôt à la conversion de la solution de 3.5 à 4.0. J'ai recréé la solution, n'effectuant aucune modification de code et le problème ne s'est jamais produit à nouveau.

Answer 2

0x8007000E est un out-of-mémoire d'erreur. Cela et le nombre de poignées semblent pointer vers une fuite de ressources. Assurez-vous de disposer de tous les objets qui implémentent IDisposable. Cela inclut les tableaux de ManualResetEvent que vous utilisez.

Si vous avez le temps, vous pouvez également convertir en utilisant la classe .NET 4.0 Task; il a été conçu pour gérer des scénarios complexes comme celui-ci beaucoup plus proprement. En définissant des objets enfant , vous pouvez réduire votre nombre de threads global (les threads sont assez chers non seulement en raison de la planification mais aussi en raison de leur espace de pile).

Answer 3

Je pense que vous devriez reconsidérer complètement votre architecture. Le fait que vous ne pouvez avoir que 3 connexions simultanées vous supplie presque d'utiliser 1 thread pour générer la liste des fichiers et 3 threads pour les traiter. Votre thread de production insère tous les fichiers dans une file d'attente et les 3 threads de consommation suppriment la file d'attente et continuent le traitement lorsque les éléments arrivent dans la file d'attente. Une file d'attente bloquante peut considérablement simplifier le code. Si vous utilisez .NET 4.0, vous pouvez utiliser la classe BlockingCollection. J'ai clairement surimplifié les choses dans l'exemple ci-dessus, mais j'espère que vous aurez l'idée générale. Notez que cela est beaucoup plus simple car il n'y a pas beaucoup de synchronisation des threads (la plupart seront intégrés dans la file d'attente de blocage) et bien sûr, il n'y a pas d'utilisation des objets WaitHandle. Évidemment, vous devrez ajouter les bons mécanismes pour fermer les threads gracieusement, mais cela devrait être assez facile.

Answer 4

Je cherche des réponses à un problème similaire (nombre de poignées augmentant avec le temps).

Je pris un coup d'œil à votre architecture d'application et comme vous suggérer quelque chose qui pourrait vous aider:

Avez-vous entendu parler de IOCP (Ports d'achèvement de sortie d'entrée). Je ne suis pas sûr de la difficulté d'implémenter ceci en utilisant C# mais en C/C++ c'est un morceau de gâteau. En utilisant cela, vous créez un pool de threads unique (Le nombre de threads dans ce pool est généralement défini comme 2 x le nombre de cœurs de processeurs ou de processeurs dans le PC ou le serveur) Vous associez ce pool à un handle IOCP et au pool fait le travail. Consultez l'aide pour les fonctions suivantes: CreateIoCompletionPort(); PostQueuedCompletionStatus(); GetQueuedCompletionStatus();

En général, la création et la sortie à la volée de threads peuvent prendre du temps et entraîner des pénalités de performances et une fragmentation de la mémoire. Il existe des milliers de publications sur IOCP dans MSDN et dans google.