פתרון פרויקט הגמר: חיבור הבטחות מקצה לקצה¶
נממש בסיס עובד שמחבר קבלה מוגבלת, פרוטוקול משותף ל־TCP ול־Unix socket, יומן עמיד, deduplication ו־shutdown. הפתרון ממחזר את מנוע היומן שכבר נבדק בפרק 6 במקום ליצור מנוע שני עם חוזה אחר. כל קוד האינטגרציה החדש מופיע כאן. ההרחבות של worker מבודד והתמחות נשארות משימת ההגנה העצמאית, והפתרון מפרט בהמשך כיצד לשלב ולבדוק אותן בלי לשנות בשקט את ההבטחה.
הכנת מודול האחסון¶
צרו פרויקט בשם amitjobs והעתיקו לקובץ src/journal.rs את קוד היומן מתוך 6 - פרויקט מסכם - פתרון פרויקט, כולל ההבאות, הקבועים, encode, scan, exclusive, Journal, Recovery, initialize והבדיקות, אך בלי הפונקציה main. הוסיפו pub(crate) ל־Journal, ל־Recovery, לשדה Recovery.records, לפונקציות Journal::open ו־Journal::append ולפונקציה initialize. אין לשנות את סדר write ו־sync, את בדיקת checksum של הכותרת או את מצב poisoned. ה־module הוא המימוש המלא מהפרק הקודם, ולא מקום ריק להשלמה.
הגדירו את Cargo.toml הבא ושמרו Cargo.lock אחרי resolution שנבדק. הגרסאות הרחבות כאן מתארות דרישות API; חבילת ההגשה חייבת לכלול את הגרסאות המדויקות שנפתרו ונבדקו.
[package]
name = "amitjobs"
version = "0.1.0"
edition = "2024"
[dependencies]
libc = "0.2"
tokio = { version = "1", features = ["rt-multi-thread", "macros", "net", "io-util", "time", "sync", "signal"] }
ייצוג התוצאה ומועד ההתחייבות¶
הקובץ src/main.rs הבא מלא. התוכן שנשמר בתוך רשומת journal כולל magic של schema, מזהה בקשה, אורך, payload ותוצאה. ה־checksum והמספור של המעטפת מגיעים ממודול journal. בזמן recovery אנחנו בודקים גם invariants עסקיים, מפני שתקינות checksum אינה מוכיחה שהרשומה הגיונית.
mod journal;
use std::collections::BTreeMap;
use std::fs;
use std::io;
use std::os::unix::fs::{DirBuilderExt, FileTypeExt, MetadataExt};
use std::path::{Path, PathBuf};
use std::sync::{Arc, Mutex};
use std::time::Duration;
use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt};
use tokio::net::{TcpListener, UnixListener, UnixStream};
use tokio::sync::{OwnedSemaphorePermit, Semaphore};
use tokio::task::JoinSet;
const MAX: usize = 4096;
const RESULTS: usize = 1024;
const ACTIVE: u32 = 16;
fn invalid(message: &'static str) -> io::Error {
io::Error::new(io::ErrorKind::InvalidData, message)
}
fn sum(bytes: &[u8]) -> u64 { bytes.iter().map(|&b| u64::from(b)).sum() }
struct Store {
journal: journal::Journal,
entries: BTreeMap<u64, (Vec<u8>, u64)>,
failed: bool,
}
impl Store {
fn open(path: &Path) -> io::Result<Self> {
let (journal, recovered) = journal::Journal::open(path)?;
let mut entries = BTreeMap::new();
for record in recovered.records {
if record.len() < 24 || &record[..4] != b"JOB1" {
return Err(invalid("unknown job schema"));
}
let id = u64::from_be_bytes(record[4..12].try_into().unwrap());
let length = u32::from_be_bytes(record[12..16].try_into().unwrap()) as usize;
if length > MAX || record.len() != 24 + length {
return Err(invalid("job length"));
}
let body = record[16..16 + length].to_vec();
let result = u64::from_be_bytes(record[16 + length..].try_into().unwrap());
if sum(&body) != result || entries.len() >= RESULTS ||
entries.insert(id, (body, result)).is_some() {
return Err(invalid("job invariant"));
}
}
Ok(Self { journal, entries, failed: false })
}
fn apply(&mut self, opcode: u8, id: u64, body: Vec<u8>) -> io::Result<(u8, u64)> {
if self.failed { return Err(invalid("store requires recovery")); }
if opcode == 2 {
return Ok(self.entries.get(&id).map_or((4, 0), |(_, result)| (0, *result)));
}
if let Some((old, result)) = self.entries.get(&id) {
return Ok(if old == &body { (0, *result) } else { (3, 0) });
}
if self.entries.len() >= RESULTS { return Ok((2, 0)); }
let result = sum(&body);
let mut record = Vec::with_capacity(24 + body.len());
record.extend_from_slice(b"JOB1");
record.extend_from_slice(&id.to_be_bytes());
record.extend_from_slice(&(body.len() as u32).to_be_bytes());
record.extend_from_slice(&body);
record.extend_from_slice(&result.to_be_bytes());
if let Err(error) = self.journal.append(&record) {
self.failed = true;
return Err(error);
}
self.entries.insert(id, (body, result));
Ok((0, result))
}
}
async fn request<S>(mut stream: S, store: Arc<Mutex<Store>>,
permit: OwnedSemaphorePermit) -> io::Result<()>
where S: AsyncRead + AsyncWrite + Unpin {
let mut header = [0_u8; 20];
stream.read_exact(&mut header).await?;
let opcode = header[5];
let length = u32::from_be_bytes(header[16..20].try_into().unwrap()) as usize;
if &header[..4] != b"AMJ1" || header[4] != 1 ||
!matches!(opcode, 1 | 2) || header[6..8] != [0, 0] ||
length > MAX || (opcode == 2 && length != 0) {
return Err(invalid("request header"));
}
let id = u64::from_be_bytes(header[8..16].try_into().unwrap());
let mut body = vec![0; length];
stream.read_exact(&mut body).await?;
let (result, _permit) = tokio::task::spawn_blocking(move || {
let result = store.lock().map_err(|_| invalid("store lock poisoned"))
.and_then(|mut store| store.apply(opcode, id, body));
(result, permit)
}).await.map_err(io::Error::other)?;
let (status, total) = match result {
Ok(reply) => reply,
Err(error) => { eprintln!("request={id} storage={error}"); (5, 0) }
};
let mut reply = [0_u8; 20];
reply[..4].copy_from_slice(b"AMR1");
reply[4] = status;
reply[8..16].copy_from_slice(&id.to_be_bytes());
reply[16..20].copy_from_slice(&8_u32.to_be_bytes());
stream.write_all(&reply).await?;
stream.write_all(&total.to_be_bytes()).await?;
stream.shutdown().await
}
fn verify_directory(path: &Path) -> io::Result<()> {
let metadata = fs::symlink_metadata(path)?;
if !metadata.is_dir() || metadata.uid() != unsafe { libc::geteuid() } ||
metadata.mode() & 0o077 != 0 {
return Err(io::Error::new(io::ErrorKind::PermissionDenied, "private owned directory required"));
}
Ok(())
}
async fn serve(directory: PathBuf) -> io::Result<()> {
verify_directory(&directory)?;
let store = Arc::new(Mutex::new(Store::open(&directory.join("journal"))?));
let socket = directory.join("service.sock");
match fs::symlink_metadata(&socket) {
Ok(metadata) if metadata.file_type().is_socket() &&
metadata.uid() == unsafe { libc::geteuid() } => fs::remove_file(&socket)?,
Ok(_) => return Err(invalid("unexpected socket path object")),
Err(error) if error.kind() == io::ErrorKind::NotFound => (),
Err(error) => return Err(error),
}
let unix = UnixListener::bind(&socket)?;
let tcp = TcpListener::bind("127.0.0.1:43001").await?;
let permits = Arc::new(Semaphore::new(ACTIVE as usize));
let mut jobs = JoinSet::new();
let mut terminate = tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate())?;
eprintln!("ready socket={} tcp=127.0.0.1:43001", socket.display());
loop {
tokio::select! {
result = tokio::signal::ctrl_c() => { result?; break; }
_ = terminate.recv() => break,
Some(result) = jobs.join_next(), if !jobs.is_empty() => {
eprintln!("finished={result:?}");
}
accepted = tcp.accept() => {
let (stream, _) = accepted?;
if let Ok(permit) = permits.clone().try_acquire_owned() {
let store = Arc::clone(&store);
jobs.spawn(async move {
tokio::time::timeout(Duration::from_secs(3), request(stream, store, permit))
.await.map_err(|_| invalid("request deadline"))?
});
}
}
accepted = unix.accept() => {
let (stream, _) = accepted?;
if let Ok(permit) = permits.clone().try_acquire_owned() {
let store = Arc::clone(&store);
jobs.spawn(async move {
tokio::time::timeout(Duration::from_secs(3), request(stream, store, permit))
.await.map_err(|_| invalid("request deadline"))?
});
}
}
}
}
drop(tcp);
drop(unix);
while let Some(result) = jobs.join_next().await { eprintln!("drained={result:?}"); }
let all = permits.acquire_many_owned(ACTIVE).await.map_err(io::Error::other)?;
drop(all);
fs::remove_file(&socket)?;
eprintln!("stopped resources=returned");
Ok(())
}
async fn client(directory: &Path, opcode: u8, id: u64, body: &[u8]) -> io::Result<()> {
if body.len() > MAX { return Err(invalid("client payload limit")); }
let mut stream = UnixStream::connect(directory.join("service.sock")).await?;
let mut header = [0_u8; 20];
header[..4].copy_from_slice(b"AMJ1");
header[4] = 1;
header[5] = opcode;
header[8..16].copy_from_slice(&id.to_be_bytes());
header[16..20].copy_from_slice(&(body.len() as u32).to_be_bytes());
stream.write_all(&header).await?;
stream.write_all(body).await?;
let mut reply = [0_u8; 28];
stream.read_exact(&mut reply).await?;
if &reply[..4] != b"AMR1" || reply[5..8] != [0, 0, 0] ||
u64::from_be_bytes(reply[8..16].try_into().unwrap()) != id ||
u32::from_be_bytes(reply[16..20].try_into().unwrap()) != 8 {
return Err(invalid("reply identity"));
}
println!("status={} result={}", reply[4],
u64::from_be_bytes(reply[20..28].try_into().unwrap()));
Ok(())
}
#[tokio::main]
async fn main() -> io::Result<()> {
let args: Vec<String> = std::env::args().collect();
if args.len() < 3 { return Err(invalid("init|serve|submit|get directory [id] [body]")); }
let directory = PathBuf::from(&args[2]);
match args[1].as_str() {
"init" => {
fs::DirBuilder::new().mode(0o700).create(&directory)?;
journal::initialize(&directory.join("journal"))
}
"serve" => serve(directory).await,
"submit" | "get" => {
let id: u64 = args.get(3).ok_or_else(|| invalid("id required"))?
.parse().map_err(|_| invalid("invalid id"))?;
let opcode = if args[1] == "submit" { 1 } else { 2 };
let body = if opcode == 1 { args.get(4).ok_or_else(|| invalid("body required"))?.as_bytes() }
else { &[] };
tokio::time::timeout(Duration::from_secs(5), client(&directory, opcode, id, body))
.await.map_err(|_| invalid("client deadline"))?
}
_ => Err(invalid("unknown command")),
}
}
הרצה ובדיקות של הבסיס¶
הריצו את השרת בחלון מסוף אחד ואת הלקוח בחלון אחר. ה־init דורש תיקייה שאינה קיימת, ו־serve מיועד לאותה תיקייה לאחר אתחול. אחרי Ctrl-C או SIGTERM ניתן להפעיל serve שוב בלי init.
cargo test --locked
cargo build --locked
./target/debug/amitjobs init ./state
./target/debug/amitjobs serve ./state
./target/debug/amitjobs submit ./state 7 abc
./target/debug/amitjobs submit ./state 7 abc
./target/debug/amitjobs submit ./state 7 abd
./target/debug/amitjobs get ./state 7
./target/debug/amitjobs get ./state 99
התוצאה של abc היא 294. שתי בקשות זהות מחזירות status 0, בקשה שונה עם אותו ID מחזירה 3, ו־lookup של ID שאינו קיים מחזיר 4. ערך 2 מציין קיבולת תוצאות מלאה, וערך 5 מציין אחסון שאינו יכול להתחייב כרגע. קריאת lookup אחרי restart צריכה להחזיר את אותה תוצאה, ו־retry לא צריך להגדיל את היומן.
ה־TCP מוגבל ללולאה המקומית אך אינו מאמת זהות משתמש, ולכן הבסיס מיועד למעבדה שבה כל משתמשי המכונה מהימנים. במוצר רב־משתמשי יש להוסיף authentication או להסיר את המאזין הזה; אין להציג loopback כגבול הרשאה. הממשק המקומי נמצא בתיקייה פרטית ומהימנה. המחיקה של socket ישן מתבצעת רק אחרי רכישת נעילת היומן, ולכן מופע שני אינו מוחק socket של מופע פעיל שמחזיק אותה נעילה.
ה־timeout עוטף את כל חיי בקשת הרשת. אם הוא מתרחש לפני spawn_blocking, אין commit. אם הוא מתרחש אחריו, ה־closure ממשיך להחזיק body, Store ו־permit עד סיום הפעולה. מכאן נובע שהלקוח יכול לקבל תוצאה לא ידועה, וה־retry עם אותו ID פותר אותה בלי יצירת פעולה חדשה. ה־shutdown ממתין לכל ההיתרים, כולל אלה שנמצאים בפעולות חוסמות שה־future שלהן כבר בוטל.
ה־drain אינו הבטחה לעצור syscall תקוע בדיסק. מנהל השירות יכול להפסיק את התהליך לאחר התקציב החיצוני שלו, והיומן יחזור ל־recovery בהפעלה הבאה. כדי לבדוק כשל אחרי append ולפני reply הוסיפו failpoint מבוקר אחרי Journal::append, עצרו את התהליך והריצו lookup אחרי restart. עבור כשל sync הזריקו שגיאה ברכיב האחסון ובדקו ש־failed מונע תשובות שנשענות על מצב לא ודאי.
הרחבת המערכת והוכחתה¶
בקבלת קובץ דרך SCM_RIGHTS נשתמש בקוד המקבל OwnedFd מפרק 9, נאמת peer ו־seals ונעביר את ה־File יחד עם permit אל worker. נשמור ביומן תיאור עבודה שאפשר לשחזר, ולא מספר fd שאינו בעל משמעות בתהליך הבא. אם הקלט צריך לשרוד restart, נעתיק או נשמר אותו במאגר עמיד לפני Accepted לפי פרוטוקול commit מתועד.
ב־worker מבודד המפקח מחזיק את הסמכות ליומן, וה־worker מקבל רק קלט ותעלת תוצאה. תוצאה שנשלחה מן ה־worker עדיין אינה אישור ללקוח; המפקח מאמת אותה ומבצע commit. כשל worker לפני commit מאפשר retry של חישוב טהור. כאשר העבודה כוללת השפעה חיצונית, נדרש idempotency key בצד השני או חוזה מפורש של תוצאה לא ידועה.
את האריזה נחבר ל־unit משיעור 14.8, תוך התאמת init להתקנה בלי לדרוס תיקיית StateDirectory שכבר קיימת. תהליך ההתקנה ייצור journal חדש באמצעות initialize רק כשאינו קיים, וישאיר מצב קיים לבדיקה ושחזור. אין להריץ init של המעבדה על נתוני ייצור קיימים. ה־runbook יתאר מעבר דורות, גיבוי, החזרה לאחור ומטריצת הפצות שנבדקה.
בדוח הביצועים נשווה את הבסיס הסדרתי תחת Mutex ל־group commit או לשירות אחסון ייעודי. כל שיפור צריך לשמר את הקשר בין sync לבין reply. בדוח ההגנה נציג trace של בקשה, ניסוי retry לאחר אובדן reply, גבול משאבים תחת לקוחות איטיים והרחבה חדשה שמוסיפה יכולת בלי לשבור את הבדיקות הישנות.