Беззахватные алгоритмы: модель «сделай, запиши,(поручи другому)»

Следуя совету хабрапублики, пробую новый вариант перевода термина "lock-free"

В прошлый раз мы видели «беззахватный по духу» алгоритм, где захват был реализован так, что поток, обращающийся к захваченным данным, не ждёт их освобождения, а отправляется «обходным путём» (вычисляет требуемый результат, не пользуясь услугами кэша). В своём следующем посте Реймонд объясняет, как данный алгоритм можно усовершенствовать на случай, когда «обходного пути» нет. Алгоритм, однако, остаётся беззахватным: каждый поток продолжает работать, не дожидаясь освобождения захваченных данных.

В общей переменной теперь нужны два служебных бита: вдобавок к флагу захвата, как в прошлом примере, — флаг «поручена новая работа»; а если порученная работа сложная, то кроме флага, нужно будет где-то хранить ещё и её параметры. Например, в общей переменной можно хранить указатель на (выравненный в памяти) объект с параметрами, а в свободных младших битах указателя — два названных флага.

Перед выполнением действия над объектом, первым делом мы его захватываем, атомарно устанавливая соответствующий флаг. Если окажется, что объект уже был захвачен, — поручим выполнение нашего действия захватившему потоку, установив второй флаг.

Если же объект удалось захватить, то после завершения работы с ним снимаем флаг захвата и одновременно проверяем, не поручили ли нам новую работу. (Т.е. не было ли обращений к объекту за то время, пока мы его держали захваченным.) Если есть работа, то мы выполним и её; и так далее, пока однажды при разблокировке объекта отложенной работы не окажется. Мы не вправе оставить объект в состоянии «не захвачен, но есть работа».

Получившаяся модель «сделай, запиши,(поручи другому)» сплетена из множества циклов на случай всевозможных неудач. Каждая атомарная операция заключается в цикл; выполнение отложенной работы осуществляется в цикле; и каждый раз, когда вызов InterlockedCompareExchange выявляет перезапись использованных данных, нужно откатить всю сделанную работу, и выполнять её с начала. Модель выходит весьма запутанная. Реймонд характеризует её так: «Во всём мире, пожалуй, лишь пятеро смогут её верно реализовать, и я к этим пятерым не отношусь.» Тем не менее, в качестве иллюстрации он приводит пример объекта под названием GroupWait, который поддерживает две операции:

• AddWait: Добавить в список новый хэндл события.
• SignalAll: Установить все события в списке, автоматически удаляя каждое событие одновременно с его установкой. Чтобы событие установилось при следующем вызове SignalAll, его нужно добавить в список снова.

Реализован объект просто как список хэндлов. Понятно, что его можно было бы реализовать и без мудрёной беззахватной модели: например, при помощи атомарных функций над списками, реализованных начиная с Windows XP. Задача выбрана лишь для того, чтобы сопроводить словесное описание примерным кодом. Не принимайте GroupWait за образец реализации потоко-безопасного списка.

Итак, младшие два бита указателя мы задействуем для хранения двух флагов: флага захвата, указывающего, что некий поток выполняет операцию над списком; и флага работы, указывающего, что пользователь потребовал установку событий в то время, пока список был захвачен.

// WARNING! IF YOU USE THIS CODE YOU ARE AN IDIOT - READ THE TEXT ABOVE

struct NODE;
NODE *Node(LONG_PTR key) { return reinterpret_cast<NODE*>(key); }

enum {
 Locked = 1,
 Signalled = 2,
};

struct NODE {
 NODE *pnNext;
 HANDLE hEvent;

 LONG_PTR Key() { return reinterpret_cast<LONG_PTR>(this); }
 NODE *Ptr() { return Node(Key() & ~(Locked | Signalled)); }
};

#define NODE_INVALID Node(-1)

class GroupWait {
public:
 GroupWait() : m_pnRoot(NULL) { }
 ~GroupWait();
 BOOL AddWait(HANDLE hEvent);
 void SignalAll();
private:
 NODE *m_pnRoot;
};

