Многопоточность в C++: std::thread, мьютексы и condition_variable

Первый поток

<thread> появился в C++11 и дал переносимые потоки — без pthreads и WinAPI. Поток запускается с callable-объектом и идёт параллельно с вызывающим:

#include <thread>
#include <iostream>

void worker(int id) {
    std::cout << "поток " << id << " работает\n";
}

int main() {
    std::thread t1(worker, 1);
    std::thread t2(worker, 2);

    t1.join();   // ждём завершения
    t2.join();
}

Обязательное правило: std::thread либо join() (дождаться), либо detach() (отпустить). Иначе деструктор вызовет std::terminate. С C++20 есть std::jthread — делает join автоматически и поддерживает остановку через stop_token.

{
    std::jthread t([]{ /* работа */ });
}   // join сам вызовется в деструкторе

Гонка данных и мьютекс

Главная угроза многопоточности — data race: два потока одновременно пишут в одну переменную. Результат непредсказуем.

int counter = 0;

void bad() {
    for (int i = 0; i < 100000; ++i) ++counter;  // не атомарно!
}
// два потока — вместо 200000 получите случайное число

Защита — мьютекс: только один поток может держать блокировку:

#include <mutex>

std::mutex m;
int counter = 0;

void good() {
    for (int i = 0; i < 100000; ++i) {
        std::lock_guard lock(m);   // захват при входе, отпуск при выходе
        ++counter;
    }
}

std::lock_guard — RAII-обёртка: деструктор отпускает мьютекс, включая выход по исключению. Для нескольких мьютексов — std::scoped_lock (C++17), который захватывает их без дедлока:

std::mutex m1, m2;
{
    std::scoped_lock lock(m1, m2);   // захватит оба атомарно
    // работаем с двумя ресурсами
}

Deadlock: как не попасться

Дедлок — потоки ждут друг друга вечно. Классика: поток A захватил m1 и хочет m2, поток B захватил m2 и хочет m1.

Профилактика:

• захватывать мьютексы всегда в одном порядке;
• использовать scoped_lock для нескольких мьютексов;
• держать блокировку минимальное время — не делать IO под мьютексом;
• не вызывать чужой код (колбэки) под блокировкой.

condition_variable: ожидание события

Когда поток должен ждать условия (например, очередь не пуста) — не крутите цикл со sleep, используйте condition_variable:

#include <condition_variable>

std::mutex m;
std::condition_variable cv;
std::queue<int> tasks;

// производитель
void produce(int v) {
    {
        std::lock_guard lock(m);
        tasks.push(v);
    }
    cv.notify_one();   // разбудить ждущего
}

// потребитель
void consume() {
    while (true) {
        std::unique_lock lock(m);
        cv.wait(lock, []{ return !tasks.empty(); });  // спим до предиката
        int v = tasks.front(); tasks.pop();
        lock.unlock();
        process(v);
    }
}

cv.wait(lock, pred) атомарно отпускает мьютекс и засыпает; при пробуждении захватывает обратно и проверяет предикат — защита от spurious wakeup.

std::async и future: задачи с результатом

Для «посчитай параллельно и верни результат» есть std::async — запускает функцию и возвращает future:

#include <future>

auto fut = std::async(std::launch::async, []{
    return heavyComputation();
});

// делаем что-то ещё...
int result = fut.get();   // ждать и забрать результат

future.get() блокирует до готовности и может перебросить исключение из потока в ваш. Есть и shared_future для нескольких ожидателей, packaged_task для явной задачи.

atomic: лёгкая синхронизация без мьютексов

Для простых счётчиков и флагов мьютекс — overkill. std::atomic даёт атомарные операции:

#include <atomic>

std::atomic<int> counter{0};
std::atomic<bool> done{false};

void worker() {
    counter.fetch_add(1);      // атомарный ++
    // или counter++ — тоже атомарно
    done.store(true);          // запись
}

// чтение
while (!done.load()) { /* ждём */ }

Атомики сложнее, чем кажется — есть модели memory ordering. Базовое правило: используйте atomic по умолчанию (sequential consistency), мьютекс — когда нужна критическая секция для нескольких операций.

Типичные ошибки

• Забыть join/detach — std::terminate при разрушении std::thread.
• Data race на простом int — ++ не атомарен.
• Захват мьютексов в разном порядке — дедлок.
• Сон в цикле вместо condition_variable — процессор впустую и задержки.
• std::ref для ссылок: std::thread t(f, std::ref(obj)) — иначе объект скопируется.
• Исключения в потоке — поймаются только внутри потока или через future::get.

Итог

Минимум для работы: std::jthread для потоков, lock_guard/scoped_lock для данных, condition_variable для событий, std::async для результатов, atomic для счётчиков и флагов. Помните: гонки данных — UB, мьютексы захватываем в одном порядке, потоки всегда join или detach.

Полный пример: пул потоков-воркеров

class ThreadPool {
    std::vector<std::thread> workers_;
    std::queue<std::function<void()>> tasks_;
    std::mutex m_;
    std::condition_variable cv_;
    bool stop_ = false;
public:
    ThreadPool(int n) {
        for (int i = 0; i < n; ++i)
            workers_.emplace_back([this] { work(); });
    }
    void post(std::function<void()> f) {
        { std::lock_guard lk(m_); tasks_.push(std::move(f)); }
        cv_.notify_one();
    }
    void work() {
        while (true) {
            std::function<void()> task;
            {
                std::unique_lock lk(m_);
                cv_.wait(lk, [this]{ return stop_ || !tasks_.empty(); });
                if (stop_ && tasks_.empty()) return;
                task = std::move(tasks_.front()); tasks_.pop();
            }
            task();
        }
    }
    ~ThreadPool() {
        { std::lock_guard lk(m_); stop_ = true; }
        cv_.notify_all();
        for (auto& t : workers_) t.join();
    }
};

Параллельные алгоритмы (C++17)

Стандартные алгоритмы могут выполняться параллельно — достаточно передать execution policy:

#include <execution>

std::for_each(std::execution::par_unseq, v.begin(), v.end(),
              [](int& x){ x = heavy(x); });

auto sum = std::reduce(std::execution::par, v.begin(), v.end(), 0LL);

par — может параллелить, par_unseq — ещё и векторизовать, seq — обычно.

Проверка гонок: ThreadSanitizer

g++ -fsanitize=thread -g app.cpp -o app -pthread
./app
# TSan покажет каждую data race со стеками обоих потоков

Атомики без мьютекса: флаги и счётчики

Для простых shared-счётчиков atomic легче и быстрее мьютекса:

std::atomic<int> processed{0};
std::atomic<bool> done{false};

// в воркерах:
++processed;
// когда всё сделано:
done.store(true, std::memory_order_release);
// в наблюдателе:
while (!done.load(std::memory_order_acquire));

memory_order_release/acquire — минимальный достаточный набор для публикации данных.

Мини-проект: параллельный downloader

Напишите класс, который берёт список URL и качает их N потоками с семафором ограничения параллелизма (std::counting_semaphore C++20) — классическая задача, где concurrency даёт измеримое ускорение.