حل بديل لـ Waithandle.Waitall 64 LIME؟

Question

يولد طلبي الكثير من خيوط العمال الصغيرة المختلفة عبر ThreadPool.QueueUserWorkItem الذي أتتبعه عبر متعدد ManualResetEvent حالات. انا استعمل ال WaitHandle.WaitAll طريقة لمنع طلبي من الإغلاق حتى تكتمل هذه الخيوط.

ومع ذلك ، لم أواجه أي مشاكل من قبل ، حيث أن طلبي يخضع لمزيد من التحميل ، أي المزيد من المواضيع التي يتم إنشاؤها ، لقد بدأت الآن في الحصول على هذا الاستثناء:

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

ما هو أفضل حل بديل لهذا؟

رمز مقتطف

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

الحل البديل

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

قم بإنشاء متغير يتتبع عدد مهام التشغيل:

int numberOfTasks = 100;

إنشاء إشارة:

ManualResetEvent signal = new ManualResetEvent(false);

انخفاض عدد المهام كلما انتهت المهمة:

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

إذا لم تكن هناك مهمة متبقية ، فقم بتعيين الإشارة:

    signal.Set();
}

وفي الوقت نفسه ، في مكان آخر ، انتظر حتى يتم تعيين الإشارة:

signal.WaitOne();

Answer 2

بدءًا من .NET 4.0 ، لديك خياران آخران (و IMO ، نظافة) متاح لك.

الأول هو استخدام CountdownEvent صف دراسي. يمنع الحاجة إلى الاضطرار إلى التعامل مع زيادة وتناقص بمفردك:

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

ومع ذلك ، هناك حل أكثر قوة ، وهذا لاستخدام Task صف دراسي, ، مثل ذلك:

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

باستخدام Task الفصل والدعوة إلى WaitAll هو أكثر نظافة ، IMO ، لأنك تنسج بدايات الخيوط أقل في جميع أنحاء الكود (إشعار ، لا توجد مقابض انتظار) ؛ ليس عليك إعداد عداد ، والتعامل مع زيادة/انخفاض ، فقط قمت بإعداد مهامك ثم انتظرها. هذا يتيح أن يكون الرمز أكثر تعبيراً في ماذا او ما لما تريد القيام به وليس بدائيات كيف (على الأقل ، من حيث إدارة موازاة ذلك).

.NET 4.5 يوفر المزيد من الخيارات ، يمكنك تبسيط توليد تسلسل Task مثيلات عن طريق الاتصال ثابتة Run طريقة على Task صف دراسي:

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

أو يمكنك الاستفادة من مكتبة TPL DataFlow (إنه في System مساحة الاسم ، لذلك فهي رسمية ، على الرغم من أنها تنزيل من Nuget ، مثل Entity Framework) واستخدم ActionBlock<TInput>, ، مثل ذلك:

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

نلاحظ أن ActionBlock<TInput> بشكل افتراضي ، يقوم عنصر واحد في وقت واحد ، لذلك إذا كنت ترغب في معالجة إجراءات متعددة في وقت واحد ، فيجب عليك تعيين عدد العناصر المتزامنة التي تريد معالجتها في المُنشئ عن طريق تمرير أ ExecutionDataflowBlockOptions مثيل ووضع ملف MaxDegreeOfParallelism منشأه:

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

إذا كان إجراءاتك آمنًا حقًا ، فيمكنك تعيين ملف MaxDegreeOfParallelsim خاصية ل DataFlowBlockOptions.Unbounded:

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

النقطة المهمة ، لديك سيطرة جيدة على الحبيبات كيف بالتوازي ، تريد أن تكون خياراتك.

بالطبع ، إذا كان لديك سلسلة من العناصر التي تريد نقلها إلى ActionBlock<TInput> مثيل ، ثم يمكنك ربط ISourceBlock<TOutput> تنفيذ لتغذية ActionBlock<TInput>, ، مثل ذلك:

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

اعتمادًا على ما تحتاج إلى القيام به ، تصبح مكتبة TPL DataFlow كثير خيار أكثر جاذبية ، لأنه يتعامل مع التزامن عبر الكل المهام مرتبطة معًا ، وتتيح لك أن تكون محددًا جدًا فقط كيف تريد أن تكون كل قطعة ، مع الحفاظ على الفصل المناسب للمخاوف لكل كتلة.

Answer 3

الحل الخاص بك غير صحيح. السبب هو أن Set و WaitOne يمكن أن يسبق السباق إذا كان عنصر العمل الأخير يتسبب في threadCount للذهاب إلى الصفر قبل كان على موضوع قائمة الانتظار فرصة لقائم الانتظار الكل عناصر العمل. الإصلاح بسيط. تعامل مع موضوع قائمة الانتظار كما لو كان عنصر عمل نفسه. تهيئة threadCount إلى 1 وقم بتناقص وإشارة عند اكتمال قائمة الانتظار.

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

كتفضيل شخصي أحب استخدام CountdownEvent الفصل للقيام بالعد بالنسبة لي.

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

Answer 4

إضافة إلى إجابة DTB ، يمكنك لفه في فصل بسيط لطيف.

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

إضافة إلى إجابة DTB عندما نريد إجراء عمليات الاسترجاعات.

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

لقد قمت بحلها ببساطة عن طريق تراكم عدد الأحداث التي يجب الانتظار دون فقدان الكثير من الأداء ، وهي تعمل بشكل مثالي على بيئة الإنتاج. يتبع الرمز:

        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

هنا حل آخر. هنا "الأحداث" هي قائمة من ManualResetEvent. يمكن أن يكون حجم القائمة أكبر من 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 كحد أقصى اثنين من Waithandles. بالنسبة للحالات ، ينتهي تطبيق Waithandles قبل الأوان.