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

Конкурентность с помощью async

В этом разделе мы применим async к некоторым из тех же задач конкурентности, которые решали с помощью потоков в главе 16. Многие ключевые идеи мы уже обсудили там, поэтому здесь сосредоточимся на том, чем потоки отличаются от футур.

Во многих случаях API для конкурентности на основе async очень похожи на API для потоков. В других — заметно отличаются. И даже когда API потоков и async выглядят похоже, ведут они себя часто по-разному — и почти всегда у них разные характеристики производительности.

Создание задачи с помощью spawn_task

Первое, что мы делали в разделе «Создание потока через spawn» главы 16, — считали в двух отдельных потоках. Сделаем то же самое с async. Крейт trpl даёт функцию spawn_task, очень похожую на API thread::spawn, и функцию sleep — асинхронную версию thread::sleep. С их помощью можно реализовать пример со счётом, как в листинге 17-6.

extern crate trpl; // нужно для mdbook test

use std::time::Duration;

fn main() {
    trpl::block_on(async {
        trpl::spawn_task(async {
            for i in 1..10 {
                println!("hi number {i} from the first task!");
                trpl::sleep(Duration::from_millis(500)).await;
            }
        });

        for i in 1..5 {
            println!("hi number {i} from the second task!");
            trpl::sleep(Duration::from_millis(500)).await;
        }
    });
}

Для начала оборачиваем код функции main в trpl::block_on, чтобы функция верхнего уровня могла быть асинхронной.

Затем пишем внутри блока два цикла, в каждом из которых вызов trpl::sleep ждёт полсекунды (500 миллисекунд) перед отправкой следующего сообщения. Один цикл помещаем в тело trpl::spawn_task, другой — в цикл for на верхнем уровне. И добавляем await после вызовов sleep.

Этот код ведёт себя похоже на реализацию с потоками — в том числе сообщения у вас в терминале могут появляться в другом порядке:

hi number 1 from the second task!
hi number 1 from the first task!
hi number 2 from the first task!
hi number 2 from the second task!
hi number 3 from the first task!
hi number 3 from the second task!
hi number 4 from the first task!
hi number 4 from the second task!
hi number 5 from the first task!

Эта версия останавливается, как только заканчивается цикл for в теле главного асинхронного блока: задача, порождённая spawn_task, завершается вместе с функцией main. Чтобы задача отработала до конца, нужно дождаться её через дескриптор. С потоками мы «блокировались» методом join, пока поток не закончит работу. В листинге 17-7 то же самое можно сделать через await: сам дескриптор задачи — это футура. Её тип Output — Result, так что после ожидания мы ещё и разворачиваем его через unwrap.

extern crate trpl; // нужно для mdbook test

use std::time::Duration;

fn main() {
    trpl::block_on(async {
        let handle = trpl::spawn_task(async {
            for i in 1..10 {
                println!("hi number {i} from the first task!");
                trpl::sleep(Duration::from_millis(500)).await;
            }
        });

        for i in 1..5 {
            println!("hi number {i} from the second task!");
            trpl::sleep(Duration::from_millis(500)).await;
        }

        handle.await.unwrap();
    });
}

Обновлённая версия работает, пока не закончатся оба цикла:

hi number 1 from the second task!
hi number 1 from the first task!
hi number 2 from the first task!
hi number 2 from the second task!
hi number 3 from the first task!
hi number 3 from the second task!
hi number 4 from the first task!
hi number 4 from the second task!
hi number 5 from the first task!
hi number 6 from the first task!
hi number 7 from the first task!
hi number 8 from the first task!
hi number 9 from the first task!

Пока похоже, что async и потоки дают одинаковый результат, только синтаксис другой: вместо вызова join у дескриптора — await, и вызовы sleep тоже ждём через await.

Главное же отличие в том, что для этого не понадобилось создавать ещё один поток операционной системы. Более того, здесь не нужно создавать даже задачу. Асинхронные блоки компилируются в анонимные футуры, поэтому можно поместить каждый цикл в асинхронный блок и попросить среду выполнения выполнить оба до конца с помощью функции trpl::join.

