Как создавать lock-free структуры данных в C# на базе CAS и Thread.Volatile

Привет, Хабр!

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

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

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

CAS — это операция, которая проверяет, совпадает ли текущее значение с ожидаемым, и если совпадает, заменяет его на новое значение. Атомарно, без возможности прерывания. Это основная операция для создания lock-free структур данных.

Думаю, многие из вас уже сталкивались с этой операцией через метод Interlocked.CompareExchange. Вот пример:

int oldValue = 10; int newValue = 20; int comparand = 10;  // Используем CAS для замены int result = Interlocked.CompareExchange(ref oldValue, newValue, comparand); Console.WriteLine(result); // вернет 10, если oldValue было равно comparand, и заменит oldValue на newValue.

В примере мы пытаемся заменить значение переменной oldValue на newValue, но только если текущее значение переменной соответствует значению, которое мы ожидаем — в данном случае, 10. Если оно не совпадает, то операция не выполняется, и мы продолжаем.

Благодаря CAS можно атомарно обновлять значения, не блокируя другие потоки. И это — наш основной инструмент для создания lock-free структур.

Стек без блокировок — первый шаг в lock-free

Представьте себе ситуацию: вы разрабатываете многозадачное приложение, где несколько потоков будут работать с данным стеком. И, о чудо, вам не нужно использовать блокировки! Будем делать всё через CAS.

Вот как будет выглядеть код для lock-free стека:

