Многопоточность в C++: Современные техники C++26

// До-C++11 многопоточность: кросс-платформенный ад
#ifdef _WIN32
    HANDLE thread = CreateThread(NULL, 0, ThreadFunction, NULL, 0, NULL);
    // Windows-специфичный код
#else
    pthread_t thread;
    pthread_create(&thread, NULL, threadFunction, NULL);
    // POSIX-специфичный код
#endif

// C++11: единый подход к многопоточности
#include <thread>
#include <mutex>
#include <iostream>
 
std::mutex print_mutex;
 
void print_message(const std::string& msg) {
    std::lock_guard<std::mutex> guard(print_mutex);
    std::cout << msg << std::endl;
}
 
int main() {
    std::thread t1(print_message, "Hello");
    std::thread t2(print_message, "World");
    t1.join();
    t2.join();
    return 0;
}

// C++17: параллельные алгоритмы
#include <algorithm>
#include <execution>
#include <vector>
 
std::vector<int> data{9, 1, 8, 2, 7, 3, 6, 4, 5};
// Параллельная сортировка
std::sort(std::execution::par, data.begin(), data.end());

// C++20: умные потоки с отменой
#include <thread>
 
void long_operation(std::stop_token token) {
    while (!token.stop_requested()) {
        // Выполняем длительную операцию
        // Периодически проверяем запрос на отмену
    }
}
 
int main() {
    std::jthread worker(long_operation);
    // Где-то в другом потоке
    worker.request_stop(); // Вежливая просьба завершиться
    // worker автоматически вызовет join() при выходе из области видимости
    return 0;
}

// Предполагаемый синтаксис в C++26
sender auto process_data = [](int value) -> int {
    return value * 2;
};
 
// Создание цепочки асинхронных операций
sender auto operation = just(42)
    | then(process_data)
    | then([](int result) { 
        std::cout << "Result: " << result << std::endl; 
    });
 
// Запуск асинхронной операции
sync_wait(std::move(operation));

#include <thread>
#include <iostream>
 
void simple_function() {
    std::cout << "Я выполняюсь в отдельном потоке!" << std::endl;
}
 
class Worker {
public:
    void operator()() {
        std::cout << "Я функтор в отдельном потоке!" << std::endl;
    }
    
    void work(int param) {
        std::cout << "Метод класса с параметром: " << param << std::endl;
    }
};
 
int main() {
    Worker worker;
    
    // Четыре способа создания потоков
    std::thread t1(simple_function);                  // Обычная функция
    std::thread t2(worker);                           // Функтор
    std::thread t3(&Worker::work, &worker, 42);       // Метод класса
    std::thread t4([]{ std::cout << "Лямбда!"; });   // Лямбда-выражение
    
    // Не забываем дождаться завершения
    t1.join(); t2.join(); t3.join(); t4.join();
    
    return 0;
}

std::mutex resource_mutex;
int shared_counter = 0;
 
void increment_counter() {
    resource_mutex.lock();
    // Критическая секция - только один поток может быть здесь
    shared_counter++;
    resource_mutex.unlock();
}

void increment_counter_safe() {
    std::lock_guard<std::mutex> guard(resource_mutex); // RAII-обёртка
    shared_counter++; // При любом выходе из функции мьютекс будет разблокирован
}

std::mutex pizza_mutex;
std::condition_variable pizza_ready;
bool pizza_is_done = false;
 
// Поток-повар
void cook_pizza() {
    // Готовим пиццу
    std::this_thread::sleep_for(std::chrono::seconds(2));
    
    // Сообщаем, что пицца готова
    {
        std::lock_guard<std::mutex> lock(pizza_mutex);
        pizza_is_done = true;
    }
    pizza_ready.notify_all(); // Будим всех ждущих
}
 
// Поток-клиент
void wait_for_pizza() {
    std::unique_lock<std::mutex> lock(pizza_mutex);
    pizza_ready.wait(lock, []{ return pizza_is_done; });
    // Кушаем пиццу
    std::cout << "Наконец-то пицца готова!" << std::endl;
}

// C++20
#include <latch>
#include <barrier>
 
std::latch completion_latch(3); // Ждём три потока
 
void worker_thread() {
    // Выполняем работу
    // ...
    completion_latch.count_down(); // Сигнализируем о завершении
}
 
void controller_thread() {
    // Ждём, пока все three потока завершат работу
    completion_latch.wait();
    // Все worker_thread завершились
}

// C++20
#include <semaphore>
 
std::counting_semaphore<5> connection_limiter(5); // Максимум 5 одновременных соединений
 
void handle_request() {
    connection_limiter.acquire(); // Захватываем "пропуск"
    // Обрабатываем запрос...
    connection_limiter.release(); // Возвращаем "пропуск"
}

// Без явной синхронизации этот код может работать некорректно
bool ready = false;
int data = 0;
 
