c# - 在多线程环境中使用 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");
        }