Конкурентность с общим состоянием
Передача сообщений – хороший способ работать с конкурентностью, но не единственный. Другой способ заключается в том, чтобы несколько потоков получали доступ к одним и тем же общим данным. Снова рассмотрим эту часть лозунга из документации языка Go: “Не общайтесь, разделяя память.”
Как выглядело бы общение через разделение памяти? Кроме того, почему сторонники передачи сообщений предостерегают от использования общей памяти?
В некотором смысле каналы в любом языке программирования похожи на единоличное владение, потому что после передачи значения по каналу вы больше не должны использовать это значение. Конкурентность с общей памятью похожа на множественное владение: несколько потоков могут получать доступ к одному и тому же участку памяти одновременно. Как вы видели в главе 15, где умные указатели сделали возможным множественное владение, множественное владение может добавлять сложность, потому что этими разными владельцами нужно управлять. Система типов и правила владения Rust сильно помогают делать это управление правильно. В качестве примера рассмотрим мьютексы, один из наиболее распространенных примитивов конкурентности для общей памяти.
Управление доступом с помощью мьютексов
Мьютекс – это сокращение от mutual exclusion, то есть взаимное исключение: мьютекс позволяет только одному потоку получать доступ к некоторым данным в любой конкретный момент времени. Чтобы получить доступ к данным в мьютексе, поток сначала должен сообщить, что хочет доступ, запросив захват блокировки мьютекса. Блокировка – это структура данных, являющаяся частью мьютекса и отслеживающая, кто в данный момент имеет исключительный доступ к данным. Поэтому говорят, что мьютекс защищает хранимые им данные с помощью системы блокировки.
Мьютексы имеют репутацию сложных в использовании, потому что нужно помнить два правила:
- Перед использованием данных нужно попытаться захватить блокировку.
- Когда вы закончили работу с данными, которые защищает мьютекс, нужно разблокировать данные, чтобы другие потоки могли захватить блокировку.
Для метафоры мьютекса из реального мира представьте панельную дискуссию на конференции, где есть только один микрофон. Прежде чем участник дискуссии сможет говорить, он должен попросить или подать сигнал, что хочет использовать микрофон. Получив микрофон, он может говорить сколько хочет, а затем передать микрофон следующему участнику, который просит слова. Если участник забудет передать микрофон, когда закончит говорить, никто другой не сможет высказаться. Если управление общим микрофоном пойдет неправильно, панельная дискуссия не будет работать как задумано!
Правильно управлять мьютексами может быть невероятно трудно, и именно поэтому так много людей с энтузиазмом относятся к каналам. Однако благодаря системе типов и правилам владения 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> в однопоточном контексте для простотыКак и со многими типами, мы создаем Mutex<T> с помощью связанной функции
new. Чтобы получить доступ к данным внутри мьютекса, мы используем метод
lock для захвата блокировки. Этот вызов заблокирует текущий поток, чтобы он
не мог выполнять никакую работу, пока не наступит наша очередь владеть
блокировкой.
Вызов lock завершился бы ошибкой, если бы другой поток, удерживающий
блокировку, вызвал панику. В таком случае никто уже не смог бы получить
блокировку, поэтому мы выбрали unwrap, чтобы этот поток вызвал панику, если
мы окажемся в такой ситуации.
После захвата блокировки мы можем обращаться с возвращаемым значением, которое
в этом случае называется num, как с изменяемой ссылкой на внутренние данные.
Система типов гарантирует, что мы захватим блокировку перед использованием
значения в m. Тип m – Mutex<i32>, а не i32, поэтому мы должны
вызвать lock, чтобы иметь возможность использовать значение i32. Мы не
можем забыть; иначе система типов не даст нам доступ к внутреннему 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());
}Mutex<T>Мы создаем переменную 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
|
help: consider moving the expression out of the loop so it is only moved once
|
8 ~ let mut value = counter.lock();
9 ~ for _ in 0..10 {
10 | let handle = thread::spawn(move || {
11 ~ let mut num = value.unwrap();
|
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());
}Rc<T>, чтобы позволить нескольким потокам владеть Mutex<T>Снова компилируем и получаем… другие ошибки! Компилятор многому нас учит:
$ 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/1159e78c4747b02ef996e55082b704c09b970588/library/std/src/thread/mod.rs:723:1
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. Компилятор также сообщает нам причину: the trait `Send` is not implemented for `Rc<Mutex<i32>>`. Мы поговорим о Send в следующем
разделе: это один из трейтов, гарантирующих, что типы, которые мы используем с
потоками, предназначены для использования в конкурентных ситуациях.
К сожалению, Rc<T> небезопасно разделять между потоками. Когда Rc<T>
управляет счетчиком ссылок, он добавляет к счетчику при каждом вызове clone
и вычитает из счетчика, когда каждый клон удаляется. Но он не использует
никаких примитивов конкурентности, чтобы гарантировать, что изменения счетчика
не могут быть прерваны другим потоком. Это могло бы привести к неправильным
счетчикам – неочевидным ошибкам, которые, в свою очередь, могли бы привести к
утечкам памяти или удалению значения до того, как мы закончили с ним работать.
Нам нужен тип, который точно такой же, как Rc<T>, но изменяет счетчик ссылок
потокобезопасным способом.
Атомарный подсчет ссылок с Arc<T>
К счастью, Arc<T> является типом вроде Rc<T>, который безопасно
использовать в конкурентных ситуациях. Буква a означает atomic, то есть
это тип с атомарным подсчетом ссылок. Атомарные операции – дополнительный
вид примитивов конкурентности, который мы не будем подробно рассматривать
здесь: подробнее смотрите документацию стандартной библиотеки для
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());
}Arc<T> для оборачивания Mutex<T>, чтобы иметь возможность совместно владеть им из нескольких потоковЭтот код напечатает следующее:
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. Так же как мы
использовали RefCell<T> в главе 15, чтобы позволить изменять содержимое
внутри Rc<T>, мы используем Mutex<T>, чтобы изменять содержимое внутри
Arc<T>.
Еще одна деталь, которую нужно отметить: Rust не может защитить вас от всех
видов логических ошибок при использовании Mutex<T>. Вспомните из главы 15,
что использование Rc<T> сопровождалось риском создания циклов ссылок, когда
два значения Rc<T> ссылаются друг на друга, вызывая утечки памяти. Подобным
образом Mutex<T> сопровождается риском создания взаимных блокировок. Они
возникают, когда операции требуется заблокировать два ресурса, а два потока уже
захватили по одной из блокировок, из-за чего они вечно ждут друг друга. Если
вам интересны взаимные блокировки, попробуйте создать Rust-программу, в
которой есть взаимная блокировка; затем изучите стратегии смягчения взаимных
блокировок для мьютексов в любом языке и попробуйте реализовать их в Rust.
Документация API стандартной библиотеки для Mutex<T> и MutexGuard содержит
полезную информацию.
Мы завершим эту главу разговором о трейтах Send и Sync и о том, как
использовать их с пользовательскими типами.