最快的直列装配自旋锁

提问者：小点点

最快的直列装配自旋锁

我正在用c编写一个多线程应用程序，其中性能至关重要。在线程之间复制小结构时，我需要使用大量锁定，为此我选择使用自旋锁。

我对此进行了一些研究和速度测试，我发现大多数实现大致相同：

微软CRITICAL_SECTION，SpinCount设置为1000，得分约为140个时间单位
使用Microsoft InterlockedCompareExchange实现此算法可获得约95个时间单位的分数
我还尝试使用一些内联程序集与__asm{}使用类似的代码，它可以获得大约70个时间单位，但我不确定是否已经创建了适当的内存屏障。

编辑：泰晤士报给出了两个线程锁定和解锁自旋锁1,000,000次所需的时间。

我知道这没有太大的区别，但是由于自旋锁是一个被大量使用的对象，人们会认为程序员会同意最快的方法来创建自旋锁。然而，谷歌搜索它会导致许多不同的方法。我认为如果使用内联汇编并使用指令CMPXCHG8B而不是比较32位寄存器，上述方法将是最快的。此外，必须考虑内存障碍，这可以通过LOCK CMPXHG8B（我认为？）来完成，它保证了内核之间共享内存的“独占权利”。最后[有人建议]对于繁忙的等待应该伴随着NOP： REP，这将使超线程处理器能够切换到另一个线程，但我不确定这是否正确？

从我对不同自旋锁的性能测试中，可以看出没有太大差异，但出于纯粹的学术目的，我想知道哪一个最快。然而，由于我在汇编语言和内存障碍方面的经验极其有限，如果有人能为我在以下模板中提供的最后一个示例编写汇编代码，我会很高兴8BLOCK CMPXCHG和适当的内存障碍：

__asm
{
     spin_lock:
         ;locking code.
     spin_unlock:
         ;unlocking code.
}

共3个答案

匿名用户

虽然已经有一个公认的答案，但有一些遗漏的东西可以用来改进所有的答案，摘自这篇Intel文章，所有这些都在快速锁定实现之上：

在易失性读取而不是原子指令上旋转，这可以避免不必要的总线锁定，尤其是在高度竞争的锁上。
对高度竞争的锁使用退避
内联锁，最好使用内部函数用于内联asm有害的编译器（基本上是MSVC）。

匿名用户

我通常不会抱怨有人努力实现快速代码：这通常是一个非常好的练习，可以更好地理解编程和更快的代码。

我也不会在这里抱怨，但我可以毫不含糊地说，快速旋转锁3指令长或更长的问题——至少在x86架构上——是徒劳的追逐。

原因如下：

使用典型的代码序列调用自旋锁

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指令提高了处理器上的锁pin，这实际上意味着我希望在接下来的几个时钟周期内使用总线。这个信号穿过缓存，一直到最慢的总线主控设备，通常是PCI总线。当每个总线主控设备完成后，锁定（锁定确认）信号被发回。然后实际的交换发生了。问题是锁定/锁定序列需要很长时间。PCI总线可能以33MHz运行，有几个延迟周期。在3.3 GHzCPU这意味着每个PCI总线周期需要一百CPU周期。

根据经验，我假设一个锁需要300到3000个CPU周期才能完成，最后我不知道我是否会拥有这个锁。所以你可以通过“快速”旋转锁保存的几个周期将是海市蜃楼，因为没有锁像下一个，这将取决于你在短时间内的公共汽车情况。

________________EDIT________________

我刚刚读到自旋锁是一个“大量使用的对象”嗯，你显然不明白自旋锁每次被调用都会消耗大量的CPU周期。或者，换句话说，每次调用它，你都会失去大量的处理能力。

使用自旋锁（或它们的大兄弟，临界区）的诀窍是尽可能少地使用它们，同时仍然实现预期的程序功能。到处使用它们很容易，结果你会得到乏善可陈的性能。

这不仅仅是编写快速代码，它还涉及组织您的数据。当您编写“在线程之间复制小结构”时，您应该意识到锁可能需要比实际复制长数百倍的时间才能完成。

________________EDIT________________

