Cél és előfeltétel
Az Arc és Mutex megosztott állapotot szervezett. Más helyzetekben egyszerűbb, ha a munkások saját értéket készítenek, majd azt egy fogadó szálnak adják át. A standard könyvtár mpsc csatornája több küldőt és egy fogadót támogat. A cél most a tulajdon útjának és a leállás feltételének megértése. A példák nem hálózati üzeneteket küldenek, és nem ígérnek tartós tárolást vagy másik folyamatot elérő kézbesítést.
Az érték nem marad két módosító gazdánál
A send tulajdonba veszi az üzenetet. Egy String átadása után ugyanazt az értéket a küldő nem használhatja tovább sajátjaként. A fogadó a sikeres fogadáskor átveszi. Ezzel elkerülhető, hogy mindkét szál közösen módosítson egyetlen szövegpuffert. Másolható értéknél, például egy kis egész számnál, másolás történhet; a szerződést az üzenet típusa határozza meg.
Az alap channel fogalmilag korlátlan pufferű csatornát készít. A küldés nem arra vár, hogy a fogadó feldolgozza az üzenetet. Ez kényelmes, de nem korlátozza a felhalmozódó adatok számát. A következő leckében ezért külön foglalkozunk a korlátos változattal. Itt még azt tanuljuk meg, hogy az üzenetek és a végpontok tulajdona hogyan hat a program befejeződésére.
A fogadó iterátora akkor ér véget, ha minden küldő megszűnt, és a pufferből is elfogytak az üzenetek. Nem elegendő, hogy az egyik munkás végzett. Egy az indító szálon felejtett küldőklón életben tarthatja a csatornát, a fogadó pedig tovább várhat. Ezért az összes klón helyét és élettartamát ugyanúgy át kell gondolni, mint korábban a JoinHandle értékekét.
1. feladat: egy szöveg átadása
Indíts szálat egy tulajdonolt szöveggel, küldd át, majd a főszálon vedd át. A munkaszálat ezután várd meg. Az üres szöveg is érvényes üzenet: ne keverd össze a csatorna végével. A függvény visszatérési értékét ellenőrizzük, nem az üzenet feltételezett érkezési idejét vagy a terminálra írt sorokat.
Megoldás
use std::{sync::mpsc, thread};
fn transfer(text: String) -> String {
let (tx, rx) = mpsc::channel();
let handle = thread::spawn(move || tx.send(text));
let received = rx.recv().unwrap();
handle.join().unwrap().unwrap();
received
}
fn main() {
assert_eq!(transfer(String::from("árvíz")), "árvíz");
assert_eq!(transfer(String::new()), "");
}A sikeres küldés még nem kész feldolgozás
A send sikere nem jelenti, hogy a fogadó üzleti művelete már befejeződött. A fogadó később meg is szűnhet az üzenet feldolgozása előtt. Ha a küldőnek tényleges befejezési visszajelzés kell, azt külön protokollal, például válaszüzenettel kell megtervezni. A példáinkban a főszál közvetlenül megkapja és ellenőrzi az adatot, ezért a teljesítést az ellenőrzések és a lezárt szálak együtt igazolják.
2. feladat: több termelő, rendezett eredmény
Minden azonosítóhoz indíts egy küldőt. Az eredeti küldővéget a klónok elkészítése után dobd el, különben az iterátor vége nem lenne elérhető. Fogadj mindent, majd várd meg a szálakat, és rendezd a listát. Azonos azonosítók megengedettek ebben a gyakorlatban: két külön üzenet marad két külön elem, nem összevonás történik.
Megoldás
use std::{sync::mpsc, thread};
fn gather(ids: Vec<u32>) -> Vec<(u32, String)> {
let (tx, rx) = mpsc::channel();
let handles: Vec<_> = ids.into_iter().map(|id| {
let sender = tx.clone();
thread::spawn(move || sender.send((id, format!("munka:{id}"))))
}).collect();
drop(tx);
let mut messages: Vec<_> = rx.into_iter().collect();
let joined: Vec<_> = handles.into_iter().map(|handle| handle.join()).collect();
for result in joined { result.unwrap().unwrap(); }
messages.sort();
messages
}
fn main() {
assert_eq!(gather(vec![2, 1, 2]), vec![
(1, String::from("munka:1")),
(2, String::from("munka:2")),
(2, String::from("munka:2")),
]);
assert!(gather(vec![]).is_empty());
}3. feladat: kézbesíthetetlen érték visszavétele
Ha a fogadó megszűnt, a küldés hibát ad, és visszaadja az el nem küldött értéket. Használd ezt arra, hogy a tulajdonolt szöveget visszanyerd. A fogadó végét még a szál indítása előtt dobd el, így az elutasítás determinisztikus. Nincs verseny arra, hogy a fogadó éppen melyik pillanatban ér véget.
Megoldás
use std::{sync::mpsc, thread};
fn rejected(text: String) -> String {
let (tx, rx) = mpsc::channel::<String>();
drop(rx);
let handle = thread::spawn(move || tx.send(text).unwrap_err().0);
handle.join().unwrap()
}
fn main() {
assert_eq!(rejected(String::from("megőrzendő")), "megőrzendő");
assert_eq!(rejected(String::new()), "");
}Üzenetsorrend és adatmodell
Egy küldő saját üzeneteinek sorrendje használható kapcsolat, több termelő közös beérkezési sorrendjéből viszont ne következtess az eredeti munkalista sorrendjére. Ha ez számít, küldj azonosítót is. Az azonosító jelentése legyen világos: sorszám, dokumentumkulcs és időbélyeg nem felcserélhető fogalom. A második feladat azonosító szerint rendez, és nem állítja, hogy ezzel időrendet állított volna helyre.
A csatorna vége nem mindig jelenti a feladat sikerét. Ha egy munkás pánikol, a küldője megszűnhet, és a fogadó ugyanúgy elérheti az iterátor végét. Ezért kell az eredmények fogadása mellett a szálak befejezését is ellenőrizni. Nagyobb feldolgozóban az elvárt darabszámot vagy az azonosítók teljességét is vizsgálhatod. Egy üres lista csak akkor siker, ha a bemenet valóban nem tartalmazott munkát.
A végpontok rövid neve helyett a szerepüket is kövesd: melyik függvény készít adatot, és melyik felel a teljes eredményért? A fogadó tulajdonosának meg kell tudnia különböztetni a szabályosan üres munkát a hiányzó eredménytől. Ha ezt a döntést a termelők közé szórjuk, nehezebb lesz a teljes futást egyetlen helyen ellenőrizni.
Elfogadási feltételek
- A tulajdonolt szöveg fogadással kerül a következő szálhoz; az üres szöveg is megmarad.
- Az eredeti és minden klónozott küldő lezárul, az összes munkaszálat megvárjuk.
- A több termelő eredményeit explicit rendezzük, a megszűnt fogadó hibájából az eredeti értéket visszanyerjük.
Továbblépés
A kapacitás és visszanyomás leckében a fogadó tempója már hatással lesz a küldőre. Ehhez a tulajdon útja mellett azt is meg kell tervezni, hol blokkolhat a program, és ki tudja közben továbbvinni a feldolgozást.