Поскольку объединяем в одном значении указатель на элемент списка и пару флагов, то для удобства стоит определить методы, преобразующие используемые типы один в другой: Node возвращает указатель, Key — числовое значение, и Ptr — годный к использованию указатель (без флагов в младших битах).

Изображать наши значения будем как битовые поля вида p|S|L: p — указатель на следующий элемент списка; S — флаг работы; и L — флаг захвата. Установленный флаг S означает, что нужно обработать все элементы списка, начиная со следующего — представьте себе его обозначенным не внутри элемента, а на исходящей стрелке.

Например:

   m_pnRoot
  +--------+-+-+
  |   *    |0|1|
  +---|----+-+-+
      |
      v
  +--------+-+-+---------+
A |   *    |1|?| hEvent1 |
  +---|----+-+-+---------+
      |
      v
  +--------+-+-+---------+
B |   *    |?|?| hEvent2 |
  +---|----+-+-+---------+
      |
      v
  +--------+-+-+---------+
C |  NULL  |?|?| hEvent3 |
  +--------+-+-+---------+

Изображён объект GroupWait, содержащий три хэндла. Сброшенный флаг S в голове списка означает, что никто не требовал установить событие hEvent1. Напротив, установленный флаг S в элементе A означает, что нужно обойти все элементы после A и установить соответствующие события — а именно, hEvent2 и hEvent3. В частности, значение флага S в элементах B и C не имеет значения; эти элементы будут обработаны в любом случае, потому что того требует флаг S в элементе A. Таким образом, значение флага S в последнем элементе списка вообще никогда не имеет значения.

Флаг L используется только в голове списка; во всех остальных элементах он не имеет значения.

Итак, приготовления завершены; добавляем в список новый хэндл.

BOOL GroupWait::AddWait(HANDLE hEvent)
{
 NODE *pnInsert = new(nothrow) NODE;
 if (pnInsert == NULL) return FALSE;
 pnInsert->hEvent = hEvent;

 NODE *pn;
 NODE *pnNew;
 do {
  pn = InterlockedReadAcquire(&m_pnRoot, NODE_INVALID);
  pnInsert->pnNext = pn;
  pnNew = Node(pnInsert->Key() | (pn->Key() & Locked));
 } while (InterlockedCompareExchangeRelease(&m_pnRoot, pnNew, pn) != pn);
 return TRUE;
}

Мы добавляем в начало списка новый элемент, и следим, чтобы флаг L из старого элемента перешёл в новый: иначе может получиться, что другой поток уже захватил список, а мы его ненароком «освободили». Флаг S в добавляемом элементе сброшен: никто ещё не требовал установить новое событие. Добавляем элемент в голову списка мы по знакомой модели «сделай, запиши,(попытайся снова)» — проверяя, что никто не изменил список за нашей спиной. Обратите внимание, что «проблемы ABA» не возникает: даже если неизменившееся значение m_pnRoot указывает уже на другой объект, то мы всё равно используем только сам указатель, а не содержимое объекта.

Простота метода AddWait необычна для модели «сделай, запиши,(поручи другому)»: в случае неудачи нам нечего перепоручить — вся работа заключается в одной записи в память. Расплачиваться за эту простоту будут остальные методы, в которых придётся предусмотреть обработку элементов, добавленных в список за то время, пока он был захвачен.

Метод SignalAll получается настолько сложным, что читать его лучше по частям.

