Минимальный неблокирующий стек на AtomicReference

Минимальный неблокирующий стек на AtomicReference

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

Основные концепции

Lock-free означает, что потоки не блокируют друг друга явно (synchronized, lock). Вместо этого используется optimistic locking с помощью Compare-And-Set (CAS) операций.

CAS операция:

• Сравнивает текущее значение
• Если оно совпадает с ожидаемым, устанавливает новое значение
• Возвращает true/false в зависимости от результата

Реализация Lock-Free стека

import java.util.concurrent.atomic.AtomicReference;

public class LockFreeStack<T> {
    private static class Node<T> {
        final T value;
        Node<T> next;
        
        Node(T value) {
            this.value = value;
        }
    }
    
    private final AtomicReference<Node<T>> head = new AtomicReference<>();
    
    /**
     * Добавляет элемент в стек (в начало).
     * Использует CAS в цикле для обеспечения безопасности.
     */
    public void push(T value) {
        if (value == null) {
            throw new NullPointerException("Стек не принимает null");
        }
        
        Node<T> newNode = new Node<>(value);
        Node<T> oldHead;
        
        // Бесконечный цикл CAS
        while (true) {
            oldHead = head.get();  // получаем текущую вершину
            newNode.next = oldHead;  // новый узел указывает на старую вершину
            
            // пытаемся установить новый узел в качестве вершины
            if (head.compareAndSet(oldHead, newNode)) {
                return;  // успех!
            }
            // иначе повторяем (другой поток изменил head)
        }
    }
    
    /**
     * Удаляет и возвращает элемент из вершины стека.
     * Возвращает null, если стек пуст.
     */
    public T pop() {
        Node<T> oldHead;
        Node<T> newHead;
        
        // Бесконечный цикл CAS
        while (true) {
            oldHead = head.get();  // получаем текущую вершину
            
            if (oldHead == null) {
                return null;  // стек пуст
            }
            
            newHead = oldHead.next;  // новая вершина это следующий узел
            
            // пытаемся установить новую вершину
            if (head.compareAndSet(oldHead, newHead)) {
                return oldHead.value;  // успех, возвращаем значение
            }
            // иначе повторяем (другой поток изменил head)
        }
    }
    
    /**
     * Проверяет, пуст ли стек.
     */
    public boolean isEmpty() {
        return head.get() == null;
    }
    
    /**
     * Возвращает размер стека (неточно в многопоточной среде).
     * O(n) операция!
     */
    public int size() {
        int count = 0;
        for (Node<T> node = head.get(); node != null; node = node.next) {
            count++;
        }
        return count;
    }
}

Пример использования

public class Main {
    public static void main(String[] args) {
        LockFreeStack<Integer> stack = new LockFreeStack<>();
        
        // Добавляем элементы
        stack.push(1);
        stack.push(2);
        stack.push(3);
        stack.push(4);
        stack.push(5);
        
        System.out.println("Размер: " + stack.size());  // 5
        
        // Удаляем элементы
        System.out.println(stack.pop());  // 5
        System.out.println(stack.pop());  // 4
        System.out.println(stack.pop());  // 3
        System.out.println(stack.pop());  // 2
        System.out.println(stack.pop());  // 1
        System.out.println(stack.pop());  // null (стек пуст)
    }
}

Многопоточный пример

public class MultiThreadedExample {
    public static void main(String[] args) throws InterruptedException {
        LockFreeStack<Integer> stack = new LockFreeStack<>();
        final int NUM_THREADS = 4;
        final int NUM_ITEMS = 1000;
        
        // Создаем потоки-производители
        Thread[] producers = new Thread[NUM_THREADS];
        for (int t = 0; t < NUM_THREADS; t++) {
            final int threadNum = t;
            producers[t] = new Thread(() -> {
                for (int i = 0; i < NUM_ITEMS; i++) {
                    int value = threadNum * NUM_ITEMS + i;
                    stack.push(value);
                }
            });
        }
        
        // Создаем потоки-потребители
        Thread[] consumers = new Thread[NUM_THREADS];
        int[] consumedCounts = new int[NUM_THREADS];
        
        for (int t = 0; t < NUM_THREADS; t++) {
            final int threadNum = t;
            consumers[t] = new Thread(() -> {
                int count = 0;
                while (count < NUM_ITEMS || !stack.isEmpty()) {
                    Integer value = stack.pop();
                    if (value != null) {
                        count++;
                    }
                }
                consumedCounts[threadNum] = count;
            });
        }
        
        // Запускаем все потоки
        for (Thread t : producers) t.start();
        for (Thread t : consumers) t.start();
        
        // Ждем завершения
        for (Thread t : producers) t.join();
        for (Thread t : consumers) t.join();
        
        System.out.println("Стек пуст: " + stack.isEmpty());
        System.out.println("Всего обработано: " + 
            java.util.Arrays.stream(consumedCounts).sum());
    }
}

