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

Использование потоков для одновременного выполнения кода

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

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

• Состояния гонки, при которых потоки обращаются к данным или ресурсам в несогласованном порядке
• Взаимные блокировки, при которых два потока ждут друг друга, не давая обоим потокам продолжить выполнение
• Ошибки, которые возникают только в определенных ситуациях и которые трудно надежно воспроизвести и исправить

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 в переменной. Возвращаемый тип 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() перед циклом for в 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));
        }
    });

    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-замыканий с потоками

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

Обратите внимание, что в листинге 16-1 замыкание, которое мы передаем в thread::spawn, не принимает аргументов: мы не используем никаких данных из основного потока в коде порожденного потока. Чтобы использовать данные из основного потока в порожденном потоке, замыкание порожденного потока должно захватить нужные ему значения. Листинг 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); // oh no!

    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); // oh no!
   |          ^ 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 переопределяет консервативное поведение Rust по умолчанию, при котором используется заимствование; оно не позволяет нарушать правила владения.

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