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

Потоки: одновременное выполнение кода

В большинстве современных операционных систем код запущенной программы выполняется в процессе (process), и операционная система управляет сразу многими процессами. Внутри программы тоже могут быть независимые части, работающие одновременно. То, что выполняет эти независимые части, называется потоками (threads). Например, у веб-сервера может быть несколько потоков, чтобы отвечать сразу на несколько запросов.

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

• состояниям гонки (race conditions), когда потоки обращаются к данным или ресурсам в непредсказуемом порядке;
• взаимным блокировкам (deadlocks), когда два потока ждут друг друга и ни один не может продолжить работу;
• багам, которые проявляются лишь в определённых ситуациях, и их трудно надёжно воспроизвести и исправить.

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

Языки программирования реализуют потоки по-разному, и многие операционные системы предоставляют API, через который язык может создавать новые потоки. Стандартная библиотека Rust использует модель потоков 1:1: на каждый поток языка приходится один поток операционной системы. Есть крейты, реализующие другие модели потоков, с другими компромиссами по сравнению с моделью 1:1. (Асинхронность в Rust, с которой мы познакомимся в следующей главе, — тоже другой подход к конкурентности.)

Создание потока через spawn

Чтобы создать новый поток, вызывают функцию thread::spawn и передают ей замыкание (о замыканиях мы говорили в главе 13) с кодом, который нужно выполнить в новом потоке. В примере из листинга 16-1 главный поток выводит один текст, а новый поток — другой.

use std::thread;
use std::time::Duration;

fn main() {
    thread::spawn(|| {
        for i in 1..10 {
            println!("hi number {i} from the spawned thread!");
            thread::sleep(Duration::from_millis(1));
        }
    });

    for i in 1..5 {
        println!("hi number {i} from the main thread!");
        thread::sleep(Duration::from_millis(1));
    }
}

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

hi number 1 from the main thread!
hi number 1 from the spawned thread!
hi number 2 from the main thread!
hi number 2 from the spawned thread!
hi number 3 from the main thread!
hi number 3 from the spawned thread!
hi number 4 from the main thread!
hi number 4 from the spawned thread!
hi number 5 from the spawned thread!

Вызовы thread::sleep заставляют поток ненадолго приостановиться и дают поработать другому потоку. Скорее всего, потоки будут чередоваться, но это не гарантировано: всё зависит от того, как операционная система планирует потоки. В этом запуске главный поток вывел текст первым, хотя в коде вывод из порождённого потока стоит раньше. И хотя мы велели порождённому потоку выводить, пока i не станет 9, он успел дойти лишь до 5, прежде чем главный поток завершился.

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

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

С кодом из листинга 16-1 две проблемы. Во-первых, порождённый поток обычно останавливается раньше времени — потому что завершается главный. Во-вторых, порядок выполнения потоков не гарантирован, так что нельзя гарантировать даже то, что порождённый поток вообще успеет поработать!

Чтобы это исправить, сохраним в переменную значение, которое возвращает thread::spawn. Его тип — JoinHandle<T>. JoinHandle<T> — значение, которым мы владеем; если вызвать у него метод join, он дождётся завершения своего потока. В листинге 16-2 показано, как использовать JoinHandle<T> потока из листинга 16-1 и вызвать join, чтобы порождённый поток закончился раньше, чем завершится main.

use std::thread;
use std::time::Duration;

fn main() {
    let handle = thread::spawn(|| {
        for i in 1..10 {
            println!("hi number {i} from the spawned thread!");
            thread::sleep(Duration::from_millis(1));
        }
    });

    for i in 1..5 {
        println!("hi number {i} from the main thread!");
        thread::sleep(Duration::from_millis(1));
    }

    handle.join().unwrap();
}

Вызов join у дескриптора блокирует текущий поток, пока не завершится поток, которому принадлежит дескриптор. Блокировка потока означает, что поток не может ни выполнять работу, ни завершиться. Мы поставили вызов join после цикла for главного потока, поэтому листинг 16-2 выведет примерно такое:

hi number 1 from the main thread!
hi number 2 from the main thread!
hi number 1 from the spawned thread!
hi number 3 from the main thread!
hi number 2 from the spawned thread!
hi number 4 from the main thread!
hi number 3 from the spawned thread!
hi number 4 from the spawned thread!
hi number 5 from the spawned thread!
hi number 6 from the spawned thread!
hi number 7 from the spawned thread!
hi number 8 from the spawned thread!
hi number 9 from the spawned thread!

Потоки по-прежнему чередуются, но главный поток из-за вызова handle.join() ждёт и не завершается, пока не закончит порождённый.

А теперь посмотрим, что будет, если перенести handle.join() в main перед циклом for:

use std::thread;
use std::time::Duration;

fn main() {
    let handle = thread::spawn(|| {
        for i in 1..10 {
            println!("hi number {i} from the spawned thread!");
            thread::sleep(Duration::from_millis(1));
        }
    });

    handle.join().unwrap();

    for i in 1..5 {
        println!("hi number {i} from the main thread!");
        thread::sleep(Duration::from_millis(1));
    }
}

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

hi number 1 from the spawned thread!
hi number 2 from the spawned thread!
hi number 3 from the spawned thread!
hi number 4 from the spawned thread!
hi number 5 from the spawned thread!
hi number 6 from the spawned thread!
hi number 7 from the spawned thread!
hi number 8 from the spawned thread!
hi number 9 from the spawned thread!
hi number 1 from the main thread!
hi number 2 from the main thread!
hi number 3 from the main thread!
hi number 4 from the main thread!

