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

Трейты асинхронности поближе

На протяжении главы мы по-разному пользовались трейтами Future, Stream и StreamExt. Но до сих пор не углублялись в то, как они устроены и как сочетаются друг с другом, — для повседневной работы на Rust это обычно и не нужно. Однако иногда встречаются ситуации, когда нужно понимать чуть больше подробностей об этих трейтах, а также о типе Pin и трейте Unpin. В этом разделе мы копнём ровно настолько, чтобы помочь в таких ситуациях, а по-настоящему глубокое погружение оставим другой документации.

Трейт Future

Начнём с того, как устроен трейт Future. Вот как он определён в Rust:

#![allow(unused)]
fn main() {
use std::pin::Pin;
use std::task::{Context, Poll};

pub trait Future {
    type Output;

    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output>;
}
}

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

Во-первых, ассоциированный тип Output у Future говорит, во что разрешается футура. Это аналог ассоциированного типа Item у трейта Iterator. Во-вторых, у Future есть метод poll, который принимает особую ссылку Pin в качестве параметра self и изменяемую ссылку на тип Context, а возвращает Poll<Self::Output>. О Pin и Context поговорим чуть позже. А пока сосредоточимся на том, что возвращает метод, — на типе Poll:

#![allow(unused)]
fn main() {
pub enum Poll<T> {
    Ready(T),
    Pending,
}
}

Тип Poll похож на Option: у него есть вариант со значением, Ready(T), и вариант без значения, Pending. Но смысл у Poll совсем другой! Вариант Pending означает, что футуре ещё есть что делать, и вызывающему нужно будет проверить её позже. А вариант Ready означает, что Future закончила работу и значение T готово.

Код, где используется await, Rust под капотом компилирует в код, вызывающий poll. Вернитесь к листингу 17-4, где мы выводили заголовок страницы для одного URL, когда тот был получен. Rust компилирует его во что-то примерно (хотя и не точно) такое:

match page_title(url).poll() {
    Ready(page_title) => match page_title {
        Some(title) => println!("The title for {url} was {title}"),
        None => println!("{url} had no title"),
    }
    Pending => {
        // А что здесь?
    }
}

Что делать, если футура ещё в состоянии Pending? Нужно как-то пробовать снова, и снова, и снова, пока футура наконец не будет готова. Иначе говоря, нужен цикл:

let mut page_title_fut = page_title(url);
loop {
    match page_title_fut.poll() {
        Ready(value) => match page_title {
            Some(title) => println!("The title for {url} was {title}"),
            None => println!("{url} had no title"),
        }
        Pending => {
            // продолжаем
        }
    }
}

Но если бы Rust компилировал код ровно в это, каждый await был бы блокирующим — прямая противоположность тому, чего мы добивались! Вместо этого Rust делает так, чтобы цикл мог передать управление тому, кто может приостановить работу над этой футурой, поработать над другими, а потом снова проверить эту. Как мы уже видели, этот «кто-то» — асинхронная среда выполнения, и такое планирование и координация — одна из её главных задач.

В разделе «Передача данных между задачами через сообщения» мы говорили об ожидании rx.recv. Вызов recv возвращает футуру, а ожидание футуры означает её опрос. Мы отмечали, что среда выполнения приостанавливает футуру, пока та не будет готова — с Some(message) или с None, когда канал закроется. Теперь, лучше понимая трейт Future и в особенности Future::poll, мы видим, как это работает. Среда выполнения знает, что футура не готова, когда та возвращает Poll::Pending. И наоборот, среда выполнения знает, что футура готова, и продвигает её дальше, когда poll возвращает Poll::Ready(Some(message)) или Poll::Ready(None).

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

Тип Pin и трейт Unpin

В листинге 17-13 мы ждали три футуры с помощью макроса trpl::join!. Но часто бывает коллекция — например, вектор — с каким-то количеством футур, которое становится известно только во время выполнения. Изменим листинг 17-13 на код из листинга 17-23: положим три футуры в вектор и вызовем функцию trpl::join_all. Пока этот код не скомпилируется.

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_secs(1)).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_secs(1)).await;
            }
        };

        let futures: Vec<Box<dyn Future<Output = ()>>> =
            vec![Box::new(tx1_fut), Box::new(rx_fut), Box::new(tx_fut)];

        trpl::join_all(futures).await;
    });
}

Каждую футуру мы помещаем в Box, чтобы превратить их в трейт-объекты, как в разделе «Возвращаем ошибки из run» главы 12. (Подробно о трейт-объектах — в главе 18.) Трейт-объекты позволяют обращаться со всеми анонимными футурами, которые создают эти блоки, как с одним типом: все они реализуют трейт Future.

