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

Передача управления среде выполнения

Вспомните из раздела «Наша первая асинхронная программа», что в каждой точке await Rust дает среде выполнения возможность приостановить задачу и переключиться на другую, если ожидаемый future еще не готов. Верно и обратное: Rust приостанавливает async-блоки и возвращает управление среде выполнения только в точке await. Все между точками await является синхронным.

Это означает, что если вы выполняете большой объем работы в async-блоке без точки await, такой future не даст другим futures продвигаться вперед. Иногда можно услышать, что это называется голоданием (starving) других futures одним future. В некоторых случаях это может быть не так важно. Однако если вы выполняете какую-то дорогостоящую настройку или долгую работу, или если у вас есть future, который будет бесконечно выполнять какую-то конкретную задачу, вам нужно подумать, когда и где возвращать управление среде выполнения.

Давайте смоделируем долгую операцию, чтобы проиллюстрировать проблему голодания, а затем рассмотрим, как ее решить. Листинг 17-14 вводит функцию slow.

extern crate trpl; // required for mdbook test

use std::{thread, time::Duration};

fn main() {
    trpl::block_on(async {
        // We will call `slow` here later
    });
}

fn slow(name: &str, ms: u64) {
    thread::sleep(Duration::from_millis(ms));
    println!("'{name}' ran for {ms}ms");
}

Этот код использует std::thread::sleep вместо trpl::sleep, так что вызов slow будет блокировать текущий поток на некоторое число миллисекунд. Мы можем использовать slow как замену реальным операциям, которые одновременно являются долгими и блокирующими.

В листинге 17-15 мы используем slow, чтобы имитировать выполнение такой работы, ограниченной CPU, в паре futures.

extern crate trpl; // required for mdbook test

use std::{thread, time::Duration};

fn main() {
    trpl::block_on(async {
        let a = async {
            println!("'a' started.");
            slow("a", 30);
            slow("a", 10);
            slow("a", 20);
            trpl::sleep(Duration::from_millis(50)).await;
            println!("'a' finished.");
        };

        let b = async {
            println!("'b' started.");
            slow("b", 75);
            slow("b", 10);
            slow("b", 15);
            slow("b", 350);
            trpl::sleep(Duration::from_millis(50)).await;
            println!("'b' finished.");
        };

        trpl::select(a, b).await;
    });
}

fn slow(name: &str, ms: u64) {
    thread::sleep(Duration::from_millis(ms));
    println!("'{name}' ran for {ms}ms");
}

Каждый future возвращает управление среде выполнения только после выполнения набора медленных операций. Если вы запустите этот код, то увидите такой вывод:

'a' started.
'a' ran for 30ms
'a' ran for 10ms
'a' ran for 20ms
'b' started.
'b' ran for 75ms
'b' ran for 10ms
'b' ran for 15ms
'b' ran for 350ms
'a' finished.

Как и в листинге 17-5, где мы использовали trpl::select, чтобы устроить состязание futures, загружающих два URL, select все еще завершается, как только завершается a. Однако между вызовами slow в двух futures нет чередования. Future a выполняет всю свою работу, пока не будет достигнуто ожидание вызова trpl::sleep; затем future b выполняет всю свою работу, пока не будет достигнуто ожидание его собственного вызова trpl::sleep; и, наконец, future a завершается. Чтобы оба futures могли продвигаться вперед между своими медленными задачами, нам нужны точки await, чтобы мы могли возвращать управление среде выполнения. Значит, нам нужно что-то, что можно ожидать!

Мы уже видим такую передачу управления в листинге 17-15: если бы мы удалили trpl::sleep в конце future a, он завершился бы, а future b не выполнился бы вообще. Давайте попробуем использовать функцию trpl::sleep как отправную точку для того, чтобы операции могли по очереди продвигаться вперед, как показано в листинге 17-16.

extern crate trpl; // required for mdbook test

use std::{thread, time::Duration};

