Привет всем! Сегодня мы будем говорить о потоках и разбирать многопоточность в C++.

Представим простую задачу: нужно увеличить переменную миллион раз в каждом из десяти потоков.

Звучит так, будто нам всего-то достаточно создать потоки, запустить цикл и дождаться их завершения. Но если все они работают с одной переменной, то и результат может оказаться совсем не тем, который мы ожидали.

Разберём эту задачу на C++ и увидим, какие варианты решения у нас есть. Заодно выясним, почему замена обычного int на std::atomic — это не всегда конец истории.

Начнём с обычного счётчика

Для начала напишем код без синхронизации:

#include <iostream>
#include <thread>
#include <vector>
#include <cstdlib>

auto main() -> int {
    constexpr int thread_count = 10;
    constexpr int increments = 1'000'000;

    int counter = 0;
    std::vector<std::thread> threads;

    for (int i = 0; i < thread_count; i++) {
        threads.emplace_back([&counter]() {
            for (int j = 0; j < increments; j++) {
                counter++;
            }
        });
    }

    for (std::thread& thread : threads) {
        thread.join();
    }

    std::cout << "Counter: " << counter << '\n';
    return EXIT_SUCCESS;
}

Вроде бы всё логично, десять потоков выполняют по миллиону операций. Значит, в конце должно получиться:

Counter: 10000000

Но гарантировать что будет такой результат нельзя.

Проблема в этой строке:

counter++;

Но на первый взгляд это одна операция. А на деле увеличение значения включает чтение, изменение и запись. Несколько потоков могут одновременно прочитать одно и то же значение, после чего записать результат поверх друг друга.

Например, значение счётчика равно 5:

• Первый поток читает 5.

• Второй поток тоже читает 5.

• Первый записывает 6.

• Второй записывает 6.

Произошло два увеличения, а счётчик изменился только на единицу.

Это упрощённая иллюстрация гонки. В C++ такая ситуация называется data race, и здесь есть важный момент: гонка данных в программе на C++ приводит к неопределённому поведению. Поэтому нельзя рассчитывать даже на то, что программа всегда будет выдавать просто немного заниженный результат.

Вариант первый: std::mutex

Самое логичное решение — разрешить только одному потоку одновременно изменять счётчик.

Для этого используем std::mutex.

#include <iostream>
#include <mutex>
#include <thread>
#include <vector>

int main() {
    int counter = 0;
    std::mutex counter_mutex;

    constexpr int thread_count = 10;
    constexpr int iterations = 1'000'000;

    std::vector<std::thread> threads;

    for (int i = 0; i < thread_count; ++i) {
        threads.emplace_back([&]() {
            for (int j = 0; j < iterations; ++j) {
                std::lock_guard<std::mutex> lock(counter_mutex);
                ++counter;
            }
        });
    }

    for (auto& thread : threads) {
        thread.join();
    }

    std::cout << "Counter: " << counter << '\n';
}

Теперь перед изменением счётчика поток захватывает мьютекс. Пока он удерживает его блокировку, остальные потоки не могут войти в данный участок кода.

Здесь используется std::lock_guard. Он захватывает мьютекс при создании объекта и освобождает его при выходе из области видимости. Даже если внутри произойдёт исключение, блокировка не останется случайно захваченной.

Решение рабочее, но есть нюанс.

Мы захватываем мьютекс на каждой итерации цикла. Получается что миллион попыток захвата блокировки на один поток. Но при большом количестве потоков у них будет конкуренция за один и тот-же ресурс

Для простой задачи это может оказаться лишней работой.

Можно ли обойтись без мьютекса?

Вариант второй: std::atomic

Для операций над отдельными значениями в C++ есть атомарные типы.

Попробуем заменить обычный int на std::atomic<int>:

counter++;

fetch_add() атомарно увеличивает значение. Два потока больше не смогут потерять обновление, как это происходило с обычным int.

Отдельно стоит обратить внимание на:

std::memory_order_relaxed

Для нашего счётчика этого достаточно и нам нужна атомарность самого увеличения, но мы не используем счётчик для передачи другим потокам каких-либо дополнительных данных.

relaxed не означает, что операция становится небезопасной. Атомарность сохраняется, просто мы не запрашиваем дополнительные гарантии порядка операций с памятью.

Но есть ещё один вопрос: если atomic проще и не требует явного мьютекса, значит ли это - что он всегда быстрее?

Нет.

Атомарная операция тоже имеет стоимость. Если много потоков постоянно изменяют одну и ту же переменную, они конкурируют за одну область памяти. В зависимости от процессора и нагрузки это может стать узким местом.

Поэтому наличие atomic само по себе не гарантирует хорошую производительность.

Вариант третий: не делить общий счётчик вообще

А теперь посмотрим на задачу с другой стороны.

Зачем каждому потоку обращаться к одной переменной? Ведь каждый из них может вести свой собственный подсчёт, а результаты их можно сложить после завершения работы.

#include <iostream>
#include <numeric>
#include <thread>
#include <vector>
#include <cstdlib>