Это может удивить. Ведь ни один из асинхронных блоков ничего не возвращает, так что каждый даёт Future<Output = ()>. Но помните: Future — это трейт, а компилятор создаёт для каждого асинхронного блока уникальное перечисление, даже если типы результатов совпадают. Как нельзя положить в Vec две разные написанные вручную структуры, так нельзя и смешивать перечисления, сгенерированные компилятором.

Затем мы передаём коллекцию футур в функцию trpl::join_all и ждём результат. Однако код не компилируется. Вот важная часть сообщений об ошибке:

error[E0277]: `dyn Future<Output = ()>` cannot be unpinned
  --> src/main.rs:48:33
   |
48 |         trpl::join_all(futures).await;
   |                                 ^^^^^ the trait `Unpin` is not implemented for `dyn Future<Output = ()>`
   |
   = note: consider using the `pin!` macro
           consider using `Box::pin` if you need to access the pinned value outside of the current scope
   = note: required for `Box<dyn Future<Output = ()>>` to implement `Future`
note: required by a bound in `futures_util::future::join_all::JoinAll`
  --> file:///home/.cargo/registry/src/index.crates.io-1949cf8c6b5b557f/futures-util-0.3.30/src/future/join_all.rs:29:8
   |
27 | pub struct JoinAll<F>
   |            ------- required by a bound in this struct
28 | where
29 |     F: Future,
   |        ^^^^^^ required by this bound in `JoinAll`

Заметка в сообщении советует закрепить (pin) значения макросом pin!, то есть поместить их в тип Pin, который гарантирует, что значения не будут перемещаться в памяти. Сообщение говорит, что закрепление нужно, потому что dyn Future<Output = ()> должен реализовывать трейт Unpin, а сейчас не реализует.

Функция trpl::join_all возвращает структуру JoinAll. Эта структура обобщена по типу F, который обязан реализовывать трейт Future. Когда футуру ждут напрямую через await, она неявно закрепляется. Поэтому pin! не нужно писать везде, где мы ждём футуры.

Но здесь мы ждём не футуру напрямую. Мы создаём новую футуру, JoinAll, передавая коллекцию футур в функцию join_all. Сигнатура join_all требует, чтобы типы элементов коллекции реализовывали трейт Future, а Box<T> реализует Future, только если обёрнутый им T — футура, реализующая трейт Unpin.

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

#![allow(unused)]
fn main() {
use std::pin::Pin;
use std::task::{Context, Poll};

pub trait Future {
    type Output;

    // Обязательный метод
    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output>;
}
}

Параметр cx и его тип Context — ключ к тому, как среда выполнения узнаёт, когда проверять ту или иную футуру, оставаясь при этом ленивой. Подробности опять же выходят за рамки этой главы, и думать об этом обычно приходится, только когда пишешь собственную реализацию Future. Сосредоточимся на типе self: мы впервые видим метод, у которого self аннотирован типом. Аннотация типа у self работает как аннотации типов других параметров функции, но с двумя важными отличиями:

• Она говорит Rust, каким должен быть тип self, чтобы метод можно было вызвать.
• Это не может быть любой тип. Допускается только тип, для которого реализован метод, ссылка или умный указатель на этот тип либо Pin, обёрнутый вокруг ссылки на этот тип.

Подробнее этот синтаксис мы увидим в главе 18. Пока достаточно знать: чтобы опросить футуру и проверить, Pending она или Ready(Output), нужна изменяемая ссылка на тип, обёрнутая в Pin.

Pin — обёртка для типов, похожих на указатели: &, &mut, Box, Rc. (Формально Pin работает с типами, реализующими трейты Deref или DerefMut, но на практике это равносильно работе только со ссылками и умными указателями.) Pin сам по себе не указатель, и, в отличие от Rc и Arc, у него нет собственного поведения вроде подсчёта ссылок. Это чисто инструмент, с помощью которого компилятор накладывает ограничения на использование указателей.

То, что await реализован через вызовы poll, отчасти объясняет ошибку, которую мы видели. Но там речь шла об Unpin, а не о Pin. Как же связаны Pin и Unpin и почему Future требует, чтобы для вызова poll значение self было обёрнуто в Pin?

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

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

Когда мы перемещаем футуру — кладём её в структуру данных, чтобы перебирать в join_all, или возвращаем из функции, — на самом деле перемещается конечный автомат, который для нас создал Rust. И, в отличие от большинства других типов в Rust, у футур, которые Rust создаёт для асинхронных блоков, в полях вариантов могут оказаться ссылки на самих себя, как показано на упрощённой схеме на рисунке 17-4.

