Макрос try_join

#![allow(unused)]
fn main() {
macro_rules! try_join {
    ($(biased;)? $($future:expr),*) => { ... };
}
}

Доступно только с флагом функции macros.

Ожидает завершения нескольких конкурентных ветвей, возвращая результат, когда все ветви завершаются с Ok(_) или при первой Err(_).

Описание

Макрос try_join! должен использоваться внутри асинхронных функций, замыканий и блоков.

Аналогично join!, макрос try_join! принимает список асинхронных выражений и выполняет их конкурентно в одной и той же задаче. Каждое асинхронное выражение вычисляется в future, и future из каждого выражения мультиплексируются в текущей задаче. Макрос try_join! возвращает результат, когда все ветви возвращают Ok или когда первая ветвь возвращает Err.

Примечания

Предоставленные future хранятся inline и не требуют выделения Vec.

Характеристики времени выполнения

Поскольку все асинхронные выражения выполняются в текущей задаче, выражения могут выполняться конкурентно, но не параллельно. Это означает, что все выражения выполняются в одном потоке, и если одна ветвь блокирует поток, все другие выражения не смогут продолжить выполнение. Если требуется параллелизм, порождайте каждое асинхронное выражение с помощью tokio::spawn и передавайте дескриптор соединения в try_join!.

Честность (Fairness)

По умолчанию сгенерированный future макроса try_join! чередует порядок опроса содержащихся future при каждом пробуждении.

Это поведение можно переопределить, добавив biased; в начало использования макроса. Это заставит try_join опрашивать future в порядке их следования сверху вниз.

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

Но у этого режима есть важное предостережение. Вы сами отвечаете за обеспечение справедливого порядка опроса ваших future. Если, например, вы объединяете поток и future завершения, и поток имеет огромный объем сообщений, обработка которых занимает много времени за один опрос, вам следует поместить future завершения раньше в списке try_join!, чтобы гарантировать, что он всегда будет опрашиваться и не будет задерживаться из-за того, что future потока долго возвращает Poll::Pending.

Примеры

Базовый try_join с двумя ветвями

#![allow(unused)]
fn main() {
async fn do_stuff_async() -> Result<(), &'static str> {
    // асинхронная работа
    Ok(())
}

async fn more_async_work() -> Result<(), &'static str> {
    // больше асинхронной работы
    Ok(())
}

let res = tokio::try_join!(
    do_stuff_async(),
    more_async_work()
);

match res {
    Ok((first, second)) => {
        // делаем что-то со значениями
        println!("Обе операции завершились успешно");
    }
    Err(err) => {
        println!("Обработка завершилась ошибкой; ошибка = {}", err);
    }
}
}

Использование try_join! с порожденными задачами

#![allow(unused)]
fn main() {
use tokio::task::JoinHandle;

async fn do_stuff_async() -> Result<String, &'static str> {
    // асинхронная работа
    Ok("результат".to_string())
}

async fn more_async_work() -> Result<i32, &'static str> {
    // больше асинхронной работы
    Ok(42)
}

async fn flatten<T>(handle: JoinHandle<Result<T, &'static str>>) -> Result<T, &'static str> {
    match handle.await {
        Ok(Ok(result)) => Ok(result),
        Ok(Err(err)) => Err(err),
        Err(err) => Err("обработка задачи завершилась ошибкой"),
    }
}

let handle1 = tokio::spawn(do_stuff_async());
let handle2 = tokio::spawn(more_async_work());

match tokio::try_join!(flatten(handle1), flatten(handle2)) {
    Ok((text, number)) => {
        println!("Успех: текст = {}, число = {}", text, number);
    }
    Err(err) => {
        println!("Не удалось выполнить с ошибкой: {}.", err);
    }
}
}

Использование режима biased; для контроля порядка опроса

#![allow(unused)]
fn main() {
async fn do_stuff_async() -> Result<(), &'static str> {
    // асинхронная работа
    Ok(())
}

async fn more_async_work() -> Result<(), &'static str> {
    // больше асинхронной работы
    Ok(())
}

let res = tokio::try_join!(
    biased;
    do_stuff_async(),
    more_async_work()
);

match res {
    Ok((first, second)) => {
        // делаем что-то со значениями
    }
    Err(err) => {
        println!("обработка завершилась ошибкой; ошибка = {}", err);
    }
}
}

Практический пример с различными типами результатов

#![allow(unused)]
fn main() {
async fn fetch_user() -> Result<String, &'static str> {
    // имитация запроса к API
    tokio::time::sleep(tokio::time::Duration::from_millis(100)).await;
    Ok("Иван Иванов".to_string())
}

async fn fetch_balance() -> Result<f64, &'static str> {
    // имитация запроса к базе данных
    tokio::time::sleep(tokio::time::Duration::from_millis(150)).await;
    Ok(1000.50)
}

async fn validate_session() -> Result<bool, &'static str> {
    // проверка сессии
    Ok(true)
}

let result = tokio::try_join!(
    fetch_user(),
    fetch_balance(),
    validate_session()
);

match result {
    Ok((user, balance, is_valid)) => {
        println!("Пользователь: {}, Баланс: {:.2}, Сессия действительна: {}", 
                 user, balance, is_valid);
    }
    Err(err) => {
        eprintln!("Ошибка при загрузке данных: {}", err);
    }
}
}

Обработка ошибок с досрочным завершением

#![allow(unused)]
fn main() {
async fn process_step1() -> Result<String, &'static str> {
    // шаг 1 обработки
    Ok("шаг1 завершен".to_string())
}

async fn process_step2() -> Result<i32, &'static str> {
    // шаг 2 обработки - имитация ошибки
    Err("ошибка на шаге 2")
}

async fn process_step3() -> Result<bool, &'static str> {
    // этот шаг не будет выполнен из-за ошибки в step2
    Ok(true)
}

let result = tokio::try_join!(
    process_step1(),
    process_step2(),
    process_step3()
);

match result {
    Ok(_) => {
        println!("Все шаги завершены успешно");
    }
    Err(err) => {
        println!("Обработка прервана на шаге с ошибкой: {}", err);
        // step3 не будет выполнен из-за досрочного завершения
    }
}
}

Использование с разными типами ошибок

#![allow(unused)]
fn main() {
use std::convert::From;

// Функция для приведения разных типов ошибок к общему типу
fn map_error<E: ToString>(err: E) -> String {
    err.to_string()
}

async fn api_call() -> Result<String, reqwest::Error> {
    // вызов API
    Ok("данные API".to_string())
}

async fn db_query() -> Result<i32, sqlx::Error> {
    // запрос к базе данных
    Ok(42)
}

let result = tokio::try_join!(
    api_call().map_err(map_error),
    db_query().map_err(map_error)
);

match result {
    Ok((api_data, db_data)) => {
        println!("API данные: {}, DB данные: {}", api_data, db_data);
    }
    Err(err) => {
        eprintln!("Ошибка: {}", err);
    }
}
}