public class LockFreeStack<T> {     private class Node     {         public T Value;         public Node Next;     }      private Node head;      public void Push(T item)     {         Node newNode = new Node { Value = item };  // Создаём новый узел         Node oldHead;                  do         {             oldHead = head;  // Сохраняем текущую вершину стека             newNode.Next = oldHead;  // Новый узел будет указывать на старую вершину         }          while (Interlocked.CompareExchange(ref head, newNode, oldHead) != oldHead); // Атромарно обновляем head     }      public bool TryPop(out T result)     {         Node oldHead;                  do         {             oldHead = head;  // Сохраняем текущую вершину стека             if (oldHead == null)             {                 result = default(T);  // Если стек пуст, возвращаем false                 return false;             }         }          while (Interlocked.CompareExchange(ref head, oldHead.Next, oldHead) != oldHead); // Атромарно обновляем head          result = oldHead.Value;         return true;     } }

Все сделали с помощью атомарной операции CAS. Когда поток выполняет Push или Pop, он атомарно меняет ссылку на вершину стека, что предотвращает блокировки.

А теперь представим, что несколько потоков одновременно выполняют операции Push и Pop в стеке. В условиях многозадачности важно, чтобы операции были атомарными и не приводили к состояниям гонки.

Пример с параллельным запуском потоков для тестирования lock-free стека:

using System; using System.Threading;  public class Program {     public static void Main()     {         var stack = new LockFreeStack<int>();          // Запуск 10 потоков для выполнения Push         Thread[] threads = new Thread[10];         for (int i = 0; i < 10; i++)         {             int threadId = i;             threads[i] = new Thread(() =>             {                 stack.Push(threadId);                 Console.WriteLine($"Thread {threadId} pushed.");             });             threads[i].Start();         }          // Ждем завершения всех потоков         foreach (var thread in threads)         {             thread.Join();         }          // Попытка извлечь элементы из стека         for (int i = 0; i < 10; i++)         {             if (stack.TryPop(out int result))             {                 Console.WriteLine($"Popped value: {result}");             }         }     } }

Но все не так просто, как кажется. Бывают ситуации, когда стек может быть поврежден из-за проблемы ABA.

Это одна из самых больших проблем при использовании CAS. Допустим поток А читает значение (например, вершину стека), поток B изменяет это значение, а затем возвращает его в прежнее состояние. Поток А видит, что значение не изменилось, но на самом деле оно прошло через несколько изменений.

Как мы можем бороться с этим? Один из способов — это использовать счетчики версий или ссылки на атомарные указатели.

Вот пример, как можно добавить версионность в структуру данных, чтобы избежать проблемы ABA:

public class LockFreeStackWithVersion<T> {     private class Node     {         public T Value;         public Node Next;         public int Version;  // Счётчик версий     }      private Node head;      public void Push(T item)     {         Node newNode = new Node { Value = item };         Node oldHead;                  do         {             oldHead = head;             newNode.Next = oldHead;             newNode.Version = oldHead?.Version + 1 ?? 0;  // Устанавливаем версию на основе текущей вершины         }         while (Interlocked.CompareExchange(ref head, newNode, oldHead) != oldHead);     }      public bool TryPop(out T result)     {         Node oldHead;                  do         {             oldHead = head;             if (oldHead == null)             {                 result = default(T);                 return false;             }         }         while (Interlocked.CompareExchange(ref head, oldHead.Next, oldHead) != oldHead);          result = oldHead.Value;         return true;     } }

Теперь у каждого узла есть версия, и мы отслеживаем, когда узел был изменен. Если версия изменилась, значит, кто-то уже работал с данным элементом.

Thread.Volatile

Важно упомянуть также Thread.Volatile, который помогает избежать ситуации, когда одно ядро процессора видит устаревшие данные из кэша, а другое — нет. В многозадачности это может быть критично, и с помощью Thread.Volatile можно гарантировать, что переменные читаются и записываются непосредственно в память, минуя кэш.

Пример:

public class LockFreeStackWithVolatile<T> {     private class Node     {         public T Value;         public Node Next;     }      private volatile Node head;  // Используем volatile для гарантированной видимости      public void Push(T item)     {         Node newNode = new Node { Value = item };         Node oldHead;          do         {             oldHead = head;             newNode.Next = oldHead;         }          while (Interlocked.CompareExchange(ref head, newNode, oldHead) != oldHead);     }      public bool TryPop(out T result)     {         Node oldHead;          do         {             oldHead = head;             if (oldHead == null)             {                 result = default(T);                 return false;             }         }          while (Interlocked.CompareExchange(ref head, oldHead.Next, oldHead) != oldHead);          result = oldHead.Value;         return true;     } }

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

Другие lock-free структуры

Кроме стека, есть и другие структуры данных, которые можно реализовать без блокировок. Например, очередь или двусвязный список. Рассмотрим пример с очередью:

public class LockFreeQueue<T> {     private class Node     {         public T Value;         public Node Next;     }      private Node head;     private Node tail;      public LockFreeQueue()     {         head = tail = new Node();  // Начальная пустая очередь     }      public void Enqueue(T item)     {         Node newNode = new Node { Value = item };         Node oldTail;          do         {             oldTail = tail;         }          while (Interlocked.CompareExchange(ref tail.Next, newNode, null) != null);         Interlocked.CompareExchange(ref tail, newNode, oldTail);     }      public bool TryDequeue(out T result)     {         Node oldHead;          do         {             oldHead = head;             if (oldHead.Next == null)             {                 result = default(T);                 return false;             }         }          while (Interlocked.CompareExchange(ref head, oldHead.Next, oldHead) != oldHead);          result = oldHead.Next.Value;         return true;     } }

Здесь также используется CAS для обновления хвоста и головы очереди.

Когда НЕ стоит использовать lock-free структуры данных?

Немного подумаем и о том, когда использовать lock-free структуры данных не стоит.

1. Когда работа с многозадачностью не критична: если задача не требует высокой степени параллелизма и не возникает серьезных проблем с производительностью, не стоит заморачиваться с lock-free структурами. Просто используешь обычные блокировки и не мучаешься.

2. Когда у тебя не гарантирована атомарность операций: если не можешь быть уверен, что операции, которые ты проводишь, могут быть выполнены атомарно, не стоит использовать lock-free подход.

3. Когда нужна совместимость с устаревшими решениями: Если твой проект имеет множество старых решений, которые работают с блокировками, переход на lock-free структуру может привести к дополнительным проблемам. Тут нужно четко понимать, как будет происходить миграция.

В заключение рекомендую обратить внимание на открытые уроки:

• 26 декабря — Работа с NoSQL на С#: какие виды нереляционных баз данных существуют и где их применять. Записаться

• 20 января — Высокопроизводительное программирование на C#. Записаться