Один счётчик, десять потоков и неправильный результат: разбираемся в многопоточности C++

Представим простую задачу: десять потоков должны увеличить общий счётчик по миллиону раз. Математика элементарная — в результате ожидаем десять миллионов.

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

Разберёмся, почему так происходит, как исправить проблему с помощью std::mutex и std::atomic, когда нужны std::condition_variable и std::future, а также чем std::jthread удобнее обычного std::thread.

Все примеры рассчитаны на C++20, если не указано иное.

1. Ошибка, которую легко не заметить

Начнём с программы:

#include <iostream>#include <thread>#include <vector>int main() {    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 (auto& thread : threads) {        thread.join();    }    std::cout << "Counter: " << counter << '\n';}

На первый взгляд всё правильно:

• создаём десять потоков;

• каждый миллион раз увеличивает счётчик;

• ждём завершения всех потоков;

• выводим результат.

Ожидаемое значение:

Counter: 10000000

Однако программа содержит data race — гонку данных.

Операция ++counter не является одной неделимой операцией. Упрощённо её можно представить так:

1. прочитать текущее значение;

2. увеличить его;

3. записать новое значение.

Допустим, счётчик равен 100. Два потока могут одновременно прочитать это значение, оба вычислить 101 и записать 101. Одно увеличение потеряно.

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

И ещё: join() не исправляет гонку. Он позволяет дождаться завершения потока и синхронизирует действия завершившегося потока с потоком, который вызвал join(). Но он не защищает общий счётчик во время параллельного выполнения.

Как исправить код?

2. std::mutex: защищаем общий ресурс

Самый очевидный вариант — мьютекс. Он позволяет организовать взаимное исключение: в защищённый участок кода одновременно проходит только один поток, владеющий этим мьютексом.

#include <iostream>#include <mutex>#include <thread>#include <vector>int main() {    constexpr int thread_count = 10;    constexpr int increments = 1'000'000;    int counter = 0;    std::mutex mutex;    std::vector<std::thread> threads;    for (int i = 0; i < thread_count; ++i) {        threads.emplace_back([&]() {            for (int j = 0; j < increments; ++j) {                std::lock_guard<std::mutex> lock(mutex);                ++counter;            }        });    }    for (auto& thread : threads) {        thread.join();    }    std::cout << "Counter: " << counter << '\n';}

Теперь доступ к counter защищён. std::lock_guard захватывает мьютекс при создании объекта и освобождает его при выходе из области видимости — в том числе при раскрутке стека из-за исключения.

Это намного безопаснее, чем вручную вызывать lock() и unlock(): если забыть освободить мьютекс на одном из путей выполнения, остальные потоки могут зависнуть.

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

Мьютекс не обязательно плох для производительности. Он нужен, когда защищает действительно общий изменяемый объект или несколько связанных операций. Просто здесь есть более подходящий инструмент.

3. std::atomic: когда нужна одна атомарная операция

Если задача сводится к безопасному изменению одного счётчика, можно использовать std::atomic.

#include <atomic>#include <iostream>#include <thread>#include <vector>int main() {    constexpr int thread_count = 10;    constexpr int increments = 1'000'000;    std::atomic<int> counter{0};    std::vector<std::thread> threads;    for (int i = 0; i < thread_count; ++i) {        threads.emplace_back([&]() {            for (int j = 0; j < increments; ++j) {                counter.fetch_add(1, std::memory_order_relaxed);            }        });    }    for (auto& thread : threads) {        thread.join();    }    std::cout << "Counter: " << counter.load() << '\n';}

Здесь fetch_add() атомарно увеличивает значение. Два потока не смогут потерять друг друга из-за одновременной записи.

Почему используется std::memory_order_relaxed?

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

При этом результат читается после завершения всех потоков. join() обеспечивает необходимую синхронизацию с завершившимися потоками, а атомарный объект позволяет безопасно выполнять параллельные увеличения.

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

И ещё один нюанс: атомарные операции не гарантированно быстрее мьютексов на любой платформе. Это зависит от типа операции, процессора, конкуренции потоков и реализации.

4. А можно вообще не делить один счётчик?

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

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

#include <iostream>#include <numeric>#include <thread>#include <vector>int main() {    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[i] = local_counter;        });    }    for (auto& thread : threads) {        thread.join();    }    const int total = std::accumulate(        partial_results.begin(),        partial_results.end(),        0    );    std::cout << "Counter: " << total << '\n';}

Почему здесь не нужен мьютекс?

Каждый поток изменяет собственный local_counter. А каждый элемент partial_results записывается только одним потоком. После завершения всех потоков главный поток складывает результаты.

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

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

