Самая быстрая спин-блокировка при встроенной сборке

Просто посмотрите здесь: спин-блокировка x86 с использованием cmpxchg

И благодаря Кори Нельсону

__asm{
spin_lock:
xorl %ecx, %ecx
incl %ecx
spin_lock_retry:
xorl %eax, %eax
lock; cmpxchgl %ecx, (lock_addr)
jnz spin_lock_retry
ret

spin_unlock:
movl $0 (lock_addr)
ret
}

Другой источник сообщает: http://www.geoffchappell.com/studies/windows/km/cpu/cx8.htm

       lock    cmpxchg8b qword ptr [esi]
is replaceable with the following sequence

try:
        lock    bts dword ptr [edi],0
        jnb     acquired
wait:
        test    dword ptr [edi],1
        je      try
        pause                   ; if available
        jmp     wait

acquired:
        cmp     eax,[esi]
        jne     fail
        cmp     edx,[esi+4]
        je      exchange

fail:
        mov     eax,[esi]
        mov     edx,[esi+4]
        jmp     done

exchange:
        mov     [esi],ebx
        mov     [esi+4],ecx

done:
        mov     byte ptr [edi],0

А вот обсуждение реализации lock-free и lock: http://newsgroups.derkeiler.com/Archive/Comp/comp.programming.threads/2011-10/msg00009.html

Хотя принятый ответ уже есть, есть несколько вещей, которые были упущены, которые можно было бы использовать для улучшения всех ответов, взятых из эта статья Intel, описывающая все вышесказанное о реализации быстрой блокировки:

1. Вращение на непостоянном чтении, а не на атомарной инструкции, позволяет избежать ненужных блокировок шины, особенно при блокировках с высокой степенью конкуренции.
2. Используйте откат для очень спорных замков
3. Встроенная блокировка, предпочтительно с встроенными функциями для компиляторов, где встроенный asm вреден (в основном MSVC).

Обычно я не из тех, кто жаловался на кого-то, кто стремится к быстрому написанию кода: обычно это очень хорошее упражнение, которое приводит к лучшему пониманию программирования и ускорению кода.

Я тоже не буду здесь жаловаться, но могу однозначно заявить, что вопрос о быстрой спин-блокировке длиной в три или несколько инструкций - по крайней мере, на архитектуре 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 поднимает стопорный штифт на процессоре, что фактически означает, что мне нужна шина в течение следующих нескольких тактов. Этот сигнал проходит через кеши до самого медленного устройства управления шиной, которым обычно является шина PCI. Когда каждое устройство управления шиной завершает работу, сигнал locka (подтверждение блокировки) отправляется обратно. Затем происходит собственно обмен. Проблема в том, что последовательность блокировки / блокировки занимает ОЧЕНЬ много времени. Шина PCI может работать на частоте 33 МГц с несколькими циклами задержки. На процессоре с тактовой частотой 3,3 ГГц это означает, что каждый цикл шины PCI занимает сто циклов процессора.

Как правило, я предполагаю, что блокировка займет от 300 до 3000 циклов процессора, и, в конце концов, я не знаю, буду ли я владельцем блокировки. Таким образом, несколько циклов, которые вы можете сохранить с помощью «быстрой» спин-блокировки, будут миражом, потому что никакая блокировка не похожа на следующую, она будет зависеть от ситуации с вашей шиной в течение этого короткого времени.

________________РЕДАКТИРОВАТЬ________________

Я только что прочитал, что спин-блокировка - это «часто используемый объект». Что ж, вы, очевидно, не понимаете, что спин-блокировка потребляет огромное количество циклов процессора каждый раз, когда она вызывается. Или, говоря другими словами, каждый раз, когда вы вызываете его, вы теряете значительную часть своих возможностей обработки.

Уловка при использовании спин-блокировок (или их более крупного родственника, критического раздела) состоит в том, чтобы использовать их как можно реже, при этом все же достигая намеченной программной функции. Их легко использовать повсюду, и в результате вы получите низкую производительность.

Дело не только в написании быстрого кода, но и в организации ваших данных. Когда вы пишете «копирование небольших структур между потоками», вы должны понимать, что блокировка может занять в сотни раз больше времени, чем фактическое копирование.

________________РЕДАКТИРОВАТЬ________________

Когда вы вычисляете среднее время блокировки, оно, вероятно, мало что скажет, поскольку оно измеряется на вашем компьютере, что может не являться предполагаемой целью (которая может иметь совершенно другие характеристики использования шины). Для вашей машины среднее значение будет складываться из отдельных очень коротких времен (когда действия по мастерингу шины не мешают) вплоть до очень медленных времен (когда вмешательство в мастеринг шины было значительным).

Вы можете ввести код, который определяет самый быстрый и самый медленный случаи и вычисляет частное, чтобы увидеть, насколько сильно может варьироваться время спин-блокировки.

________________РЕДАКТИРОВАТЬ________________

Обновление за май 2016 г.

Питер Кордес продвигал идею о том, что «настройка блокировки в неконкурентном случае может иметь смысл» и что время блокировки в несколько сотен тактовых циклов не возникает на современных ЦП, за исключением ситуаций, когда переменная блокировки не выровнена. Мне стало интересно, написана ли моя предыдущая тестовая программа в 32-битном Watcom C - WOW64 может помешать, поскольку он работал в 64-битной ОС: Windows 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 HT с тактовой частотой 2,7 ГГц и общей кэш-памятью третьего уровня.

Во-первых, похоже, что влияние WOW64 было незначительным.

Один поток, выполняющий неконтролируемую блокировку / разблокировку, мог делать это каждые 110 циклов. Настраивать неконтролируемую блокировку бесполезно: любой код, добавленный для улучшения одиночной инструкции XCHG, только замедлит ее.

Когда четыре HT бомбардируют переменную блокировки попытками блокировки, ситуация радикально меняется. Время, необходимое для достижения успешной блокировки, увеличивается до 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 меньше попыток блокировки), хотя это немного увеличивает сложность.

В Википедии есть хорошая статья о спин-блокировках, вот реализация x86

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

Обратите внимание, что их реализация не использует префикс «lock», потому что он избыточен на x86 для инструкции «xchg» - он неявно имеет семантику блокировки, как обсуждалось в этом обсуждении Stackoverflow:

На многоядерном x86 LOCK необходим в качестве префикса для XCHG?

REP: NOP - это псевдоним для инструкции PAUSE, вы можете узнать больше об этом здесь.

Как команда паузы x86 работает в спин-блокировке * и * можно ли ее использовать в других сценариях?

Что касается барьеров памяти, вот все, что вам может быть интересно знать

Барьеры памяти: взгляд на оборудование для программных хакеров Пол Э. Маккенни

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

Просто спрашиваю:

Прежде чем углубиться в спин-блокировку и почти безблокирующие структуры данных:

Удостоверились ли вы в своих тестах и ​​в приложении, что конкурирующие потоки гарантированно работают на разных ядрах?

В противном случае вы можете получить программу, которая отлично работает на вашей машине разработки, но отстой / терпит неудачу в полевых условиях, потому что один поток должен быть одновременно блокировщиком и разблокировщиком вашей спин-блокировки.

Чтобы дать вам цифру: в Windows у вас есть стандартный временной интервал в 10 миллисекунд. Если вы не убедитесь, что два физических потока участвуют в блокировке / разблокировке, вы получите около 500 блокировок / разблокировок в секунду, и этот результат будет очень meh