fn main() {
    trpl::block_on(async {
        let one_ms = Duration::from_millis(1);

        let a = async {
            println!("'a' started.");
            slow("a", 30);
            trpl::sleep(one_ms).await;
            slow("a", 10);
            trpl::sleep(one_ms).await;
            slow("a", 20);
            trpl::sleep(one_ms).await;
            println!("'a' finished.");
        };

        let b = async {
            println!("'b' started.");
            slow("b", 75);
            trpl::sleep(one_ms).await;
            slow("b", 10);
            trpl::sleep(one_ms).await;
            slow("b", 15);
            trpl::sleep(one_ms).await;
            slow("b", 350);
            trpl::sleep(one_ms).await;
            println!("'b' finished.");
        };

        trpl::select(a, b).await;
    });
}

fn slow(name: &str, ms: u64) {
    thread::sleep(Duration::from_millis(ms));
    println!("'{name}' ran for {ms}ms");
}

Мы добавили вызовы trpl::sleep с точками await между каждым вызовом slow. Теперь работа двух futures чередуется:

'a' started.
'a' ran for 30ms
'b' started.
'b' ran for 75ms
'a' ran for 10ms
'b' ran for 10ms
'a' ran for 20ms
'b' ran for 15ms
'a' finished.

Future a все еще выполняется некоторое время перед передачей управления future b, потому что он вызывает slow до первого вызова trpl::sleep, но после этого futures передают управление друг другу каждый раз, когда один из них доходит до точки await. В этом случае мы сделали это после каждого вызова slow, но могли бы разбить работу любым способом, который кажется нам наиболее разумным.

Однако на самом деле мы не хотим здесь спать: мы хотим продвигаться вперед как можно быстрее. Нам просто нужно вернуть управление среде выполнения. Мы можем сделать это напрямую, используя функцию trpl::yield_now. В листинге 17-17 мы заменяем все эти вызовы trpl::sleep на trpl::yield_now.

extern crate trpl; // required for mdbook test

use std::{thread, time::Duration};

fn main() {
    trpl::block_on(async {
        let a = async {
            println!("'a' started.");
            slow("a", 30);
            trpl::yield_now().await;
            slow("a", 10);
            trpl::yield_now().await;
            slow("a", 20);
            trpl::yield_now().await;
            println!("'a' finished.");
        };

        let b = async {
            println!("'b' started.");
            slow("b", 75);
            trpl::yield_now().await;
            slow("b", 10);
            trpl::yield_now().await;
            slow("b", 15);
            trpl::yield_now().await;
            slow("b", 350);
            trpl::yield_now().await;
            println!("'b' finished.");
        };

        trpl::select(a, b).await;
    });
}

fn slow(name: &str, ms: u64) {
    thread::sleep(Duration::from_millis(ms));
    println!("'{name}' ran for {ms}ms");
}

Этот код яснее выражает фактическое намерение и может быть значительно быстрее, чем использование sleep, потому что таймеры вроде того, который использует sleep, часто имеют ограничения по точности. Например, версия sleep, которую мы используем, всегда будет спать как минимум одну миллисекунду, даже если мы передадим ей Duration в одну наносекунду. Повторим: современные компьютеры быстры — они могут сделать очень много за одну миллисекунду!

Это означает, что async может быть полезен даже для задач, ограниченных вычислениями, в зависимости от того, что еще делает ваша программа, потому что он предоставляет полезный инструмент для структурирования отношений между разными частями программы (но ценой накладных расходов конечного автомата async). Это форма кооперативной многозадачности, где каждый future имеет возможность определять, когда передать управление через точки await. Следовательно, каждый future также несет ответственность за то, чтобы не блокироваться слишком долго. В некоторых встроенных операционных системах на основе Rust это единственный вид многозадачности!

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

Создание собственных async-абстракций

Мы также можем компоновать futures вместе, чтобы создавать новые шаблоны. Например, мы можем построить функцию timeout с помощью уже имеющихся у нас async-строительных блоков. Когда мы закончим, результат станет еще одним строительным блоком, который мы сможем использовать для создания новых async-абстракций.

Листинг 17-18 показывает, как мы ожидали бы, что этот timeout будет работать с медленным future.

extern crate trpl; // required for mdbook test

use std::time::Duration;

fn main() {
    trpl::block_on(async {
        let slow = async {
            trpl::sleep(Duration::from_secs(5)).await;
            "Finally finished"
        };

        match timeout(slow, Duration::from_secs(2)).await {
            Ok(message) => println!("Succeeded with '{message}'"),
            Err(duration) => {
                println!("Failed after {} seconds", duration.as_secs())
            }
        }
    });
}

