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

Передача данных между потоками через сообщения

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

Для конкурентности на основе передачи сообщений в стандартной библиотеке Rust есть реализация каналов. Канал (channel) — общее понятие в программировании: механизм, по которому данные пересылаются из одного потока в другой.

Канал в программировании можно представить как русло, по которому течёт вода, — ручей или реку. Если пустить в реку что-нибудь вроде резиновой уточки, её унесёт вниз по течению до самого конца русла.

У канала две половины: передатчик (transmitter) и приёмник (receiver). Передатчик — это место выше по течению, где вы пускаете уточку в реку, а приёмник — место ниже по течению, куда уточка в итоге приплывает. Одна часть вашего кода вызывает у передатчика методы с данными, которые нужно отправить, а другая проверяет, не пришли ли сообщения на приёмную сторону. Канал считается закрытым, если уничтожена хотя бы одна из его половин — передатчик или приёмник.

Мы постепенно напишем программу, в которой один поток генерирует значения и отправляет их по каналу, а другой получает их и выводит. Чтобы показать, как это работает, будем пересылать между потоками простые значения. Освоив этот приём, вы сможете использовать каналы для любых потоков, которым нужно общаться: например, для чата или для системы, где много потоков выполняют части вычисления и отправляют результаты одному потоку, который их собирает.

Сначала, в листинге 16-6, создадим канал, но ничего с ним делать не будем. Учтите, что код пока не скомпилируется: Rust не может понять, значения какого типа мы хотим отправлять по каналу.

use std::sync::mpsc;

fn main() {
    let (tx, rx) = mpsc::channel();
}

Новый канал создаётся функцией mpsc::channel; mpsc расшифровывается как multiple producer, single consumer — «много производителей, один потребитель». Коротко говоря, в стандартной библиотеке Rust каналы устроены так, что у канала может быть несколько отправляющих концов, производящих значения, но лишь один принимающий, который их потребляет. Представьте несколько ручьёв, впадающих в одну большую реку: всё, что пустили по любому из ручьёв, в итоге окажется в одной реке. Пока начнём с одного производителя, а когда пример заработает, добавим ещё.

Функция mpsc::channel возвращает кортеж: первый элемент — отправляющий конец, передатчик, а второй — принимающий конец, приёмник. Сокращения tx и rx во многих областях традиционно обозначают transmitter (передатчик) и receiver (приёмник), поэтому мы так и называем переменные. Мы используем инструкцию let с образцом, который разбирает кортеж; об образцах в let и деструктуризации мы поговорим в главе 19. Пока достаточно знать, что такая инструкция let — удобный способ достать части кортежа, который возвращает mpsc::channel.

Перенесём передающий конец в порождённый поток и отправим одну строку, чтобы порождённый поток пообщался с главным, как в листинге 16-7. Это как пустить уточку в реку выше по течению или отправить сообщение в чате из одного потока в другой.

use std::sync::mpsc;
use std::thread;

fn main() {
    let (tx, rx) = mpsc::channel();

    thread::spawn(move || {
        let val = String::from("hi");
        tx.send(val).unwrap();
    });
}

Снова создаём новый поток через thread::spawn и с помощью move перемещаем tx в замыкание, чтобы им владел порождённый поток. Чтобы отправлять сообщения по каналу, порождённый поток должен владеть передатчиком.

У передатчика есть метод send, который принимает отправляемое значение. Метод send возвращает Result<T, E>: если приёмник уже уничтожен и отправлять значение некуда, операция отправки вернёт ошибку. В этом примере мы вызываем unwrap, чтобы при ошибке запаниковать. Но в настоящем приложении её нужно было бы обработать как следует — стратегии правильной обработки ошибок описаны в главе 9.

В листинге 16-8 мы получаем значение из приёмника в главном потоке. Это как выловить уточку из воды в конце реки или получить сообщение в чате.

use std::sync::mpsc;
use std::thread;

fn main() {
    let (tx, rx) = mpsc::channel();

    thread::spawn(move || {
        let val = String::from("hi");
        tx.send(val).unwrap();
    });

    let received = rx.recv().unwrap();
    println!("Got: {received}");
}

У приёмника два полезных метода: recv и try_recv. Мы используем recv (от receive — получить). Он блокирует выполнение главного потока и ждёт, пока по каналу не придёт значение. Когда значение отправлено, recv вернёт его в Result<T, E>. А когда передатчик закроется, recv вернёт ошибку — знак того, что значений больше не будет.

Метод try_recv не блокирует поток, а сразу возвращает Result<T, E>: Ok с сообщением, если оно есть, и Err, если сообщений пока нет. try_recv полезен, если у потока есть и другая работа, пока он ждёт сообщений: можно написать цикл, который время от времени вызывает try_recv, обрабатывает сообщение, если оно пришло, а иначе немного занимается другой работой до следующей проверки.

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

Запустив код из листинга 16-8, увидим значение, выведенное главным потоком:

Got: hi

Отлично!

Передача владения через каналы