void producer() {
data = 42;        // Строка 1
ready = true;     // Строка 2
}
 
void consumer() {
while (!ready) {} // Ждём сигнала от producer
assert(data == 42); // Может неожиданно не сработать!
}

#include <atomic>
 
std::atomic<bool> ready{false};
int data = 0;
 
void producer() {
data = 42;        
ready.store(true, std::memory_order_release);
}
 
void consumer() {
while (!ready.load(std::memory_order_acquire)) {} 
assert(data == 42); // Теперь всегда сработает!
}

// Этот код может по-разному работать на разных архитектурах 
// при использовании memory_order_relaxed
std::atomic<int> x{0}, y{0};
 
void thread1() {
x.store(1, std::memory_order_relaxed);
y.store(1, std::memory_order_relaxed);
}
 
void thread2() {
int y_val = y.load(std::memory_order_relaxed);
int x_val = x.load(std::memory_order_relaxed);
assert(!(y_val == 1 && x_val == 0)); // Может не сработать!
}

// Пример установки барьера памяти
std::atomic<int> flag{0};
 
void thread1() {
// Выполняем какие-то операции
std::atomic_thread_fence(std::memory_order_release);
flag.store(1, std::memory_order_relaxed);
}
 
void thread2() {
while (flag.load(std::memory_order_relaxed) != 1) {}
std::atomic_thread_fence(std::memory_order_acquire);
// Теперь все операции из thread1 видны
}

std::atomic<int> counter{0};
 
void increment_counter() {
int old_value = counter.fetch_add(1, std::memory_order_relaxed);
std::cout << "Предыдущее значение: " << old_value << std::endl;
}

// Демонстрация накладных расходов атомарных операций
#include <chrono>
#include <atomic>
#include <iostream>
 
int main() {
const int iterations = 10000000;
 
// Неатомарная операция
int regular_counter = 0;
auto start = std::chrono::high_resolution_clock::now();
for (int i = 0; i < iterations; ++i) {
    regular_counter++;
}
auto end = std::chrono::high_resolution_clock::now();
std::cout << "Обычные операции: " 
          << std::chrono::duration_cast<std::chrono::milliseconds>(end - start).count() 
          << " мс\n";
 
// Атомарная операция
std::atomic<int> atomic_counter{0};
start = std::chrono::high_resolution_clock::now();
for (int i = 0; i < iterations; ++i) {
    atomic_counter.fetch_add(1, std::memory_order_relaxed);
}
end = std::chrono::high_resolution_clock::now();
std::cout << "Атомарные операции: " 
          << std::chrono::duration_cast<std::chrono::milliseconds>(end - start).count() 
          << " мс\n";
 
return 0;
}

#include <atomic>
#include <thread>
#include <vector>
#include <iostream>
 
class EventCounter {
private:
    std::atomic<size_t> counter{0};
    
public:
    void increment() {
        counter.fetch_add(1, std::memory_order_relaxed);
    }
    
    size_t get_count() const {
        return counter.load(std::memory_order_relaxed);
    }
};
 
int main() {
    const int thread_count = 4;
    const int events_per_thread = 1000000;
    
    EventCounter counter;
    std::vector<std::thread> threads;
    
    for (int i = 0; i < thread_count; ++i) {
        threads.emplace_back([&counter, events_per_thread]{
            for (int j = 0; j < events_per_thread; ++j) {
                counter.increment();
            }
        });
    }
    
    for (auto& t : threads) {
        t.join();
    }
    
    std::cout << "Final count: " << counter.get_count() << std::endl;
    // Ожидаем: thread_count * events_per_thread
    
    return 0;
}

class Singleton {
private:
    static std::atomic<Singleton*> instance;
    static std::mutex initMutex;
    
    Singleton() = default;
    
public:
    static Singleton* getInstance() {
        Singleton* p = instance.load(std::memory_order_acquire);
        if (p == nullptr) {
            std::lock_guard<std::mutex> lock(initMutex);
            p = instance.load(std::memory_order_relaxed);
            if (p == nullptr) {
                p = new Singleton();
                instance.store(p, std::memory_order_release);
            }
        }
        return p;
    }
};
 
std::atomic<Singleton*> Singleton::instance{nullptr};
std::mutex Singleton::initMutex;

// C++20
std::atomic<std::shared_ptr<MyClass>> atomic_ptr{
    std::make_shared<MyClass>()
};
 
void update_data() {
    auto new_ptr = std::make_shared<MyClass>();
    // Настраиваем новый объект...
    atomic_ptr.store(new_ptr);
}
 
void use_data() {
    auto current_ptr = atomic_ptr.load();
    // Используем объект безопасно...
}

std::atomic<bool> ready{false};
 
void producer() {
    // Готовим данные...
    ready.store(true);
    ready.notify_one(); // Будим один ждущий поток
}
 
