Solution de contournement pour la limite de handle WaitHandle.WaitAll 64?

Question

Mon application génère des charges de différentes petites threads via ThreadPool.QueueUserWorkItem qui j'ai garder une trace de la via de multiples ManualResetEvent des cas. J'utilise l' WaitHandle.WaitAll méthode pour bloquer ma demande de fermeture jusqu'à ce que ces fils ont terminé.

Je n'ai jamais eu de problèmes avant, toutefois, que ma demande est venue en vertu de charge en plus c'est à dire plus de threads en cours de création, je suis maintenant commence à prendre de cette exception:

WaitHandles must be less than or equal to 64 - missing documentation

Quelle est la meilleure solution pour cela?

Extrait De Code

List<AutoResetEvent> events = new List<AutoResetEvent>();

// multiple instances of...
var evt = new AutoResetEvent(false);
events.Add(evt);
ThreadPool.QueueUserWorkItem(delegate
{
    // do work
    evt.Set();
});

...
WaitHandle.WaitAll(events.ToArray());

Solution de contournement

int threadCount = 0;
ManualResetEvent finished = new ManualResetEvent(false);

...
Interlocked.Increment(ref threadCount);
ThreadPool.QueueUserWorkItem(delegate
{
    try
    {
         // do work
    }
    finally
    {
        if (Interlocked.Decrement(ref threadCount) == 0)
        {
             finished.Set();
        }
    }
});

...
finished.WaitOne();

Answer 1

Créez une variable qui garde la trace du nombre de tâches en cours d'exécution:

 int numberOfTasks = 100;
 

Créer un signal:

 ManualResetEvent signal = new ManualResetEvent(false);
 

Décrémentez le nombre de tâches chaque fois qu'une tâche est terminée:

 if (Interlocked.Decrement(ref numberOftasks) == 0)
{
 

S'il ne reste aucune tâche, définissez le signal:

     signal.Set();
}
 

En attendant, quelque part ailleurs, attendez que le signal soit émis:

 signal.WaitOne();
 

Answer 2

