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

В многозадачности блокировки в старом добром понимании начинают становиться узким местом. Когда мы пытаемся использовать обычные синхронизации типа 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#. Записаться