Csatornák Tokio-val: mpsc, oneshot és broadcast a gyakorlatban

Miért kellenek csatornák konkurrens Rust kódban

Amikor több tokio::spawn-nal indított task fut egymás mellett, előbb-utóbb kommunikálniuk kell. A megosztott állapot klasszikus megközelítése a Mutex<T> vagy RwLock<T>, de ez könnyen lock contention-höz és nehezen olvasható kódhoz vezet, ha sok task ír-olvas egyszerre.

A csatorna (channel) egy más filozófia: az adat átmegy egyik taskból a másikba, nincs megosztott mutábilis állapot. Ez pontosan az ownership modellbe illik — a küldő elveszti a tulajdonjogot az üzenet felett, a fogadó megkapja. Nincs lock, nincs data race, és a borrow checker is nyugton marad.

A Tokio a tokio::sync modulban ad néhány kiváló csatornatípust:

• mpsc – multiple producer, single consumer
• oneshot – egyetlen érték egyszeri elküldésére
• broadcast – multiple producer, multiple consumer, mindenki megkapja az összes üzenetet
• watch – állapotváltozás jelzésére (ezt most nem részletezzük)

tokio::sync::mpsc: több küldő, egy fogadó

Az mpsc csatorna a leggyakrabban használt eszköz task-ok közötti kommunikációra. Létrehozásakor kapunk egy Sender<T>-t és egy Receiver<T>-t. A Sender klónozható — innen a "multiple producer" —, de csak egyetlen Receiver létezhet.

use tokio::sync::mpsc;

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

    for i in 0..3 {
        let tx = tx.clone();
        tokio::spawn(async move {
            let msg = format!("üzenet a {i}. worker-től");
            if tx.send(msg).await.is_err() {
                eprintln!("a fogadó már nem érdeklődik");
            }
        });
    }

    // Az eredeti tx-et is le kell dobni, különben a csatorna
    // sosem záródik le, és a ciklus alul örökre várna.
    drop(tx);

    while let Some(msg) = rx.recv().await {
        println!("kapott: {msg}");
    }
}

Fontos részlet: a channel(32) egy bounded csatornát hoz létre, 32-es puffermérettel. Ha a puffer betelik, a send().await blokkol (pontosabban: a task felfüggesztődik), amíg a fogadó nem ürít belőle. Ez a backpressure mechanizmus lényege, amiről lentebb részletesen is szó lesz.

oneshot csatornák egyszeri válaszokhoz

Van, amikor csak egyetlen értéket kell átadni egyik taskból a másikba — tipikusan egy worker eredményének visszaküldésekor. Erre pazarlás lenne egy egész mpsc csatornát nyitni; a tokio::sync::oneshot pontosan erre a célra készült.

use tokio::sync::oneshot;

async fn heavy_computation(input: u64) -> u64 {
    input * input
}

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

    tokio::spawn(async move {
        let result = heavy_computation(21).await;
        // Ha itt hiba lenne (pl. Err(_) a send-en), az azt jelenti,
        // hogy a másik oldal (rx) már el lett dobva.
        let _ = tx.send(result);
    });

    match rx.await {
        Ok(value) => println!("az eredmény: {value}"),
        Err(_) => println!("a worker leállt válasz nélkül"),
    }
}

A oneshot::Sender::send nem async — nem lehet blokkolni rajta, egyszerűen elteszi az értéket, vagy hibát ad, ha a Receiver már megszűnt. A Receiver maga egy Future, amit .await-elhetsz.

Egy klasszikus mintázat: egy task elküld egy kérést egy mpsc csatornán, és a kéréssel együtt egy oneshot::Sender-t is átad, amin a válasz visszaérkezik. Ezt "request-response" mintának hívják, és gyakran így néz ki a kérés típusa:

enum Command {
    Compute {
        input: u64,
        responder: tokio::sync::oneshot::Sender<u64>,
    },
}

A worker task feldolgozza a Command-ot, majd a responder.send(...)-en visszaküldi az eredményt a hívónak — miközben a Command-okat egy közös mpsc::Receiver<Command>-en fogadja több küldőtől.

broadcast csatornák több fogyasztónak

Ha az üzenetet nem egy, hanem több fogadónak kell megkapnia — például egy chat szerverben minden csatlakozott kliensnek el kell jutnia egy üzenetnek —, a tokio::sync::broadcast a megoldás.

use tokio::sync::broadcast;

#[tokio::main]
async fn main() {
    let (tx, _rx) = broadcast::channel::<String>(16);

    let mut rx1 = tx.subscribe();
    let mut rx2 = tx.subscribe();

    tokio::spawn(async move {
        while let Ok(msg) = rx1.recv().await {
            println!("olvasó 1: {msg}");
        }
    });

    tokio::spawn(async move {
        while let Ok(msg) = rx2.recv().await {
            println!("olvasó 2: {msg}");
        }
    });

    tx.send("szerver újraindul 5 perc múlva".to_string()).unwrap();
    tx.send("mentsd el a munkádat!".to_string()).unwrap();

    // Adjunk időt a task-oknak, hogy feldolgozzák.
    tokio::time::sleep(std::time::Duration::from_millis(50)).await;
}