ABA-проблема

ABA-проблема — это проблема, которая может возникнуть в lock-free алгоритмах:

1. Поток 1 видит значение A
2. Поток 2 изменяет A → B → A (быстро)
3. Поток 1 думает, что ничего не изменилось и выполняет CAS
4. Но на самом деле структура данных изменилась!

Пример ABA-проблемы:

Стек: [A] → [B] → [C]

Поток 1:
1. oldHead = получить head  // [A] → [B]
2. (отвлекся, context switch)

Поток 2:
1. pop()  // [B]
2. pop()  // [A]
3. push([B])  // [B] → [C]
4. Теперь: head = [B] → [C]

Поток 1 (продолжает):
3. compareAndSet(oldHead=[A], newNode)  // УСПЕШНО!
4. Но newNode указывает на [B], которая теперь неправильная!

Решение ABA-проблемы: AtomicStampedReference

import java.util.concurrent.atomic.AtomicStampedReference;

public class LockFreeStackWithStamp<T> {
    private static class Node<T> {
        final T value;
        Node<T> next;
        
        Node(T value) {
            this.value = value;
        }
    }
    
    // используем версию (stamp) вместе со ссылкой
    private final AtomicStampedReference<Node<T>> head = 
        new AtomicStampedReference<>(null, 0);
    
    public void push(T value) {
        if (value == null) {
            throw new NullPointerException();
        }
        
        Node<T> newNode = new Node<>(value);
        int[] stamp = new int[1];  // для хранения версии
        
        while (true) {
            Node<T> oldHead = head.get(stamp);
            int oldStamp = stamp[0];
            
            newNode.next = oldHead;
            
            // compareAndSet теперь проверяет и значение, и версию
            if (head.compareAndSet(oldHead, newNode, oldStamp, oldStamp + 1)) {
                return;
            }
        }
    }
    
    public T pop() {
        int[] stamp = new int[1];
        
        while (true) {
            Node<T> oldHead = head.get(stamp);
            int oldStamp = stamp[0];
            
            if (oldHead == null) {
                return null;
            }
            
            Node<T> newHead = oldHead.next;
            
            if (head.compareAndSet(oldHead, newHead, oldStamp, oldStamp + 1)) {
                return oldHead.value;
            }
        }
    }
    
    public boolean isEmpty() {
        return head.getReference() == null;
    }
}

Сравнение подходов

*В synchronized и Lock нет ABA-проблемы, так как они блокируют доступ полностью

Тесты для lock-free стека

public class LockFreeStackTest {
    private LockFreeStack<Integer> stack;
    
    @Before
    public void setUp() {
        stack = new LockFreeStack<>();
    }
    
    @Test
    public void testPushPop() {
        stack.push(1);
        stack.push(2);
        stack.push(3);
        
        assertEquals(3, (int) stack.pop());
        assertEquals(2, (int) stack.pop());
        assertEquals(1, (int) stack.pop());
        assertNull(stack.pop());
    }
    
    @Test
    public void testEmpty() {
        assertTrue(stack.isEmpty());
        stack.push(1);
        assertFalse(stack.isEmpty());
        stack.pop();
        assertTrue(stack.isEmpty());
    }
    
    @Test(expected = NullPointerException.class)
    public void testNullValue() {
        stack.push(null);
    }
    
    @Test
    public void testConcurrentAccess() throws InterruptedException {
        final int NUM_THREADS = 10;
        final int NUM_OPS = 1000;
        List<Thread> threads = new ArrayList<>();
        
        for (int t = 0; t < NUM_THREADS; t++) {
            final int threadNum = t;
            threads.add(new Thread(() -> {
                for (int i = 0; i < NUM_OPS; i++) {
                    if (i % 2 == 0) {
                        stack.push(threadNum * NUM_OPS + i);
                    } else if (!stack.isEmpty()) {
                        stack.pop();
                    }
                }
            }));
        }
        
        for (Thread t : threads) t.start();
        for (Thread t : threads) t.join();
        
        // Проверяем, что стек функционирует корректно
        assertFalse(stack.isEmpty());
    }
}

Ключевые моменты

• Lock-free алгоритмы используют CAS для синхронизации без блокировок
• AtomicReference предоставляет compareAndSet операцию
• ABA-проблема может возникнуть при использовании только ссылок
• AtomicStampedReference решает ABA-проблему добавлением версии
• Lock-free обычно дает лучшую производительность при высокой конкуренции
• Справедливость может быть проблемой в lock-free алгоритмах (starvation)