Объяснение: будущее Rust для веб-разработки

Как я объяснил в одной из своих последних статей, фьючерсы - это часть асинхронной среды. Нам нужны синтаксис, тип и среда выполнения, чтобы иметь возможность запускать код без блокировки. Futures - это типовая часть асинхронной истории Rust.

Две точки зрения

Если вы переходите с NodeJS, Futures in Rust не имеет особого смысла. В NodeJS все происходит асинхронно. Поэтому, чтобы вы могли сказать: «Эй, мне действительно нужно дождаться ответа на этот HTTP-вызов GET», вы помещаете.then() в Promise, поэтому вы можете быть уверены, что просто выполняете код внутри .then(), когда HTTP-вызов закончен.

В Rust по умолчанию все блокируется и синхронно, поэтому вы можете спросить себя: «Зачем беспокоиться о сложности, это именно то, что я хотел в первую очередь!»

Rust - это язык системного программирования. Итак, чтобы вы могли создавать приложения в Rust, вам всегда нужно носить две шляпы. «Системная шляпа» (⛑) и «шляпа программиста» (🎩). Системная шляпа (⛑) позволяет вам подумать о том, что на самом деле лучше всего с машинной точки зрения, шляпа программиста (🎩) заботится о синтаксисе и написании программного обеспечения.

Если вы работаете с NodeJS, о системной шляпе позаботится среда выполнения Googles V8, так что вы можете сосредоточиться на синтаксисе. В Rust вам помогают ящики, хотя некоторые решения вам нужно принимать самостоятельно.

Системная шляпа - это причина, по которой мы хотим использовать фьючерсы. Следовательно, вам необходимо обрабатывать Futures как тип в вашем приложении, а затем убедиться, что вы используете среду выполнения для их фактического выполнения. Если вы потребляете фьючерсы (например, когда используемый вами ящик возвращает Future), вы должны понимать, откуда берутся данные. В этом случае необходимы шляпа программиста и системная шляпа.

Что на самом деле такое Future? 🎩

Future в Rust на самом деле является trait, который, если вы хотите его реализовать, выглядит так:

trait Future {
    type Item;
    type Error;
    fn poll(&mut self) -> Poll<Self::Item, Self::Error>;
}

poll метод важен. Это будет вызвано из вашей среды выполнения и вернет Async::Ready или Async::NotReady.

Если вы хотите получить различные выпуски журнала из удаленного места или из файловой системы, вы можете создать и вернуть для этого свои собственные Future:

struct Magazine {
    issues: Vec<u8>
}

и impl признак Future на нем:

impl Future for Magazine {
    // here we return a single byte
    type Item = u8;
    type Error = io::Error;
    // this method is getting called from the runtime. Everytime we can read
    // a byte into the buffer, we return `Async::Ready`
    fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
        let mut buffer = [0;1];
        match self.0.poll_read(&mut buf) {
            Ok(Async::Ready(_num_bytes_read)) => Ok(Async::Ready(buffer[0])),
            Ok(Async::NotReady) => Ok(Async::NotReady),
            Err(e) => Err(e)
        }
    }
}

Среда выполнения заполнит буфер запрашиваемой информацией, и Future вернет Async::Ready, когда буфер заполнится.

Это основная суть Future в Rust.

Когда использовать фьючерсы? 🎩

Реализация признака Future для типа - это ваш способ сказать машине: «Эй, это может занять некоторое время, учтите это при выполнении этой части кода».

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

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

Почему использование фьючерсов экономит время? ⛑

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

Поэтому вы создаете три метода, каждый из которых возвращает Future, собирает их и затем передает их среде выполнения (например, tokio через tokio::run).

Как использовать фьючерсы? 🎩 + ⛑

Если ящик возвращает Future, вы можете использовать .and_then() для работы с результатом, когда Future будет готов. Использование .map() на Future позволяет вам изменить тип Future.

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

Stream - это продолжение будущего. Если в Future создается одно значение, то Stream дает значения, пока они присутствуют.

• Stream: Stream, как и Future, черта, которую вы можете impl на типе. Это позволяет вам перебирать возвращаемые значения (Some(_) или None).
• Sink: для непрерывной (асинхронной) записи данных либо в сокет, либо в файловую систему.

Итак, Streams предназначены для чтения данных, а Sinks - для их записи. В нашей веб-экосистеме есть два типа потоков:

• Поток байтов (например, тела HTTP или потоки TCP)
• Поток сообщений (например, WebSocket Frames или UDP-пакет), где каждое сообщение имеет фиксированный размер.

Пример кода

Давайте посмотрим на пример использования, когда используемый вами ящик возвращает Future. При выполнении HTTP-вызовов хорошим примером является reqwest. На Future, возвращенном из reqwest, вы можете использовать .and_then() для обработки результатов (🎩):

// We have to use "r#" before "async" because "async" is a reserved keyword.
use reqwest::r#async::Client;
// The return type has to be `Future<Item=(), Error=()>` to be able
// to use `tokio::run`. 
// If it has a different type, you have to use `tokio::block_on`
fn fetch_data() -> impl Future<Item=(), Error=()> {
    Client::new()
        .get(url)
        .send()
        .and_then(|res| {
            res.into_body().concat2()
        })
        .map_err(|err| println!("request error: {}", err))
        .map(|body| {
            // here you can use the body to write it to a file
            // or return it via Ok()
            // If you return it via for example Ok(users)
            // then you need to adjust the return type in impl Future<Item=TYPE
            // Examples can be found here: 
            // https://github.com/gruberb/futures_playground
            // For now, lets just turn the body into a Vector
            let v = body.to_vec();
        })
}

После того, как вы создали метод, который возвращает Future (fetch_data()), вы должны передать его среде выполнения, например tokio (⛑):