Minden subscribe()-bal létrehozott Receiver a saját olvasási pozícióját tartja a belső ringbufferben. Ha egy fogadó lassabb, mint amilyen sebességgel az üzenetek érkeznek, és a buffer betelik, a régi üzenetek felülíródnak — ekkor a recv() egy RecvError::Lagged(n)-t ad vissza, jelezve, hogy n üzenetet kihagyott a fogadó.

Backpressure és bounded vs unbounded csatornák

A mpsc::channel(n) bounded csatornát ad, ahol n a puffer mérete. Ez a backpressure alapja: ha a fogyasztó nem tud lépést tartani a termelőkkel, a send().await egyszerűen vár, amíg hely nem lesz a pufferben. Ez visszahat a küldő oldalra, és lassítja azt — pontosan úgy, ahogy egy valós rendszerben a nyomás visszafelé is terjed egy csővezetékben.

Van egy mpsc::unbounded_channel() variáns is, ami sosem blokkol küldéskor — a puffer korlátlanul nő. Ez csábító lehet, de veszélyes:

use tokio::sync::mpsc;

#[tokio::main]
async fn main() {
    let (tx, mut rx) = mpsc::unbounded_channel::<u64>();

    tokio::spawn(async move {
        for i in 0.. {
            // Ha a fogadó lassú, ez itt korlátlanul memóriát fogyaszt!
            if tx.send(i).is_err() {
                break;
            }
        }
    });

    // Csak néhányat olvasunk ki, a többi felhalmozódik a memóriában.
    for _ in 0..5 {
        if let Some(v) = rx.recv().await {
            println!("kapott: {v}");
        }
    }
}

A megfelelő puffermélység kiválasztása empirikus feladat: túl kicsi puffer sok felesleges felfüggesztést okoz, túl nagy pedig elfedheti a valós teljesítményproblémákat. Kezdésnek egy 16-128 közötti érték legtöbbször jó kiindulópont, aztán méréssel finomíts rajta.

Kis producer-consumer alkalmazás megépítése

Rakjunk össze egy egyszerű, de valósághű rendszert: több "termelő" task számokat generál, egy "aggregátor" task összegyűjti és összesíti őket, majd egy oneshot-on visszaküldi a végeredményt annak, aki elindította a folyamatot.

use tokio::sync::{mpsc, oneshot};

#[derive(Debug)]
enum WorkerMsg {
    Value(u64),
    Done,
}

async fn producer(id: u32, tx: mpsc::Sender<WorkerMsg>) {
    for i in 1..=5 {
        let value = (id as u64) * 100 + i as u64;
        // Ha a csatorna betelt, itt várunk – ez a backpressure.
        if tx.send(WorkerMsg::Value(value)).await.is_err() {
            return;
        }
    }
}

async fn aggregator(
    mut rx: mpsc::Receiver<WorkerMsg>,
    result_tx: oneshot::Sender<u64>,
    expected_done_count: u32,
) {
    let mut sum = 0u64;
    let mut done_count = 0u32;

    while let Some(msg) = rx.recv().await {
        match msg {
            WorkerMsg::Value(v) => sum += v,
            WorkerMsg::Done => {
                done_count += 1;
                if done_count >= expected_done_count {
                    break;
                }
            }
        }
    }

    let _ = result_tx.send(sum);
}

#[tokio::main]
async fn main() {
    let worker_count = 3;
    let (tx, rx) = mpsc::channel::<WorkerMsg>(16);
    let (result_tx, result_rx) = oneshot::channel::<u64>();

    tokio::spawn(aggregator(rx, result_tx, worker_count));

    for id in 0..worker_count {
        let tx = tx.clone();
        tokio::spawn(async move {
            producer(id, tx.clone()).await;
            let _ = tx.send(WorkerMsg::Done).await;
        });
    }

    drop(tx);

    match result_rx.await {
        Ok(sum) => println!("végösszeg: {sum}"),
        Err(_) => println!("az aggregátor leállt válasz nélkül"),
    }
}

Ebben a példában három fontos mintázat találkozik:

1. mpsc hordozza az egyedi értékeket és a "kész vagyok" jelzést egyaránt egy enum-ba csomagolva.
2. Az aggregátor egy explicit Done számláló alapján dönti el, mikor zárja le a feldolgozást — ez sokkal megbízhatóbb, mint a Sender-ek dropolására várni, mert itt egyértelmű, üzenetalapú szemantikát kapunk.
3. oneshot szállítja vissza a végeredményt a main-nek — pontosan egyszer, pontosan egy értékkel.

Összefoglalás

A Tokio csatornái nem véletlenül a legtöbbet használt konkurrenciai eszközök async Rust-ban: az mpsc a munkaelosztás és aggregáció alapja, a oneshot a pontos, egyszeri válaszok tiszta módja, a broadcast pedig a több fogyasztóra szóló értesítéseket oldja meg elegánsan. A bounded csatornák backpressure-je pedig nem korlátozás, hanem védőháló — segít abban, hogy a rendszered ne fusson szét memóriafogyasztásban egy hirtelen terheléscsúcs alatt.

Ha most kezdesz Tokio-val dolgozni, érdemes a legtöbb esetben bounded mpsc-ből indulni, és csak akkor nyúlni broadcast-hoz vagy unbounded-hez, ha a konkrét probléma valóban azt igényli. A csatornák a Rust ownership-modelljéhez illő, biztonságos módot adnak arra, hogy konkurrens taskok kommunikáljanak — lock-olás és data race-ek nélkül.