Асинхронщина в Rust: Стандартная библиотека и async/.await01.06.2020 10:32

Вариант №2 (адапторы)

use std::ops::AddAssign;

use futures::Future;

fn compound_future<'a, Fut1, Fut2, Fut3, T, U, F>(
    fut1: Fut1,
    fut2: Fut2,
    fut3: Fut3,
    mut f: F,
) -> impl Future + 'a
where
    Fut1: Future + 'a,
    Fut2: Future + 'a,
    Fut3: Future + 'a,
    F: FnMut(T) -> U + 'a,
    U: AddAssign,
{
        fut1.join3(fut2, fut3).map(move |(a, b, c)| {
                let mut result = f(a);
                result += f(b);
                result += f(c);
                result
        })
}

Вариант №3 (async/.await)

use std::{future::Future, ops::AddAssign};

async fn compound_future(
    fut1: Fut1,
    fut2: Fut2,
    fut3: Fut3,
    mut f: F,
) -> U
where
    Fut1: Future
,
    Fut2: Future
,
    Fut3: Future
,
    F: FnMut(T) -> U,
    U: AddAssign,
{
    let mut result = f(fut1.await);
    result += f(fut2.await);
    result += f(fut3.await);
    result
}

Анализ

В первом варианте мы просто инкапсулировали асинхронные операции в CompoundFuture и в каждом вызове poll делаем прогресс текущей операции вручную. Не сложно догадаться, что это чревато ошибками, ведь в конечном итоге придётся писать своё подобие конечного автомата из Future (что мы и сделали) и много-много повторяющегося кода. Данный подход почти невозможно встретить в пользовательском коде, но полностью исключать его не следует, т.к. он всё ещё используется во внутренностях основополагающих библиотек (futures, Tokio и других).

Второй подход уместился всего в 24 строки, что в 3.5 раза меньше решения с ручным управлением, и, как следствие, понижается риск ошибки, повышается читабельность и поддерживаемость. Он использует понятие адапторов (комбинаторов). Дадим нестрогое определение адапторам:

Адапторы в контексте асинхронных операций — это методы, определённые на типе Fut1, отображающие (возможно, с дополнительными аргументами, влияющими на поведение) Fut1 в Fut2<..., Fut1, ...>, где Fut1: Future, Fut2: Future.

Путём последовательного вызова адапторов генерируются новые асинхронные операции (форма ленивых вычислений), каждая из которых обладает собственным поведением. Например, метод futures::future::Future::join3 возвращает футуру, вычисляющую результаты трёх переданных асинхронных операций, а futures::future::Future::map возвращает футуру, результатом которой есть значение нового типа, полученное заданным отображением результата переданной футуры.

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

= note: expected type ()
        found type futures::future::and_then::AndThen, futures::future::or_else::OrElse, [closure@src\main.rs:139:22: 144:14]>, std::result::Result<(contract::User, std::string::String, i64, i64), http::response::Response>, [closure@src\main.rs:145:23: 162:14]>, futures::future::map_err::MapErr>, [closure@src\main.rs:211:51: 224:46 telegram_client:_, chat_id:_, text:_]>, futures::future::result_::FutureResult<(), telegram_client::TelegramClientError>>, [closure@src\main.rs:173:90: 230:30 file_id:_, ext:_, user:_, message_id:_, dbs:_, chat_id:_, telegram_client:_]>, futures::future::result_::FutureResult<(), telegram_client::TelegramClientError>>, [closure@src\main.rs:166:31: 235:22 user:_, chat_id:_, message_id:_, telegram_client:_, file_id:_, dbs:_]>, [closure@src\main.rs:236:30: 242:22]>, [closure@src\main.rs:163:23: 243:14 telegram_client:_, dbs:_]>, [closure@src\main.rs:245:18: 248:14]>, std::result::Result, hyper::error::Error>, fn(http::response::Response) -> std::result::Result, hyper::error::Error> {std::result::Result, hyper::error::Error>::Ok}>, [closure@src\main.rs:136:54: 250:6 telegram_client:_, dbs:_]>

[Взято отсюда]

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

Проблемы предыдущих подходов решает синтаксис async/.await, силами которого можно писать асинхронный код в синхронном стиле. Асинхронный участок кода (async fn, async { ... } или async move { ... }) компилятором трансформируется в объект, реализующий Future, а fut.await внутри этих участков асинхронно ждёт выполнения fut. Это означает, что в итоговом методе poll код, расположенный после fut.await, выполняться не будет, пока fut не вернёт Poll::Ready(Output). Все приведённые выше реализации одинаковы по смыслу (но итоговый код может разный сгенерироваться). Асинхронный участок кода порождает объект с неявным типом в противовес комбинаторам, которые наслаивают их друг на друга.

Ещё одно неочевидное преимущество async/.await над адапторным/комбинаторным подходом — возможность занимать данные между «вызовами» .await — устраняется нужда в клонировании или использовании сырых (небезопасных) указателей.

Как и с любым нетривиальным решением в дизайне языка, особенно совмещающего системное программирование со сравнительно мощной системой типов, существуют несколько противоположных мнений. Вот некоторые аргументы против async/.await:

• Слишком дорогое нововведение. Попытки реализовать async/.await начались ещё в 2018 году (если не раньше), в то время как можно было обойтись старым дизайном с адапторами/комбинаторами (как это сделано во многих других языках, например, Scala, Java, …) и потратить время на более важные вещи, например, на систему типов, при условии, что компилятор научится более доходчиво объяснять несовпадения в типах, встречающиеся в том числе в ряде других мест (те же итераторы, стримы (речь о них в следующих статьях)).

  
  