В разделе «Ждём завершения всех потоков» главы 16 мы показали, как вызывать метод join у типа JoinHandle, который возвращает std::thread::spawn. Функция trpl::join похожа на него, только для футур. Если передать ей две футуры, она создаст одну новую. Эта новая футура завершается, когда завершатся обе переданные, и выдаёт кортеж с их результатами. Поэтому в листинге 17-8 мы с помощью trpl::join ждём завершения и fut1, и fut2. Ждём мы при этом не сами fut1 и fut2, а новую футуру, которую создала trpl::join. Результат мы игнорируем: это просто кортеж из двух единичных значений.

extern crate trpl; // нужно для mdbook test

use std::time::Duration;

fn main() {
    trpl::block_on(async {
        let fut1 = async {
            for i in 1..10 {
                println!("hi number {i} from the first task!");
                trpl::sleep(Duration::from_millis(500)).await;
            }
        };

        let fut2 = async {
            for i in 1..5 {
                println!("hi number {i} from the second task!");
                trpl::sleep(Duration::from_millis(500)).await;
            }
        };

        trpl::join(fut1, fut2).await;
    });
}

Запустив код, увидим, что обе футуры выполняются до конца:

hi number 1 from the first task!
hi number 1 from the second task!
hi number 2 from the first task!
hi number 2 from the second task!
hi number 3 from the first task!
hi number 3 from the second task!
hi number 4 from the first task!
hi number 4 from the second task!
hi number 5 from the first task!
hi number 6 from the first task!
hi number 7 from the first task!
hi number 8 from the first task!
hi number 9 from the first task!

Теперь порядок каждый раз будет одинаковым — совсем не то, что мы видели с потоками и с trpl::spawn_task в листинге 17-7. Дело в том, что функция trpl::join справедливая (fair): она проверяет каждую футуру одинаково часто, чередуя их, и никогда не даёт одной вырваться вперёд, если другая готова. С потоками решает операционная система — какой поток проверить и сколько дать ему поработать. В асинхронном Rust решает среда выполнения — какую задачу проверить. (На практике всё сложнее: асинхронная среда выполнения может под капотом использовать потоки операционной системы для управления конкурентностью, так что гарантировать справедливость ей бывает труднее — но всё же возможно!) Среды выполнения не обязаны гарантировать справедливость для каждой операции и часто дают разные API, позволяя выбрать, нужна ли вам справедливость.

Попробуйте разные варианты ожидания футур и посмотрите, что получится:

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

А в качестве дополнительного упражнения попробуйте угадать вывод в каждом случае до запуска кода!

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

Обмен данными между футурами тоже будет знакомым: снова воспользуемся передачей сообщений, но на этот раз с асинхронными версиями типов и функций. Мы пойдём немного другим путём, чем в разделе «Передача данных между потоками через сообщения» главы 16, чтобы показать ключевые различия между конкурентностью на потоках и на футурах. В листинге 17-9 начнём с одного асинхронного блока — не порождая отдельную задачу, как порождали отдельный поток.

extern crate trpl; // нужно для mdbook test

fn main() {
    trpl::block_on(async {
        let (tx, mut rx) = trpl::channel();

        let val = String::from("hi");
        tx.send(val).unwrap();

        let received = rx.recv().await.unwrap();
        println!("received '{received}'");
    });
}

Здесь используется trpl::channel — асинхронная версия API канала «много производителей, один потребитель», которым мы пользовались с потоками в главе 16. Асинхронная версия API лишь немного отличается от версии для потоков: приёмник rx здесь изменяемый, а метод recv возвращает футуру, которую нужно дождаться, а не само значение. Теперь можно отправлять сообщения от отправителя к приёмнику. Заметьте: ни отдельного потока, ни даже задачи порождать не нужно — достаточно дождаться вызова rx.recv.

Синхронный метод Receiver::recv из std::mpsc::channel блокирует поток, пока не получит сообщение. А trpl::Receiver::recv — нет, потому что он асинхронный. Вместо блокировки он возвращает управление среде выполнения, пока не придёт сообщение или пока не закроется отправляющая сторона канала. Вызов send, напротив, мы не ждём, потому что он не блокирует. Ему это и не нужно: канал, в который мы отправляем, неограниченный (unbounded).

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

