为什么这个多线程代码会在某些时候打印6？

Question

我正在创建两个线程,并向它们传递一个函数,该函数执行下面显示的代码10,000,000次.

大多数情况下,"5"打印到控制台.有时它是"3"或"4".很清楚为什么会这样.

但是,它也打印"6".这怎么可能？

class Program
{
    private static int _state = 3;

    static void Main(string[] args)
    {
        Thread firstThread = new Thread(Tr);
        Thread secondThread = new Thread(Tr);

        firstThread.Start();
        secondThread.Start();

        firstThread.Join();
        secondThread.Join();

        Console.ReadLine();
    }

    private static void Tr()
    {
        for (int i = 0; i < 10000000; i++)
        {
            if (_state == 3)
            {
                _state++;
                if (_state != 4)
                {
                    Console.Write(_state);
                }
                _state = 3;
            }
        }
    }
}

这是输出:

Answer 1

我想我已经弄清了导致这个问题的一系列事件:

线程1进入 if (_state == 3)

上下文切换

线程2进入if (_state == 3)
线程2增量状态(state = 4)

上下文切换

线程1 读 _state作4

上下文切换

线程2设置_state = 3
线程2进入if (_state == 3)

上下文切换

线程1执行 _state = 4 + 1

上下文切换

线程2 _state在5
线程2执行时读取_state = 5 + 1;

Answer 2

这是典型的竞争条件.编辑:事实上,有多种竞争条件.

它可以在_state3的任何时间发生,并且两个线程都if可以通过单个核心中的上下文切换同时到达语句,或者同时并行地在多个核心中.

这是因为++操作员首先读取_state然后递增它.在第一个if声明它会读5或甚至6 之后,有可能会有足够的时间.

编辑:如果你为N个线程推广这个例子,你可能会观察到一个高达3 + N + 1的数字.

当线程开始运行或者刚刚设置_state为3 时,这可能是正确的.

要避免这种情况,请在if语句周围使用锁定,或使用Interlocked访问权限_state,例如if (System.Threading.Interlocked.CompareAndExchange(ref _state, 3, 4) == 3)和System.Threading.Interlocked.Exchange(ref _state, 3).

如果你想保持竞争条件,你应该声明_state为volatile,否则你冒险在每个线程看到_state本地没有来自其他线程的更新.

或者,如果您将实现切换为变量和捕获该变量的闭包,则可以使用System.Threading.Volatile.Read和System.Threading.Volatile.Write,因为局部变量不能(并且将无法)声明.在这种情况下,甚至必须使用易失性写入进行初始化._stateTrvolatile

编辑:如果我们通过扩展每个读取稍微更改代码,也许竞争条件更明显:

    // Without some sort of memory barrier (volatile, lock, Interlocked.*),
    // a thread is allowed to see _state as if other threads hadn't touched it
    private static volatile int _state = 3;

// ...

        for (int i = 0; i < 10000000; i++)
        {
            int currentState;
            currentState = _state;
            if (currentState == 3)
            {
                // RACE CONDITION: re-read the variable
                currentState = _state;
                currentState = currentState + 1:
                // RACE CONDITION: non-atomic write
                _state = currentState;

                currentState = _state;
                if (currentState != 4)
                {
                    // RACE CONDITION: re-read the variable
                    currentState = _state;
                    Console.Write(currentState);
                }
                _state = 3;
            }
        }

我在_state可能与之前的变量读取语句假设不同的地方添加了注释.

这是一个很长的图表,它显示甚至可以连续两次打印6次,每次打印一次,就像op发布的图像一样.请记住,线程可能无法同步运行,通常是由于抢先上下文切换,缓存停顿或核心速度差异(由于省电或临时turbo速度):

这个类似于原始类,但它使用Volatile类,其中state现在是闭包捕获的变量.易失性访问的数量和顺序变得明显:

    static void Main(string[] args)
    {
        int state = 3;

        ThreadStart tr = () =>
        {
            for (int i = 0; i < 10000000; i++)
            {
                if (Volatile.Read(ref state) == 3)
                {
                    Volatile.Write(ref state, Volatile.Read(state) + 1);
                    if (Volatile.Read(ref state) != 4)
                    {
                        Console.Write(Volatile.Read(ref state));
                    }
                    Volatile.Write(ref state, 3);
                }
            }
        };

        Thread firstThread = new Thread(tr);
        Thread secondThread = new Thread(tr);

        firstThread.Start();
        secondThread.Start();

        firstThread.Join();
        secondThread.Join();

        Console.ReadLine();
    }

一些线程安全的方法:

    private static object _lockObject;

// ...

        // Do not allow concurrency, blocking
        for (int i = 0; i < 10000000; i++)
        {
            lock (_lockObject)
            {
                // original code
            }
        }

        // Do not allow concurrency, non-blocking
        for (int i = 0; i < 10000000; i++)
        {
            bool lockTaken = false;
            try
            {
                Monitor.TryEnter(_lockObject, ref lockTaken);
                if (lockTaken)
                {
                    // original code
                }
            }
            finally
            {
                if (lockTaken) Monitor.Exit(_lockObject);
            }
        }

        // Do not allow concurrency, non-blocking
        for (int i = 0; i < 10000000; i++)
        {
            // Only one thread at a time will succeed in exchanging the value
            try
            {
                int previousState = Interlocked.CompareExchange(ref _state, 4, 3);
                if (previousState == 3)
                {
                    // Allow race condition on purpose (for no reason)
                    int currentState = Interlocked.CompareExchange(ref _state, 0, 0);
                    if (currentState != 4)
                    {
                        // This branch is never taken
                        Console.Write(currentState);
                    }
                }
            }
            finally
            {
                Interlocked.CompareExchange(ref _state, 3, 4);
            }
        }

        // Allow concurrency
        for (int i = 0; i < 10000000; i++)
        {
            // All threads increment the value
            int currentState = Interlocked.Increment(ref _state);
            if (currentState == 4)
            {
                // But still, only one thread at a time enters this branch
                // Allow race condition on purpose (it may actually happen here)
                currentState = Interlocked.CompareExchange(ref _state, 0, 0);
                if (currentState != 4)
                {
                    // This branch might be taken with a maximum value of 3 + N
                    Console.Write(currentState);
                }
            }
            Interlocked.Decrement(ref _state);
        }

这个有点不同,它需要_state在增量之后的最后已知值执行某些操作:

        // Allow concurrency
        for (int i = 0; i < 10000000; i++)
        {
            // All threads increment the value
            int currentState = Interlocked.Increment(ref _state);
            if (currentState != 4)
            {
                // Only the thread that incremented 3 will not take the branch
                // This can happen indefinitely after the first increment for N > 1
                // This branch might be taken with a maximum value of 3 + N
                Console.Write(currentState);
            }
            Interlocked.Decrement(ref _state);
        }

请注意,Interlocked.Increment/ Interlocked.Decrement示例并不安全,与lock/ Monitor和Interlocked.CompareExchange示例不同,因为没有可靠的方法来了解增量是否成功.

一种常见的方法是递增,然后使用try/ finally在finally块中递减的位置.但是,可能会抛出异步异常(例如ThreadAbortException)

异步异常可能会抛出到意外的位置,可能是每个机器指令:ThreadAbortException,StackOverflowException和OutOfMemoryException.

另一种方法是初始化currentState为低于3的值并在finally块中有条件地减少.但同样,在Interlocked.Increment返回和currentState分配给结果之间,可能会发生异步异常,因此currentState即使Interlocked.Increment成功,仍然可以具有初始值.