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

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

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

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

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

Будем разбираться с вами на практических примерах.

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.