void GroupWait::SignalAll()
{
 NODE *pnCapture;
 NODE *pnNew;
 do {
  pnCapture = InterlockedReadAcquire(&m_pnRoot, NODE_INVALID);

  if (pnCapture->Key() & Locked) {
   pnNew = Node(pnCapture->Key() | Signaled);
  } else {
   pnNew = Node(Locked);
  }
 } while (InterlockedCompareExchangeAcquire(&m_pnRoot,
                              pnNew, pnCapture) != pnCapture);

 if (pnCapture->Key() & Locked) return;

 ...

Если список захвачен, то единственное, что от нас требуется — установить в его голове флаг S. Если же список свободен, то попытаемся его захватить, и одновременно «украдём» все элементы списка, записывая вместо головы «заглушку» NULL|0|1. В обоих случаях перезапись головы списка осуществляется по модели «сделай, запиши,(попытайся снова)» — повторяем попытки, пока запись не удастся.

Установив флаг S, мы перепоручили нашу работу потоку, захватившему список. Установленный флаг S в голове списка означает, что обработать нужно все следующие за головой элементы списка — т.е. вообще все элементы списка; как раз то, что нам нужно. Поток, захвативший список, проверит флаг при освобождении списка, и выполнит работу за нас.

Если же список не был захвачен, то нашим действием мы его захватили. «Украденные» элементы не видны другим потокам, так что одновременный вызов SignalAll с разных потоков не приведёт к многократной установке событий.

 ...
 NODE *pnNext;
 NODE *pn;
 for (pn = pnCapture->Ptr(); pn; pn = pnNext) {
  SetEvent(pn->hEvent);
  pnNext = pn->pnNext->Ptr();
  delete pn;
 }
 ...

Обход «украденного» списка реализован очень прямолинейно, безо всякой заботы о потоко-безопасности; просто берём элемент за элементом, устанавливаем событие, и удаляем элемент. Единственная особенность — вызов Ptr для удаления флагов из указателя на следующий элемент.

Теперь список надо разблокировать. Для начала:

 ...
 pnCapture = pnNew;
 ...

В начале метода мы записали pnNew в m_pnRoot, и если это значение так и сохранилось в голове списка — значит, мы легко отделались: никто не обращался к списку за то время, пока мы с ним работали.

 ...
 for (;;) {
  pnNew = Node(pnCapture->Key() & ~Locked);
  if (InterlockedCompareExchangeRelease(&m_pnRoot,
                      pnNew, pnCapture) == pnCapture) {
   return;
  }
 ...

Вначале проверим, изменился ли список: если нет, достаточно его лишь разблокировать — и готово. Иначе начинаем новый цикл: нужно выполнить всю отложенную работу, накопившуюся, пока список был захвачен.

 ...
  pnCapture = InterlockedReadAcquire(&m_pnRoot, NODE_INVALID);

  NODE *pnNew = Node(pnCapture->Key() & ~(Locked | Signaled));
  NODE **ppn = &pnNew;
  NODE *pn;
  NODE *pnNext;

  BOOL fSignalSeen = FALSE;
  for (pn = pnNew; pn->Ptr(); pn = pnNext) {
   pnNext = pn->Ptr()->pnNext;
   if (fSignalSeen) {
    SetEvent(pn->Ptr()->hEvent);
    delete pn->Ptr();
   } else if (pn->Key() & Signaled) {
    fSignalSeen = TRUE;
    (*ppn) = Node(Locked); // украсть, оставив захваченным
    SetEvent(pn->Ptr()->hEvent);
    delete pn->Ptr();
   } else {
    ppn = &pn->Ptr()->pnNext;
   }
  }
 } // снова попытаемся разблокировать
} // конец функции

Чтобы выполнить отложенную работу, мы обходим список, пока не найдём установленный бит S. В первом элементе, в котором бит S установлен, мы обнуляем исходящий указатель, чтобы «украсть» остаток списка; затем, обходя этот остаток, мы устанавливаем каждое событие, и удаляем элемент. Как и прежде, мы «крадём» список затем, чтобы одновременный вызов с нескольких потоков не привёл к многократной установке события. В конце, когда работа выполнена, мы снова пытаемся разблокировать список — в расчёте на то, что однажды новой работы не окажется, и мы сможем с полным правом выйти из функции.

---

Как видите, основная идея модели «сделай, запиши,(поручи другому)» достаточно проста, но от её реализации с учётом всех тонкостей может разболеться голова. Лучше всего оставить реализацию таких штук системным программистам, у которых достаточно времени, терпения и способностей, чтобы со всем этим справиться. Например, в интервью "Arun Kishan: Inside Windows 7 — Farewell to the Windows Kernel Dispatcher Lock" архитектор, работавший над ядром Windows, рассказывают о применении именно этой модели.