Займёмся первым: отправим серию сообщений с паузами между ними, как в листинге 17-10.

extern crate trpl; // нужно для mdbook test

use std::time::Duration;

fn main() {
    trpl::block_on(async {
        let (tx, mut rx) = trpl::channel();

        let vals = vec![
            String::from("hi"),
            String::from("from"),
            String::from("the"),
            String::from("future"),
        ];

        for val in vals {
            tx.send(val).unwrap();
            trpl::sleep(Duration::from_millis(500)).await;
        }

        while let Some(value) = rx.recv().await {
            println!("received '{value}'");
        }
    });
}

Сообщения нужно не только отправить, но и получить. Здесь мы знаем, сколько придёт сообщений, так что можно было бы вручную четыре раза вызвать rx.recv().await. Но в реальности обычно ждёшь заранее неизвестное число сообщений, и ждать нужно, пока не станет ясно, что сообщений больше не будет.

В листинге 16-10 мы обрабатывали все элементы, полученные из синхронного канала, циклом for. Но для последовательности элементов, которые появляются асинхронно, цикла for в Rust пока нет, поэтому понадобится цикл, которого мы ещё не видели, — условный цикл while let. Это циклическая версия конструкции if let из раздела «Краткое управление потоком: if let и let...else» главы 6. Цикл выполняется, пока указанный в нём образец продолжает совпадать со значением.

Вызов rx.recv возвращает футуру, которую мы ждём. Среда выполнения приостанавливает футуру, пока та не будет готова. Каждый раз, когда приходит сообщение, футура разрешается в Some(message). Когда канал закрывается — неважно, пришло ли хоть одно сообщение, — футура разрешается в None: это значит, что значений больше нет и опрашивать (то есть ждать) больше не нужно.

Цикл while let сводит всё это вместе. Если результат rx.recv().await — Some(message), мы получаем сообщение и можем использовать его в теле цикла, как с if let. Если результат — None, цикл заканчивается. На каждом новом витке цикл снова доходит до точки ожидания, и среда выполнения снова приостанавливает его до прихода следующего сообщения.

Теперь код успешно отправляет и получает все сообщения. К сожалению, осталась пара проблем. Во-первых, сообщения приходят не с интервалом в полсекунды, а все сразу — через 2 секунды (2000 миллисекунд) после запуска программы. Во-вторых, программа никогда не завершается! Она вечно ждёт новых сообщений. Остановить её придётся сочетанием ctrl-C.

Код внутри одного асинхронного блока выполняется последовательно

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

В листинге 17-10 всего один асинхронный блок, поэтому всё в нём выполняется последовательно. Конкурентности по-прежнему нет. Сначала выполняются все вызовы tx.send вперемешку со всеми вызовами trpl::sleep и их точками ожидания. И только потом цикл while let добирается до точек ожидания на вызовах recv.

Чтобы получить нужное поведение — паузу между сообщениями, — операции с tx и rx нужно разнести по отдельным асинхронным блокам, как в листинге 17-11. Тогда среда выполнения сможет выполнять их по отдельности с помощью trpl::join, как в листинге 17-8. И снова ждём мы результат вызова trpl::join, а не отдельные футуры. Если бы мы ждали отдельные футуры по очереди, то просто вернулись бы к последовательному выполнению — ровно к тому, чего пытаемся избежать.

extern crate trpl; // нужно для mdbook test

use std::time::Duration;

fn main() {
    trpl::block_on(async {
        let (tx, mut rx) = trpl::channel();

        let tx_fut = async {
            let vals = vec![
                String::from("hi"),
                String::from("from"),
                String::from("the"),
                String::from("future"),
            ];

            for val in vals {
                tx.send(val).unwrap();
                trpl::sleep(Duration::from_millis(500)).await;
            }
        };

        let rx_fut = async {
            while let Some(value) = rx.recv().await {
                println!("received '{value}'");
            }
        };

        trpl::join(tx_fut, rx_fut).await;
    });
}

С обновлённым кодом из листинга 17-11 сообщения выводятся с интервалом в 500 миллисекунд, а не все разом через 2 секунды.