tokio::run(fetch_data());

Обзор высокого уровня

1. Вы получаете Future из внешнего ящика
2. Future, скорее всего, вернет Stream значений, поэтому вам нужно преобразовать это Stream в тип, с которым вы можете работать синхронно (например, вектор или строка).
3. Вы возвращаете все Future из метода через -> impl Future<Item=(), Error=()>, где фигурные скобки () являются заполнителями для фактического типа, который вы хотите вернуть.
4. Вы передаете метод среде выполнения, такой как tokio, через tokio::run(method())
5. Вызов run запустит среду выполнения, настроит необходимые ресурсы, а затем поместит это будущее в пул потоков и начнет опрос вашего будущего.
6. Затем он попытается передать работу операционной системе.
7. Каждый раз, когда среда выполнения опрашивает ваш Future, если базовый ресурс ввода-вывода, которого ожидает ваш Future, не готов, он вернет NotReady​. Среда выполнения видит это NotReady​ возвращаемое значение и переводит Future в спящий режим
8. Как только приходит событие от базового ресурса ввода-вывода, среда выполнения проверяет, связан ли этот ресурс ввода-вывода с вашим Future, и снова начинает опрос. На этот раз ваш Future сможет вернуть Ready со значением, поскольку базовый ресурс ввода-вывода предоставил значение
9. Затем среда выполнения установит состояние Future в состояние готовности и обработает .and_then() часть кода.

Future, в отличие от NodeJS, выполняется через tokio::run, а не раньше. В Node, как только вы напишете Promise, объект будет немедленно возвращен.

Что такого особенного или сложного в Futures? ⛑

Давайте рассмотрим наш пример выше:

• Создаем нового клиента с Client::new() и .send() нашим запросом
• Получим обратно Response:

pub struct Response {
    status: StatusCode,
    headers: HeaderMap,
    url: Box<Url>,
    body: Decoder,
    ...
}

• Само тело - это декодер, который можно превратить в Body через .into_body().
• Body сам реализует Поток (как упоминалось ранее).
• Теперь мы можем заглянуть в Futures API от Rust и выяснить: мы можем превратить поток байтов в единый элемент с помощью .concat2()

...
        .and_then(|res| {
            res.into_body().concat2()
        })
...

• Мы используем этот единственный элемент в .map() части как body
• С помощью вашего редактора кода вы узнаете, что этот body на самом деле является Chunk, возвращенным из библиотеки Hyper
• Теперь мы можем превратить этот Chunk в Vector

...
        .map(|body| {
            let v = body.to_vec();
            // do whatever with v
        })
...

С этого момента мы вернулись в «нормальную» страну Rust и можем забыть, что только что произошло 🙃.

Вы можете найти полный пример в этом репозитории GitHub. Там я получаю JSON и записываю его в файл.

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

Эти мысленные шаги помогут вам не потеряться при работе с Futures:

1. Какой возвращаемый тип или значение я получаю от этой библиотеки?
2. Как я могу получить доступ к Stream значений в этом ответе?
3. Во что библиотека превратит этот Stream, когда я соберу все значения через .concat2()?
4. Как я могу превратить этот новый тип в векторный или другой формат Rust std, чтобы передать его синхронным методам?

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

Как выполнить более одного Future? 🎩

Обычно вы хотите собирать свои значения как Streams, чтобы для каждого элемента, превышающего Stream, вы могли создать новый Future для его обработки.

В Rust Futures API есть метод под названием FuturesUnordered, который можно использовать для добавления более одного Future:

use futures::stream::futures_unordered::FuturesUnordered;
use hyper::{client::ResponseFuture, Client};
fn setup_requests() -> FuturesUnordered<ResponseFuture> {
    let mut list_of_futures = FuturesUnordered::new();
    let client = Client::new();
    let first = client.get(URL);
    list_of_futures.push(first);
    let second = client.get(URL);
    list_of_futures.push(second);
    list_of_futures
}

В этом примере мы использовали hyper для наших HTTP-вызовов. Вы можете найти остальной код здесь, на Github.

Синтаксис будет немного отличаться, если вы используете reqwest. Здесь вы .join() несколько запросов и возвращаете их как «одно будущее».

fn fetch() -> impl Future<Item=(), Error=()> {
    let client = Client::new();
    let json = |mut res : Response | {
        res.json::<STRUCT_TYPE>()
    };
    let request1 =
        client
            .get(URL)
            .send()
            .and_then(json);
    let request2 =
        client
            .get(URL)
            .send()
            .and_then(json);
    request1.join(request2)
        .map(|(res1, res2)|{
            println!("{:?}", res1);
            println!("{:?}", res2);
        })
        .map_err(|err| {
            println!("stdout error: {}", err);
        })
}

Полный код также можно найти на GitHub.

Какое будущее у Futures? 🎩 + ⛑

Фьючерсы переходят на стабильный Rust на уровне 1,37, то есть примерно в июне. Также есть изменения в синтаксисе и времени выполнения, которые увеличивают объем кода, который вам нужно написать, чтобы получить этот поток значений из Future в синхронный формат Rust.

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

Резюме

Если вы выполняете асинхронные операции, такие как выборка файлов из операционной системы или отправка HTTP-запросов на удаленный сервер, Futures позволяет вам работать с возвращаемыми значениями неблокирующим образом.

Если это будет синхронно, вам придется заблокировать поток, в котором выполняется операция, и дождаться результата, пока вы не продолжите. Чтобы сделать это асинхронным способом, у нас есть среда выполнения, которая сама создает потоки и берет на себя Futures. Оно заполнит значение внутри Future, когда операционная система вернет значение среде выполнения.

После выполнения Future среда выполнения устанавливает Async::Ready, и .and_then() часть кода будет выполнена.