Как реализовать класс Barrier из функциональности .NET 4 в .NET 3.5

Question

По некоторым причинам я должен придерживаться .NET 3.5, и мне нужна функциональность класса Barrier из .NET 4. У меня есть несколько потоков, которые выполняют некоторую работу, и я хочу, чтобы они ждали друг друга, пока все не будет сделано , Когда все будет сделано, я хочу, чтобы они делали работу снова и снова подобным образом. Воодушевленный потоком Разница между Barrier в C # 4.0 и WaitHandle в C # 3.0? Я решил реализовать функциональность Barrier с классами AutoResetEvent и WaitHandle. Хотя я столкнулся с проблемой с моим кодом:

class Program
{
    const int numOfThreads = 3;

    static AutoResetEvent[] barrier = new AutoResetEvent[numOfThreads];
    static Random random = new Random(System.DateTime.Now.Millisecond);

    static void barriers2(object barrierObj)
    {
        AutoResetEvent[] barrierLocal = (AutoResetEvent[])barrierObj;
        string name = Thread.CurrentThread.Name;
        for (int i = 0; i < 10; i++)
        {
            int sleepTime = random.Next(2000, 10000);
            System.Console.Out.WriteLine("Thread {0} at the 'barrier' will sleep for {1}.", name, sleepTime);
            Thread.Sleep(sleepTime);
            System.Console.Out.WriteLine("Thread {0} at the 'barrier' with time {1}.", name, sleepTime);
            int currentId = Convert.ToInt32(name);
            //for(int z = 0; z < numOfThreads; z++)
                barrierLocal[currentId].Set();
            WaitHandle.WaitAll(barrier);
            /*
            for (int k = 0; k < numOfThreads; k++)
            {
                if (k == currentId)
                {
                    continue;
                }
                System.Console.Out.WriteLine("Thread {0} is about to wait for the singla from thread: {1}", name, k);
                barrierLocal[k].WaitOne();
                System.Console.Out.WriteLine("Thread {0} is about to wait for the singla from thread: {1}. done", name, k);
            }
            */
        }
    }

    static void Main(string[] args)
    {
        for (int i = 0; i < numOfThreads; i++)
        {
            barrier[i] = new AutoResetEvent(false);
        }
        for (int i = 0; i < numOfThreads; i++)
        {
            Thread t = new Thread(Program.barriers2);
            t.Name = Convert.ToString(i);
            t.Start(barrier);
        }
    }
}

Вывод, который я получаю, выглядит следующим образом:

Поток 0 у «барьера» будет спать на 7564 Поток 1 у «барьера» будет спать 5123 Поток 2 у «барьера» будет спать 4237 Резьба 2 у «барьера» со временем 4237 Поток 1 у «барьера» со временем 5123 Нить 0 у «барьера» со временем 7564 Поток 0 у «барьера» будет спать 8641 Нить 0 у «барьера» со временем 8641

И это все. После последней строки больше нет вывода, и приложение не завершает работу. Похоже, что есть какой-то тупик. Однако не могу найти проблему. Любая помощь приветствуется.

Спасибо!

Answer 1

Это потому, что вы используете AutoResetEvent. Один из вызовов WaitAll () потока будет завершен первым. Что автоматически вызывает Reset () для всех ARE. Что не позволяет другим потокам завершать вызовы WaitAll ().

Здесь требуется ManualResetEvent.

Answer 2

Вот моя реализация, которую я использую для своей игры XNA .Когда я писал это, барьер был недоступен, и я все еще застрял в .Net 3.5.Требуется три набора ManualResetEvents и массив счетчиков для поддержания фазы.

using System;
using System.Threading;