auto main() -> int {
    constexpr int thread_count = 10;
    constexpr int increments = 1'000'000;

    std::vector<int> partial_results(thread_count, 0);
    std::vector<std::thread> threads;

    for (int i = 0; i < thread_count; ++i) {
        threads.emplace_back([&, i]() {
            int local_counter = 0;

            for (int j = 0; j < increments; ++j) {
                local_counter++;
            }

            partial_results.at(i) = local_counter;
        });
    }

    for (std::thread& thread : threads) {
        thread.join();
    }

    const int total = std::accumulate(
        partial_results.begin(),
        partial_results.end(),
        0
    );

    std::cout << "Counter: " << total << '\n';
    return EXIT_SUCCESS;
}

В этом варианте у каждого потока своя переменная local_counter. и поэтому потоки не обращаются к одному общему счётчику.

Каждый записывает результат в отдельный элемент results. И после join() основной поток собирает все значения.

Здесь нет необходимости использовать ни mutex, ни atomic для самих счётчиков.

Я думаю, что именно этот пример хорошо показывает одну из важных идей многопоточности: иногда лучше изменить организацию работы, чем добавлять синхронизацию.

Но да, такой подход он подходит не для каждой задачи. Если результат одного потока должен быть доступен остальным прямо сейчас во время выполнения, локального счётчика может быть недостаточно.

Но если задача допускает независимую обработку данных, разделение работы часто оказывается хорошим решением.

Что выбрать: mutex, atomic или локальные данные?

Лучшего ответа нет. Всё зависит от того, что именно мы синхронизируем.

Я бы не выбирал atomic только потому что он выглядит современнее mutex. Если нужно согласованно менять несколько полей объекта, то мьютекс обычно более понятнее и безопаснее.

И наоборот, если использовать мьютекс для простого атомарного счётчика то не всегда имеет смысл.

std::thread и std::jthread

До этого момента мы создавали потоки через std::thread.

Но у него есть особенность: перед уничтожением объекта потока нужно обязательно вызвать join() или detach(). Иначе программа завершится через std::terminate().

В C++20 появился std::jthread, который упрощает управление временем жизни потока.

#include <iostream>
#include <thread>
#include <cstdlib>

auto main() -> int {
    std::jthread worker([]() {
        std::cout << "Worker is running" << '\n';
    });

    // При выходе worker автоматически присоединится.
    return EXIT_SUCCESS;
}

Когда наш объект worker выходит из данной области видимости, его деструктор автоматически выполняет join().

У jthread также есть механизм кооперативной остановки через std::stop_token:

#include <chrono>
#include <iostream>
#include <thread>
#include <cstdlib>

using namespace std::chrono_literals;

auto main() -> int {
    std::jthread worker([](std::stop_token stop_token) {
        while (!stop_token.stop_requested()) {
            std::cout << "Working..." << '\n';
            std::this_thread::sleep_for(200ms);
        }

        std::cout << "Stopping" << '\n';
    });

    std::this_thread::sleep_for(1s);
    worker.request_stop();
    return EXIT_SUCCESS;
}

Здесь поток сам периодически проверяет запрос на остановку. И это важно, request_stop() не убивает поток принудительно, он только отправляет запрос, который рабочий поток должен обработать.

Для небольших задач jthread часто удобнее обычного thread, особенно когда важно не забыть корректно завершить поток.

А что с std::condition_variable?

mutex и atomic решают разные задачи, но даже иногда нам нужно не просто защитить данные, а заставить этот поток подождать определённого события.

Например, есть очередь задач. Один поток добавляет задачи, а другой забирает их. Если очередь пустая, то нет смысла постоянно проверять её в цикле.

Для этого можно использовать std::condition_variable.

Упрощённый пример:

#include <future>
#include <iostream>
#include <cstdlib>

auto calculate() -> uint64_t {
    uint64_t result = 0;

    for (uint64_t i = 1; i <= 1'000'000; i++) {
        result += i;
    }

    return result;
}

auto main() -> int {
    auto result = std::async(
        std::launch::async,
        calculate
    );

    std::cout << result.get() << '\n';
    return EXIT_SUCCESS;
}

Рабочий поток ждёт, пока ready станет true, и после изменения состояния основной поток вызывает notify_one(), и ожидающий поток получает возможность продолжить работу.

Обратите внимание вот на что, на предикат внутри wait().

Ожидание может завершиться не только после уведомления, поэтому условие нужно проверять. Вариант с предикатом делает это за нас.

condition_variable особенно полезен при реализации очередей задач, пулов потоков и других механизмов, где один поток должен ждать действий другого.

Что в итоге?

На примере одного счётчика мы посмотрели несколько способов организации совместной работы потоков.

Самый простой код не всегда является безопасным, и поэтому мьютекс решает проблему гонки данных, но добавляет синхронизацию. Атомарные операции удобные для отдельных значений, однако не гарантируют автоматического ускорения. А иногда лучше вообще отказаться от общего изменяемого состояния.

Последний вариант мне кажется ну особенно интересным: прежде чем добавлять блокировки, стоит подумать, действительно ли нашим нескольким потокам нужно изменять одни и те же данные?

В следующей статье хочу посмотреть на похожие задачи, но уже со стороны Rust. Там компилятор проверяет многие ограничения при работе с потоками ещё до запуска программы. Будет интересно сравнить, какие решения мы можем написать в C++ и какие ограничения встретит тот же код в Rust.