Мелкие детали — например, где вызван join, — влияют на то, будут ли ваши потоки работать одновременно.

Замыкания move и потоки

С замыканиями, которые передаются в thread::spawn, часто используют ключевое слово move: тогда замыкание забирает владение значениями, которые использует из окружения, и тем самым владение передаётся из одного потока в другой. Мы обсуждали move применительно к замыканиям в разделе «Захват ссылок или перемещение владения» главы 13. Теперь сосредоточимся на том, как move взаимодействует с thread::spawn.

Обратите внимание: замыкание, которое мы передаём в thread::spawn в листинге 16-1, не принимает аргументов — в коде порождённого потока мы не используем никаких данных из главного. Чтобы использовать в порождённом потоке данные главного, замыкание порождённого потока должно захватить нужные значения. В листинге 16-3 — попытка создать вектор в главном потоке и использовать его в порождённом. Но пока это не сработает, как вы сейчас увидите.

use std::thread;

fn main() {
    let v = vec![1, 2, 3];

    let handle = thread::spawn(|| {
        println!("Here's a vector: {v:?}");
    });

    handle.join().unwrap();
}

Замыкание использует v, поэтому захватит v и сделает его частью своего окружения. thread::spawn выполняет замыкание в новом потоке, так что внутри нового потока к v вроде бы можно обратиться. Но при компиляции получаем ошибку:

$ cargo run
   Compiling threads v0.1.0 (file:///projects/threads)
error[E0373]: closure may outlive the current function, but it borrows `v`, which is owned by the current function
 --> src/main.rs:6:32
  |
6 |     let handle = thread::spawn(|| {
  |                                ^^ may outlive borrowed value `v`
7 |         println!("Here's a vector: {v:?}");
  |                                     - `v` is borrowed here
  |
note: function requires argument type to outlive `'static`
 --> src/main.rs:6:18
  |
6 |       let handle = thread::spawn(|| {
  |  __________________^
7 | |         println!("Here's a vector: {v:?}");
8 | |     });
  | |______^
help: to force the closure to take ownership of `v` (and any other referenced variables), use the `move` keyword
  |
6 |     let handle = thread::spawn(move || {
  |                                ++++

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

Rust сам выводит, как захватить v, и поскольку println! нужна только ссылка на v, замыкание пытается позаимствовать v. Но есть проблема: Rust не может знать, сколько будет работать порождённый поток, а значит, не знает, останется ли ссылка на v действительной.

В листинге 16-4 сценарий, где ссылка на v с ещё большей вероятностью окажется недействительной.

use std::thread;

fn main() {
    let v = vec![1, 2, 3];

    let handle = thread::spawn(|| {
        println!("Here's a vector: {v:?}");
    });

    drop(v); // ой-ой!

    handle.join().unwrap();
}

Если бы Rust разрешил запустить этот код, порождённый поток мог бы сразу уйти в фон, так и не начав работу. Внутри у него ссылка на v, а главный поток тут же уничтожает v функцией drop, о которой мы говорили в главе 15. Тогда, когда порождённый поток начнёт работу, v уже недействителен — а значит, недействительна и ссылка на него. Беда!

Чтобы исправить ошибку компиляции из листинга 16-3, воспользуемся советом из сообщения об ошибке:

help: to force the closure to take ownership of `v` (and any other referenced variables), use the `move` keyword
  |
6 |     let handle = thread::spawn(move || {
  |                                ++++

Добавив ключевое слово move перед замыканием, мы заставляем его забрать владение используемыми значениями, а не позволяем Rust решить, что их нужно позаимствовать. Изменённый листинг 16-3, показанный в листинге 16-5, скомпилируется и будет работать как задумано.

use std::thread;

fn main() {
    let v = vec![1, 2, 3];

    let handle = thread::spawn(move || {
        println!("Here's a vector: {v:?}");
    });

    handle.join().unwrap();
}

Может возникнуть соблазн так же исправить и код из листинга 16-4, где главный поток вызывал drop, — замыканием move. Но это не сработает: то, что пытается сделать листинг 16-4, запрещено по другой причине. Если добавить к замыканию move, v переместится в окружение замыкания, и вызвать для него drop в главном потоке уже не получится. Вместо этого будет такая ошибка компиляции:

$ cargo run
   Compiling threads v0.1.0 (file:///projects/threads)
error[E0382]: use of moved value: `v`
  --> src/main.rs:10:10
   |
 4 |     let v = vec![1, 2, 3];
   |         - move occurs because `v` has type `Vec<i32>`, which does not implement the `Copy` trait
 5 |
 6 |     let handle = thread::spawn(move || {
   |                                ------- value moved into closure here
 7 |         println!("Here's a vector: {v:?}");
   |                                     - variable moved due to use in closure
...
10 |     drop(v); // ой-ой!
   |          ^ value used here after move
   |
help: consider cloning the value before moving it into the closure
   |
 6 ~     let value = v.clone();
 7 ~     let handle = thread::spawn(move || {
 8 ~         println!("Here's a vector: {value:?}");
   |

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

Правила владения Rust снова нас спасли! Код из листинга 16-3 выдавал ошибку, потому что Rust действовал осторожно и лишь заимствовал v для потока, а значит, главный поток теоретически мог сделать ссылку порождённого потока недействительной. Когда мы говорим Rust передать владение v порождённому потоку, мы тем самым гарантируем Rust, что главный поток больше не будет пользоваться v. А если так же изменить листинг 16-4, то, попытавшись использовать v в главном потоке, мы нарушим правила владения. Ключевое слово move отменяет осторожное поведение по умолчанию — заимствование, — но не позволяет нарушать правила владения.

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