namespace Colin.Threading
{
    /// <summary>
    /// Threading primitive for "barrier" sync, where N threads must stop at certain points 
    /// and wait for all their bretheren before continuing.
    /// </summary>
    public sealed class NThreadGate
    {
        public int mNumThreads;
        private ManualResetEvent[] mEventsA;
        private ManualResetEvent[] mEventsB;
        private ManualResetEvent[] mEventsC;
        private ManualResetEvent[] mEventsBootStrap;
        private Object mLockObject;
        private int[] mCounter;
        private int mCurrentThreadIndex = 0;

        public NThreadGate(int numThreads)
        {
            this.mNumThreads = numThreads;

            this.mEventsA = new ManualResetEvent[this.mNumThreads];
            this.mEventsB = new ManualResetEvent[this.mNumThreads];
            this.mEventsC = new ManualResetEvent[this.mNumThreads];
            this.mEventsBootStrap = new ManualResetEvent[this.mNumThreads];
            this.mCounter = new int[this.mNumThreads];
            this.mLockObject = new Object();

            for (int i = 0; i < this.mNumThreads; i++)
            {
                this.mEventsA[i] = new ManualResetEvent(false);
                this.mEventsB[i] = new ManualResetEvent(false);
                this.mEventsC[i] = new ManualResetEvent(false);
                this.mEventsBootStrap[i] = new ManualResetEvent(false);
                this.mCounter[i] = 0;
            }
        }

        /// <summary>
        /// Adds a new thread to the gate system.
        /// </summary>
        /// <returns>Returns a thread ID for this thread, to be used later when waiting.</returns>
        public int AddThread()
        {
            lock (this.mLockObject)
            {
                this.mEventsBootStrap[this.mCurrentThreadIndex].Set();
                this.mCurrentThreadIndex++;
                return this.mCurrentThreadIndex - 1;
            }
        }

        /// <summary>
        /// Stop here and wait for all the other threads in the NThreadGate. When all the threads have arrived at this call, they
        /// will unblock and continue.
        /// </summary>
        /// <param name="myThreadID">The thread ID of the caller</param>
        public void WaitForOtherThreads(int myThreadID)
        {
            // Make sure all the threads are ready.
            WaitHandle.WaitAll(this.mEventsBootStrap);

            // Rotate between three phases.
            int phase = this.mCounter[myThreadID];
            if (phase == 0)        // Flip
            {
                this.mEventsA[myThreadID].Set();
                WaitHandle.WaitAll(this.mEventsA);
                this.mEventsC[myThreadID].Reset();
            }
            else if (phase == 1)    // Flop
            {
                this.mEventsB[myThreadID].Set();
                WaitHandle.WaitAll(this.mEventsB);
                this.mEventsA[myThreadID].Reset();
            }
            else    // Floop
            {
                this.mEventsC[myThreadID].Set();
                WaitHandle.WaitAll(this.mEventsC);
                this.mEventsB[myThreadID].Reset();
                this.mCounter[myThreadID] = 0;
                return;
            }

            this.mCounter[myThreadID]++;
        }
    }
}

Настройка шлюза потока:

private void SetupThreads()
{
    // Make an NThreadGate for N threads.
    this.mMyThreadGate = new NThreadGate(Environment.ProcessorCount);

    // Make some threads...
    // e.g. new Thread(new ThreadStart(this.DoWork);
}

Метод рабочего потока:

private void DoWork()
{
    int localThreadID = this.mMyThreadGate.AddThread();

    while (this.WeAreStillRunning)
    {
        // Signal this thread as waiting at the barrier
        this.mMyThreadGate.WaitForOtherThreads(localThreadID);

        // Synchronized work here...

        // Signal this thread as waiting at the barrier
        this.mMyThreadGate.WaitForOtherThreads(localThreadID);

        // Synchronized work here...

        // Signal this thread as waiting at the barrier
        this.mMyThreadGate.WaitForOtherThreads(localThreadID);
    }
}

Answer 3

Загрузите Reactive Extensions backport для .NET 3.5. Вы найдете класс Barrier вместе с другими полезными параллельными структурами данных и механизмами синхронизации, выпущенными в .NET 4.0.