Язык программирования Rust

Конкурентность с общим состоянием

Передача сообщений — хороший способ организовать конкурентность, но не единственный. Другой способ — дать нескольким потокам доступ к одним и тем же общим данным. Вспомните ещё раз часть лозунга из документации Go: «Не обменивайтесь данными через общую память».

Как выглядел бы обмен данными через общую память? И почему сторонники передачи сообщений предостерегают от общей памяти?

В каком-то смысле каналы в любом языке похожи на единоличное владение: отправив значение по каналу, пользоваться им больше нельзя. А конкурентность с общей памятью похожа на множественное владение: несколько потоков могут одновременно обращаться к одному и тому же участку памяти. Как вы видели в главе 15, где умные указатели сделали множественное владение возможным, оно усложняет дело: разными владельцами нужно управлять. Система типов и правила владения Rust сильно помогают сделать это правильно. Для примера рассмотрим мьютексы — один из самых распространённых примитивов конкурентности для общей памяти.

Управление доступом с помощью мьютексов

Мьютекс (mutex) — сокращение от mutual exclusion, «взаимное исключение»: мьютекс в каждый момент времени даёт доступ к данным только одному потоку. Чтобы обратиться к данным внутри мьютекса, поток должен сначала заявить, что хочет доступа, — попросить захватить блокировку (lock) мьютекса. Блокировка — это структура данных внутри мьютекса, которая отмечает, у кого сейчас исключительный доступ к данным. Поэтому говорят, что мьютекс охраняет свои данные с помощью системы блокировок.

У мьютексов репутация сложного инструмента, потому что нужно помнить два правила:

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

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

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

API Mutex<T>

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

use std::sync::Mutex;

fn main() {
    let m = Mutex::new(5);

    {
        let mut num = m.lock().unwrap();
        *num = 6;
    }

    println!("m = {m:?}");
}

Как и многие типы, Mutex<T> создаётся ассоциированной функцией new. Чтобы добраться до данных внутри мьютекса, захватываем блокировку методом lock. Этот вызов блокирует текущий поток: тот не может ничего делать, пока не придёт наша очередь владеть блокировкой.

Вызов lock завершится неудачей, если другой поток, державший блокировку, запаниковал. Тогда блокировку уже никто никогда не получит, поэтому мы решили вызвать unwrap и в такой ситуации запаниковать и в этом потоке.

Когда блокировка захвачена, с возвращённым значением — здесь оно называется num — можно обращаться как с изменяемой ссылкой на данные внутри. Система типов гарантирует, что мы захватим блокировку, прежде чем пользоваться значением в m. Тип m — Mutex<i32>, а не i32, поэтому, чтобы воспользоваться значением i32, обязательно нужно вызвать lock. Забыть не получится: иначе система типов не даст добраться до внутреннего i32.

Вызов lock возвращает тип MutexGuard, обёрнутый в LockResult, который мы обработали вызовом unwrap. Тип MutexGuard реализует Deref и указывает на внутренние данные; а ещё у него есть реализация Drop, которая автоматически снимает блокировку, когда MutexGuard выходит из области видимости, — в нашем случае в конце внутренней области. Поэтому мы не рискуем забыть снять блокировку и не дать другим потокам пользоваться мьютексом: блокировка снимается сама.

Сняв блокировку, мы выводим значение мьютекса и видим, что внутреннее i32 удалось изменить на 6.

Общий доступ к Mutex<T>

Теперь попробуем с помощью Mutex<T> разделить значение между несколькими потоками. Запустим 10 потоков, и пусть каждый увеличит счётчик на 1, чтобы счётчик дошёл от 0 до 10. Пример из листинга 16-13 не скомпилируется, и по этой ошибке мы узнаем больше о том, как пользоваться Mutex<T> и как Rust помогает делать это правильно.

use std::sync::Mutex;
use std::thread;

fn main() {
    let counter = Mutex::new(0);
    let mut handles = vec![];

    for _ in 0..10 {
        let handle = thread::spawn(move || {
            let mut num = counter.lock().unwrap();

            *num += 1;
        });
        handles.push(handle);
    }

    for handle in handles {
        handle.join().unwrap();
    }

    println!("Result: {}", *counter.lock().unwrap());
}

