Csatornák (mpsc, oneshot)

Az előző leckében megtanultad, hogyan indíthatsz párhuzamos taskokat tokio::spawn-nal, és hogyan várd meg az eredményüket JoinHandle-lel. De mi van, ha két task nem csak egyszer ad vissza egy értéket, hanem folyamatosan kell beszélniük egymással? Erre valók a csatornák: a tokio::sync::mpsc és a tokio::sync::oneshot, amelyek biztonságos, ownership-alapú üzenetküldést tesznek lehetővé taskok között, megosztott, zárolt állapot nélkül.

Miért csatorna, és nem megosztott memória?

Ha két task ugyanazt az adatot módosítaná, a borrow checker azonnal beleszólna: két task nem tarthat egyszerre &mut T-t ugyanarra az adatra. A megoldás nem az, hogy "kikerüljük" ezt a szabályt, hanem az, hogy másképp gondolkodunk: az adatot elküldjük egyik taskból a másikba, ownership-átadással. Aki megkapja az üzenetet, az a kizárólagos tulajdonosa – nincs két hivatkozás, nincs versenyhelyzet.

Ez a gondolkodásmód sok más nyelvben is bevált ("ne osztott állapotot védj lakattal, hanem üzenetet küldj"), és a Rust ownership-modelljével különösen jól illik: a csatornán elküldött érték move-ol, a küldő fél elveszti hozzá a hozzáférést.

Az mpsc csatorna: több küldő, egy fogadó

A név a "multi-producer, single-consumer" rövidítése: sok task küldhet üzenetet, de csak egy task olvashatja azokat. A mpsc::channel::<T>(capacity) egy (Sender<T>, Receiver<T>) páros ad vissza. A capacity a csatorna belső pufferének mérete – ha betelik, a send(...).await addig vár, amíg a fogadó fel nem szabadít helyet. Ezt nevezzük backpressure-nek: a lassú fogadó automatikusan lefékezi a gyors küldőket.

A Sender<T> Clone, tehát több task is küldhet a csatornára a saját klónozott másolatával. A Receiver<T> viszont nem Clone – pontosan egy fogadó lehet, hiszen minden üzenetet csak egyszer szabad kiolvasni.

use tokio::sync::mpsc;

#[tokio::main]
async fn main() {
    // legfeljebb 8 elem várhat a csatornában, mielőtt a send blokkolna
    let (tx, mut rx) = mpsc::channel::<i32>(8);

    // a fogadó oldal: összegzi a beérkező számokat
    let receiver_task = tokio::spawn(async move {
        let mut sum = 0;
        while let Some(value) = rx.recv().await {
            sum += value;
        }
        sum
    });

    // három "termelő" task, mindegyik a saját klónozott Sender-ét használja
    for worker_id in 0..3 {
        let tx_clone = tx.clone();
        tokio::spawn(async move {
            for i in 1..=5 {
                tx_clone
                    .send(worker_id * 10 + i)
                    .await
                    .expect("a receiver task még él");
            }
        });
    }

    // fontos: az eredeti tx-et is el kell engedni, különben a csatorna sosem zár le
    drop(tx);

    let sum = receiver_task.await.expect("a receiver task ne paniceljen");
    println!("Az összeg: {sum}");
}

A rx.recv().await Option<T>-et ad vissza: Some(érték), amíg jön valami, és None, amikor az összes Sender (az eredeti és minden klón) elszállt. Ezért kellett a drop(tx) sor – amíg a fő taskban élt az eredeti Sender, a csatorna sosem zárt volna le, és a receiver_task örökre a recv().await-nél ragadt volna.

A oneshot csatorna: pontosan egy üzenet

A tokio::sync::oneshot egy sokkal egyszerűbb eszköz: pontosan egyetlen érték átadására való, pontosan egy küldő és egy fogadó között. Tipikus felhasználás a "kérés-válasz" minta: elindítasz egy taskot, és onnan pontosan egy választ vársz vissza.

use tokio::sync::oneshot;

async fn square_async(n: i32) -> i32 {
    n * n
}

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

    tokio::spawn(async move {
        let result = square_async(7).await;
        // a oneshot Sender::send NEM async: egyetlen értéket ad át, nincs mire várni
        let _ = tx.send(result);
    });

    match rx.await {
        Ok(value) => println!("A négyzet: {value}"),
        Err(_) => println!("A küldő task elszállt, mielőtt válaszolt volna"),
    }
}

Fontos különbség az mpsc-hez képest: a oneshot::Sender::send nem async függvény, mert nincs mire várnia – a csatornának pontosan egy "rekesze" van. Ha a küldő task elszáll (pl. panicol) anélkül, hogy send-elt volna, a Sender drop-olódik válasz nélkül, és a rx.await Err-t ad vissza. Ezért érdemes a fogadó oldalon match-csel kezelni ezt az esetet, nem csak expect-tel elhasalni.

Üzenetküldő minta: worker task kérés-válasz csatornával

A két csatornatípus kombinálásával egy nagyon hasznos mintát kapunk: egy "worker" task, amely egy mpsc csatornán fogad parancsokat, és minden parancshoz mellékelt oneshot::Sender-en válaszol. Ez az alapja sok komolyabb async architektúrának – egyetlen task "birtokolja" az állapotot, mások pedig üzenettel kérnek tőle adatot vagy módosítást, közös zárolás nélkül.

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

enum Command {
    Add {
        a: i32,
        b: i32,
        respond_to: oneshot::Sender<i32>,
    },
    Shutdown,
}

async fn worker(mut rx: mpsc::Receiver<Command>) {
    while let Some(command) = rx.recv().await {
        match command {
            Command::Add { a, b, respond_to } => {
                let _ = respond_to.send(a + b);
            }
            Command::Shutdown => break,
        }
    }
}