Но по умолчанию любой объект со ссылкой на самого себя небезопасно перемещать: ссылки всегда указывают на фактический адрес в памяти того, на что ссылаются (рисунок 17-5). Если переместить саму структуру данных, внутренние ссылки продолжат указывать на старое место. А это место в памяти теперь недействительно. Во-первых, его значение не будет обновляться при изменении структуры. Во-вторых — и это важнее, — компьютер теперь волен использовать эту память для чего-то другого! Позже можно прочитать оттуда совершенно посторонние данные.

Теоретически компилятор Rust мог бы при каждом перемещении объекта обновлять все ссылки на него, но это могло бы сильно ударить по производительности — особенно если обновлять нужно целую сеть ссылок. Если же гарантировать, что структура данных не перемещается в памяти, обновлять ссылки не придётся. Именно для этого и нужен анализатор заимствований Rust: в безопасном коде он не даёт переместить элемент, на который есть активная ссылка.

Pin опирается на это и даёт ровно ту гарантию, которая нам нужна. Если закрепить значение, обернув указатель на него в Pin, оно больше не сможет перемещаться. То есть если у вас Pin<Box<SomeType>>, закрепляется значение SomeType, а не указатель Box. Это показано на рисунке 17-6.

На самом деле сам указатель Box по-прежнему может свободно перемещаться. Помните: нам важно, чтобы на месте оставались данные, на которые в конечном счёте указывает ссылка. Если указатель перемещается, а данные, на которые он указывает, остаются на месте, как на рисунке 17-7, никакой проблемы нет. (В качестве самостоятельного упражнения изучите документацию к этим типам и к модулю std::pin и попробуйте понять, как сделать это с Pin, обёрнутым вокруг Box.) Главное — сам самоссылающийся тип переместиться не может: он по-прежнему закреплён.

Однако большинство типов совершенно безопасно перемещать, даже если они оказались за указателем Pin. О закреплении нужно думать, только когда у элементов есть внутренние ссылки. Примитивные значения вроде чисел и логических значений безопасны: внутренних ссылок у них, очевидно, нет. Как и у большинства типов, с которыми вы обычно работаете в Rust. Например, Vec можно перемещать без опаски. Но, судя по тому, что мы видели до сих пор, с Pin<Vec<String>> пришлось бы делать всё через безопасные, но ограниченные API Pin, хотя Vec<String> всегда безопасно перемещать, если на него нет других ссылок. Нужен способ сказать компилятору, что в таких случаях перемещать элементы можно, — и здесь в игру вступает Unpin.

Unpin — трейт-маркер, как трейты Send и Sync из главы 16, и собственной функциональности у него нет. Трейты-маркеры существуют лишь для того, чтобы сообщить компилятору, что тип, реализующий трейт, безопасно использовать в определённом контексте. Unpin говорит компилятору, что данному типу не нужно соблюдать никаких гарантий насчёт того, можно ли безопасно перемещать значение.

Как и в случае с Send и Sync, компилятор автоматически реализует Unpin для всех типов, для которых может доказать, что это безопасно. Особый случай — опять же как с Send и Sync — когда Unpin для типа не реализован. Записывается это как impl !Unpin for SomeType, где SomeType — имя типа, которому нужно соблюдать эти гарантии, чтобы оставаться безопасным, когда указатель на него используется в Pin.

Иначе говоря, о связи Pin и Unpin нужно помнить две вещи. Во-первых, Unpin — это «обычный» случай, а !Unpin — особый. Во-вторых, реализует тип Unpin или !Unpin, важно только тогда, когда вы используете закреплённый указатель на этот тип вроде Pin<&mut SomeType>.

Для наглядности возьмём String: у неё есть длина и символы Юникода, из которых она состоит. String можно обернуть в Pin, как на рисунке 17-8. Но String автоматически реализует Unpin, как и большинство других типов в Rust.

Поэтому с String можно делать то, что было бы недопустимо, реализуй она !Unpin, — например, заменить одну строку другой ровно на том же месте в памяти, как на рисунке 17-9. Договор Pin это не нарушает: у String нет внутренних ссылок, из-за которых её было бы небезопасно перемещать. Именно поэтому она реализует Unpin, а не !Unpin.

Теперь мы знаем достаточно, чтобы понять ошибки при вызове join_all из листинга 17-23. Изначально мы пытались переместить футуры, созданные асинхронными блоками, в Vec<Box<dyn Future<Output = ()>>>, но, как мы видели, у этих футур могут быть внутренние ссылки, поэтому Unpin они автоматически не реализуют. Если же их закрепить, получившиеся значения Pin можно класть в Vec, не опасаясь, что данные внутри футур куда-то переместятся. В листинге 17-24 показано, как исправить код: вызвать макрос pin! там, где определена каждая из трёх футур, и поправить тип трейт-объекта.

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