Мы создаём переменную counter, которая хранит i32 внутри Mutex<T>, как в листинге 16-12. Затем создаём 10 потоков, перебирая диапазон чисел. Через thread::spawn даём всем потокам одно и то же замыкание: оно перемещает счётчик в поток, захватывает блокировку Mutex<T> методом lock и прибавляет 1 к значению в мьютексе. Когда поток заканчивает выполнять замыкание, num выходит из области видимости и снимает блокировку, чтобы её мог захватить другой поток.

В главном потоке мы собираем все дескрипторы потоков. Затем, как в листинге 16-2, вызываем join у каждого, чтобы дождаться завершения всех потоков. После этого главный поток захватывает блокировку и выводит результат программы.

Мы намекали, что этот пример не скомпилируется. Посмотрим почему!

$ cargo run
   Compiling shared-state v0.1.0 (file:///projects/shared-state)
error[E0382]: borrow of moved value: `counter`
  --> src/main.rs:21:29
   |
 5 |     let counter = Mutex::new(0);
   |         ------- move occurs because `counter` has type `std::sync::Mutex<i32>`, which does not implement the `Copy` trait
...
 8 |     for _ in 0..10 {
   |     -------------- inside of this loop
 9 |         let handle = thread::spawn(move || {
   |                                    ------- value moved into closure here, in previous iteration of loop
...
21 |     println!("Result: {}", *counter.lock().unwrap());
   |                             ^^^^^^^ value borrowed here after move

For more information about this error, try `rustc --explain E0382`.
error: could not compile `shared-state` (bin "shared-state") due to 1 previous error

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

Множественное владение в нескольких потоках

В главе 15 мы давали значению нескольких владельцев с помощью умного указателя Rc<T>, создавая значение с подсчётом ссылок. Сделаем то же самое здесь и посмотрим, что получится. В листинге 16-14 мы оборачиваем Mutex<T> в Rc<T> и клонируем Rc<T>, прежде чем передать владение в поток.

use std::rc::Rc;
use std::sync::Mutex;
use std::thread;

fn main() {
    let counter = Rc::new(Mutex::new(0));
    let mut handles = vec![];

    for _ in 0..10 {
        let counter = Rc::clone(&counter);
        let handle = thread::spawn(move || {
            let mut num = counter.lock().unwrap();

            *num += 1;
        });
        handles.push(handle);
    }

    for handle in handles {
        handle.join().unwrap();
    }

    println!("Result: {}", *counter.lock().unwrap());
}

Снова компилируем и получаем… другие ошибки! Компилятор многому нас учит:

$ cargo run
   Compiling shared-state v0.1.0 (file:///projects/shared-state)
error[E0277]: `Rc<std::sync::Mutex<i32>>` cannot be sent between threads safely
  --> src/main.rs:11:36
   |
11 |           let handle = thread::spawn(move || {
   |                        ------------- ^------
   |                        |             |
   |  ______________________|_____________within this `{closure@src/main.rs:11:36: 11:43}`
   | |                      |
   | |                      required by a bound introduced by this call
12 | |             let mut num = counter.lock().unwrap();
13 | |
14 | |             *num += 1;
15 | |         });
   | |_________^ `Rc<std::sync::Mutex<i32>>` cannot be sent between threads safely
   |
   = help: within `{closure@src/main.rs:11:36: 11:43}`, the trait `Send` is not implemented for `Rc<std::sync::Mutex<i32>>`
note: required because it's used within this closure
  --> src/main.rs:11:36
   |
11 |         let handle = thread::spawn(move || {
   |                                    ^^^^^^^
note: required by a bound in `spawn`
  --> /rustc/88d9e12ae178fab0fb5cc050a94da85685d449ea/library/std/src/thread/functions.rs:125:0

For more information about this error, try `rustc --explain E0277`.
error: could not compile `shared-state` (bin "shared-state") due to 1 previous error

Ничего себе многословное сообщение! Вот на чём стоит сосредоточиться: `Rc<Mutex<i32>>` cannot be sent between threads safely — «Rc<Mutex<i32>> нельзя безопасно передавать между потоками». Компилятор называет и причину: the trait `Send` is not implemented for `Rc<Mutex<i32>>` — «для Rc<Mutex<i32>> не реализован трейт Send». О Send мы поговорим в следующем разделе: это один из трейтов, гарантирующих, что типы, которые мы используем с потоками, предназначены для конкурентных ситуаций.

К сожалению, делить Rc<T> между потоками небезопасно. Управляя счётчиком ссылок, Rc<T> увеличивает его при каждом вызове clone и уменьшает при уничтожении каждого клона. Но при этом он не использует никаких примитивов конкурентности, которые гарантировали бы, что изменение счётчика не прервёт другой поток. Это может привести к неверным значениям счётчика — коварным багам, которые, в свою очередь, ведут к утечкам памяти или к тому, что значение уничтожается раньше, чем мы закончили с ним работать. Нам нужен тип, ровно такой же, как Rc<T>, но изменяющий счётчик ссылок потокобезопасно.

Атомарный подсчёт ссылок: Arc<T>

К счастью, Arc<T> — именно такой тип, как Rc<T>, но безопасный в конкурентных ситуациях. Буква a означает atomic (атомарный): это тип с атомарным подсчётом ссылок (atomically reference-counted). Атомарные типы — ещё один вид примитивов конкурентности, подробно разбирать их мы не будем; подробности — в документации стандартной библиотеки к std::sync::atomic. Пока достаточно знать, что атомарные типы работают как примитивные, но их безопасно делить между потоками.

Возможно, вы спросите: почему тогда не все примитивные типы атомарные и почему типы стандартной библиотеки не используют Arc<T> по умолчанию? Дело в том, что за потокобезопасность приходится платить производительностью, а платить стоит только тогда, когда это действительно нужно. Если вы работаете со значениями в одном потоке, код будет быстрее без гарантий, которые дают атомарные типы.

Вернёмся к примеру. У Arc<T> и Rc<T> одинаковый API, так что исправить программу можно, поменяв строку use, вызов new и вызов clone. Код из листинга 16-15 наконец скомпилируется и заработает.

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

fn main() {
    let counter = Arc::new(Mutex::new(0));
    let mut handles = vec![];

    for _ in 0..10 {
        let counter = Arc::clone(&counter);
        let handle = thread::spawn(move || {
            let mut num = counter.lock().unwrap();

            *num += 1;
        });
        handles.push(handle);
    }

    for handle in handles {
        handle.join().unwrap();
    }

    println!("Result: {}", *counter.lock().unwrap());
}

Этот код выведет:

Result: 10

Получилось! Мы посчитали от 0 до 10 — вроде бы не так уж впечатляюще, зато многое узнали о Mutex<T> и потокобезопасности. По такой же схеме можно делать и операции сложнее, чем увеличение счётчика. Например, разбить вычисление на независимые части, раздать их потокам, а затем с помощью Mutex<T> дать каждому потоку дописать свою часть в итоговый результат.

Учтите: для простых числовых операций есть типы попроще Mutex<T> — в модуле std::sync::atomic стандартной библиотеки. Они дают безопасный конкурентный атомарный доступ к примитивным типам. В этом примере мы взяли Mutex<T> с примитивным типом, чтобы сосредоточиться на том, как работает Mutex<T>.

RefCell<T>/Rc<T> и Mutex<T>/Arc<T>

Вы, наверное, заметили, что counter неизменяемый, но мы смогли получить изменяемую ссылку на значение внутри него. Значит, Mutex<T> даёт внутреннюю изменяемость — как семейство типов Cell. Как в главе 15 мы с помощью RefCell<T> меняли содержимое внутри Rc<T>, так и здесь с помощью Mutex<T> меняем содержимое внутри Arc<T>.

Учтите и ещё одно: когда вы используете Mutex<T>, Rust не может защитить вас от всех логических ошибок. Вспомните из главы 15, что с Rc<T> есть риск создать цикл ссылок, когда два значения Rc<T> ссылаются друг на друга, и память утекает. Точно так же с Mutex<T> есть риск создать взаимную блокировку (deadlock). Она возникает, когда операции нужно захватить два ресурса, а два потока захватили по одной блокировке и вечно ждут друг друга. Если вам интересны взаимные блокировки, попробуйте написать программу на Rust со взаимной блокировкой, затем изучите стратегии борьбы с ними для мьютексов в любом языке и попробуйте реализовать их на Rust. Полезные сведения есть в документации стандартной библиотеки к API Mutex<T> и MutexGuard.

Завершим главу разговором о трейтах Send и Sync и о том, как использовать их с собственными типами.