void consumer() {
    // Ждём, пока флаг не станет true
    ready.wait(false); // Блокируемся, пока ready == false
    // Данные готовы к использованию
}

// Пример lock-free стека
template<typename T>
class LockFreeStack {
private:
    struct Node {
        T data;
        Node* next;
        Node(const T& value) : data(value), next(nullptr) {}
    };
 
    std::atomic<Node*> head{nullptr};
 
public:
    void push(const T& value) {
        Node* new_node = new Node(value);
        Node* old_head = head.load();
        do {
            new_node->next = old_head;
        } while (!head.compare_exchange_weak(old_head, new_node));
    }
 
    bool pop(T& result) {
        Node* old_head = head.load();
        while (old_head && !head.compare_exchange_weak(old_head, old_head->next)) {}
        
        if (!old_head) return false;
        
        result = old_head->data;
        delete old_head;
        return true;
    }
};

#include <future>
#include <iostream>
#include <thread>
 
long fibonacci(unsigned n) {
if (n < 2) return n;
return fibonacci(n-1) + fibonacci(n-2);
}
 
int main() {
// Запускаем вычисление в отдельном потоке
std::future<long> result = std::async(fibonacci, 42);
 
// Можем делать что-то ещё, пока вычисляется число Фибоначчи
std::cout << "Вычисляем 42-е число Фибоначчи..." << std::endl;
std::cout << "Пока займёмся другими делами." << std::endl;
 
// Когда результат нужен, просто запрашиваем его
std::cout << "Результат: " << result.get() << std::endl;
return 0;
}

#include <future>
#include <thread>
#include <iostream>
 
void calculate(std::promise<int> result_promise) {
try {
    // Имитация сложных вычислений
    std::this_thread::sleep_for(std::chrono::seconds(2));
    
    // Установка результата
    result_promise.set_value(42);
} catch (...) {
    // Если что-то пошло не так, передаём исключение в future
    result_promise.set_exception(std::current_exception());
}
}
 
int main() {
std::promise<int> promise;
std::future<int> future = promise.get_future();
 
// Запускаем задачу в отдельном потоке
std::thread calculation_thread(calculate, std::move(promise));
 
// Ждём результата
std::cout << "Ожидаем результат..." << std::endl;
try {
    int result = future.get(); // Блокируемся до получения результата
    std::cout << "Результат: " << result << std::endl;
} catch (const std::exception& e) {
    std::cout << "Исключение: " << e.what() << std::endl;
}
 
calculation_thread.join();
return 0;
}

void calculate_with_exception(std::promise<int> promise) {
try {
    throw std::runtime_error("Ой, что-то пошло не так!");
} catch (...) {
    promise.set_exception(std::current_exception());
}
}
 
// И в main:
try {
int result = future.get();
} catch (const std::runtime_error& e) {
std::cout << "Поймано исключение из другого потока: " << e.what() << std::endl;
}

std::promise<int> promise;
std::shared_future<int> shared_future = promise.get_future().share();
 
// Теперь разные потоки могут получить результат
std::thread t1([shared_future]() {
    std::cout << "Поток 1: " << shared_future.get() << std::endl;
});
 
std::thread t2([shared_future]() {
    std::cout << "Поток 2: " << shared_future.get() << std::endl;
});
 
// Устанавливаем значение, которое получат оба потока
promise.set_value(42);
t1.join();
t2.join();

void old_way() {
std::thread t([]{ 
    /* работа */ 
});
// Если забыть вызвать t.join(), программа аварийно завершится
t.join();
}
 
void new_way() {
std::jthread t([]{ 
    /* работа */ 
});
// Когда t выйдет из области видимости, join() вызовется автоматически
}

#include <thread>
#include <chrono>
#include <iostream>
 
int main() {
std::jthread worker([](std::stop_token stoken) {
    int counter = 0;
    while (!stoken.stop_requested() && counter < 10) {
        std::cout << "Работаем... " << counter++ << std::endl;
        std::this_thread::sleep_for(std::chrono::seconds(1));
    }
    std::cout << "Поток завершается" << std::endl;
});
 
// Даём потоку поработать 3 секунды
std::this_thread::sleep_for(std::chrono::seconds(3));
 
// Вежливо просим завершиться
std::cout << "Просим поток завершиться" << std::endl;
worker.request_stop();
 
// join() будет вызван автоматически при уничтожении worker
}

#include <latch>
#include <barrier>
#include <thread>
#include <vector>
#include <iostream>
#include <functional>
 
