Korlátos kapacitás és visszanyomás

Cél és előfeltétel

A csatornák tulajdonátadása után felmerül a kérdés: mi történik, ha a termelő gyorsabban készít adatot, mint ahogy a fogadó dolgozik? Korlátlan puffer mellett az üzenetek felhalmozódhatnak. A sync_channel véges számú üzenetet tárol, és a blokkoló küldés szükség esetén vár. A cél az ilyen visszanyomás megértése, időmérés és kitalált ütemezési feltételezések nélkül. A programok kizárólag beágyazott adatokon futnak.

Mit korlátoz a kapacitás?

A kapacitás a csatornapuffer üzenetszámát határozza meg. Nem az üzenetek teljes bájtméretét, nem az előzetesen betöltött bemenetet, és nem a már dolgozó munkások számát. Egyetlen üzenet lehet nagy vektor is. Ha valódi memóriakorlát a cél, az üzenetek méretét, a termelés helyét és a feldolgozás közben tartott adatokat együtt kell vizsgálni. A véges puffer hasznos építőelem, de önmagában nem teljes erőforrásterv.

A blokkoló send helyet várhat a pufferben. Ez akkor működik, ha közben egy másik szereplő képes fogadni. Ha a főszál előbb megvárná a telt csatornára küldő munkás végét, és csak utána kezdene fogadni, körkörös várakozást alakíthatna ki. A helyes életciklusban a fogadás már folyamatban van, amikor a küldőnek szüksége lehet rá. A feladatokban ezért a fogadást a küldő join hívása elé tesszük.

A try_send nem vár a felszabaduló helyre. Telt puffer esetén Full, megszűnt fogadónál Disconnected hibát ad, és mindkettő megőrzi az átadni próbált értéket. A két eset más döntést kívánhat. A telt puffer átmeneti torlódás, a megszűnt fogadó viszont azt jelzi, hogy ezen a csatornán már nincs, aki átvegye az adatot. A hiba figyelmen kívül hagyása mindkét esetben elveszítené a feladat szándékát.

1. feladat: telt és lezárt csatorna

Egyetlen főszálon hozz létre egyhelyes puffert. Az első üzenet férjen be, a második visszaadott Full érték legyen. Miután átvetted az elsőt, szüntesd meg a fogadót, és ellenőrizd a következő küldés külön hibáját. Mivel a teszt alatt nincs párhuzamos fogadó, a telt állapot nem véletlen időzítés eredménye.

use std::sync::mpsc::{sync_channel, TrySendError};
fn main() {
    let (tx, rx) = sync_channel(1);
    tx.try_send("első").unwrap();
    assert!(matches!(tx.try_send("második"), Err(TrySendError::Full("második"))));
    assert_eq!(rx.recv().unwrap(), "első");
    drop(rx);
    assert!(matches!(tx.try_send("harmadik"),
        Err(TrySendError::Disconnected("harmadik"))));
}

A visszanyomás legyen viselkedési döntés

A Full láttán nem kötelező végtelenül újrapróbálkozni. Az alkalmazás dönthet várakozásról, az új munka elutasításáról vagy egy dokumentált összevonási szabályról. Fontos, hogy a döntés látható legyen a hívónak. Egy szoros, állandóan újrapróbáló ciklus fölösleges processzorterhelést okozhat, miközben a fogadónak nem ad több időt vagy jobb leállítási szerződést. Ebben a leckében a normál adatút blokkoló küldést használ, és közben tényleges fogadót működtet.

2. feladat: egyhelyes adatfolyam lezárása

Egy munkaszál továbbítson több számot az egyhelyes csatornán. A főszál addig fogadjon, amíg a küldő megszűnik, és csak utána várja meg a szálat. A vektorba gyűjtés itt tesztcélú végpont, ezért a teljes eredmény memóriában marad. Az üres bemenetnél a küldő üzenet nélkül zárul, a fogadó pedig szabályosan üres listát ad.

use std::{sync::mpsc, thread};
fn forward(values: Vec<u32>) -> Vec<u32> {
    let (tx, rx) = mpsc::sync_channel(1);
    let handle = thread::spawn(move || {
        for value in values { tx.send(value)?; }
        Ok::<(), mpsc::SendError<u32>>(())
    });
    let received: Vec<_> = rx.into_iter().collect();
    handle.join().unwrap().unwrap();
    received
}
fn main() {
    assert_eq!(forward(vec![3, 1, 4]), vec![3, 1, 4]);
    assert_eq!(forward(vec![0]), vec![0]);
    assert!(forward(vec![]).is_empty());
}

Nulla kapacitás: közvetlen találkozás

A nulla kapacitás nem használhatatlan vagy automatikusan hibás csatorna. A küldő és a fogadó közvetlen átadásnál találkozik: a blokkoló küldés csak az átvételhez kapcsolódva fejeződik be. Ettől a fogadó további üzleti feldolgozása még nem lesz kész. Ha például a kapott szöveget ezután számolná át, ahhoz továbbra is külön befejezési visszajelzés kellene.

3. feladat: nulla pufferű szövegátadás

Készíts közvetlen átadást tulajdonolt szöveggel. A fogadás előzze meg a küldő megvárását. Az eredményt és a küldés sikerét is ellenőrizd, és az üres szöveget érvényes üzenetként kezeld. Ne próbálj abból következtetni a helyességre, hogy a két szál valamelyike szemmel láthatóan előbb indult vagy előbb írt volna a terminálra.

use std::{sync::mpsc, thread};
fn handoff(text: String) -> String {
    let (tx, rx) = mpsc::sync_channel(0);
    let handle = thread::spawn(move || tx.send(text));
    let received = rx.recv().unwrap();
    handle.join().unwrap().unwrap();
    received
}
fn main() {
    assert_eq!(handoff(String::from("átadás")), "átadás");
    assert_eq!(handoff(String::new()), "");
}

A csatorna lezárásához a kapacitástól függetlenül minden küldő élettartama számít. A második példában a küldő kizárólag a munkaszálhoz kerül, és a lezárás visszatérésekor megszűnik. Emiatt az üres bemenethez sem kell külön végjelző üzenet. Más protokollban lehet helye explicit végjelzőnek, de akkor azt is tisztázni kell, hogy több termelő közül melyik küldheti, és történhet-e utána további adatküldés. Egyetlen, világos lezárási szabály könnyebben ellenőrizhető, mint a végpontok megszűnésének és több különleges üzenetnek a keveréke.

Elfogadási feltételek

1. A Full és a Disconnected két külön ellenőrzött kimenet, az elutasított érték mindkettőben megmarad.
2. A blokkoló küldő mellett tényleges fogadás folyik; minden elindított szál befejezését megvárjuk.
3. A nulla kapacitású változat és az üres bemenet is végesen lezárul, alvás vagy időzítési állítás nélkül.

Továbblépés

Az összesítő miniprojektben már külön korlátot adunk a munkásszámnak és az eredménycsatorna kapacitásának. A két korlát eltérő erőforrást szabályoz, ezért a dokumentációban és a tesztekben is külön nevezzük meg őket.