En commençant avec .NET 4.0, vous avez deux de plus (et de l'OMI, plus propre) options disponibles pour vous.

La première est d'utiliser l' CountdownEvent classe. Ça évite d'avoir à gérer l'incrémentation et de décrémentation sur votre propre:

int tasks = <however many tasks you're performing>;

// Dispose when done.
using (var e = new CountdownEvent(tasks))
{
    // Queue work.
    ThreadPool.QueueUserWorkItem(() => {
        // Do work
        ...

        // Signal when done.
        e.Signal();
    });

    // Wait till the countdown reaches zero.
    e.Wait();
}

Cependant, il y a une même solution plus robuste, et que l'utilisation de l' Task classe, comme ceci:

// The source of your work items, create a sequence of Task instances.
Task[] tasks = Enumerable.Range(0, 100).Select(i =>
    // Create task here.
    Task.Factory.StartNew(() => {
        // Do work.
    }

    // No signalling, no anything.
).ToArray();

// Wait on all the tasks.
Tasks.WaitAll(tasks);

À l'aide de l' Task de la classe et de l'appel à WaitAll est beaucoup plus propre, de l'OMI, comme vous allez le tissage moins primitives de thread tout au long de votre code (avis, pas d'attente poignées); vous n'avez pas à mettre en place un compteur de gérer l'incrémentation/décrémentation, que vous venez de configurer vos tâches et puis attendre sur eux. Cela permet au code d'être plus expressif dans la ce qui de ce que vous voulez faire et de ne pas les primitives de la façon dont (au moins, en termes de gestion de la parallélisation de celui-ci).

.NET 4.5 offre encore plus d'options, vous pouvez simplifier la génération de la séquence de Task des occurrences de l'appel de la statique Run méthode sur l' Task classe:

// The source of your work items, create a sequence of Task instances.
Task[] tasks = Enumerable.Range(0, 100).Select(i =>
    // Create task here.
    Task.Run(() => {
        // Do work.
    }

    // No signalling, no anything.
).ToArray();

// Wait on all the tasks.
Tasks.WaitAll(tasks);

Ou, vous pouvez prendre avantage de la TPL bibliothèque de Flux de données (c'est dans l' System d'espace de noms, donc, c'est officiel, même si c'est un téléchargement à partir de NuGet, comme Entity Framework) et utiliser un ActionBlock<TInput>, comme suit:

// Create the action block.  Since there's not a non-generic
// version, make it object, and pass null to signal, or
// make T the type that takes the input to the action
// and pass that.
var actionBlock = new ActionBlock<object>(o => {
    // Do work.
});

// Post 100 times.
foreach (int i in Enumerable.Range(0, 100)) actionBlock.Post(null);

// Signal complete, this doesn't actually stop
// the block, but says that everything is done when the currently
// posted items are completed.
actionBlock.Complete();

// Wait for everything to complete, the Completion property
// exposes a Task which can be waited on.
actionBlock.Completion.Wait();

Notez que l' ActionBlock<TInput> par des processus par défaut un élément à la fois, donc si vous voulez l'avoir procédé à plusieurs actions à la fois, vous devez définir le nombre de connexions simultanées éléments que vous souhaitez traiter dans le constructeur par le passage d'un ExecutionDataflowBlockOptions de l'instance et le réglage de la MaxDegreeOfParallelism de la propriété:

var actionBlock = new ActionBlock<object>(o => {
    // Do work.
}, new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = 4 });

Si votre action est vraiment pas thread-safe, vous pouvez alors régler le MaxDegreeOfParallelsim de la propriété d' DataFlowBlockOptions.Unbounded:

var actionBlock = new ActionBlock<object>(o => {
    // Do work.
}, new ExecutionDataflowBlockOptions { 
    MaxDegreeOfParallelism = DataFlowBlockOptions.Unbounded
});

Le point de l'être, vous disposez d'un contrôle précis sur la façon parallèle, vous voulez que vos options.

Bien sûr, si vous avez une séquence d'éléments que vous souhaitez passé dans votre ActionBlock<TInput> exemple, vous pouvez lier un ISourceBlock<TOutput> de la mise en œuvre de nourrir l' ActionBlock<TInput>, comme suit:

// The buffer block.
var buffer = new BufferBlock<int>();

// Create the action block.  Since there's not a non-generic
// version, make it object, and pass null to signal, or
// make T the type that takes the input to the action
// and pass that.
var actionBlock = new ActionBlock<int>(o => {
    // Do work.
});

// Link the action block to the buffer block.
// NOTE: An IDisposable is returned here, you might want to dispose
// of it, although not totally necessary if everything works, but
// still, good housekeeping.
using (link = buffer.LinkTo(actionBlock, 
    // Want to propagate completion state to the action block.
    new DataflowLinkOptions {
        PropagateCompletion = true,
    },
    // Can filter on items flowing through if you want.
    i => true)
{ 
    // Post 100 times to the *buffer*
    foreach (int i in Enumerable.Range(0, 100)) buffer.Post(i);

    // Signal complete, this doesn't actually stop
    // the block, but says that everything is done when the currently
    // posted items are completed.
    actionBlock.Complete();

    // Wait for everything to complete, the Completion property
    // exposes a Task which can be waited on.
    actionBlock.Completion.Wait();
}

En fonction de ce que vous devez faire, le TPL Dataflow bibliothèque devient un beaucoup plus attrayant, en ce qu'il traite de l'accès concurrentiel à travers toutes les tâches liées ensemble, et il vous permet d'être très précis sur juste comment parallèle, vous voulez que chaque morceau, tout en conservant une bonne séparation des préoccupations de chaque bloc.

Answer 3

Votre solution de contournement n'est pas correct. La raison en est que l' Set et WaitOne pourraient course si le dernier élément de travail provoque l' threadCount aller à zéro avant la mise en attente du thread a eu de la chance de file d'attente de tous les éléments de travail. La solution est simple. Traiter votre mise en attente du thread, comme si il s'agissait d'un élément de travail lui-même. Initialiser threadCount à 1 et ne une diminution du signal et quand la queue est terminée.

int threadCount = 1;
ManualResetEvent finished = new ManualResetEvent(false);
...
Interlocked.Increment(ref threadCount); 
ThreadPool.QueueUserWorkItem(delegate 
{ 
    try 
    { 
         // do work 
    } 
    finally 
    { 
        if (Interlocked.Decrement(ref threadCount) == 0) 
        { 
             finished.Set(); 
        } 
    } 
}); 
... 
if (Interlocked.Decrement(ref threadCount) == 0)
{
  finished.Set();
}
finished.WaitOne(); 

Comme une question de préférence personnelle j'ai comme l'utilisation de la CountdownEvent classe pour faire le calcul pour moi.

var finished = new CountdownEvent(1);
...
finished.AddCount();
ThreadPool.QueueUserWorkItem(delegate 
{ 
    try 
    { 
         // do work 
    } 
    finally 
    { 
      finished.Signal();
    } 
}); 
... 
finished.Signal();
finished.Wait(); 

Answer 4

En ajoutant à la réponse de dtb, vous pouvez envelopper ceci dans une classe simple et agréable.

 public class Countdown : IDisposable
{
    private readonly ManualResetEvent done;
    private readonly int total;
    private long current;

    public Countdown(int total)
    {
        this.total = total;
        current = total;
        done = new ManualResetEvent(false);
    }

    public void Signal()
    {
        if (Interlocked.Decrement(ref current) == 0)
        {
            done.Set();
        }
    }

    public void Wait()
    {
        done.WaitOne();
    }

    public void Dispose()
    {
        ((IDisposable)done).Dispose();
    }
}
 

Answer 5

Windows XP SP3 prend en charge deux WaitHandles au maximum. Dans les cas plus de 2 WaitHandles, l'application se termine prématurément.