Abhilfe für die WaitHandle.WaitAll 64 Griffgrenze?

Question

Meine Anwendung Spawns Lasten der verschiedenen kleinen Arbeitsthreads über ThreadPool.QueueUserWorkItem, die ich den Überblick über Instanzen über mehrere ManualResetEvent halten. Ich verwende die WaitHandle.WaitAll Methode meine Anwendung zu blockieren, zu schließen, bis diese Fäden abgeschlossen haben.

Ich habe noch nie irgendwelche Probleme hatte, aber, wie meine Anwendung unter mehr Last kommt das heißt mehr Threads erstellt wird, bin ich jetzt beginnen, diese Ausnahme zu erhalten:

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

Was ist die beste alternative Lösung für das?

Code-Snippet

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());

Umgehung

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

Erstellen Sie eine Variable, die den Überblick über die Anzahl der laufenden Tasks hält:

int numberOfTasks = 100;

Erstellen Sie ein Signal:

ManualResetEvent signal = new ManualResetEvent(false);

Decrement die Anzahl der Aufgaben, wenn eine Aufgabe beendet:

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

Wenn es keine Aufgabe bleibt, setzen das Signal:

    signal.Set();
}

Zwischenzeit woanders, Warten auf das Signal gesetzt werden:

signal.WaitOne();

Answer 2

Beginnend mit .NET 4.0, haben Sie zwei weitere (und IMO, Reiniger) Möglichkeiten, die Ihnen.

Das ist zunächst die CountdownEvent Klasse zu verwenden. Es verhindert das Bedürfnis, das Inkrementieren und Dekrementieren auf eigene Faust zu behandeln:

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();
}

Es gibt jedoch eine noch robustere Lösung, und das ist das Task Klasse , etwa so:

// 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.
Task.WaitAll(tasks);

die Task-Klasse verwenden und den Anruf an WaitAll ist viel sauberer, IMO, wie Sie ‚re Weben weniger Einfädeln Primitiven im gesamten Code (Ankündigung, keine Wartegriffe); Sie müssen nicht einen Zähler, Griff Inkrementieren / Dekrementieren, die Sie gerade Ihre Aufgaben einrichten und warten auf sie einrichten. Auf diese Weise können Sie den Code sein ausdrucks in der was von dem, was Sie tun wollen und nicht die Grundelemente von wie (zumindest in Bezug auf die Parallelisierung der es verwaltet).

.NET 4.5 bietet noch mehr Möglichkeiten, können Sie die Erzeugung der Folge von Task Instanzen vereinfachen, indem die statische Run Methode auf der Task Klasse :

// 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);

Oder Sie nutzen die TPL Dataflow-Bibliothek (es ist im System Namespace , es ist so offiziell, auch wenn es ein Download von NuGet, wie Entity Framework ist) und verwendet ein ActionBlock<TInput> , etwa so:

// 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();

Beachten Sie, dass die ActionBlock<TInput> standardmäßig verarbeitet ein Element zu einem Zeitpunkt, also wenn Sie es verarbeiten mehrere Aktionen auf einmal haben wollen, müssen Sie die Anzahl der gleichzeitig Elemente setzen Sie, indem ein zu Prozess im Konstruktor wollen < a href = "http://msdn.microsoft.com/en-us/library/system.threading.tasks.dataflow.executiondataflowblockoptions.aspx"> ExecutionDataflowBlockOptions Instanz und die MaxDegreeOfParallelism Eigenschaft :

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

Wenn die Aktion wirklich sicher ist, fädeln, dann können Sie die MaxDegreeOfParallelsim Eigenschaft auf DataFlowBlockOptions.Unbounded :

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

Der Punkt ist, haben Sie feinkörnige Kontrolle über wie parallel Sie Ihre Optionen.

Natürlich, wenn Sie eine Folge von Elementen haben, dass Sie in Ihre ActionBlock<TInput> Instanz übergeben wollen, dann können Sie eine ISourceBlock<TOutput> Umsetzung der ActionBlock<TInput> zu füttern, etwa so:

// 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();
}

Je nachdem, was Sie tun müssen, wird die TPL Datenfluß-Bibliothek eine viel attraktivere Option, indem sie die Parallelität über Griffe alle die Aufgaben miteinander verknüpft, und ermöglicht es Ihnen, sehr spezifisch über nur , wie parallel Sie jedes Stück, sein wollen, zu sein, während die richtige Trennung von Bedenken für jeden Block beibehalten wird.

Answer 3