• Неконсистентность синтаксиса. Future можно было бы выразить монадой, что позволило бы нам обращаться с асинхронными операциями при помощи do-нотации, в прочем, как и с Result, Option, Either и так далее. Однако в текущей стабильной версии Rust это оказывается невозможным сделать по-человечески без полной поддержки HKT — Higher-Kinded Types. Более подробную аргументацию смотреть в статье Алексея Жуковского «Монады как паттерн переиспользования кода». Если заинтересовала тема HKT, также будет интересно к прочтению

  
  

Наличие разницы между async move { ... } и async { ... } может сперва показаться неочевидной, ведь у версии простого блока ({ ... }) не существует move-аналога. Дело в том, что первый асинхронный вариант (с move) овладевает захваченными переменными, а второй вариант (без move) — сначала пытается заимствовать среду иммутабельно, если не получилось — заимствовать мутабельно, если и это не получилось — овладевает (вспомним, что асинхронные блоки преобразуются компилятором в анонимные типы, реализующие трейт Future).

Вот простой пример, демонстрирующий различие. Функция foo() два раза выводит строку с abc на экран:

async fn foo() {
    let string = String::from("abc");

    async { dbg!(&string); }.await;
    async { dbg!(&string); }.await;
}

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

error[E0382]: use of moved value: `string`
 --> src/lib.rs:5:16
  |
2 |     let string = String::from("abc");
  |         ------ move occurs because `string` has type `std::string::String`, which does not implement the `Copy` trait
3 |     
4 |     async move { dbg!(&string); }.await;
  |                ------------------
  |                |       |
  |                |       variable moved due to use in generator
  |                value moved here
5 |     async move { dbg!(&string); }.await;
  |                ^^^^^^^^------^^^^
  |                |       |
  |                |       use occurs due to use in generator
  |                value used here after move

Рассмотрим ещё один пример:

async fn foo() -> String {
    let string = String::from("abc");

    async { string }.await
}

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

И наконец, рассмотрим пример с мутабельным заимствованием:

async fn foo() {
    let mut string = String::from("abc");

    async { string.push_str("def"); }.await;

    dbg!(string);
}

Заимствовать иммутабельно не вышло, ведь сигнатура у String::push_str требует &mut self, значит заимствуем мутабельно — компилируется успешно. Если же присоединить move, то, как вы уже догадались, произойдёт ошибка компиляции, ведь на последней строке функции располжен вызов макроса dbg!, требующего владения string:

error[E0382]: use of moved value: `string`
 --> src/lib.rs:6:10
  |
2 |     let mut string = String::from("abc");
  |         ---------- move occurs because `string` has type `std::string::String`, which does not implement the `Copy` trait
3 |     
4 |     async move { string.push_str("def"); }.await;
  |                ---------------------------
  |                | |
  |                | variable moved due to use in generator
  |                value moved here
5 |     
6 |     dbg!(string);
  |          ^^^^^^ value used here after move

Попробуем скомпилировать следующий код:

fn main() {
    let closure = async || {
        dbg!();
    };
}

Вывод:

error[E0658]: async closures are unstable
 --> src/main.rs:2:19
  |
2 |     let closure = async || {
  |                   ^^^^^
  |
  = note: see issue #62290  for more information

То есть асинхронные замыкания ещё не стабилизировались (RFC). Не путайте асинхронные замыкания и замыкания, возвращающие Future:

• Первое — async [move] |...| { ... } (нестабилизированное)
• Второе — |...| async [move] { ... } (стабилизированное)

Как вы уже могли заметить, в сигнатуре метода Future::poll присутствует self: Pin<&mut Self>. Что это значит? Это даёт гарантию компилятору, что асинхронная операция не переместится в памяти во время выполнения. Зачем это нужно? Представим ситуацию, где на вход компилятору подаётся такая конструкция (взято отсюда):

async {
    let mut x = [0; 128];
    let read_into_buf_fut = read_into_buf(&mut x);
    read_into_buf_fut.await;
    println!("{:?}", x);
}

Должно быть сгенерировано следующее:

struct ReadIntoBuf<'a> {
    buf: &'a mut [u8], // Указывает на `x` снизу
}

struct AsyncFuture {
    x: [u8; 128],
    read_into_buf_fut: ReadIntoBuf<'self>,
}

Если объект AsyncFuture переместится в памяти, то указатель в read_into_buf_fut.buf инвалидируется (станет указывать на неправильное значение), что вызовет UB. Решение простое — запретить перемещение в памяти асинхронных операций во время их выполнения, что и делает self: Pin<&mut Self>.

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

Future — трейт для асинхронных (конкурентных) вычислений с poll-based кооперативной моделью исполнения. Объекты Future являются программными компонентами первого класса: мы можем с ними обращаться как нам вздумается, а сами по себе они ничего не делают — чтобы их выполнить, необходимо вызывать метод Future::poll до того момента, когда он вернёт Poll::Ready(Output). Обычно этим занимается отдельный модуль, называемый экзекьютором (или планировщиком).

async/.await в Rust — синтаксис для построения анонимных асинхронных операций. Код которых смотрится так, будто выполнен в синхронном стиле.

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

Ключевое слово await, которое синтаксически ведёт себя как поле, асинхронно ожидает футуру fut (fut.await). Остальное содержимое асинхронного блока будет выполнено лишь после завершения fut. Комбинаторный аналог — futures::Future::and_then (futures 0.1.x).

Далее — про то, что позволяет безжизненные футуры превратить в работающий код: про асинхронную среду исполнения Tokio.

За ревью спасибо @blandger, также другим людям за поддержку в русскоязычном Телеграм-чате @rust_async, куда вы тоже можете задавать вопросы, связанные с асинхронным программированием в Rust.

© Habrahabr.ru