Правила владения играют в передаче сообщений ключевую роль: они помогают писать безопасный конкурентный код. Предотвращение ошибок в конкурентном программировании — та выгода, которую даёт привычка думать о владении во всех программах на Rust. Проведём эксперимент и покажем, как каналы и владение вместе предотвращают проблемы: попробуем использовать значение val в порождённом потоке после того, как отправили его по каналу. Скомпилируйте код из листинга 16-9 и посмотрите, почему так нельзя.

use std::sync::mpsc;
use std::thread;

fn main() {
    let (tx, rx) = mpsc::channel();

    thread::spawn(move || {
        let val = String::from("hi");
        tx.send(val).unwrap();
        println!("val is {val}");
    });

    let received = rx.recv().unwrap();
    println!("Got: {received}");
}

Здесь мы пытаемся вывести val после того, как отправили его по каналу через tx.send. Разрешать такое было бы плохой идеей: как только значение отправлено в другой поток, тот может изменить или уничтожить его раньше, чем мы снова им воспользуемся. Изменения в другом потоке могли бы привести к ошибкам или неожиданным результатам из-за несогласованных или уже несуществующих данных. Но при попытке скомпилировать код из листинга 16-9 Rust выдаёт ошибку:

$ cargo run
   Compiling message-passing v0.1.0 (file:///projects/message-passing)
error[E0382]: borrow of moved value: `val`
  --> src/main.rs:10:27
   |
 8 |         let val = String::from("hi");
   |             --- move occurs because `val` has type `String`, which does not implement the `Copy` trait
 9 |         tx.send(val).unwrap();
   |                 --- value moved here
10 |         println!("val is {val}");
   |                           ^^^ value borrowed here after move

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

Наша ошибка конкурентности стала ошибкой компиляции. Функция send забирает владение своим параметром, а когда значение перемещается, владельцем становится приёмник. Это не даёт случайно снова воспользоваться значением после отправки: система владения проверяет, что всё в порядке.

Отправка нескольких значений

Код из листинга 16-8 скомпилировался и заработал, но не показал наглядно, что два отдельных потока общаются по каналу.

В листинге 16-10 мы внесли изменения, которые докажут, что код из листинга 16-8 работает конкурентно: теперь порождённый поток отправляет несколько сообщений и делает паузу в секунду между ними.

use std::sync::mpsc;
use std::thread;
use std::time::Duration;

fn main() {
    let (tx, rx) = mpsc::channel();

    thread::spawn(move || {
        let vals = vec![
            String::from("hi"),
            String::from("from"),
            String::from("the"),
            String::from("thread"),
        ];

        for val in vals {
            tx.send(val).unwrap();
            thread::sleep(Duration::from_secs(1));
        }
    });

    for received in rx {
        println!("Got: {received}");
    }
}

На этот раз у порождённого потока есть вектор строк, которые нужно отправить главному потоку. Мы перебираем их, отправляем по одной и между отправками делаем паузу, вызывая thread::sleep со значением Duration в одну секунду.

В главном потоке мы больше не вызываем recv явно, а обращаемся с rx как с итератором: каждое полученное значение выводим. Когда канал закроется, перебор закончится.

Запустив код из листинга 16-10, вы увидите следующий вывод — строки будут появляться с паузой в секунду:

Got: hi
Got: from
Got: the
Got: thread

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

Несколько производителей

Выше мы говорили, что mpsc — сокращение от multiple producer, single consumer. Воспользуемся этим и расширим код из листинга 16-10: создадим несколько потоков, которые отправляют значения одному и тому же приёмнику. Для этого передатчик можно клонировать, как в листинге 16-11.

use std::sync::mpsc;
use std::thread;
use std::time::Duration;

fn main() {
    // --пропуск--

    let (tx, rx) = mpsc::channel();

    let tx1 = tx.clone();
    thread::spawn(move || {
        let vals = vec![
            String::from("hi"),
            String::from("from"),
            String::from("the"),
            String::from("thread"),
        ];

        for val in vals {
            tx1.send(val).unwrap();
            thread::sleep(Duration::from_secs(1));
        }
    });

    thread::spawn(move || {
        let vals = vec![
            String::from("more"),
            String::from("messages"),
            String::from("for"),
            String::from("you"),
        ];

        for val in vals {
            tx.send(val).unwrap();
            thread::sleep(Duration::from_secs(1));
        }
    });

    for received in rx {
        println!("Got: {received}");
    }

    // --пропуск--
}

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

Запустив код, вы увидите примерно такой вывод:

Got: hi
Got: more
Got: from
Got: messages
Got: for
Got: the
Got: thread
Got: you

В зависимости от системы значения могут прийти в другом порядке. Именно это делает конкурентность одновременно интересной и сложной. Если поэкспериментировать с thread::sleep, задавая разные значения в разных потоках, каждый запуск будет ещё менее предсказуемым и давать разный вывод.

Теперь, когда мы посмотрели, как работают каналы, рассмотрим другой подход к конкурентности.