int main() {
constexpr int num_threads = 4;
std::latch start_signal(1); // Контролируется одним потоком
std::latch finish_signal(num_threads); // Ждём все потоки
 
std::vector<std::jthread> threads;
for (int i = 0; i < num_threads; ++i) {
    threads.emplace_back([i, &start_signal, &finish_signal]() {
        std::cout << "Поток " << i << " готов" << std::endl;
        
        // Ждём сигнала старта
        start_signal.wait();
        
        // Выполняем работу
        std::cout << "Поток " << i << " работает" << std::endl;
        
        // Сообщаем о завершении
        finish_signal.count_down();
    });
}
 
// Даём всем потокам подготовиться
std::this_thread::sleep_for(std::chrono::seconds(1));
 
// Запускаем всех одновременно
std::cout << "Старт!" << std::endl;
start_signal.count_down();
 
// Ждём завершения всех потоков
finish_signal.wait();
std::cout << "Все потоки завершились" << std::endl;
 
// threads уничтожаются автоматически, вызывая join()
}

std::barrier sync_point(num_threads, []{ 
    std::cout << "Итерация завершена" << std::endl; 
});
 
// В каждом потоке
for (int iteration = 0; iteration < 5; ++iteration) {
// Выполняем работу итерации
std::cout << "Поток " << id << " выполняет итерацию " << iteration << std::endl;
 
// Ждём, пока все закончат
sync_point.arrive_and_wait();
 
// Все потоки синхронно переходят к следующей итерации
}

// Мьютекс: для защиты сложных операций
std::mutex balance_mutex;
void transfer_money(Account& from, Account& to, int amount) {
std::lock_guard<std::mutex> lock(balance_mutex);
from.balance -= amount;
to.balance += amount;
}
 
// Атомарная операция: для простых обновлений
std::atomic<int> counter{0};
void count_event() {
counter.fetch_add(1, std::memory_order_relaxed);
}

#include <chrono>
#include <thread>
#include <atomic>
#include <mutex>
#include <shared_mutex>
#include <iostream>
 
constexpr int ITERATIONS = 10000000;
constexpr int THREADS = 4;
 
// Вариант 1: std::mutex
void test_mutex() {
int counter = 0;
std::mutex mtx;
 
auto worker = [&]() {
    for (int i = 0; i < ITERATIONS / THREADS; ++i) {
        std::lock_guard<std::mutex> lock(mtx);
        ++counter;
    }
};
 
std::vector<std::thread> threads;
for (int i = 0; i < THREADS; ++i) {
    threads.emplace_back(worker);
}
 
for (auto& t : threads) {
    t.join();
}
}
 
// Вариант 2: std::atomic
void test_atomic() {
std::atomic<int> counter{0};
 
auto worker = [&]() {
    for (int i = 0; i < ITERATIONS / THREADS; ++i) {
        counter.fetch_add(1, std::memory_order_relaxed);
    }
};
 
std::vector<std::thread> threads;
for (int i = 0; i < THREADS; ++i) {
    threads.emplace_back(worker);
}
 
for (auto& t : threads) {
    t.join();
}
}
 
// Запускаем тесты и засекаем время...

class ThreadSafeCache {
private:
std::unordered_map<std::string, std::string> cache;
mutable std::shared_mutex mutex;
 
public:
std::string get(const std::string& key) {
    std::shared_lock<std::shared_mutex> lock(mutex); // Блокировка для чтения
    auto it = cache.find(key);
    return (it != cache.end()) ? it->second : "";
}
 
void set(const std::string& key, const std::string& value) {
    std::unique_lock<std::shared_mutex> lock(mutex); // Блокировка для записи
    cache[key] = value;
}
};

template<typename T>
class ThreadSafeQueue {
private:
std::queue<T> queue;
std::mutex mutex;
std::condition_variable cv;
 
public:
void push(T value) {
    std::lock_guard<std::mutex> lock(mutex);
    queue.push(std::move(value));
    cv.notify_one(); // Уведомляем одного ждущего потребителя
}
 
T pop() {
    std::unique_lock<std::mutex> lock(mutex);
    cv.wait(lock, [this] { return !queue.empty(); });
    T value = queue.front();
    queue.pop();
    return value;
}
};

std::atomic<bool> data_ready{false};
std::atomic<int> counter{0};
 
void producer() {
// Готовим данные...
counter.store(42, std::memory_order_relaxed);
data_ready.store(true, std::memory_order_release);
data_ready.notify_one(); // Уведомляем одного потребителя
}
 
void consumer() {
data_ready.wait(false); // Ждём, пока data_ready != false
// Теперь counter гарантированно установлен
int value = counter.load(std::memory_order_acquire);
}

// Синхронизация старта потоков с latch
std::latch starting_line(1); // Контролируется одним потоком
 
void worker(int id) {
std::cout << "Поток " << id << " готов" << std::endl;
starting_line.wait(); // Ждём сигнала старта
std::cout << "Поток " << id << " побежал!" << std::endl;
}
 