Ihre Abhilfe ist nicht korrekt. Der Grund dafür ist, dass die Set und WaitOne könnten laufen, wenn das letzte Arbeitselement die threadCount auf Null gehen verursacht vor die Warteschlangen Thread ist dem Zufall zu Warteschlange hat alle Workitem. Die Lösung ist einfach. Behandeln Sie Ihre Warteschlangen Faden, als ob es ein Workitem waren selbst. Initialisieren threadCount 1 und tun eine Abnahme- und Signal, wenn die Warteschlangen abgeschlossen ist.

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(); 

Als persönliche Präferenz Ich mag die CountdownEvent-Klasse für mich das Zählen zu tun.

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

Answer 4

Zusätzlich zu dtb Antwort können Sie diese in eine schöne einfache Klasse wickeln.

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

Zusätzlich zu dtb Antwort, wenn wir Rückrufe haben wollen.

using System;
using System.Runtime.Remoting.Messaging;
using System.Threading;

class Program
{
    static void Main(string[] args)
    {
        Main m = new Main();
        m.TestMRE();
        Console.ReadKey();

    }
}

class Main
{
    CalHandler handler = new CalHandler();
    int numberofTasks =0;
    public void TestMRE()
    {

        for (int j = 0; j <= 3; j++)
        {
            Console.WriteLine("Outer Loop is :" + j.ToString());
            ManualResetEvent signal = new ManualResetEvent(false);
            numberofTasks = 4;
            for (int i = 0; i <= 3; i++)
            {
                CalHandler.count caller = new CalHandler.count(handler.messageHandler);
                caller.BeginInvoke(i, new AsyncCallback(NumberCallback),signal);
            }
            signal.WaitOne();
        }

    }

    private void NumberCallback(IAsyncResult result)
    {
        AsyncResult asyncResult = (AsyncResult)result;

        CalHandler.count caller = (CalHandler.count)asyncResult.AsyncDelegate;

        int num = caller.EndInvoke(asyncResult);

        Console.WriteLine("Number is :"+ num.ToString());

        ManualResetEvent mre = (ManualResetEvent)asyncResult.AsyncState;
        if (Interlocked.Decrement(ref numberofTasks) == 0)
        {
            mre.Set();
        }
    }

}
public class CalHandler
{
    public delegate int count(int number);

    public int messageHandler ( int number )
    {
        return number;
    }

}

Answer 6

protected void WaitAllExt(WaitHandle[] waitHandles)
{
    //workaround for limitation of WaitHandle.WaitAll by <=64 wait handles
    const int waitAllArrayLimit = 64;
    var prevEndInd = -1;
    while (prevEndInd < waitHandles.Length - 1)
    {
        var stInd = prevEndInd + 1;
        var eInd = stInd + waitAllArrayLimit - 1;
        if (eInd > waitHandles.Length - 1)
        {
            eInd = waitHandles.Length - 1;
        }
        prevEndInd = eInd;

        //do wait
        var whSubarray = waitHandles.Skip(stInd).Take(eInd - stInd + 1).ToArray();
        WaitHandle.WaitAll(whSubarray);
    }

}

Answer 7

habe ich es einfach Paginieren die Anzahl der Ereignisse gelöst, ohne viel performace warten verloren, und es funktioniert perfekt auf die Produktionsumgebung. Folgt den Code:

        var events = new List<ManualResetEvent>();

        // code omited

        var newEvent = new ManualResetEvent(false);
        events.Add(newEvent);
        ThreadPool.QueueUserWorkItem(c => {

            //thread code
            newEvent.Set();
        });

        // code omited

        var wait = true;
        while (wait)
        {
            WaitHandle.WaitAll(events.Take(60).ToArray());
            events.RemoveRange(0, events.Count > 59 ? 60 : events.Count);
            wait = events.Any();

        }

Answer 8

Hier ist eine weitere Lösung. Hier ist die „Ereignisse“ wird eine Liste des Manual. Die Größe der Liste kann größer sein als 64 (MAX_EVENTS_NO).

int len = events.Count;
if (len <= MAX_EVENTS_NO)
    {
        WaitHandle.WaitAll(events.ToArray());
    } else {
        int start = 0;
        int num = MAX_EVENTS_NO;
        while (true)
            {
                if(start + num > len)
                {
                   num = len - start;
                }
                List<ManualResetEvent> sublist = events.GetRange(start, num);
                WaitHandle.WaitAll(sublist.ToArray());
                start += num;
                if (start >= len)
                   break;
           }
   }

Answer 9

Windows XP SP3 unterstützt maximal zwei waithandles. Für Fälle mehr als 2 waithandles Anwendung vorzeitig beendet wird.