Давайте реализуем это! Для начала подумаем об API для timeout:

• Он сам должен быть async-функцией, чтобы мы могли его ожидать.
• Его первым параметром должен быть future, который нужно запустить. Мы можем сделать его обобщенным, чтобы он работал с любым future.
• Его вторым параметром будет максимальное время ожидания. Если мы используем Duration, его будет легко передать дальше в trpl::sleep.
• Он должен возвращать Result. Если future успешно завершится, Result будет Ok со значением, произведенным этим future. Если первым истечет тайм-аут, Result будет Err с длительностью, которую мы ждали до истечения тайм-аута.

Листинг 17-19 показывает это объявление.

extern crate trpl; // required for mdbook test

use std::time::Duration;

fn main() {
    trpl::block_on(async {
        let slow = async {
            trpl::sleep(Duration::from_secs(5)).await;
            "Finally finished"
        };

        match timeout(slow, Duration::from_secs(2)).await {
            Ok(message) => println!("Succeeded with '{message}'"),
            Err(duration) => {
                println!("Failed after {} seconds", duration.as_secs())
            }
        }
    });
}

async fn timeout<F: Future>(
    future_to_try: F,
    max_time: Duration,
) -> Result<F::Output, Duration> {
    // Here is where our implementation will go!
}

Это удовлетворяет нашим целям для типов. Теперь подумаем о поведении, которое нам нужно: мы хотим устроить состязание переданного future с длительностью. Мы можем использовать trpl::sleep, чтобы создать future-таймер из длительности, и использовать trpl::select, чтобы запустить этот таймер вместе с future, который передает вызывающая сторона.

В листинге 17-20 мы реализуем timeout, сопоставляя результат ожидания trpl::select.

extern crate trpl; // required for mdbook test

use std::time::Duration;

use trpl::Either;

// --snip--

fn main() {
    trpl::block_on(async {
        let slow = async {
            trpl::sleep(Duration::from_secs(5)).await;
            "Finally finished"
        };

        match timeout(slow, Duration::from_secs(2)).await {
            Ok(message) => println!("Succeeded with '{message}'"),
            Err(duration) => {
                println!("Failed after {} seconds", duration.as_secs())
            }
        }
    });
}

async fn timeout<F: Future>(
    future_to_try: F,
    max_time: Duration,
) -> Result<F::Output, Duration> {
    match trpl::select(future_to_try, trpl::sleep(max_time)).await {
        Either::Left(output) => Ok(output),
        Either::Right(_) => Err(max_time),
    }
}

Реализация trpl::select несправедлива: она всегда опрашивает аргументы в том порядке, в котором они переданы (другие реализации select будут случайным образом выбирать, какой аргумент опросить первым). Поэтому мы передаем future_to_try в select первым, чтобы он получил шанс завершиться, даже если max_time — очень короткая длительность. Если future_to_try завершается первым, select вернет Left с результатом от future_to_try. Если первым завершится timer, select вернет Right с выводом таймера ().

Если future_to_try успешно завершается и мы получаем Left(output), мы возвращаем Ok(output). Если вместо этого истекает таймер сна и мы получаем Right(()), мы игнорируем () с помощью _ и возвращаем Err(max_time).

Теперь у нас есть работающий timeout, построенный из двух других async-помощников. Если мы запустим наш код, он напечатает сценарий ошибки после тайм-аута:

Failed after 2 seconds

Поскольку futures компонуются с другими futures, вы можете строить действительно мощные инструменты, используя меньшие async-строительные блоки. Например, можно использовать тот же подход, чтобы объединять тайм-ауты с повторными попытками, а затем применять их к операциям вроде сетевых вызовов (таких как в листинге 17-5).

На практике вы обычно будете работать напрямую с async и await, а во вторую очередь — с функциями вроде select и макросами вроде join!, чтобы управлять тем, как исполняются самые внешние futures.

Теперь мы увидели несколько способов работать с несколькими futures одновременно. Далее мы рассмотрим, как можно работать с несколькими futures в последовательности во времени с помощью streams.