当您计算平均锁定时间时，它可能会说很少，因为它是在您的机器上测量的，这可能不是预期的目标（可能具有完全不同的总线使用特征）。对于您的机器，平均值将由单个非常快的时间（当总线控制活动没有干扰时）一直到非常慢的时间（当总线控制干扰很大时）组成。

您可以引入确定最快和最慢情况的代码并计算商以查看自旋锁时间的变化有多大。

________________EDIT________________

2016年5月更新。

彼得·科德斯提倡“在无竞争的情况下调整锁是有意义的”，并且在现代CPU上不会出现数百个时钟周期的锁定时间，除非锁变量不对齐。我开始怀疑我之前用32位Watcom C编写的测试程序是否会受到WOW64的阻碍，因为它运行在64位OS：视窗7上。

所以我写了一个64位程序，并用TDM的gcc 5.3编译它。该程序使用隐式总线锁定指令变体“XCHG r， m”进行锁定，并使用简单的赋值“MOV m，r”进行解锁。在一些锁变体中，锁变量经过预先测试，以确定尝试锁定是否可行（使用简单的比较“CMP r，m”，可能不会冒险离开L3）。这里是：

// compiler flags used:

// -O1 -m64 -mthreads -mtune=k8 -march=k8 -fwhole-program -freorder-blocks -fschedule-insns -falign-functions=32 -g3 -Wall -c -fmessage-length=0

#define CLASSIC_BUS_LOCK
#define WHILE_PRETEST
//#define SINGLE_THREAD

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];    /* 32*8=256 bytes for each thread's parameters should result in them residing in different cache lines */

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;

//rwfence ();

  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 *)&lts[0]->threadId);
#ifndef SINGLE_THREAD
  lts[1]->threadHandle = CreateThread (NULL, 65536u, (void *)locking_thread, (void *)lts[1], 0, (void *)&lts[1]->threadId);
  lts[2]->threadHandle = CreateThread (NULL, 65536u, (void *)locking_thread, (void *)lts[2], 0, (void *)&lts[2]->threadId);
  lts[3]->threadHandle = CreateThread (NULL, 65536u, (void *)locking_thread, (void *)lts[3], 0, (void *)&lts[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 HTs、2.7GHz和通用L3缓存。

首先，WOW64的影响似乎可以忽略不计。

执行无竞争锁/解锁的单个线程能够每110个周期执行一次。调整无竞争锁是无用的：为增强单个XCHG指令而添加的任何代码只会使其变慢。

随着四个HT用锁尝试轰炸锁变量，情况发生了根本性的变化。获得成功锁所需的时间跃升到994个周期，其中很大一部分可以归因于2.2次失败的锁尝试。换句话说，在高竞争情况下，平均必须尝试3.2次锁才能获得成功的锁。显然，110个周期没有变成110*3.2，而是更接近110*9。因此，其他机制在这里发挥作用，就像在旧机器上的测试一样。此外，平均994个周期包含716到1157之间的范围

实现预测试的锁变体需要最简单变体（XCHG）大约95%的循环。平均而言，他们将执行17次CMP，以发现尝试1.75个锁是可行的，其中1个成功。我建议使用预测试不仅是因为它更快：它对总线锁定机制施加的压力更小（3.2-1.75=1.45次锁定尝试），尽管它略微增加了复杂性。

匿名用户

维基百科有一篇关于自旋锁的好文章，这里是x86实现

http://en.wikipedia.org/wiki/Spinlock#Example_implementation

请注意，它们的实现没有使用“lock”前缀，因为它在x86上对于“xchg”指令是多余的——它隐含地具有锁语义学，如Stackoverflow讨论中所述：

在多核x86上，锁是否需要作为XCHG的前缀？

REP： NOP是PAUSE指令的别名，您可以在此处了解更多信息

x86暂停指令如何在自旋锁*中工作，*它可以在其他场景中使用吗？

关于记忆障碍的问题，这是你可能想知道的一切

内存障碍：软件黑客的硬件视图Paul E. McKenney

http://irl.cs.ucla.edu/~yingdi/paperreading/whymb.2010.06.07c.pdf