在多线程环境中结合使用Interlocked.Decrement、monitor.wait和monitor.pulse时的错误行为

Question

我正在尝试实现一个多线程库，它将使用线程池同时运行任务。基本上，它将从接收到的收集参数中向线程池添加任务，然后等待，直到正在处理的最后一个任务发送脉冲信号。我在早期的测试中取得了成功，但当我想要测试那些处理起来非常短的任务时，我遇到了一个奇怪的问题。不知怎么的，要么脉冲信号是在等待命令在主线程中发生之前发送的，要么就是发生了一些我根本看不到的事情，无论我如何努力进行同步。

为了纠正我的问题，我实现了另一个“不太可取的”解决方案，因为我权衡了潜在的性能好处，到目前为止工作得很好，但我想知道为什么我的第一种方法在这种情况下首先不起作用，即使在性能方面两者之间没有太大的区别。

为了说明这一点，我在简化下面的过程后添加了这两个解决方案。有人能帮我指出哪里出了问题吗？

提前谢谢。

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading;
using System.Diagnostics;

namespace TestcodeBenchmark
{
    class Program
    {
        static int remainingTasks = 10000000;
        static Stopwatch casioF91W = new Stopwatch();
        static Random rg = new Random();
        static readonly object waitObject = new object();


        static void Main(string[] args)
        {
            TestLoop(30, remainingTasks);
            Console.ReadKey();
        }

        private static void TestLoop(int loopCount, int remainingCountResetNumber)
        {
            for (int i = 0; i < loopCount; i++)
            {
                remainingTasks = remainingCountResetNumber;
                //When this method is called it eventualy stuck at Monitor.Wait line
                TestInterlocked();

                remainingTasks = remainingCountResetNumber;
                //When this method is called it processes stuff w/o any issues.
                TestManualLock();
                Console.WriteLine();
            }
        }

        private static void TestInterlocked()
        {
            casioF91W.Restart();
            //for (int i = 0; i < remainingTasks; i++)
            //{
            //    ThreadPool.QueueUserWorkItem(delegate { TestInterlockedDecrement(); });
            //}
            int toStart = remainingTasks;
            //for (int i = 0; i < remainingTasks; i++)
            for (int i = 0; i < toStart; i++)
            {
                if (!ThreadPool.QueueUserWorkItem(delegate { TestInterlockedDecrement(); }))
                    Console.WriteLine("Queue failed");
            }
            //lock waitObject to be able to call Monitor.Wait
            lock (waitObject)
            {
                //if waitObject is locked then no worker thread should be able to send a pulse signal
                //however, if pulse signal was sent before locking here remainingTasks should be
                //zero so don't wait if all tasks are processed already
                if (remainingTasks != 0)
                {
                    //release the lock on waitObject and wait pulse signal from the worker thread that 
                    //finishes last task
                    Monitor.Wait(waitObject);
                }
            }
            casioF91W.Stop();
            Console.Write("Interlocked:{0}ms ", casioF91W.ElapsedMilliseconds);
        }

        private static void TestInterlockedDecrement()
        {
            //process task
            //TestWork();
            //Once processing finishes decrement 1 from remainingTasks using Interlocked.Decrement
            //to make sure it is atomic and therefore thread safe. If resulting value is zero
            //send pulse signal to wake main thread.            
            if (Interlocked.Decrement(ref remainingTasks) == 0)
            {
                //Acquire a lock on waitObject to be able to send pulse signal to main thread. If main 
                //thread acquired the lock earlier, this will wait until main thread releases it
                lock (waitObject)
                {
                    //send a pulse signal to main thread to continue
                    Monitor.PulseAll(waitObject);
                }
            }
        }

        private static void TestManualLock()
        {
            casioF91W.Restart();

            //Acquire the lock on waitObject and don't release it until all items are added and
            //Wait method is called. This will ensure wait method is called in main thread
            //before any worker thread can send pulse signal by requiring worker threads to
            //lock waitObject to be able to modify remainingTasks            
            lock (waitObject)
            {
                for (int i = 0; i < remainingTasks; i++)
                {
                    ThreadPool.QueueUserWorkItem(delegate { TestManualDecrement(); });
                }
                Monitor.Wait(waitObject);
            }
            casioF91W.Stop();
            Console.Write("ManualLock:{0}ms ", casioF91W.ElapsedMilliseconds);
        }

        private static void TestManualDecrement()
        {
            //TestWork();
            //try to acquire lock on wait object.
            lock (waitObject)
            {
                //if lock is acquired, decrement remaining tasks by and then check
                //whether resulting value is zero.
                if (--remainingTasks == 0)
                {
                    //send a pulse signal to main thread to continue
                    Monitor.PulseAll(waitObject);
                }
            }
        }

        private static void TestWork()
        {
            //Uncomment following to simulate some work.
            //int i = rg.Next(100, 110);
            //for (int j = 0; j < i; j++)
            //{

            //}
        }
    }
}

Answer 1

当你开始你的任务时，你的循环就是启动remainingTasks任务。然而，当你接近10000的时候，一些任务已经完成，并将这个数字减少到10000以下，所以你没有开始适当数量的任务。如果我修改您的循环以保存应该启动的任务数，代码将成功运行。(请注意，您还应该检查QueueUserWorkItem的返回值。)

        int toStart = remainingTasks;
        for (int i = 0; i < toStart; i++)
        {
            if (!ThreadPool.QueueUserWorkItem(delegate { TestInterlockedDecrement(); }))
                Console.WriteLine("Queue failed");
        }