int main() {
std::vector<std::jthread> threads;
for (int i = 0; i < 5; ++i) {
    threads.emplace_back(worker, i);
}
std::this_thread::sleep_for(std::chrono::seconds(1));
std::cout << "На старт! Внимание! Марш!" << std::endl;
starting_line.count_down(); // Даём сигнал всем потокам стартовать
}

class ConnectionPool {
private:
std::vector<Connection> connections;
std::counting_semaphore<16> available{16}; // Максимум 16 соединений
 
public:
Connection acquire() {
    available.acquire(); // Блокируемся, если нет свободных соединений
    // Возвращаем свободное соединение...
}
 
void release(Connection conn) {
    // Возвращаем соединение в пул...
    available.release(); // Увеличиваем счётчик доступных соединений
}
};

std::once_flag init_flag;
Expensive_Resource* resource = nullptr;
 
void use_resource() {
std::call_once(init_flag, []() {
    resource = new Expensive_Resource();
    // Выполнится только один раз, при первом вызове
});
 
// Теперь можно безопасно использовать resource
}

#include <algorithm>
#include <execution>
#include <vector>
 
std::vector<int> data(10000000);
// Заполняем данными...
 
// Обычная последовательная сортировка
std::sort(data.begin(), data.end());
 
// Параллельная сортировка - используем все доступные ядра!
std::sort(std::execution::par, data.begin(), data.end());

#include <numeric>
#include <execution>
#include <vector>
 
std::vector<int> data(10000000, 1); // 10М единиц
 
// Последовательное суммирование
int sum1 = std::accumulate(data.begin(), data.end(), 0);
 
// Параллельное суммирование
int sum2 = std::reduce(std::execution::par, data.begin(), data.end(), 0);

std::vector<Product> products = /* ... */;
 
// Вычисляем общую стоимость всех товаров параллельно
double total_value = std::transform_reduce(
std::execution::par,
products.begin(), products.end(),
0.0,
std::plus<>(),
[](const Product& p) { return p.price * p.quantity; }
);

// Опасно! Гонка данных!
int sum = 0;
std::for_each(std::execution::par, data.begin(), data.end(), 
[&sum](int x) { sum += x; });
 
// Безопасно, используем атомарные операции
std::atomic<int> atomic_sum = 0;
std::for_each(std::execution::par, data.begin(), data.end(), 
[&atomic_sum](int x) { atomic_sum.fetch_add(x, std::memory_order_relaxed); });

// C++20 с ranges
std::vector<int> result;
auto view = data 
| std::views::filter([](int x) { return x > 0; })
| std::views::transform([](int x) { return x * x; });
 
// Параллельное копирование результатов
std::copy(std::execution::par, view.begin(), view.end(), 
std::back_inserter(result));

// C++23
std::vector<int> result;
auto filtered_squared = data 
| std::views::filter([](int x) { return x > 0; })
| std::views::transform([](int x) { return x * x; });
 
// Параллельное копирование с использованием ranges напрямую
std::ranges::copy(std::execution::par, filtered_squared, 
std::back_inserter(result));

class ThreadPool {
private:
std::vector<std::thread> workers;
std::queue<std::function<void()>> tasks;
std::mutex queue_mutex;
std::condition_variable condition;
bool stop;
 
public:
ThreadPool(size_t threads) : stop(false) {
for (size_t i = 0; i < threads; ++i) {
    workers.emplace_back([this] {
        while (true) {
            std::function<void()> task;
            {
                std::unique_lock<std::mutex> lock(queue_mutex);
                condition.wait(lock, [this] { 
                    return stop || !tasks.empty(); 
                });
                
                if (stop && tasks.empty()) return;
                
                task = std::move(tasks.front());
                tasks.pop();
            }
            task();
        }
    });
}
}
 
template<class F>
void enqueue(F&& f) {
{
    std::unique_lock<std::mutex> lock(queue_mutex);
    tasks.emplace(std::forward<F>(f));
}
condition.notify_one();
}
 
~ThreadPool() {
{
    std::unique_lock<std::mutex> lock(queue_mutex);
    stop = true;
}
condition.notify_all();
for (std::thread& worker : workers) {
    worker.join();
}
}
};

template<class F, class... Args>
auto enqueue(F&& f, Args&&... args) 
-> std::future<typename std::result_of<F(Args...)>::type> {
using return_type = typename std::result_of<F(Args...)>::type;
 
auto task = std::make_shared<std::packaged_task<return_type()>>(
    std::bind(std::forward<F>(f), std::forward<Args>(args)...)
);
 
std::future<return_type> result = task->get_future();
{
    std::unique_lock<std::mutex> lock(queue_mutex);
    if (stop) throw std::runtime_error("enqueue on stopped ThreadPool");
    tasks.emplace([task](){ (*task)(); });
}
condition.notify_one();
return result;
}

ThreadPool pool(4); // Создаем пул из 4 потоков
std::vector<std::future<int>> results;
 
