int thisIndex = Interlocked.Increment(ref messagesIndex) & indexMask;
定义如下:
定义如下:
static int messagesIndex = -1;
public const int MaxMessages = 0x10000;
const int indexMask = MaxMessages-1;
messagesIndex
没有被其他代码访问。
如果我在单个线程中运行该代码数十亿次,不会出现任何错误。
如果我在几个线程上运行上面的代码,我会得到相同的数字两次,每1x千次跳过另一个数字。
我已经在6个线程上运行了下面的代码数十亿次,从未出现过错误:
int thisIndex = Interlocked.Increment(ref messagesIndex);
结论和问题
看起来 Interlocked.Increment()
单独使用是按预期工作的,但是 Interlocked.Increment()
和 indexMask
结合使用时却不行 :-(
有没有什么办法可以让它始终正常工作,而不仅仅是99.99%的时间?
我尝试将 Interlocked.Increment(ref messagesIndex)
分配给一个易失性整数变量,并在该变量上执行 "& indexMask"
操作:
[ThreadStatic]
volatile static int nextIncrement;
nextIncrement = Interlocked.Increment(ref mainIndexIncrementModTest);
indexes[testThreadIndex++] = nextIncrement & maskIncrementModTest;
当我将代码写成一行时,它会引起与将其写在多行中相同的问题。
反汇编
也许有人可以从反汇编中猜出编译器引入的问题:
indexes[testThreadIndex++] = Interlocked.Increment(ref mainIndexIncrementTest);
0000009a mov eax, dword ptr [ebp-48h]
0000009d mov dword ptr [ebp-58h], eax
000000a0 inc dword ptr [ebp-48h]
000000a3 mov eax, dword ptr [ebp-44h]
000000a6 mov dword ptr [ebp-5Ch], eax
000000a9 lea ecx, ds:[00198F84h]
000000af call 6D758403
000000b4 mov dword ptr [ebp-60h], eax
000000b7 mov eax, dword ptr [ebp-58h]
000000ba mov edx, dword ptr [ebp-5Ch]
000000bd cmp eax, dword ptr [edx+4]
000000c0 jb 000000C7
000000c2 call 6D9C2804
000000c7 mov ecx, dword ptr [ebp-60h]
000000ca mov dword ptr [edx+eax*4+8], ecx
indexes[testThreadIndex++] = Interlocked.Increment(ref mainIndexIncrementModTest) & maskIncrementModTest;
0000009a mov eax, dword ptr [ebp-48h]
0000009d mov dword ptr [ebp-58h], eax
000000a0 inc dword ptr [ebp-48h]
000000a3 mov eax, dword ptr [ebp-44h]
000000a6 mov dword ptr [ebp-5Ch], eax
000000a9 lea ecx,ds:[001D8F88h]
000000af call 6D947C8B
000000b4 mov dword ptr [ebp-60h], eax
000000b7 mov eax, dword ptr [ebp-60h]
000000ba and eax, 0FFFh
000000bf mov edx, dword ptr [ebp-58h]
000000c2 mov ecx, dword ptr [ebp-5Ch]
000000c5 cmp edx, dword ptr [ecx+4]
000000c8 jb 000000CF
000000ca call 6DBB208C
000000cf mov dword ptr [ecx+edx*4+8], eax
漏洞检测
为了发现漏洞,我会无限运行问题行并在6个线程中每个线程将返回的整数写入巨大的整数数组中。一段时间后,我停止线程并搜索所有六个整数数组,以确保每个数字只被返回一次(当然,我允许 "& indexMask" 操作)。
using System;
using System.Text;
using System.Threading;
namespace RealTimeTracer
{
class Test
{
#region Test Increment Multi Threads
// ----------------------------
const int maxThreadIndexIncrementTest = 0x200000;
static int mainIndexIncrementTest = -1; //the counter gets incremented before its use
static int[][] threadIndexThraces;
private static void testIncrementMultiThread()
{
const int maxTestThreads = 6;
Thread.CurrentThread.Name = "MainThread";
//start writer test threads
Console.WriteLine("start " + maxTestThreads + " test writer threads.");
Thread[] testThreads = testThreads = new Thread[maxTestThreads];
threadIndexThraces = new int[maxTestThreads][];
int testcycle = 0;
do
{
testcycle++;
Console.WriteLine("testcycle " + testcycle);
for (int testThreadIndex = 0; testThreadIndex < maxTestThreads; testThreadIndex++)
{
Thread testThread = new Thread(testIncrementThreadBody);
testThread.Name = "TestThread " + testThreadIndex;
testThreads[testThreadIndex] = testThread;
threadIndexThraces[testThreadIndex] = new int[maxThreadIndexIncrementTest+1]; //last int will be never used, but easier for programming
}
mainIndexIncrementTest = -1; //the counter gets incremented before its use
for (int testThreadIndex = 0; testThreadIndex < maxTestThreads; testThreadIndex++)
{
testThreads[testThreadIndex].Start(testThreadIndex);
}
//wait for writer test threads
Console.WriteLine("wait for writer threads.");
foreach (Thread testThread in testThreads)
{
testThread.Join();
}
//verify that EVERY index is used exactly by one thread.
Console.WriteLine("Verify");
int[] threadIndexes = new int[maxTestThreads];
for (int counter = 0; counter < mainIndexIncrementTest; counter++)
{
int threadIndex = 0;
for (; threadIndex < maxTestThreads; threadIndex++)
{
if (threadIndexThraces[threadIndex][threadIndexes[threadIndex]]==counter)
{
threadIndexes[threadIndex]++;
break;
}
}
if (threadIndex==maxTestThreads)
{
throw new Exception("Could not find index: " + counter);
}
}
} while (!Console.KeyAvailable);
}
public static void testIncrementThreadBody(object threadNoObject)
{
int threadNo = (int)threadNoObject;
int[] indexes = threadIndexThraces[threadNo];
int testThreadIndex = 0;
try
{
for (int counter = 0; counter < maxThreadIndexIncrementTest; counter++)
{
indexes[testThreadIndex++] = Interlocked.Increment(ref mainIndexIncrementTest);
}
}
catch (Exception ex)
{
OneTimeTracer.Trace(Thread.CurrentThread.Name + ex.Message);
}
}
#endregion
#region Test Increment Mod Multi Threads
// --------------------------------
const int maxThreadIndexIncrementModTest = 0x200000;
static int mainIndexIncrementModTest = -1; //the counter gets incremented before its use
const int maxIncrementModTest = 0x1000;
const int maskIncrementModTest = maxIncrementModTest - 1;
private static void testIncrementModMultiThread()
{
const int maxTestThreads = 6;
Thread.CurrentThread.Name = "MainThread";
//start writer test threads
Console.WriteLine("start " + maxTestThreads + " test writer threads.");
Thread[] testThreads = testThreads = new Thread[maxTestThreads];
threadIndexThraces = new int[maxTestThreads][];
int testcycle = 0;
do
{
testcycle++;
Console.WriteLine("testcycle " + testcycle);
for (int testThreadIndex = 0; testThreadIndex < maxTestThreads; testThreadIndex++)
{
Thread testThread = new Thread(testIncrementModThreadBody);
testThread.Name = "TestThread " + testThreadIndex;
testThreads[testThreadIndex] = testThread;
threadIndexThraces[testThreadIndex] = new int[maxThreadIndexIncrementModTest+1]; //last int will be never used, but easier for programming
}
mainIndexIncrementModTest = -1; //the counter gets incremented before its use
for (int testThreadIndex = 0; testThreadIndex < maxTestThreads; testThreadIndex++)
{
testThreads[testThreadIndex].Start(testThreadIndex);
}
//wait for writer test threads
Console.WriteLine("wait for writer threads.");
foreach (Thread testThread in testThreads)
{
testThread.Join();
}
//verify that EVERY index is used exactly by one thread.
Console.WriteLine("Verify");
int[] threadIndexes = new int[maxTestThreads];
int expectedIncrement = 0;
for (int counter = 0; counter < mainIndexIncrementModTest; counter++)
{
int threadIndex = 0;
for (; threadIndex < maxTestThreads; threadIndex++)
{
if (threadIndexes[threadIndex]<maxThreadIndexIncrementModTest &&
threadIndexThraces[threadIndex][threadIndexes[threadIndex]]==expectedIncrement)
{
threadIndexes[threadIndex]++;
expectedIncrement++;
if (expectedIncrement==maxIncrementModTest)
{
expectedIncrement = 0;
}
break;
}
}
if (threadIndex==maxTestThreads)
{
StringBuilder stringBuilder = new StringBuilder();
for (int threadErrorIndex = 0; threadErrorIndex < maxTestThreads; threadErrorIndex++)
{
int index = threadIndexes[threadErrorIndex];
if (index<0)
{
stringBuilder.AppendLine("Thread " + threadErrorIndex + " is empty");
}
else if (index==0)
{
stringBuilder.AppendLine("Thread " + threadErrorIndex + "[0]=" +
threadIndexThraces[threadErrorIndex][0]);
}
else if (index>=maxThreadIndexIncrementModTest)
{
stringBuilder.AppendLine("Thread " + threadErrorIndex + "[" + (index-1) + "]=" +
threadIndexThraces[threadErrorIndex][maxThreadIndexIncrementModTest-2] + ", " +
threadIndexThraces[threadErrorIndex][maxThreadIndexIncrementModTest-1]);
}
else
{
stringBuilder.AppendLine("Thread " + threadErrorIndex + "[" + (index-1) + "]=" +
threadIndexThraces[threadErrorIndex][index-1] + ", " +
threadIndexThraces[threadErrorIndex][index]);
}
}
string exceptionString = "Could not find index: " + expectedIncrement + " for counter " + counter + Environment.NewLine + stringBuilder.ToString();
Console.WriteLine(exceptionString);
return;
//throw new Exception(exceptionString);
}
}
} while (!Console.KeyAvailable);
}
public static void testIncrementModThreadBody(object threadNoObject)
{
int threadNo = (int)threadNoObject;
int[] indexes = threadIndexThraces[threadNo];
int testThreadIndex = 0;
try
{
for (int counter = 0; counter < maxThreadIndexIncrementModTest; counter++)
{
// indexes[testThreadIndex++] = Interlocked.Increment(ref mainIndexIncrementModTest) & maskIncrementModTest;
int nextIncrement = Interlocked.Increment(ref mainIndexIncrementModTest);
indexes[testThreadIndex++] = nextIncrement & maskIncrementModTest;
}
}
catch (Exception ex)
{
OneTimeTracer.Trace(Thread.CurrentThread.Name + ex.Message);
}
}
#endregion
}
}
这会产生以下错误:
6个int数组的内容(每个线程一个)
线程0 [30851] = 2637,2641
线程1 [31214] = 2639,2644
线程2 [48244] = 2638,2643
线程3 [26512] = 2635,2642
线程4 [0] = 2636,2775
线程5 [9173] = 2629,2636 解释:
线程4使用了2636
线程5也使用了2636 !!!! 这不应该发生
线程0使用了2637
线程2使用了2638
线程1使用了2639
2640没有被任何线程使用 !!! 这就是测试检测到的错误
线程0使用了2641
线程3使用了2642
&
是两种操作,它们组合后不一定是原子性的。使用锁来解决这种操作的组合。 - Candide