Cél és előfeltétel
A miniprojekt memóriában kapott számlistákat dolgoz fel. Minden lista önálló munka, saját sorszámmal és ellenőrzött részösszeggel. Legfeljebb négy munkásszál indulhat, az eredménycsatornában legfeljebb két üzenet várakozik. A korlátos csatornák és a szálak lezárása már ismerős; most egyetlen követhető adatútba rendezzük őket. Nincs fájlírás, felhasználói bemenet vagy időzítésre épülő ellenőrzés.
Külön terv a munkára és az eredményre
A munkákat körkörösen osztjuk szét a munkások között. Ez egyszerű és determinisztikus, de nem ígér egyenlő futási időt: az egyik lista sokkal hosszabb lehet a másiknál. A feladat célja a tulajdon és a befejezés helyessége. Dinamikus munkasor bevezetése új szinkronizációs kérdéseket nyitna, ezért azt itt nem rejtjük el néhány véletlenszerűen megosztott változó mögött.
Minden munkás saját kosarat birtokol, így a bemeneti adatokhoz nem kell közös mutex. Az eredményeket egyetlen fogadó gyűjti. A sorszámot még a szálindítás előtt a bemenetből állítjuk elő, ezért utólag ellenőrizhető, hogy minden munka pontosan egyszer szerepel-e. A rendezés nem a befejezés idejét rekonstruálja, hanem a bemenet eredeti sorrendjét állítja helyre.
Három hibaszintet különböztetünk meg. A hibás munkásszám vagy a szál rendellenes befejezése a teljes futás hibája. Egy részösszeg túlcsordulása a megfelelő sor eredménye marad. Végül a sikeres részösszegek együtt is túl nagyok lehetnek a végösszeghez. Emiatt a jelentés a sorokat és a végösszeg saját Result értékét egyaránt megtartja; egy elutasított összesítés nem tünteti el a már elkészült részleteket.
1. feladat: stabil munkakiosztás
Készíts kosarakat egytől négyig megengedett munkásszámhoz. A sorszám a bemenet bejárási sorrendjéből származzon. Három munka két kosárnál a nulladik és második munkát az első kosárba, az elsőt a másodikba helyezi. Üres bemenetnél is ugyanannyi üres kosár készüljön. A nulla munkásszámot még az osztási művelet előtt utasítsd el.
Megoldás
type Job = (usize, Vec<u64>);
fn partition(jobs: Vec<Vec<u64>>, workers: usize) -> Result<Vec<Vec<Job>>, &'static str> {
if !(1..=4).contains(&workers) { return Err("munkásszám"); }
let mut buckets = vec![Vec::new(); workers];
for (id, values) in jobs.into_iter().enumerate() { buckets[id % workers].push((id, values)); }
Ok(buckets)
}
fn main() {
let buckets = partition(vec![vec![1], vec![2], vec![3]], 2).unwrap();
assert_eq!(buckets, vec![vec![(0, vec![1]), (2, vec![3])], vec![(1, vec![2])]]);
assert_eq!(partition(vec![], 2).unwrap(), vec![vec![], vec![]]);
assert!(partition(vec![], 0).is_err());
assert!(partition(vec![], 5).is_err());
}A részfeladat maradjon tiszta művelet
Mielőtt szálak közé teszed a számítást, önálló függvényként ellenőrizd. Így a számtani hiba nem keveredik a csatorna lezárásával. Az üres lista részösszege nulla, a túlcsordulás viszont hiba. A try_fold első hibánál megáll az adott listán, de a következő munka feldolgozása tovább folytatható. Ezzel a helyi hiba határa egyértelmű marad.
2. feladat: egy kosár ellenőrzött eredményei
A kosár minden munkájához készíts azonosítót és részösszeget tartalmazó sort. Ne hagyd el a hibás sort, és ne cseréld a hibát nullára. Az első és utolsó teszt azt is ellenőrzi, hogy üres munka és üres munkalista eltérő szerkezet: előbbinél van eredménysor, utóbbinál nincs.
Megoldás
type Row = (usize, Result<u64, &'static str>);
fn batch(jobs: Vec<(usize, Vec<u64>)>) -> Vec<Row> {
jobs.into_iter().map(|(id, values)| {
let sum = values.into_iter().try_fold(0_u64, |a, b|
a.checked_add(b).ok_or("részösszeg"));
(id, sum)
}).collect()
}
fn main() {
assert_eq!(batch(vec![(0, vec![]), (1, vec![2, 3])]), vec![(0, Ok(0)), (1, Ok(5))]);
assert_eq!(batch(vec![(4, vec![u64::MAX, 1]), (5, vec![7])]),
vec![(4, Err("részösszeg")), (5, Ok(7))]);
assert!(batch(vec![]).is_empty());
}3. feladat: teljes futás és rendezett jelentés
Kapcsold össze a kiosztást, a szálakat és a két üzenetes csatornát. Előbb fogadj minden eredményt, utána vizsgáld meg az összes handle-t. Egyetlen szálhibánál se térj vissza a még hátralévő handle-ek bevárása előtt. A végén rendezz, ellenőrizd a sorok számát és azonosítóit, majd számold ki a végösszeget. A kétféle túlcsordulás külön tesztadatot kapjon.
Megoldás
use std::{sync::mpsc, thread};
type Row = (usize, Result<u64, &'static str>);
#[derive(Debug, PartialEq)]
struct Report { rows: Vec<Row>, total: Result<u64, &'static str> }
fn summarize(jobs: Vec<Vec<u64>>, workers: usize) -> Result<Report, &'static str> {
if !(1..=4).contains(&workers) { return Err("munkásszám"); }
let expected = jobs.len();
let mut buckets = vec![Vec::new(); workers];
for (id, values) in jobs.into_iter().enumerate() { buckets[id % workers].push((id, values)); }
let (tx, rx) = mpsc::sync_channel(2);
let handles: Vec<_> = buckets.into_iter().map(|bucket| {
let sender = tx.clone();
thread::spawn(move || -> Result<(), &'static str> {
for (id, values) in bucket {
let sum = values.into_iter().try_fold(0_u64, |a, b|
a.checked_add(b).ok_or("részösszeg"));
sender.send((id, sum)).map_err(|_| "fogadó megszűnt")?;
}
Ok(())
})
}).collect();
drop(tx);
let mut rows: Vec<Row> = rx.into_iter().collect();
let mut failed = false;
for handle in handles {
match handle.join() { Ok(Ok(())) => {}, _ => failed = true }
}
if failed { return Err("munkásszál"); }
rows.sort_unstable_by_key(|row| row.0);
if rows.len() != expected || rows.iter().enumerate().any(|(id, row)| id != row.0) {
return Err("hiányos eredmény");
}
let total = rows.iter().try_fold(0_u64, |sum, row| {
let value = row.1?;
sum.checked_add(value).ok_or("végösszeg")
});
Ok(Report { rows, total })
}
fn main() {
let normal = summarize(vec![vec![1, 2], vec![], vec![3]], 2).unwrap();
assert_eq!(normal.rows, vec![(0, Ok(3)), (1, Ok(0)), (2, Ok(3))]);
assert_eq!(normal.total, Ok(6));
let local = summarize(vec![vec![u64::MAX, 1], vec![5]], 2).unwrap();
assert_eq!(local.rows, vec![(0, Err("részösszeg")), (1, Ok(5))]);
assert_eq!(local.total, Err("részösszeg"));
let global = summarize(vec![vec![u64::MAX], vec![1]], 1).unwrap();
assert_eq!(global.rows, vec![(0, Ok(u64::MAX)), (1, Ok(1))]);
assert_eq!(global.total, Err("végösszeg"));
assert_eq!(summarize(vec![], 4).unwrap(), Report { rows: vec![], total: Ok(0) });
assert_eq!(summarize(vec![], 0), Err("munkásszám"));
assert_eq!(summarize(vec![], 5), Err("munkásszám"));
}A korlátok pontos jelentése
A program minden bemeneti listát előre megkap, és a jelentés sorait is megőrzi. A véges csatorna ezért nem teszi állandó memóriaigényűvé a teljes megoldást. A munkásszám viszont legfeljebb négy, függetlenül a feladatok számától. A sorok tartalmának és darabszámának ellenőrzése többet mond a helyességről, mint egy véletlenül megfelelő végösszeg, amelyből két egymást kiegyenlítő hiba akár rejtve is maradhatna.
A mintában a koordinátor és a munkások szerepe szándékosan aszimmetrikus. A munkás csak számol és küld, a koordinátor fogad, rendez és ellenőriz. Egy munkásnak nem kell ismernie más kosarak tartalmát vagy a végösszeget. Ez szűkíti azt az állapotot, amelyről egyszerre gondolkodnia kell. A hibás részösszeg sem szakítja meg a kosár további munkáit, ezért a jelentés a többi bemenetről is érdemi visszajelzést adhat.
A munkásszám ellenőrzése megelőzi a felosztást és a szálindítást. Hibás beállításnál így nem marad részben elindított folyamat, és a nulla osztó sem juthat el a maradékképzéshez. Ez ugyanaz a határelv, amelyet a típusos API-knál is használtunk: a használhatatlan állapotot minél korábban, egyértelmű eredménnyel utasítjuk el. A későbbi szálhibákat ettől külön kezeljük, mert azok már egy megkezdett futás lezárásakor jelennek meg.
Elfogadási feltételek
- Egytől négyig állítható munkásszám mellett minden bemeneti sorszám pontosan egyszer szerepel, rendezett kimenettel.
- A részösszeg hibája megmarad a megfelelő sorban, a végösszeg túlcsordulása külön eredmény.
- Az eredeti küldő lezárul, a fogadás és az összes join befejeződik, az üres bemenet is ellenőrzött.
Továbblépés
A szálak és join alapelvei itt teljes feldolgozási szerződéssé álltak össze. Bővítés előtt nevezd meg, mi változik: a bemenet tárolása, a munkakiosztás vagy a hibapolitika. Egyik új igény sem indokolja a teljességellenőrzés vagy a szálak rendezett lezárásának elhagyását.