C# plusieurs threads en attente d'un ManualResetEvent

Question

Je suis en train de jouer avec le multithreading et de créer un moteur de tâches. L'idée est que le moteur peut avoir une quantité configurable de threads en attente et quand une nouvelle tâche arrive, le premier thread libre le récupère et l'exécute.

Le problème est que quelque chose 2 threads ramasser la même tâche en quelque sorte. J'ai regardé à travers et je pense que ce code devrait fonctionner, mais de toute évidence pas. Si j'ajoute le sommeil 10ms où il est maintenant commenté, ça marche, mais je ne suis pas sûr de comprendre pourquoi. Il semble que la fonction .Reset() retourne avant de réinitialiser l'événement?

Quelqu'un peut-il expliquer? Existe-t-il un meilleur moyen de ne laisser qu'un seul thread continuer lorsqu'il y a plusieurs files d'attente?

Merci

using System; 
using System.Collections.Concurrent; 
using System.Collections.Generic; 
using System.Linq; 
using System.Text; 
using System.Threading; 
using System.Threading.Tasks; 

namespace TaskTest 
{ 
    public class Engine 
    { 
     private ManualResetEvent taskEvent; 
     private ConcurrentQueue<Task> tasks; 
     private bool running; 
     private List<Thread> threads; 
     private int threadAmount; 
     private int threadsBusy = 0; 

     public Engine(int amountOfThreads) 
     { 
      taskEvent = new ManualResetEvent(false); 
      tasks = new ConcurrentQueue<Task>(); 
      threads = new List<Thread>(); 

      threadAmount = amountOfThreads; 
     } 

     public void Start() 
     { 
      running = true; 
      for (var i = 0; i < threadAmount; i++) 
      { 
       var thread = new Thread(Process); 
       thread.Name = "Thread " + i; 
       threads.Add(thread); 
       thread.Start(); 
      } 
     } 

     public void Stop() 
     { 
      running = false; 
      taskEvent.Set(); 
      threads.ForEach(t => t.Join()); 
     } 

     private void Process() 
     { 
      while (running) 
      { 
       lock (taskEvent) 
       { 
        // Lock it so only a single thread is waiting on the event at the same time 
        taskEvent.WaitOne(); 
        taskEvent.Reset(); 
        //Thread.Sleep(10); 
       } 

       if (!running) 
       { 
        taskEvent.Set(); 
        return; 
       } 

       threadsBusy += 1; 
       if (threadsBusy > 1) 
        Console.WriteLine("Failed"); 

       Task task; 
       if (tasks.TryDequeue(out task)) 
        task.Execute(); 

       threadsBusy -= 1; 
      } 
     } 

     public void Enqueue(Task t) 
     { 
      tasks.Enqueue(t); 
      taskEvent.Set(); 
     } 
    } 
} 

EDIT Reste du code:

namespace TaskTest 
{ 
    public class Start 
    { 
     public static void Main(params string[] args) 
     { 
      var engine = new Engine(4); 
      engine.Start(); 

      while (true) 
      { 
       Console.Read(); 
       engine.Enqueue(new Task()); 
      } 
     } 
    } 
} 


namespace TaskTest 
{ 
    public class Task 
    { 
     public void Execute() 
     { 
      Console.WriteLine(Thread.CurrentThread.Name); 
     } 
    } 
} 

Answer 1

Lorsque vous utilisez Console.Read() sur une touche, deux caractères sont lus à partir de l'entrée. Vous devriez plutôt utiliser Console.ReadLine().

Notez que votre code peut être beaucoup simplifié en utilisant un BlockingCollection pour gérer la synchronisation:

public class Engine 
{ 
    private BlockingCollection<Task> tasks; 
    private List<Thread> threads; 
    private int threadAmount; 

    public Engine(int amountOfThreads) 
    { 
     tasks = new BlockingCollection<Task>(); 
     threads = new List<Thread>(); 

     threadAmount = amountOfThreads; 
    } 

    public void Start() 
    { 
     for (var i = 0; i < threadAmount; i++) 
     { 
      var thread = new Thread(Process); 
      thread.Name = "Thread " + i; 
      threads.Add(thread); 
      thread.Start(); 
     } 
    } 

    public void Stop() 
    { 
     tasks.CompleteAdding(); 
     threads.ForEach(t => t.Join()); 
    } 

    private void Process() 
    { 
     foreach (var task in tasks.GetConsumingEnumerable()) 
     { 
      task.Execute(); 
     } 
    } 

    public void Enqueue(Task t) 
    { 
     tasks.Add(t); 
    } 
}