// Отправляем 8 задач
for (int i = 0; i < 8; ++i) {
results.emplace_back(
    pool.enqueue([i] {
        std::this_thread::sleep_for(std::chrono::seconds(1));
        return i * i;
    })
);
}
 
// Получаем результаты
for (auto& result : results) {
std::cout << result.get() << ' ';
}

// Опасный код!
void dangerous() {
std::vector<int> data{1, 2, 3};
ThreadPool pool(4);
 
pool.enqueue([&data] {
    std::this_thread::sleep_for(std::chrono::seconds(1));
    for (auto& item : data) { // Обращение к уже уничтоженному вектору!
        std::cout << item << " ";
    }
});
} // data уничтожается, но задача всё ещё в очереди!

unsigned int cores = std::thread::hardware_concurrency(); // Получаем число ядер
ThreadPool pool(cores); // Идеально для CPU-интенсивных задач

ThreadPool io_pool(cores * 2); // Больше потоков для I/O-операций

class ModernThreadPool {
private:
std::vector<std::jthread> workers; // Автоматический join при уничтожении
// ... остальной код аналогичен
};

// Счётчик с использованием мьютекса
std::mutex mtx;
int counter = 0;
 
void increment_with_mutex() {
    std::lock_guard<std::mutex> guard(mtx);
    counter++;
}
 
// Lock-free счётчик
std::atomic<int> atomic_counter{0};
 
void increment_lockfree() {
    atomic_counter.fetch_add(1, std::memory_order_relaxed);
}

bool update_data(std::atomic<int>& target, int expected, int new_value) {
    return target.compare_exchange_strong(expected, new_value);
    // Если target == expected, устанавливает new_value и возвращает true
    // Иначе, обновляет expected и возвращает false
}

template<typename T>
class LockFreeStack {
private:
    struct Node {
        T data;
        Node* next;
        Node(const T& value) : data(value), next(nullptr) {}
    };
 
    std::atomic<Node*> head{nullptr};
 
public:
    void push(const T& value) {
        Node* new_node = new Node(value);
        
        // Используем CAS цикл для безопасного обновления head
        Node* current_head = head.load();
        do {
            new_node->next = current_head;
        } while (!head.compare_exchange_weak(current_head, new_node));
    }
 
    bool pop(T& result) {
        Node* current_head = head.load();
        
        while (current_head && 
               !head.compare_exchange_weak(current_head, current_head->next)) {
            // Если head изменился, current_head обновляется CAS операцией
        }
        
        if (!current_head) return false; // Стек пуст
        
        result = current_head->data;
        // Внимание: тут есть проблема с освобождением памяти!
        delete current_head; // В реальном коде так делать нельзя
        return true;
    }
};

template<typename T>
class LockFreeStackWithCounter {
private:
    struct Node {
        T data;
        Node* next;
        Node(const T& value) : data(value), next(nullptr) {}
    };
    
    struct TaggedPointer {
        Node* ptr;
        size_t tag;
        
        bool operator==(const TaggedPointer& other) const {
            return ptr == other.ptr && tag == other.tag;
        }
    };
    
    std::atomic<TaggedPointer> head;
    
    // ... остальная реализация, использующая TaggedPointer
};

// Пример wait-free инкремента счётчика (упрощённый)
std::atomic<int> counter{0};
 
void wait_free_increment() {
    counter.fetch_add(1, std::memory_order_relaxed);
    // fetch_add — атомарный и wait-free на большинстве платформ
}

int regular_value = 42;
std::atomic_ref<int> atomic_view{regular_value};
atomic_view.fetch_add(1, std::memory_order_relaxed);
assert(regular_value == 43);

// Упрощённый пример executor-а (синтаксис может измениться)
struct ThreadPoolExecutor {
    // Запускает задачу на выполнение в пуле потоков
    template<typename Function>
    void execute(Function&& f) {
        thread_pool.enqueue(std::forward<Function>(f));
    }
    
private:
    ThreadPool thread_pool{4}; // Используем пул из 4 потоков
};

// Пример композиции отправителей (предположительный синтаксис)
sender auto process_image(std::string filename) {
    return just(filename) 
        | then([](std::string name) { return load_image(name); })
        | then([](Image img) { return apply_filter(img); })
        | then([](Image img) { return save_image(img); });
}
 
// Запуск асинхронной операции
sync_wait(process_image("photo.jpg"));

sender auto operation = just(42)
    | then([](int value) {
        if (value < 0) throw std::invalid_argument("Must be positive");
        return value * 2;
    })
    | upon_error([](std::exception_ptr eptr) {
        try {
            std::rethrow_exception(eptr);
        } catch (const std::exception& e) {
            std::cerr << "Caught: " << e.what() << std::endl;
            return 0; // Восстановление после ошибки
        }
    });