Однако это не универсальное решение: если потокам действительно нужно совместно изменять один объект, локальные счётчики не всегда помогут. Кроме того, для сложных задач стоит учитывать расположение данных в памяти и возможный false sharing.

5. std::thread и std::jthread: кто отвечает за завершение?

В предыдущих примерах используется std::thread. Он позволяет запустить отдельный поток выполнения, но требует аккуратного управления временем его жизни.

Если объект std::thread уничтожается, пока поток ещё остаётся присоединяемым, программа вызывает std::terminate(). Поэтому необходимо явно вызвать join() или detach().

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

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

#include <iostream>#include <thread>int main() {    std::jthread worker([]() {        std::cout << "Worker is running\n";    });    // При выходе worker автоматически присоединится.}

При уничтожении std::jthread, если его поток ещё присоединяемый, объект запрашивает остановку и присоединяет поток. В этом примере функция ничего не проверяет и быстро завершается, поэтому автоматическое присоединение достаточно.

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

Чтобы показать кооперативную остановку, можно написать так:

#include <chrono>#include <iostream>#include <thread>using namespace std::chrono_literals;int main() {    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();}

Поток периодически проверяет stop_token и сам завершает работу после запроса остановки. В более сложных приложениях остановку нужно согласовывать с ожиданием данных, блокирующими операциями и временем жизни ресурсов.

6. std::async и std::future: когда нужен результат вычисления

Не всегда требуется самостоятельно создавать и хранить объекты потоков. Иногда задача звучит проще: выполнить вычисление и получить его результат.

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

#include <future>#include <iostream>int calculate() {    int result = 0;    for (int i = 1; i <= 1'000'000; ++i) {        result += i;    }    return result;}int main() {    auto result = std::async(        std::launch::async,        calculate    );    std::cout << result.get() << '\n';}

std::launch::async просит библиотеку запустить функцию асинхронно. Объект std::future представляет будущий результат вычисления.

Вызов get() получает результат и ждёт его готовности, если необходимо. Если функция завершилась с исключением, оно будет повторно выброшено при получении результата через get().

Если не указывать политику запуска, std::async может выбрать отложенное выполнение. Поэтому в примере политика указана явно.

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

7. std::condition_variable: поток должен не крутиться, а ждать

Представим другую задачу: один поток производит данные, а второй ждёт, пока данные появятся.

Можно постоянно проверять флаг в цикле, но это приводит к активному ожиданию: поток тратит процессорное время, снова и снова проверяя условие.

Для подобных сценариев существует std::condition_variable. Она позволяет потоку ждать уведомления об изменении состояния.

#include <condition_variable>#include <iostream>#include <mutex>#include <thread>int main() {    std::mutex mutex;    std::condition_variable cv;    bool ready = false;    std::thread worker([&]() {        std::unique_lock<std::mutex> lock(mutex);        cv.wait(lock, [&]() {            return ready;        });        std::cout << "Condition satisfied\n";    });    {        std::lock_guard<std::mutex> lock(mutex);        ready = true;    }    cv.notify_one();    worker.join();}

Здесь важны три детали.

Во-первых, условие защищено мьютексом. Поток, который ожидает, проверяет ready под тем же мьютексом, под которым другой поток меняет значение.

Во-вторых, используется версия wait с предикатом. Ожидание может завершиться без ожидаемого уведомления — например, из-за ложного пробуждения. Предикат заставляет повторно проверить условие и продолжать ждать, если оно не выполнено.

В-третьих, уведомление не заменяет само условие. Вызов notify_one() лишь сообщает ожидающему потоку, что стоит снова проверить состояние. Именно значение ready, а не уведомление само по себе, определяет, можно ли продолжать работу.

Такой механизм применяется в очередях задач, пулах потоков и схемах «производитель — потребитель».

8. Как проверить, что код действительно безопасен?

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

Для C++ есть ThreadSanitizer — инструмент обнаружения гонок данных во время выполнения.

Например, в окружении с поддержкой Clang и ThreadSanitizer можно собрать программу так:

clang++ -std=c++20 -O1 -g -fsanitize=thread -pthread main.cpp -o main

Затем запустить её и посмотреть, сообщает ли инструмент о гонках.

Поддержка ThreadSanitizer зависит от компилятора, операционной системы и целевой платформы. Также инструмент не гарантирует обнаружение всех возможных ошибок: он проверяет те пути выполнения, которые реально были выполнены во время запуска.

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

Что в итоге использовать?

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

Мьютекс не делает программу автоматически медленной, атомарная переменная не решает любую проблему синхронизации, а jthread не останавливает бесконечную функцию принудительно. У каждого инструмента своя задача.

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