use std::pin::{Pin, pin};

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

use std::time::Duration;

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

        let tx1 = tx.clone();
        let tx1_fut = pin!(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_secs(1)).await;
            }
        });

        let rx_fut = pin!(async {
            // --пропуск--
            while let Some(value) = rx.recv().await {
                println!("received '{value}'");
            }
        });

        let tx_fut = pin!(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_secs(1)).await;
            }
        });

        let futures: Vec<Pin<&mut dyn Future<Output = ()>>> =
            vec![tx1_fut, rx_fut, tx_fut];

        trpl::join_all(futures).await;
    });
}

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

Pin и Unpin важны в основном при написании низкоуровневых библиотек или самих сред выполнения, а не в повседневном коде на Rust. Но теперь, встретив эти трейты в сообщениях об ошибках, вы будете лучше понимать, как исправить код!

Трейт Stream

Теперь, когда вы глубже понимаете Future, Pin и Unpin, обратимся к трейту Stream. Как вы узнали раньше в этой главе, стримы похожи на асинхронные итераторы. Но, в отличие от Iterator и Future, на момент написания книги у Stream нет определения в стандартной библиотеке. Зато есть очень распространённое определение из крейта futures, которое используется по всей экосистеме.

Прежде чем смотреть, как трейт Stream может объединить трейты Iterator и Future, вспомним их определения. От Iterator берётся идея последовательности: его метод next выдаёт Option<Self::Item>. От Future — идея готовности во времени: его метод poll выдаёт Poll<Self::Output>. Чтобы представить последовательность элементов, которые становятся готовыми со временем, определим трейт Stream, объединяющий эти возможности:

#![allow(unused)]
fn main() {
use std::pin::Pin;
use std::task::{Context, Poll};

trait Stream {
    type Item;

    fn poll_next(
        self: Pin<&mut Self>,
        cx: &mut Context<'_>
    ) -> Poll<Option<Self::Item>>;
}
}

Трейт Stream определяет ассоциированный тип Item — тип элементов, которые выдаёт стрим. Это похоже на Iterator, где элементов может быть от нуля до сколь угодно многих, и не похоже на Future, где Output всегда ровно один, пусть даже это единичный тип ().

Ещё Stream определяет метод получения этих элементов. Мы назвали его poll_next, чтобы было видно: он опрашивает так же, как Future::poll, и выдаёт последовательность элементов так же, как Iterator::next. Его возвращаемый тип сочетает Poll и Option. Внешний тип — Poll, потому что, как и у футуры, нужно проверять готовность. Внутренний — Option, потому что, как и итератору, нужно сообщать, будут ли ещё сообщения.

Нечто очень похожее на это определение, скорее всего, в итоге войдёт в стандартную библиотеку Rust. А пока оно есть в инструментарии большинства сред выполнения, так что на него можно полагаться, и всё, о чём пойдёт речь дальше, в целом применимо!

Однако в примерах из раздела «Стримы: последовательности футур» мы использовали не poll_next и не Stream, а next и StreamExt. Конечно, можно было бы работать напрямую с API poll_next, вручную написав собственные конечные автоматы Stream, — так же, как можно работать с футурами напрямую через их метод poll. Но с await гораздо приятнее, и трейт StreamExt даёт метод next, чтобы так и делать:

#![allow(unused)]
fn main() {
use std::pin::Pin;
use std::task::{Context, Poll};

trait Stream {
    type Item;
    fn poll_next(
        self: Pin<&mut Self>,
        cx: &mut Context<'_>,
    ) -> Poll<Option<Self::Item>>;
}

trait StreamExt: Stream {
    async fn next(&mut self) -> Option<Self::Item>
    where
        Self: Unpin;

    // другие методы...
}
}

fn next(&mut self) -> Next<'_, Self> where Self: Unpin;

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

В версии StreamExt из крейта trpl трейт не только определяет метод next, но и даёт его реализацию по умолчанию, которая правильно обрабатывает все детали вызова Stream::poll_next. Значит, даже если вам понадобится написать собственный потоковый тип данных, реализовать нужно только Stream — и все, кто пользуется вашим типом, автоматически смогут применять к нему StreamExt и его методы.

На этом с низкоуровневыми подробностями этих трейтов всё. В завершение посмотрим, как сочетаются друг с другом футуры (включая стримы), задачи и потоки!