#[tokio::main]
async fn main() {
    let (tx, rx) = mpsc::channel::<Command>(8);
    tokio::spawn(worker(rx));

    let (respond_to, response) = oneshot::channel();
    tx.send(Command::Add { a: 3, b: 4, respond_to })
        .await
        .expect("a worker task fut");

    let result = response.await.expect("a worker válaszolt");
    println!("3 + 4 = {result}");

    tx.send(Command::Shutdown).await.expect("a worker task fut");
}

Ennek a mintának az a lényege, hogy a worker függvény zárt, önálló egységként birtokolja a saját állapotát (itt épp nincs is állapota, de simán lehetne egy struct mezőjében), és minden kommunikáció explicit üzeneteken keresztül zajlik. Nincs &mut megosztás, nincs zárolás – csak move-olt értékek egyik taskból a másikba.

Mit üzen a fordító, ha rossz irányba indulsz?

Gyakori kezdő hiba, hogy valaki a Receiver-t is Clone-oznia szeretné, mert megszokta, hogy a Sender az. Nézzük meg, mit mond ehhez a fordító:

use tokio::sync::mpsc;

#[tokio::main]
async fn main() {
    let (_tx, rx) = mpsc::channel::<i32>(4);
    let _rx2 = rx.clone();
}

error[E0599]: no method named `clone` found for struct `tokio::sync::mpsc::Receiver<i32>` in the current scope

Ez nem véletlen: minden üzenetet pontosan egy fogadónak kell megkapnia, különben a rendszer nem tudná eldönteni, melyik klón kapja meg melyik elemet. Ha valóban több helyen szeretnél reagálni egy-egy eseményre, azt más eszközzel (pl. több különálló csatornával, vagy egy dedikált "broadcast" jellegű megoldással) kell megoldani – ezekre egy későbbi, haladóbb leckében térünk majd vissza, itt elég annyit rögzíteni, hogy az mpsc Receiver szigorúan egyszemélyes.

Próbáld ki!

Vegyél elő a worker/Command mintát, és bővítsd egy Multiply { a: i32, b: i32, respond_to: oneshot::Sender<i32> } variánssal. A main-ben küldj el egymás után két Add és egy Multiply kérést, mindegyikhez külön oneshot::channel()-t létrehozva, majd írd ki mindhárom választ. Figyelj arra, hogy a Shutdown parancs a végén menjen, ne előbb – különben a worker leáll, mielőtt a többi kérést feldolgozná.

Önellenőrzés

1. Miért lehet klónozni egy mpsc::Sender-t, de nem lehet egy mpsc::Receiver-t?

Megoldás

Mert bármennyi task küldhet üzenetet ugyanarra a csatornára – ez nem okoz konfliktust, mindegyik `send` csak új elemet tesz a sorba. A fogadó oldalon viszont minden üzenetet pontosan egy helyen szabad kiolvasni; ha több `Receiver` versenyezne ugyanazokért az elemekért, nem lenne egyértelmű, melyik kapja meg melyiket, ezért a Rust API-ja tervezésileg kizárja ezt: csak egy `Receiver` létezhet egy csatornához.

2. Mi történik, ha egy oneshot::Receiver-en .await-elsz, de a hozzá tartozó Sender úgy szállt el (pl. panicolt a taskja), hogy közben nem hívta meg a send-et?

Megoldás

A `rx.await` `Err`-t ad vissza (a `RecvError` típus egy értékét). A `oneshot::Receiver` `Future`-ként viselkedik, és amikor a `Sender` elszáll `send` hívása nélkül, a csatorna "lezáródik" válasz nélkül – ezt a `Future` egy hibaértékkel jelzi, nem panickal. Éppen ezért érdemes `match`-csel vagy `?`-tel kezelni, nem csak vakon `unwrap`-elni.

3. Az mpsc-s példában miért kellett drop(tx)-et hívni a main-ben, ha egyébként a workerek is a saját klónozott Sender-jeiket használják?

Megoldás

A `recv().await` csak akkor ad vissza `None`-t, ha **minden** `Sender` klón (az eredetivel együtt) elszállt vagy explicit `drop`-olódott. Ha a `main` megtartja az eredeti `tx`-et, akkor – bár a workerek befejezték a küldést és a taskjaik véget értek – a csatorna szemszögéből még mindig van egy élő `Sender`, így a `receiver_task` örökre a `recv().await`-nél várna, sosem térne vissza.

Összefoglalás és mi jön legközelebb

• Az mpsc csatorna sok küldőt (Sender, Clone) és egyetlen fogadót (Receiver, nem Clone) enged meg, korlátozott kapacitással és beépített backpressure-rel.
• A oneshot csatorna pontosan egy üzenet átvitelére való, tipikusan kérés-válasz mintában – a send itt nem async, mert nincs mire várni.
• A csatorna "lezárt" állapota (fogadó oldalon None, oneshot esetén Err) akkor jön létre, amikor az összes küldő fél elszállt vagy drop-olódott.
• Az mpsc és a oneshot kombinálásával egy önálló "worker" task köré üzenetküldő (parancs-válasz) architektúrát lehet építeni, megosztott, zárolt állapot nélkül.
• Legközelebb elkezdjük az Axum webkeretrendszert: a "Router és handler-ek" leckében megnézzük, hogyan regisztrálunk útvonalakat, és hogyan írunk függvényeket, amelyek HTTP-kéréseket dolgoznak fel.

További olvasmány:

• tokio.rs – Channels tutorial
• docs.rs – tokio::sync::mpsc
• docs.rs – tokio::sync::oneshot