Dev Blog

Атомарні операції та асинхронні результати ​

Атомарні операції ​

Шаблон std::atomic<T> (https://en.cppreference.com/w/cpp/atomic/atomic) виконує операції над цілими числами, покажчиками й bool атомарно, без м’ютекса, за допомогою команд процесора (lock xadd, lock cmpxchg). Оператори ++, += і методи fetch_add, exchange атомарні, а вираз x = x + 1 складається з двох окремих атомарних операцій і гонитву не усуває.

Основа неблокувальних алгоритмів – порівняння з обміном (compare-and-swap, CAS). Метод compare_exchange_weak(expected, desired) записує desired, лише якщо поточне значення дорівнює expected; інакше записує фактичне значення в expected і повертає false. Версія weak може хибно повернути false, тому її викликають у циклі:

cpp
std::atomic<int> maximum = 0;

void update_max(int value) {
    int current = maximum.load();
    while (value > current &&
           !maximum.compare_exchange_weak(current, value)) {
        // current оновлено фактичним значенням – повторити
    }
}

Методи wait(old) і notify_one() / notify_all() (C++20) дозволяють чекати на зміну атомарної змінної без активного очікування, наприклад ready.wait(false) до встановлення прапорця.

Хибне розділення (тема 4) виникає й у C++, коли лічильники різних потоків потрапляють в одну кеш-лінію. Константа std::hardware_destructive_interference_size (на x86-64 – 64 байти) задає мінімальну відстань, на яку слід розносити такі дані:

cpp
struct Counter {
    alignas(std::hardware_destructive_interference_size)
    std::atomic<long> value = 0;          // sizeof(Counter) == 64
};

Counter counters[16];                     // кожен – у своїй кеш-лінії

Асинхронні результати: async, future, promise ​

Потік std::thread не повертає значення. Для обміну результатом між потоками є майбутнє значення std::future<T> (аналог Task<T>) і пов’язані з ним засоби (https://en.cppreference.com/w/cpp/thread/async):

  • std::async(policy, f, args…) виконує функцію й повертає future. Політика std::launch::async запускає новий потік, std::launch::deferred – відкладає виконання до виклику get() у потоці, що його викликав. Без політики реалізація обирає сама, тому для паралелізму політику async вказують явно;
  • future.get() чекає результату й повертає його (лише один раз) або повторно кидає виняток, що виник у функції; wait_for чекає з тайм-аутом; std::shared_future дозволяє читати результат кільком потокам;
  • std::promise<T> – «обіцянка», через яку потік сам встановлює значення (set_value) або виняток (set_exception) для свого future; аналог TaskCompletionSource;
  • std::packaged_task<R(Args…)> – обгортка функції, яка після виклику записує результат у future; зручна для черг задач і пулів.
cpp
// fib(n) – рекурсивне обчислення числа Фібоначчі
auto f1 = std::async(std::launch::async, fib, 35);  // новий потік
long long own = fib(32);                            // паралельно
std::println("{} {}", f1.get(), own);

std::promise<double> promise;
std::future<double> reading = promise.get_future();
std::jthread sensor([p = std::move(promise)]() mutable {
    try {
        throw std::runtime_error("датчик не відповідає");
    } catch (...) {
        p.set_exception(std::current_exception());
    }
});
try {
    reading.get();                      // виняток із потоку sensor
} catch (const std::exception& e) {
    std::println("{}", e.what());       // датчик не відповідає
}

Увага

Деструктор future, отриманого від std::async, чекає завершення задачі. Виклик std::async(std::launch::async, f); без збереження результату виконується послідовно: тимчасовий future руйнується одразу, і програма чекає, доки f завершиться.

Пул потоків ​

Створення потоку операційної системи триває десятки мікросекунд і потребує пам’яті для стека, тому для багатьох дрібних задач потоки використовують повторно в пулі потоків (тема 2). Стандартна бібліотека C++ пулу не має (його роль у C++26 виконують планувальники std::execution), тому пул пишуть самостійно або беруть з бібліотек (TBB, Boost.Asio). Класичний пул (рис. 9.6) складається з вектора потоків std::jthread, черги задач std::queue<std::move_only_function<void()>> під м’ютексом, умовної змінної з прапорцем завершення, методу submit(f), який загортає функцію в std::packaged_task, ставить її в чергу, викликає notify_one() і повертає future, та деструктора, який встановлює прапорець, викликає notify_all() і дочікується потоків. Тип std::move_only_function (C++23) потрібен тому, що packaged_task не можна копіювати. Задачу виконують після звільнення м’ютекса, інакше робітники працювали б по черзі. Повна реалізація – у прикладі «Пул потоків».