Сравнение с обменом

/* Псевдокод работы инструкции, возвращающей булево значение в синтаксисе языка C */
int cas( int* addr, int old, int new )
{
  if ( *addr != old )
    return 0;

  *addr = new;
  return 1;
}

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

Подход атомарных инструкций противостоит подходу с «условной записью» (англ. load-link/store-conditional).

Инструкция для сравнения с обменом в процессорах x86 называется CMPXCHG и выполняется следующим образом:

1. Процессор читает область памяти, указанную в команде первым операндом, не снимая по завершении чтения блокировку шины.

2. Процессор сравнивает прочтённое значение со значением в аккумуляторе (регистр AL, AX, EAX или RAX). Флагу ZF присваивается значение в зависимости от результата сравнения (1 — если значение в памяти равно значению в аккумуляторе, 0 — если они различаются).

3. Если значение в памяти было равно значению в аккумуляторе, процессор записывает значение из второго операнда в область памяти, указанную первым операндом (особенность реализации x86: запись происходит всегда, но если сравнение на шаге 2 показало неравенство, в аккумулятор записывается то значение, что было прочтено из памяти на шаге 1). По завершении записи блокировка шины снимается.

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

Для SPARC инструкции для данной операции называются CASA и CASXA.

Казалось бы, на однопроцессорной машине можно отключить прерывания и тогда состояние памяти точно ничего не изменит. Но тут есть две проблемы. Первая проблема — отключение прерываний - процедура сравнительно дорогостоящая. Вторая проблема — если отключить прерывания, а поток войдет в бесконечный цикл, то программу нельзя будет завершить без перезагрузки компьютера. Поэтому Linux требует высоких прав доступа для кода с этой инструкцией, что может доставить немало неудобств пользователям программы.

На многопроцессорной же машине отключение прерываний вообще не поможет, так как в ситуации:

1-й процессор проверил, что память не заблокирована
2-й процессор проверил, что память не заблокирована
1-й процессор заблокировал память
2-й процессор заблокировал память
1-й процессор изменил память
2-й процессор изменил память
1-й процессор разблокировал память
2-й процессор разблокировал память

изменения 1-го процессора будут потеряны, а программа будет думать, что все нормально.

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

До начала основной работы назначим им уникальные номера от 0 до n-1.

Выберем ячейку памяти, которая будет указывать, какой процессор сейчас использует ресурс. Значение −1 будет указывать, что ресурс никем не занят. Поместим в неё −1.

При основной работе каждый процессор должен проверить, что в ячейке находится −1, и если это так, то записать в неё свой номер. Если же ячейка занята — процессор обязан ждать, пока она не освободится (мы договорились, что он будет ждать, и не будем писать программы, которые не выполняют это требование).

То есть программа может выглядеть следующим образом:

; блокировка
m_wait:
mov eax, -1
mov ecx, 5               ; номер нашего процессора 5
cmpxchg 105BA9D2, ecx    ; адрес ячейки 105BA9D2
jnz m_wait ; если ресурс заблокирован
; работа с общим ресурсом
...

; снятие блокировки
...

Инструкции атомарного сравнения с обменом не входили в стандарты языков C/C++, поэтому они либо реализуются через ассемблер, либо через нестандартные расширения языка. В стандарте C++11 введена библиотека атомарных операций, в которой есть сравнение с обменом[1]. Также существует несколько библиотек с реализациями C/C++ интерфейсов к подобным инструкциям, например: Intel TBB, boost.atomic, Open Portable Atomics, Glib.

uint32_t* ptr;
uint32_t ret_val, old_val, new_val;

asm volatile("lock\n\tcmpxchgl %1,%2"
  : "=a"(ret_val)
  : "r"(new_val), "m"(*ptr), "0"(old_val)
  : "memory"
  );

uint64_t* ptr;
uint64_t ret_val, old_val, new_val;

asm volatile("lock\n\tcmpxchgq %1,%2"
  :"=a"(ret_val)
  :"r"(new_val), "m"(*ptr), "0"(old_val)
  :"memory"
  );

TYPE __sync_val_compare_and_swap(TYPE* ptr, TYPE oldval, TYPE newval);

Существование подобной инструкции открывает огромные горизонты в повышении производительности кода за счет возможности ухода вообще от блокировок как таковых.

Рассмотрим такой участок псевдокода:

читаем значение переменной;
производим некоторую обработку;
записываем новое значение переменной;

