Mi az a strukturált konkurrencia, és miért érdekel minket?

Ha eddig dolgoztál már többszálú vagy aszinkron kóddal, valószínűleg találkoztál azzal a problémával, hogy egy elindított feladat "elszabadul": a hívó függvény visszatér, a task viszont valahol a háttérben tovább él, esetleg soha nem fejeződik be, vagy pont akkor dob hibát, amikor senki sem figyel rá. Ez a klasszikus "unstructured concurrency".

A strukturált konkurrencia lényege, hogy a konkurrens feladatok életciklusa mindig egy jól definiált scope-hoz (blokkhoz, függváhíváshoz) van kötve. Ha a szülő "kilép", a gyerek feladatoknak is be kell fejeződniük — vagy explicit módon megszakítjuk őket, vagy megvárjuk őket. Nincs elárvult task, nincs meglepetés.

A Tokio ökoszisztémában három eszköz adja a strukturált konkurrencia gerincét:

  • tokio::select! — több async művelet közül az elsőre reagálunk, a többit eldobjuk
  • tokio::join! — több async műveletet egyszerre futtatunk, és mindegyikre megvárjuk az eredményt
  • tokio::task::JoinSet — dinamikus számú task kezelése egyetlen struktúrában

Ezek mindegyike megőrzi azt a garanciát, hogy a konkurrens munka nem szökik ki a hívó kód hatóköréből.

ℹ
Megjegyzés

A cikkben feltételezem, hogy a Cargo.toml-ban van egy tokio = { version = "1", features = ["full", "macros", "rt-multi-thread"] } sor. A #[tokio::main] makróval indítjuk az async runtime-ot.

A select! makró: verseny az események között

A select! makró akkor jön képbe, amikor több async műveletet indítasz el, de valójában csak az elsőre vagy vagy egy timeoutra vagy egy megszakítási jelre vagy... szóval amikor verseny van a lehetőségek között, és az első győztes dönt.

Nézzünk egy tipikus mintát: egy hálózati kérést szeretnénk kiváltani, de közben figyeljük egy leállítási jelzést is.

use tokio::sync::oneshot;
use tokio::time::{sleep, Duration};

async fn fetch_data() -> String {
    sleep(Duration::from_secs(2)).await;
    "kész adat".to_string()
}

#[tokio::main]
async fn main() {
    let (tx, rx) = oneshot::channel::<()>();

    // Egy kis idő után jelezzük a leállítást
    tokio::spawn(async move {
        sleep(Duration::from_millis(500)).await;
        let _ = tx.send(());
    });

    tokio::select! {
        data = fetch_data() => {
            println!("Megjött az adat: {data}");
        }
        _ = rx => {
            println!("Leállítási jelzés érkezett, megszakítjuk a lekérést!");
        }
    }

    println!("A select! után folytatódik a főprogram.");
}

Ami itt fontos: a select! a nem győztes branch future-jét drop-olja. Ha a fetch_data() future belül tartalmaz erőforrásokat (pl. nyitott socket), azok a Drop implementáció révén automatikusan felszabadulnak. Ez a strukturált konkurrencia egyik alappillére — nincs "elfelejtett" munka.

⚠
Figyelem

A select! ágakban használt future-öknek Unpin-nek vagy pin-elhetőnek kell lenniük, és figyelj arra, hogy a nem Copy értékek (pl. csatornák) move kulcsszóval kerüljenek be a closure-ökbe, különben borrow checker hibát kapsz.

A select! továbbá támogat feltételes ágakat is a if guard szintaxissal, ami nagyon hasznos state machine-szerű logikák megírásához:

use tokio::time::{interval, Duration};

#[tokio::main]
async fn main() {
    let mut tick = interval(Duration::from_millis(300));
    let mut counter = 0;

    loop {
        tokio::select! {
            _ = tick.tick(), if counter < 5 => {
                counter += 1;
                println!("Tick #{counter}");
            }
            else => {
                println!("Kész, kilépünk a ciklusból.");
                break;
            }
        }
    }
}

A join! makró: párhuzamos futás, közös várakozás

Amíg a select! versenyt jelent, a join! egyidejű futtatást jelent — mindegyik future-t elindítjuk, és megvárjuk, amíg mindegyik befejeződik. Ez ideális, ha több független, de egyaránt szükséges műveletet kell elvégezni, mielőtt folytatnánk.

use tokio::time::{sleep, Duration};

async fn get_user_profile() -> String {
    sleep(Duration::from_millis(400)).await;
    "profil adatok".to_string()
}

async fn get_user_orders() -> Vec<u32> {
    sleep(Duration::from_millis(600)).await;
    vec![101, 102, 103]
}

#[tokio::main]
async fn main() {
    let (profile, orders) = tokio::join!(
        get_user_profile(),
        get_user_orders()
    );

    println!("Profil: {profile}");
    println!("Rendelések: {orders:?}");
}

Fontos különbség a spawn-hoz képest: a join!-nal indított future-ök ugyanazon a task-on futnak, egyetlen szálon váltogatva egymást minden .await pontnál. Ez kisebb overhead-et jelent, mint a tokio::spawn, de nem ad valódi párhuzamosságot több CPU magon — csak konkurrenciát egy szálon belül. Ha valódi multi-thread párhuzamosságra van szükséged (pl. CPU-intenzív munkára), akkor tokio::spawn-t és JoinHandle-öket kell kombinálnod.