// Предположительный пример асинхронного файлового I/O
sender auto read_file_async(std::string path) {
    return open_file_async(path)
        | then([](file_handle handle) {
            return read_all_async(handle);
        })
        | then([](std::vector<char> data) {
            return std::string(data.begin(), data.end());
        });
}
 
// Асинхронная работа с сетью
sender auto fetch_url(std::string url) {
    return connect_async(url)
        | then([](connection conn) {
            return send_request_async(conn, "GET", "/");
        })
        | then([](http_response resp) {
            return resp.body;
        });
}

// Пример с использованием libunifex (доступна сейчас)
namespace ex = unifex;
 
auto async_task = ex::schedule(my_scheduler)
    | ex::then([](auto&&...) {
        return hard_computation();
    })
    | ex::then([](int result) {
        return result * 2;
    });
 
int result = ex::sync_wait(async_task);

// Концептуальный пример распараллеливания на основе зависимостей данных
sender auto process_data(std::vector<Data> items) {
    // Создаём отправителя для каждого элемента
    std::vector<sender auto> item_senders;
    for (auto& item : items) {
        item_senders.push_back(just(item) | then(process_item));
    }
    
    // Запускаем все обработчики параллельно и собираем результаты
    return when_all(item_senders) 
        | then([](std::vector<Result> results) {
            return combine_results(results);
        });
}

#include <iostream>
#include <thread>
#include <vector>
#include <atomic>
#include <string>
#include <sys/socket.h>
#include <netinet/in.h>
#include <unistd.h>
 
class SimpleServer {
private:
    int server_fd;
    std::atomic<bool> running{true};
    std::vector<std::thread> client_threads;
 
    void handle_client(int client_socket) {
        char buffer[1024] = {0};
        
        while (running) {
            // Читаем данные от клиента
            int bytes_read = read(client_socket, buffer, sizeof(buffer) - 1);
            
            if (bytes_read <= 0) {
                // Клиент отключился или ошибка
                break;
            }
            
            buffer[bytes_read] = '\0';
            std::cout << "Получено: " << buffer << std::endl;
            
            // Отправляем ответ
            std::string response = "Сервер получил: ";
            response += buffer;
            send(client_socket, response.c_str(), response.length(), 0);
        }
        
        close(client_socket);
    }
 
public:
    SimpleServer(int port) {
        // Создаём серверный сокет
        server_fd = socket(AF_INET, SOCK_STREAM, 0);
        if (server_fd < 0) {
            throw std::runtime_error("Ошибка при создании сокета");
        }
        
        // Настройка адреса сервера
        sockaddr_in address;
        address.sin_family = AF_INET;
        address.sin_addr.s_addr = INADDR_ANY;
        address.sin_port = htons(port);
        
        // Привязываем сокет к адресу
        if (bind(server_fd, (struct sockaddr*)&address, sizeof(address)) < 0) {
            close(server_fd);
            throw std::runtime_error("Ошибка при привязке сокета");
        }
        
        // Начинаем прослушивание
        if (listen(server_fd, SOMAXCONN) < 0) {
            close(server_fd);
            throw std::runtime_error("Ошибка при прослушивании");
        }
    }
 
    void start() {
        std::cout << "Сервер запущен, ожидание подключений..." << std::endl;
        
        while (running) {
            // Принимаем новое подключение
            sockaddr_in client_addr;
            socklen_t client_addr_len = sizeof(client_addr);
            
            int client_socket = accept(server_fd, (struct sockaddr*)&client_addr, 
                                       &client_addr_len);
            
            if (client_socket < 0) {
                std::cerr << "Ошибка при принятии подключения" << std::endl;
                continue;
            }
            
            std::cout << "Новое подключение принято" << std::endl;
            
            // Создаём новый поток для обработки клиента
            client_threads.emplace_back(&SimpleServer::handle_client, this, client_socket);
        }
    }
 
    ~SimpleServer() {
        running = false;
        close(server_fd);
        
        // Ждём завершения всех клиентских потоков
        for (auto& thread : client_threads) {
            if (thread.joinable()) {
                thread.join();
            }
        }
    }
};
 
int main() {
    try {
        SimpleServer server(8080);
        server.start();
    }
    catch (const std::exception& e) {
        std::cerr << "Ошибка: " << e.what() << std::endl;
        return 1;
    }
    
    return 0;
}

class ThreadPoolServer {
private:
    int server_fd;
    std::atomic<bool> running{true};
    ThreadPool pool; // Наш пул потоков из предыдущей главы
    
    void handle_client(int client_socket) {
        // Тот же код обработки, что и раньше
    }
 
public:
    ThreadPoolServer(int port, size_t thread_count) 
        : pool(thread_count) { // Создаём пул с заданным числом потоков
        // Инициализация сокета как в предыдущем примере
    }
 
