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

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

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

Значит, если в асинхронном блоке выполняется много работы без единой точки ожидания, эта футура не даст продвигаться никаким другим футурам. Иногда говорят, что одна футура морит голодом (starving) остальные. Порой в этом нет ничего страшного. Но если вы выполняете какую-то дорогую подготовку или долгую работу либо у вас есть футура, которая бесконечно выполняет какую-то задачу, нужно продумать, когда и где возвращать управление среде выполнения.

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

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

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

fn main() {
    trpl::block_on(async {
        // Позже здесь будем вызывать `slow`
    });
}

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 имитируем такую работу, ограниченную процессором, в паре футур.

extern crate trpl; // нужно для 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");
}

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

'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 устраивали гонку двух загрузок), select завершается, как только готова a. Но вызовы slow в двух футурах не перемежаются. Футура a делает всю свою работу, пока не дойдёт до ожидания trpl::sleep; затем футура b делает всю свою, пока не дойдёт до своего trpl::sleep; и наконец завершается a. Чтобы обе футуры могли продвигаться между своими медленными задачами, нужны точки ожидания, в которых управление возвращается среде выполнения. А значит, нужно что-то, чего можно ждать!

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

extern crate trpl; // нужно для 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");
}

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

'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.

Футура a всё ещё немного работает, прежде чем передать управление b: она вызывает slow раньше, чем trpl::sleep. Но после этого футуры передают управление друг другу каждый раз, когда одна из них доходит до точки ожидания. Здесь мы делаем это после каждого вызова slow, но разбить работу можно так, как удобнее нам.

Только вот спать нам здесь на самом деле не нужно: мы хотим продвигаться как можно быстрее. Нужно лишь вернуть управление среде выполнения. Это можно сделать напрямую — функцией trpl::yield_now. В листинге 17-17 мы заменяем все вызовы trpl::sleep на trpl::yield_now.

extern crate trpl; // нужно для 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 может пригодиться даже для задач, ограниченных процессором, — в зависимости от того, что ещё делает программа. Он даёт удобный инструмент для организации взаимодействия разных частей программы (хотя и ценой накладных расходов на асинхронный конечный автомат). Это форма кооперативной многозадачности (cooperative multitasking): каждая футура сама решает, когда передать управление, — через точки ожидания. А значит, каждая футура и отвечает за то, чтобы не блокировать слишком надолго. В некоторых встраиваемых операционных системах на Rust это единственный вид многозадачности!

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

Собственные асинхронные абстракции

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

В листинге 17-18 показано, как мы хотим пользоваться timeout с медленной футурой.

extern crate trpl; // нужно для 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:

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

Это объявление — в листинге 17-19.

extern crate trpl; // нужно для 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> {
    // Здесь будет наша реализация!
}

С типами разобрались. Теперь продумаем нужное поведение: мы хотим устроить гонку между переданной футурой и заданной длительностью. Из длительности можно сделать футуру-таймер с помощью trpl::sleep, а с помощью trpl::select запустить таймер наперегонки с футурой, которую передал вызывающий код.

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

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

use std::time::Duration;

use trpl::Either;

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

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 выбирают, какой аргумент опросить первым, случайно). Поэтому мы передаём в select первой future_to_try — чтобы у неё был шанс завершиться, даже если max_time очень короткий. Если первой закончит future_to_try, select вернёт Left с её результатом. Если первым сработает timer, select вернёт Right с результатом таймера — ().

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

Вот и готова работающая timeout, собранная из двух других асинхронных помощников. Если запустить код, после истечения времени он выведет сообщение о неудаче:

Failed after 2 seconds

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

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

Мы рассмотрели несколько способов работать с несколькими футурами одновременно. Дальше посмотрим, как работать с последовательностью футур во времени — со стримами.