✦
Tipp

Ha valamelyik join!-ba tett future hibázhat (pl. Result-ot ad vissza), célszerű a ? operátort a join! blokkon kívül, minden egyes résznél külön kezelni, mert a join! maga nem "short-circuitel" hibára — mindegyik future lefut a végéig.

JoinSet: amikor a task-ok száma futásidőben derül ki

A join! és a select! remekül működik fix számú, névvel ismert feladatnál. De mi van, ha nem tudod fordítási időben, hány task-ot kell elindítani? Például egy lista minden elemére indítasz egy lekérést, és a lista mérete csak futásidőben ismert.

Itt jön képbe a tokio::task::JoinSet. Ez egy olyan gyűjtemény, amelybe dinamikusan spawn-olhatsz task-okat, és amely garantálja, hogy amikor a JoinSet elhagyja a scope-ot (drop-olódik), az összes benne lévő, még futó task megszakításra kerül. Pontosan ez a strukturált konkurrencia lényege: a gyerek task-ok élettartama a szülő struktúrához van kötve.

use tokio::task::JoinSet;
use tokio::time::{sleep, Duration};

async fn process_item(id: u32) -> u32 {
    sleep(Duration::from_millis(100 * id as u64)).await;
    id * id
}

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

    for id in 1..=5 {
        set.spawn(process_item(id));
    }

    let mut osszeg = 0;

    while let Some(res) = set.join_next().await {
        match res {
            Ok(value) => {
                println!("Egy task lefutott, eredmény: {value}");
                osszeg += value;
            }
            Err(e) => {
                eprintln!("Egy task elszállt (panic vagy megszakítás): {e}");
            }
        }
    }

    println!("Összesített eredmény: {osszeg}");
}

A join_next() metódus mindig a legkorábban befejeződő task eredményét adja vissza, tehát nem a spawn sorrendjében, hanem a tényleges befejezési sorrendben kapod meg az eredményeket. Ez sok gyakorlati esetben pontosan az, amire szükség van — például amikor egyszerre több szervertől kérünk le adatot, és ahogy megérkezik egy válasz, azonnal feldolgozzuk.

✓
Jó tudni

A JoinSet Drop implementációja automatikusan abort()-olja az összes benne lévő, még be nem fejezett task-ot. Ez az egyik legfontosabb biztonsági háló strukturált konkurrenciában: ha kilépsz egy függvényből (akár hiba, akár return miatt), a benne létrehozott JoinSet gondoskodik a takarításról.

Feladatok megszakítása és erőforrás-felszabadítás

A strukturált konkurrencia egyik alapkérdése: mi történik, ha egy task-ot meg kell szakítani félúton? Tokio-ban ehhez a JoinHandle::abort() metódus áll rendelkezésre, amelyet a JoinSet is belsőleg használ.

use tokio::time::{sleep, Duration};

#[tokio::main]
async fn main() {
    let handle = tokio::spawn(async {
        loop {
            println!("Dolgozom...");
            sleep(Duration::from_millis(200)).await;
        }
    });

    sleep(Duration::from_millis(700)).await;
    handle.abort();

    match handle.await {
        Ok(_) => println!("A task rendesen befejeződött."),
        Err(e) if e.is_cancelled() => println!("A task megszakításra került."),
        Err(e) => println!("Egyéb hiba: {e}"),
    }
}

Az abort() a task futásának a legközelebbi .await pontján szakítja meg — nem szakítja meg szinkron, blokkoló kódot azonnal. Ezért fontos, hogy CPU-intenzív, hosszú szinkron blokkokat kerüljünk async függvényekben, mert azok nem reagálnak az abort()-ra időben.

Az erőforrás-felszabadítás szempontjából jó hír, hogy a megszakított task future-je is Drop-olódik, tehát a benne allokált struktúrák (fájlkezelők, mutex guard-ok, csatornák) rendesen felszabadulnak — feltéve, hogy helyesen implementálják a Drop-ot, vagy RAII mintát követnek.

⚠
Figyelem

Az abort() nem garantálja, hogy a task azonnal leáll — csak azt, hogy legkésőbb a következő .await-nél megszakad. Ha kritikus, hogy egy erőforrás egy adott pillanatban biztosan felszabaduljon, kombináld a select!-et egy explicit leállítási csatornával (pl. tokio::sync::watch vagy oneshot), és a task belsejében rendszeresen ellenőrizd azt.

Összefoglalás

A strukturált konkurrencia nem egy konkrét API, hanem egy gondolkodásmód: minden konkurrens munkának legyen egyértelmű "tulajdonosa", és amikor a tulajdonos scope-ja lezárul, a munka is befejeződjön vagy megszakadjon. Tokio-ban ezt háromféleképpen érhetjük el kényelmesen:

  • select! — versengő eseményekre reagálunk, a veszteseket eldobjuk
  • join! — egyszerre futtatunk több, egymástól független future-t egy task-on belül
  • JoinSet — dinamikusan spawn-olt task-ok csoportját kezeljük, automatikus takarítással

Ezek kombinálásával olyan async kódot írhatsz, amiben nincs "elárvult" háttérfolyamat, nincs erőforrás-szivárgás, és a program viselkedése jóval kiszámíthatóbb marad — még bonyolult, sok konkurrens ágból álló rendszerekben is.