    void start() {
        std::cout << "Сервер запущен с пулом из " 
                  << pool.size() << " потоков" << std::endl;
        
        while (running) {
            // Принимаем новое подключение
            int client_socket = accept(server_fd, /*...*/);
            
            if (client_socket < 0) {
                continue;
            }
            
            // Отправляем задачу в пул потоков вместо создания нового потока
            pool.enqueue([this, client_socket] {
                this->handle_client(client_socket);
            });
        }
    }
    
    // Деструктор аналогичен предыдущему
};

// Упрощённый пример с использованием epoll
class EPollServer {
private:
    int server_fd;
    int epoll_fd;
    std::atomic<bool> running{true};
    ThreadPool pool;
    
    // Инициализация и другие методы...
 
    void process_events() {
        const int MAX_EVENTS = 64;
        struct epoll_event events[MAX_EVENTS];
        
        while (running) {
            int nfds = epoll_wait(epoll_fd, events, MAX_EVENTS, -1);
            
            for (int i = 0; i < nfds; i++) {
                if (events[i].data.fd == server_fd) {
                    // Новое подключение
                    accept_connection();
                } else {
                    // Данные от существующего клиента
                    pool.enqueue([this, fd = events[i].data.fd] {
                        handle_client_data(fd);
                    });
                }
            }
        }
    }
};

// Классический пример трудноуловимого бага
std::map<std::string, int> shared_cache; // Общий кэш без защиты
 
void update_stats(const std::string& key) {
// Гонка данных! Вероятность проблемы мала, но она есть
shared_cache[key]++;
}

WARNING: ThreadSanitizer: data race (pid=12345)
  Write of size 4 at 0x7fff52d1c078 by thread T1:
    #0 update_stats cache.cpp:42
    #1 operator() thread_func.cpp:27
 
  Previous read of size 4 at 0x7fff52d1c078 by main thread:
    #0 get_stats cache.cpp:36
    #1 main main.cpp:15

std::mutex mtx;
 
void function_with_bug() {
mtx.lock();
// Какой-то код
if (error_condition) {
    return; // Упс! Забыли разблокировать мьютекс
}
// Больше кода
mtx.unlock();
}

void test_threaded_queue() {
for (int i = 0; i < 10000; ++i) {
    ThreadSafeQueue<int> queue;
    std::vector<std::thread> threads;
    
    // 10 потоков добавляют элементы
    for (int t = 0; t < 10; ++t) {
        threads.emplace_back([&queue, t] {
            // Случайная задержка перед операцией
            std::this_thread::sleep_for(
                std::chrono::microseconds(rand() % 100));
            queue.push(t);
        });
    }
    
    // 10 потоков извлекают элементы
    std::vector<int> results[10];
    for (int t = 0; t < 10; ++t) {
        threads.emplace_back([&queue, &results, t] {
            for (int j = 0; j < 5; ++j) {
                int value;
                if (queue.try_pop(value)) {
                    results[t].push_back(value);
                }
                std::this_thread::sleep_for(
                    std::chrono::microseconds(rand() % 100));
            }
        });
    }
    
    for (auto& t : threads) {
        t.join();
    }
    
    // Проверка результатов
    // ...
}
}

// Потенциальное ложное разделение
struct ThreadData {
std::atomic<int> counter1; // Используется потоком 1
std::atomic<int> counter2; // Используется потоком 2
};

// Предотвращение ложного разделения
struct alignas(64) PaddedCounter {
std::atomic<int> value;
char padding[60]; // Заполняем остаток кэш-линии
};
 
struct ThreadData {
PaddedCounter counter1;
PaddedCounter counter2;
};

// C++17 решение
struct ThreadData {
std::atomic<int> counter1;
alignas(std::hardware_destructive_interference_size) std::atomic<int> counter2;
};

// Грубозернистая блокировка (плохо для производительности)
class Database {
private:
std::mutex mtx;
std::map<std::string, UserData> users;
std::map<int, SessionData> sessions;
 
public:
void update_user(const std::string& username, const UserData& data) {
    std::lock_guard<std::mutex> lock(mtx); // Блокируем всю БД
    users[username] = data;
}
 
void update_session(int session_id, const SessionData& data) {
    std::lock_guard<std::mutex> lock(mtx); // Блокируем всю БД
    sessions[session_id] = data;
}
};
 
// Тонкозернистая блокировка (лучше для параллелизма)
class ImprovedDatabase {
private:
std::mutex users_mutex;
std::mutex sessions_mutex;
std::map<std::string, UserData> users;
std::map<int, SessionData> sessions;
 
public:
void update_user(const std::string& username, const UserData& data) {
    std::lock_guard<std::mutex> lock(users_mutex); // Блокируем только users
    users[username] = data;
}
 
void update_session(int session_id, const SessionData& data) {
    std::lock_guard<std::mutex> lock(sessions_mutex); // Блокируем только sessions
    sessions[session_id] = data;
}
};