Многопоточность в Rust: как Ownership, Send и Sync помогают бороться с гонками данных

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

В Rust мы с вами попробуем решить ту же задачу.

Главное отличие заключается не в том, что Rust умеет создавать потоки каким-то особенным способом, а в модели владения и системе типов: компилятор не позволяет безопасному коду произвольно разделять изменяемые данные между потоками.

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

Разберёмся на практических примерах.

Все примеры рассчитаны на стабильный Rust и edition 2021.

1. Создаём несколько потоков

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

use std::thread;

fn main() {
    const THREAD_COUNT: usize = 10;
    const INCREMENTS: usize = 1_000_000;

    let mut handles = Vec::new();

    for _ in 0..THREAD_COUNT {
        let handle = thread::spawn(move || {
            let mut counter = 0;

            for _ in 0..INCREMENTS {
                counter += 1;
            }
            counter
        });
        handles.push(handle);
    }

    let total: usize = handles
        .into_iter()
        .map(|handle| handle.join().expect("Поток завершился с паникой"))
        .sum();

    println!("Counter: {total}");
}

Результат:

Counter: 10000000

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

Посмотрите на move: Здесь move позволяет замыканию владеть захваченными значениями, если такие значения есть. Сам counter при этом вообще не передаётся из главного потока — он создаётся отдельно внутри каждого потока.

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

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

2. Почему нельзя просто разделить изменяемый счётчик?

Попробуем сделать задачу иначе:

use std::thread;

fn main() {
    let mut counter = 0;
    let mut handles = Vec::new();

    for _ in 0..10 {
        handles.push(thread::spawn(|| {
            counter += 1;
        }));
    }
    for handle in handles {
        handle.join().unwrap();
    }
}

Такой код не будет компилироваться.

Проблема в том, что thread::spawn требует, чтобы замыкание могло потенциально жить дольше текущей функции. А здесь оно пытается заимствовать counter из main(). В частности, захваченные данные должны удовлетворять необходимым ограничениям времени жизни и потокобезопасности.

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

Тут замыкание пытается использовать изменяемую переменную из внешней области, а Rust не разрешает такое разделение данных между независимыми потоками.

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

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

3. Arc и Mutex: безопасно разделяем общий ресурс

А что делать, если общий счётчик действительно необходим?

В Rust для этого можно использовать два инструмента безопасной работы:

• Arc — совместное владение данными с подсчётом ссылок, безопасным для нескольких потоков.

• Mutex — взаимное исключение, которое позволяет защищать изменяемое значение.

Пример:

use std::sync::{Arc, Mutex};
use std::thread;

fn main() {
    const THREAD_COUNT: usize = 10;
    const INCREMENTS: usize = 1_000_000;

    let counter = Arc::new(Mutex::new(0usize));
    let mut handles = Vec::new();

    for _ in 0..THREAD_COUNT {
        let counter = Arc::clone(&counter);

        handles.push(thread::spawn(move || {
            for _ in 0..INCREMENTS {
                let mut value = counter.lock().expect("Mutex poisoned");
                *value += 1;
            }
        }));
    }

    for handle in handles {
        handle.join().expect("Поток завершился с паникой");
    }

    println!("Counter: {}", *counter.lock().expect("Mutex poisoned"));
}

Здесь каждый поток получает собственный клон Arc, который и указывает на один и тот же объект с мьютексом.

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

Почему используется Arc, а не обычный Rc? Потому что Arc предназначен для совместного владения между потоками. Обычный Rc не обеспечивает потоко-безопасного подсчёта ссылок и не подходит для этого.

Есть и нюанс: Mutex в стандартной библиотеке Rust может оказаться в состоянии poisoning, если поток запаниковал, удерживая блокировку. В примере мы используем expect, поэтому при попытке получить poisoned Mutex программа завершится с panic. В реальном приложении обработку этого состояния нужно продумать отдельно.

4. AtomicUsize: счётчик без мьютекса

Для простого счётчика мьютекс может оказаться избыточным. В Rust есть атомарные типы, например AtomicUsize.

use std::sync::{
    Arc,
    atomic::{AtomicUsize, Ordering},
};
use std::thread;

fn main() {
    const THREAD_COUNT: usize = 10;
    const INCREMENTS: usize = 1_000_000;

    let counter = Arc::new(AtomicUsize::new(0));
    let mut handles = Vec::new();

    for _ in 0..THREAD_COUNT {
        let counter = Arc::clone(&counter);

        handles.push(thread::spawn(move || {
            for _ in 0..INCREMENTS {
                counter.fetch_add(1, Ordering::Relaxed);
            }
        }));
    }

    for handle in handles {
        handle.join().expect("Поток завершился с паникой");
    }

    println!("Counter: {}", counter.load(Ordering::Relaxed));
}

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

Что означает Ordering::Relaxed?

Он гарантирует атомарность операции, но не устанавливает дополнительный порядок доступа к другим данным. В этом примере счётчик используется только для подсчёта, а результаты читаются после присоединения всех потоков через join().

Для такой задачи этого достаточно.

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

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

5. Что такое Send и Sync?

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

Send означает, что значение соответствующего типа можно безопасно передать в другой поток.

Sync означает, что ссылки &T на значение этого типа можно безопасно использовать из нескольких потоков. Формально тип T является Sync, если &T реализует Send.

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

Например Arc<T> позволяет совместно владеть значением, но сам по себе не делает произвольный T потокобезопасным для изменения. Чтобы разделять Arc<T> между потоками, тип T должен удовлетворять необходимым ограничениям, в частности Send и Sync для стандартного сценария передачи Arc<T> между потоками.

Именно поэтому Arc<Mutex<T>> — распространённый шаблон: Arc организует совместное владение, а Mutex предоставляет контролируемый доступ к изменяемому значению.

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

6. А как ждать появления данных?

В C++ для подобных задач существует std::condition_variable. В Rust похожий механизм предоставляет std::sync::Condvar.

Представим, что поток должен дождаться, пока другой поток установит флаг готовности.

use std::sync::{Arc, Condvar, Mutex};
use std::thread;

fn main() {
    let state = Arc::new((Mutex::new(false), Condvar::new()));

    let worker_state = Arc::clone(&state);

    let worker = thread::spawn(move || {
        let (lock, cv) = &*worker_state;
        let mut ready = lock.lock().unwrap();

        while !*ready {
            ready = cv.wait(ready).unwrap();
        }

        println!("Condition satisfied");
    });

    let (lock, cv) = &*state;

    {
        let mut ready = lock.lock().unwrap();
        *ready = true;
    }

    cv.notify_one();

    worker.join().unwrap();
}

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

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

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

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

7. Есть ли в Rust аналог std::jthread?

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

В стандартном Rust нет полностью похожего аналога std::jthread.

std::thread::JoinHandle позволяет дождаться завершения потока через join(), но простое уничтожение JoinHandle отсоединяет поток, а не ждёт его завершения.

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

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

8. Какой подход выбрать?

Что Rust гарантирует, а что нет?

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

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

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

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

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