Наиболее обычным способом сделать данный код многопоточным является введение синхронизирующих примитивов (mutex'ы, спинлоки, и т. д.) следующим образом:

производим блокировку;
читаем значение переменной;
производим некоторую обработку;
записываем новое значение переменной;
отпускаем блокировку;

Однако, зачастую применим более элегантный метод:

читаем значение переменной;
производим некоторую обработку;
производим cmpxchg новое значение переменной в предположении что значение все еще равно старому;
если значение было изменено другим потоком - повторяем обработку;

Рассмотрим классический пример структуры данных — связный список.

struct ll_node {
  int key; // некоторый ключ
  int val; // некоторое значение
  struct ll_node* next; // указатель на следующий
};

Функция вставки в связанный список узла new_node после указанного узла node выглядит следующим образом:

 void ll_insert_after(struct ll_node* node, struct ll_node* new_node)
 {
   new_node->next = node->next;
   node->next = new_node; // (!!!) - обратим внимание на данную строку
 }

Очевидно, код будет работать корректно только в предположении, что значение node->next не было изменено другим потоком к моменту работы строки, отмеченной «(!!!)», в противном случае мы рискуем потерять изменения остальных потоков, породив узлы, не относящиеся к списку (Утечка памяти).

Существует 3 основных способа борьбы с этим:

1) Закрыть весь связный список мьютексом. С точки зрения производительности, это самый худший способ. Во-первых, со связным списком в данный момент времени будет работать только один поток, что само по себе сводит на нет все плюсы многопоточности. Во вторых, на практике ситуация обстоит намного хуже, чем можно было бы предположить: более-менее интенсивная одновременная работа с таким связным списком будет порождать огромное (тысячи, десятки тысяч, а в отдельных, особо интенсивных случаях даже миллионы) переключений контекста, что само по себе способно убить производительность не только самого приложения, но и системы в целом (эффект «просаживания производительности» квадратично возрастает от числа вычислительных ядер).

2) Более интеллектуальный способ. Заменить Mutex на спинлок. Несколько холостых циклов ожидания занятости обходятся намного «дешевле», чем системный вызов и порожденное им переключение контекста. Дает реальный эффект на SMP-системах, однако на одноядерных системах порождает «редкий, но меткий» убой производительности за счет длительного простоя.

3) Алгоритм без блокировки. Перепишем функцию вставки следующим образом: сделаем предположение о неизменности значения node->next явным. Его мы в явном виде и будем проверять с помощью команды cmpxchg:

 void ll_insert_after(struct ll_node* node, struct ll_node* new_node)
 {
   struct ll_node* old_val = node->next; // запоминаем старое значение
   while (1) {
     new_node->next = old_val;
     old_val = cmpxchg(&node->next, old_val, new_node);
     if (old_val == new_node)
       break; // Другие потоки не меняли node->next. Успех операции, выход.
     // Иначе повторим попытку
   }
 }

«Ядром» безблокировочной логики данной функции служит команда CMPXCHG. Она атомарно проверяет, что содержимое ячейки памяти &node->next все еще содержит значение old_val, и если это так (а вероятность этого лучшего случая крайне высока), записывает значение указателя new_node и выходит из цикла. В случае коллизии совместного доступа мы получаем обновленное значение old_val и выходим на новую итерацию цикла.

В случае рассматриваемого выше связного списка вероятность коллизии крайне мала. Формально она равна P_кол=(n/N)*p_функ , где N — к-во записей в списке, n — к-во одновременных потоков, p_функ — отношение времени, которое каждый поток проводит внутри функции вставки, к общему времени работы потока.

Главным недостатком, сдерживающим применение команды cmpxchg в безблокировочных алгоритмах, состоит в том, что команда заменяет всего лишь одно значение. В случае, когда это только значение указателя или целочисленная переменная, этого достаточно. Однако, очень часто требуется атомарно условно заменить две связанные переменные. Например: указатель на буфер и его длину, указатель на начало и конец данных и т. д. Для этих целей в процессорах Intel введены команды CMPXCHG (32-bit), CMPXCHG8B (64-bit) и CMPXCHG16B (x86 64). Кстати, требование поддержки процессором команды CMPXCHG16B появилось в ОС MS Windows версии 8.1 x64.

В процессорах SPARC эти функции выполняют команды CASA и CASXA.

• Проблема ABA

• Intel 64 and IA-32 architectures software developer’s manuals
• Отдельная страница по команде CMPXCHG
• SPARC assembly language reference manual