Перемещение владения в асинхронный блок

Но программа по-прежнему не завершается — из-за того, как цикл while let взаимодействует с trpl::join:

• Футура, которую возвращает trpl::join, завершается только тогда, когда завершились обе переданные ей футуры.
• Футура tx_fut завершается, когда заканчивается пауза после отправки последнего сообщения из vals.
• Футура rx_fut не завершится, пока не закончится цикл while let.
• Цикл while let не закончится, пока ожидание rx.recv не вернёт None.
• Ожидание rx.recv вернёт None, только когда другой конец канала закроется.
• Канал закроется, только если вызвать rx.close или когда будет уничтожена отправляющая сторона, tx.
• rx.close мы нигде не вызываем, а tx не будет уничтожен, пока не закончится самый внешний асинхронный блок, переданный в trpl::block_on.
• А этот блок не может закончиться, потому что заблокирован в ожидании trpl::join — и мы возвращаемся к началу списка.

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

В листинге 17-12 мы меняем блок, отправляющий сообщения, с async на async move.

extern crate trpl; // нужно для mdbook test

use std::time::Duration;

fn main() {
    trpl::block_on(async {
        let (tx, mut rx) = trpl::channel();

        let tx_fut = async move {
            // --пропуск--
            let vals = vec![
                String::from("hi"),
                String::from("from"),
                String::from("the"),
                String::from("future"),
            ];

            for val in vals {
                tx.send(val).unwrap();
                trpl::sleep(Duration::from_millis(500)).await;
            }
        };

        let rx_fut = async {
            while let Some(value) = rx.recv().await {
                println!("received '{value}'");
            }
        };

        trpl::join(tx_fut, rx_fut).await;
    });
}

Эта версия кода аккуратно завершается после того, как последнее сообщение отправлено и получено. Теперь посмотрим, что нужно изменить, чтобы данные отправляли несколько футур.

Объединение нескольких футур макросом join!

Этот асинхронный канал тоже поддерживает нескольких производителей, так что, чтобы отправлять сообщения из нескольких футур, можно вызвать clone у tx, как в листинге 17-13.

extern crate trpl; // нужно для mdbook test

use std::time::Duration;

fn main() {
    trpl::block_on(async {
        let (tx, mut rx) = trpl::channel();

        let tx1 = tx.clone();
        let tx1_fut = async move {
            let vals = vec![
                String::from("hi"),
                String::from("from"),
                String::from("the"),
                String::from("future"),
            ];

            for val in vals {
                tx1.send(val).unwrap();
                trpl::sleep(Duration::from_millis(500)).await;
            }
        };

        let rx_fut = async {
            while let Some(value) = rx.recv().await {
                println!("received '{value}'");
            }
        };

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

            for val in vals {
                tx.send(val).unwrap();
                trpl::sleep(Duration::from_millis(1500)).await;
            }
        };

        trpl::join!(tx1_fut, tx_fut, rx_fut);
    });
}

Сначала клонируем tx, создавая tx1 за пределами первого асинхронного блока, и перемещаем tx1 в этот блок — как раньше tx. Затем перемещаем исходный tx в новый асинхронный блок, где отправляем ещё сообщения с чуть большей задержкой. Мы поставили этот новый блок после блока приёма сообщений, но он мог бы стоять и до него. Важен порядок, в котором мы ждём футуры, а не порядок, в котором их создаём.

Оба отправляющих асинхронных блока должны быть блоками async move, чтобы и tx, и tx1 уничтожались, когда эти блоки заканчиваются. Иначе мы снова окажемся в том же бесконечном цикле, с которого начинали.

Наконец, переходим с trpl::join на trpl::join!, чтобы справиться с ещё одной футурой: макрос join! ждёт произвольное количество футур, если их число известно при компиляции. Как ждать коллекцию из заранее неизвестного числа футур, мы обсудим дальше в этой главе.

Теперь мы видим все сообщения от обеих отправляющих футур, а поскольку после отправки они ждут немного разное время, сообщения и приходят с разными интервалами:

received 'hi'
received 'more'
received 'from'
received 'the'
received 'messages'
received 'future'
received 'for'
received 'you'

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