我通常不会抱怨有人努力实现快速代码:这通常是一个非常好的练习,可以带来对编程的更好理解和更快的代码。
我在这里也不会抱怨,但我可以明确地说,在x86架构上,追求长度为3条指令或更多的快速自旋锁是徒劳无功的。
原因如下:
使用典型代码序列调用自旋锁
lock_variable DW 0 ; 0 <=> free
mov ebx,offset lock_variable
mov eax,1
xchg eax,[ebx]
; if eax contains 0 (no one owned it) you own the lock,
; if eax contains 1 (someone already does) you don't
释放自旋锁很简单。
mov ebx,offset lock_variable
mov dword ptr [ebx],0
xchg指令会在处理器上引发锁定信号,这实际上意味着我想在接下来的几个时钟周期内使用总线。该信号通过高速缓存传播并到达最慢的总线主设备,通常是PCI总线。当每个总线主设备完成锁定确认信号时,会发送回一个locka信号。然后进行实际的交换。问题是,锁定/锁定确认序列需要非常长的时间。PCI总线可能以33MHz的速度运行,具有多个周期的延迟。在3.3 GHz的CPU上,这意味着每个PCI总线周期需要100个CPU周期。
作为经验法则,我假设锁定需要300到3000个CPU周期才能完成,最终我不知道是否拥有锁定。因此,“快速”自旋锁可以节省的几个周期将是一种幻觉,因为没有两个锁是相同的,它将取决于您在那段短时间内的总线情况。
______________编辑______________
我刚刚读到自旋锁是“频繁使用的对象”。嗯,显然你不明白每次调用自旋锁都会消耗大量的CPU周期。或者换句话说,每次调用自旋锁都会损失您的大量处理能力。
使用自旋锁(或其更大的兄弟——临界区)的技巧是尽可能地少用它们,同时仍然实现预期的程序功能。到处使用它们很容易,结果会导致性能不佳。
这不仅仅是关于编写快速代码,还包括组织数据。当你写“在线程之间复制小结构”时,你应该意识到锁可能需要比实际复制花费数百倍的时间才能完成。
________________编辑________________
当您计算平均锁定时间时,它可能几乎没有任何作用,因为它是在您的机器上测量的,而这可能不是预期的目标(可能具有完全不同的总线使用特性)。对于您的机器,平均值由单个非常快的时间(当总线主控活动不干扰时)组成,一直到非常慢的时间(当总线主控干扰显著时)。
您可以引入代码来确定最快和最慢的情况,并计算商以查看自旋锁时间可以变化多大。
________________编辑________________
2016年5月更新。
Peter Cordes提出了“在没有竞争的情况下调整锁可能是有意义的”的想法,并且现代CPU上除非锁变量未对齐,否则不会出现许多数百个时钟周期的锁定时间。我开始怀疑我的
以前的测试程序 - 使用32位Watcom C编写 - 是否会受到WOW64的阻碍,因为它在64位操作系统Windows 7上运行。
因此,我编写了一个64位程序,并使用TDM的gcc 5.3进行编译。该程序利用隐式总线锁定指令变体“XCHG r,m”进行锁定,使用简单的赋值“MOV m,r”进行解锁。在某些锁变体中,将预先测试锁变量以确定是否有可能尝试锁定(使用简单的比较“CMP r,m”,可能不会越过L3)。这就是它:
#define CLASSIC_BUS_LOCK
#define WHILE_PRETEST
typedef unsigned char u1;
typedef unsigned short u2;
typedef unsigned long u4;
typedef unsigned int ud;
typedef unsigned long long u8;
typedef signed char i1;
typedef short i2;
typedef long i4;
typedef int id;
typedef long long i8;
typedef float f4;
typedef double f8;
#define usizeof(a) ((ud)sizeof(a))
#define LOOPS 25000000
#include <stdio.h>
#include <windows.h>
#ifndef bool
typedef signed char bool;
#endif
u8 CPU_rdtsc (void)
{
ud tickl, tickh;
__asm__ __volatile__("rdtsc":"=a"(tickl),"=d"(tickh));
return ((u8)tickh << 32)|tickl;
}
volatile u8 bus_lock (volatile u8 * block, u8 value)
{
__asm__ __volatile__( "xchgq %1,%0" : "=r" (value) : "m" (*block), "0" (value) : "memory");
return value;
}
void bus_unlock (volatile u8 * block, u8 value)
{
__asm__ __volatile__( "movq %0,%1" : "=r" (value) : "m" (*block), "0" (value) : "memory");
}
void rfence (void)
{
__asm__ __volatile__( "lfence" : : : "memory");
}
void rwfence (void)
{
__asm__ __volatile__( "mfence" : : : "memory");
}
void wfence (void)
{
__asm__ __volatile__( "sfence" : : : "memory");
}
volatile bool LOCK_spinlockPreTestIfFree (const volatile u8 *lockVariablePointer)
{
return (bool)(*lockVariablePointer == 0ull);
}
volatile bool LOCK_spinlockFailed (volatile u8 *lockVariablePointer)
{
return (bool)(bus_lock (lockVariablePointer, 1ull) != 0ull);
}
void LOCK_spinlockLeave (volatile u8 *lockVariablePointer)
{
*lockVariablePointer = 0ull;
}
static volatile u8 lockVariable = 0ull,
lockCounter = 0ull;
static volatile i8 threadHold = 1;
static u8 tstr[4][32];
struct LOCKING_THREAD_STRUCTURE
{
u8 numberOfFailures, numberOfPreTests;
f8 clocksPerLock, failuresPerLock, preTestsPerLock;
u8 threadId;
HANDLE threadHandle;
ud idx;
} *lts[4] = {(void *)tstr[0], (void *)tstr[1], (void *)tstr[2], (void *)tstr[3]};
DWORD WINAPI locking_thread (struct LOCKING_THREAD_STRUCTURE *ltsp)
{
ud n = LOOPS;
u8 clockCycles;
SetThreadAffinityMask (ltsp->threadHandle, 1ull<<ltsp->idx);
while (threadHold) {}
clockCycles = CPU_rdtsc ();
while (n)
{
Sleep (0);
#ifdef CLASSIC_BUS_LOCK
while (LOCK_spinlockFailed (&lockVariable)) {++ltsp->numberOfFailures;}
#else
#ifdef WHILE_PRETEST
while (1)
{
do
{
++ltsp->numberOfPreTests;
} while (!LOCK_spinlockPreTestIfFree (&lockVariable));
if (!LOCK_spinlockFailed (&lockVariable)) break;
++ltsp->numberOfFailures;
}
#else
while (1)
{
++ltsp->numberOfPreTests;
if (LOCK_spinlockPreTestIfFree (&lockVariable))
{
if (!LOCK_spinlockFailed (&lockVariable)) break;
++ltsp->numberOfFailures;
}
}
#endif
#endif
++lockCounter;
LOCK_spinlockLeave (&lockVariable);
#ifdef CLASSIC_BUS_LOCK
while (LOCK_spinlockFailed (&lockVariable)) {++ltsp->numberOfFailures;}
#else
#ifdef WHILE_PRETEST
while (1)
{
do
{
++ltsp->numberOfPreTests;
} while (!LOCK_spinlockPreTestIfFree (&lockVariable));
if (!LOCK_spinlockFailed (&lockVariable)) break;
++ltsp->numberOfFailures;
}
#else
while (1)
{
++ltsp->numberOfPreTests;
if (LOCK_spinlockPreTestIfFree (&lockVariable))
{
if (!LOCK_spinlockFailed (&lockVariable)) break;
++ltsp->numberOfFailures;
}
}
#endif
#endif
--lockCounter;
LOCK_spinlockLeave (&lockVariable);
n-=2;
}
clockCycles = CPU_rdtsc ()-clockCycles;
ltsp->clocksPerLock = (f8)clockCycles/ (f8)LOOPS;
ltsp->failuresPerLock = (f8)ltsp->numberOfFailures/(f8)LOOPS;
ltsp->preTestsPerLock = (f8)ltsp->numberOfPreTests/(f8)LOOPS;
ltsp->idx = 4u;
ExitThread (0);
return 0;
}
int main (int argc, char *argv[])
{
u8 processAffinityMask, systemAffinityMask;
memset (tstr, 0u, usizeof(tstr));
lts[0]->idx = 3;
lts[1]->idx = 2;
lts[2]->idx = 1;
lts[3]->idx = 0;
GetProcessAffinityMask (GetCurrentProcess(), &processAffinityMask, &systemAffinityMask);
SetPriorityClass (GetCurrentProcess(), HIGH_PRIORITY_CLASS);
SetThreadAffinityMask (GetCurrentThread (), 1ull);
lts[0]->threadHandle = CreateThread (NULL, 65536u, (void *)locking_thread, (void *)lts[0], 0, (void *)<s[0]->threadId);
#ifndef SINGLE_THREAD
lts[1]->threadHandle = CreateThread (NULL, 65536u, (void *)locking_thread, (void *)lts[1], 0, (void *)<s[1]->threadId);
lts[2]->threadHandle = CreateThread (NULL, 65536u, (void *)locking_thread, (void *)lts[2], 0, (void *)<s[2]->threadId);
lts[3]->threadHandle = CreateThread (NULL, 65536u, (void *)locking_thread, (void *)lts[3], 0, (void *)<s[3]->threadId);
#endif
SetThreadAffinityMask (GetCurrentThread (), processAffinityMask);
threadHold = 0;
#ifdef SINGLE_THREAD
while (lts[0]->idx<4u) {Sleep (1);}
#else
while (lts[0]->idx+lts[1]->idx+lts[2]->idx+lts[3]->idx<16u) {Sleep (1);}
#endif
printf ("T0:%1.1f,%1.1f,%1.1f\n", lts[0]->clocksPerLock, lts[0]->failuresPerLock, lts[0]->preTestsPerLock);
printf ("T1:%1.1f,%1.1f,%1.1f\n", lts[1]->clocksPerLock, lts[1]->failuresPerLock, lts[1]->preTestsPerLock);
printf ("T2:%1.1f,%1.1f,%1.1f\n", lts[2]->clocksPerLock, lts[2]->failuresPerLock, lts[2]->preTestsPerLock);
printf ("T3:%1.1f,%1.1f,%1.1f\n", lts[3]->clocksPerLock, lts[3]->failuresPerLock, lts[3]->preTestsPerLock);
printf ("T*:%1.1f,%1.1f,%1.1f\n", (lts[0]->clocksPerLock+ lts[1]->clocksPerLock+ lts[2]->clocksPerLock+ lts[3]->clocksPerLock)/ 4.,
(lts[0]->failuresPerLock+lts[1]->failuresPerLock+lts[2]->failuresPerLock+lts[3]->failuresPerLock)/4.,
(lts[0]->preTestsPerLock+lts[1]->preTestsPerLock+lts[2]->preTestsPerLock+lts[3]->preTestsPerLock)/4.);
printf ("LC:%u\n", (ud)lockCounter);
return 0;
}
该程序在一台基于DELL i5-4310U计算机上运行,具有DDR3-800、2个内核/2个超线程,可达到2.7GHz和一个普通的L3缓存。
首先,WOW64的影响似乎微不足道。
执行无争用锁定/解锁的单个线程能够每110个周期执行一次。调整无争用锁定是没有意义的:任何添加以增强单个XCHG指令的代码都只会使其变慢。
当四个超线程向锁变量发送锁定尝试时,情况发生了根本性的变化。实现成功锁定所需的时间跳至994个周期,其中相当大的一部分可以归因于2.2个失败的锁定尝试。换句话说,在高争用情况下,平均需要尝试3.2次锁定才能实现成功锁定。显然,110个周期没有变成110*3.2,而更接近于110*9。因此,其他机制在这里起作用,就像对旧机器的测试一样。此外,平均994个周期包括从716到1157的范围。
实现预测试的锁变体需要大约95%的周期,而最简单的变体(XCHG)所需的周期。平均而言,它们将执行17个CMP以确定尝试1.75个锁是否可行,其中1个成功。我建议不仅使用预测试,因为它更快:即使它稍微增加了复杂性,但它对总线锁定机制的压力较小(3.2-1.75 = 1.45个较少的锁定尝试)。
rep nop
又称pause
,它可以避免P4在离开自旋循环时出现不必要的行为。Intel的手册明确推荐在自旋等待循环中使用它。您可以使用XACQUIRE
和XRELEASE
吗(这些功能直到Haswell才可用)? - harold