Блог*

#prog #rust #rustasync

Хозяйке на заметку

(или "Антон читает за вас документацию tokio")

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

Базовый пример использования:

use tokio::task::JoinSet;

#[tokio::main]
async fn main() {
    let mut set = JoinSet::new();

    for i in 0..10 {
        // таски спавнятся при помощи spawn, поэтому
        // семантика та же: сразу начинают выполняться
        // и паникует вне асинхронного контекста tokio
        set.spawn(async move { i });
    }

    let mut seen = [false; 10];
    // мы не обязаны получать все таски до конца
    while let Some(res) = set.join_next().await {
        let idx = res.unwrap();
        seen[idx] = true;
    }

    for i in 0..10 {
        assert!(seen[i]);
    }
}

Также можно отменить скопом все таски в наборе через abort_all. Также можно переиспользовать JoinSet, используя detach_all. Это уберёт все таски из JoinSet, но они продолжат исполняться рантаймом.

2. Если у нас есть несколько тасок, которые отсылают результаты своей работы через канал, и мы хотим ограничить число одновременно исполняемых тасок, то это можно сделать при помощи канала ограниченной ёмкости и API, описанного в статье Mutex without lock, Queue without push: cancel safety in lilos (о которой я уже рассказывал):

let (tx, mut rx) = tokio::sync::mpsc::channel(limit);

for _ in 0..n_tasks {
    let work = async {
        // какая-то полезная работа
    };
    let tx = tx.clone();
    let task = async move {
        let send_permit = tx.reserve_owned().unwrap();
        let result = work.await;
        let _ = send_permit.send(result);
    };
    tokio::spawn(task);
}

// дропаем свою копию Sender-а, чтобы не заблокироваться
// на чтении из канала, в который никто не пишет
drop(tx);
while let Some(res) = rx.recv().await {
    // обрабатываем результат
}

И надо предупредить (хотя навряд ли вы в такое вляпаетесь намеренно): очевидно, если Permit-ов утечёт достаточно много, вы можете оказаться в дурацкой ситуации, когда в канале есть место, но он не работает ни за отправку, ни на получение.

docs.rs

JoinSet in tokio::task - Rust

A collection of tasks spawned on a Tokio runtime.

👍3

926 views19:37