9 - פרויקט מסכם - פתרון פרויקט
פתרון השירות - אחריות אחת לכל בקשה¶
הפתרון משתמש בטוקיו לניהול חיבורים ומחזיק היתר לכל בקשה שהתקבלה לאחריות השירות. בממשק המקומי קבלת המתאר מבוצעת במצב לא חוסם, וקריאת הקובץ מועברת לעבודה חוסמת שמחזיקה בבעלות על הקובץ ועל ההיתר. אם המשימה שמחכה לה מתבטלת, הבעלות אינה נעלמת יחד איתה.
מבנה החבילה¶
הגדירו את החבילה הבאה. הוסיפו מודול בשם rights.rs שמכיל את פונקציות control_storage, send_fd ו־receive_fd מהרצאה 9.2 ואת ההבאות שלהן, בלי הפונקציה הראשית של הניסוי. החליפו את פרמטר הסוקט בשתי הפונקציות בפרמטר &impl AsRawFd, והגדירו את שתיהן כציבוריות בתוך החבילה. גוף הפונקציות אינו משתנה: הוא כבר משתמש רק ב־as_raw_fd ולכן מתאים גם לסוקט זרם מקומי.
[package]
name = "bounded_transform"
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. השרת יוצר תיקייה חדשה ופרטית בתוך תיקיית המעבדה, ומסרב להשתמש בשם שכבר קיים. כתובת הרשת מוגבלת לממשק הלולאה המקומית כברירת מחדל; במעבדת המרחבים מעבירים כתובת מתאימה כפרמטר.
mod rights;
use std::fs::{DirBuilder, File};
use std::io;
use std::os::fd::AsRawFd;
use std::os::unix::fs::{DirBuilderExt, FileExt};
use std::path::PathBuf;
use std::sync::Arc;
use std::time::Duration;
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::io::unix::AsyncFd;
use tokio::net::{TcpListener, TcpStream, UnixListener, UnixStream};
use tokio::sync::{OwnedSemaphorePermit, Semaphore};
use tokio::task::JoinSet;
const MAX: usize = 4096;
const ACTIVE: u32 = 16;
fn invalid(message: &str) -> io::Error {
io::Error::new(io::ErrorKind::InvalidData, message)
}
fn transform(mut bytes: Vec<u8>) -> Vec<u8> {
bytes.make_ascii_uppercase();
bytes
}
async fn network(mut stream: TcpStream, _permit: OwnedSemaphorePermit) -> io::Result<()> {
let mut header = [0_u8; 20];
stream.read_exact(&mut header).await?;
if &header[..4] != b"AMT\0" || header[4] != 1 || header[5] != 1 || header[6..8] != [0, 0] {
return Err(invalid("invalid request header"));
}
let length = u32::from_be_bytes(header[16..20].try_into().unwrap()) as usize;
if length > MAX { return Err(invalid("body too large")); }
let mut body = vec![0; length];
stream.read_exact(&mut body).await?;
let body = transform(body);
header[5] = 2;
stream.write_all(&header).await?;
stream.write_all(&body).await?;
stream.shutdown().await
}
fn check_peer(stream: &std::os::unix::net::UnixStream) -> io::Result<()> {
let mut credentials = std::mem::MaybeUninit::<libc::ucred>::uninit();
let mut length = std::mem::size_of::<libc::ucred>() as libc::socklen_t;
if unsafe { libc::getsockopt(stream.as_raw_fd(), libc::SOL_SOCKET, libc::SO_PEERCRED,
credentials.as_mut_ptr().cast(), &mut length) } < 0 {
return Err(io::Error::last_os_error());
}
if length as usize != std::mem::size_of::<libc::ucred>() { return Err(invalid("credential size")); }
let credentials = unsafe { credentials.assume_init() };
if credentials.uid != unsafe { libc::geteuid() } {
return Err(io::Error::new(io::ErrorKind::PermissionDenied, "peer uid rejected"));
}
Ok(())
}
fn read_sealed(file: File) -> io::Result<Vec<u8>> {
let required = libc::F_SEAL_WRITE | libc::F_SEAL_GROW | libc::F_SEAL_SHRINK;
let seals = unsafe { libc::fcntl(file.as_raw_fd(), libc::F_GET_SEALS) };
if seals < 0 { return Err(io::Error::last_os_error()); }
if seals & required != required { return Err(invalid("required seals absent")); }
let length = file.metadata()?.len();
if length > MAX as u64 { return Err(invalid("file too large")); }
let mut bytes = vec![0; length as usize];
let mut offset = 0;
while offset < bytes.len() {
match file.read_at(&mut bytes[offset..], offset as u64) {
Ok(0) => return Err(io::Error::new(io::ErrorKind::UnexpectedEof, "short sealed file")),
Ok(count) => offset += count,
Err(error) if error.kind() == io::ErrorKind::Interrupted => continue,
Err(error) => return Err(error),
}
}
Ok(transform(bytes))
}
async fn local(stream: UnixStream, permit: OwnedSemaphorePermit) -> io::Result<()> {
let stream = stream.into_std()?;
check_peer(&stream)?;
let descriptor = AsyncFd::new(stream)?;
let file = loop {
let mut ready = descriptor.readable().await?;
match ready.try_io(|inner| rights::receive_fd(inner.get_ref())) {
Ok(result) => break File::from(result?),
Err(_) => continue,
}
};
let mut stream = UnixStream::from_std(descriptor.into_inner())?;
let (result, _permit) = tokio::task::spawn_blocking(move || {
(read_sealed(file), permit)
}).await.map_err(io::Error::other)?;
let body = result?;
stream.write_all(&(body.len() as u32).to_be_bytes()).await?;
stream.write_all(&body).await?;
stream.shutdown().await
}
struct EndpointFiles { directory: PathBuf, socket: PathBuf }
impl Drop for EndpointFiles {
fn drop(&mut self) {
let _ = std::fs::remove_file(&self.socket);
let _ = std::fs::remove_dir(&self.directory);
}
}
#[tokio::main]
async fn main() -> io::Result<()> {
let address = std::env::args().nth(1).unwrap_or_else(|| "127.0.0.1:43000".into());
let directory = PathBuf::from(format!("amit-socket-{}", std::process::id()));
DirBuilder::new().mode(0o700).create(&directory)?;
let files = EndpointFiles { socket: directory.join("service.sock"), directory };
let unix = UnixListener::bind(&files.socket)?;
let tcp = TcpListener::bind(address).await?;
let permits = Arc::new(Semaphore::new(ACTIVE as usize));
let mut jobs = JoinSet::new();
println!("tcp={}, unix={}", tcp.local_addr()?, files.socket.display());
loop {
tokio::select! {
signal = tokio::signal::ctrl_c() => { signal?; break; }
result = jobs.join_next(), if !jobs.is_empty() => {
match result {
Some(Ok(Ok(()))) => println!("completed"),
Some(Ok(Err(error))) => eprintln!("request failed: {error}"),
Some(Err(error)) => eprintln!("task failed: {error}"),
None => {}
}
}
accepted = tcp.accept() => {
let (stream, _) = accepted?;
match permits.clone().try_acquire_owned() {
Ok(permit) => { jobs.spawn(async move {
tokio::time::timeout(Duration::from_secs(3), network(stream, permit))
.await.map_err(|_| io::Error::new(io::ErrorKind::TimedOut, "request deadline"))?
}); }
Err(_) => { eprintln!("network request rejected: capacity"); }
}
}
accepted = unix.accept() => {
let (stream, _) = accepted?;
match permits.clone().try_acquire_owned() {
Ok(permit) => { jobs.spawn(async move {
tokio::time::timeout(Duration::from_secs(3), local(stream, permit))
.await.map_err(|_| io::Error::new(io::ErrorKind::TimedOut, "request deadline"))?
}); }
Err(_) => { eprintln!("local request rejected: capacity"); }
}
}
}
}
drop(tcp);
drop(unix);
let drain = async {
while let Some(result) = jobs.join_next().await { eprintln!("drain: {result:?}"); }
};
if tokio::time::timeout(Duration::from_secs(5), drain).await.is_err() {
jobs.abort_all();
while let Some(result) = jobs.join_next().await { eprintln!("aborted: {result:?}"); }
}
let all = permits.acquire_many_owned(ACTIVE).await.map_err(io::Error::other)?;
drop(all);
println!("all request resources returned");
Ok(())
}
מדוע ההיתר חוזר עם התוצאה¶
בעבודה המקומית ההיתר עובר לתוך הפונקציה החוסמת וחוזר יחד עם התוצאה. אם ההורה ממשיך, הוא מחזיק את ההיתר עד סיום התגובה. אם ההורה בוטל, העבודה החוסמת עדיין מחזיקה אותו עד סיומה, והתוצאה שננטשה משחררת אותו לאחר מכן. כך אי אפשר לקבל עוד ועוד עבודות רק מפני שהמתקשרים מפסיקים לחכות.
בסגירה רוכשים את כל ההיתרים כדי לוודא שאין עבודה שחוסמת עדיין מחזיקה משאבים. ההמתנה הזאת אינה מוצגת כהבטחה לעצירה בכוח של ספרייה חוסמת. עבור מקור קלט שאינו קובץ קטן וחתום נדרש חוזה חזק יותר או תהליך עובד שניתן לפקח עליו בנפרד.
לקוח בדיקה לממשק הרשת¶
הקוד הבא הוא בינארי בדיקה עצמאי. הוא בודק את זהות התשובה ואת התוכן, ולא רק את הצלחת החיבור.
use std::io::{self, Read, Write};
use std::net::TcpStream;
fn main() -> io::Result<()> {
let mut stream = TcpStream::connect("127.0.0.1:43000")?;
let mut header = [0_u8; 20];
header[..4].copy_from_slice(b"AMT\0");
header[4] = 1;
header[5] = 1;
header[8..16].copy_from_slice(&42_u64.to_be_bytes());
header[16..20].copy_from_slice(&5_u32.to_be_bytes());
for byte in header { stream.write_all(&[byte])?; }
stream.write_all(b"hello")?;
let mut response = [0_u8; 20];
stream.read_exact(&mut response)?;
assert_eq!(response[5], 2);
assert_eq!(u64::from_be_bytes(response[8..16].try_into().unwrap()), 42);
assert_eq!(u32::from_be_bytes(response[16..20].try_into().unwrap()), 5);
let mut body = [0_u8; 5];
stream.read_exact(&mut body)?;
assert_eq!(&body, b"HELLO");
Ok(())
}
שליחת הכותרת בית אחר בית בודקת שהשרת אינו מניח גבול קריאה מסוים. מערכת ההפעלה עדיין יכולה לאחד חלק מהבתים, ולכן בדיקות הפענוח הטהורות מהרצאה 9.3 משלימות את כיסוי כל נקודות החיתוך.
פתרון הבדיקה המקומית והשלילות¶
שמרו את הלקוח המקומי הבא בקובץ src/bin/local-client.rs והעבירו לו את נתיב הסוקט שהשרת הדפיס. הוא משתמש באותו מודול זכויות, אך מחזיק בקצה של סוקט זרם מקומי. המתאר המקורי נשאר בבעלות הלקוח ואפשר לסגור אותו לאחר השליחה.
#[path = "../rights.rs"]
mod rights;
use std::fs::File;
use std::io::{self, Read, Write};
use std::os::fd::{AsRawFd, FromRawFd};
use std::os::unix::net::UnixStream;
use std::time::Duration;
fn main() -> io::Result<()> {
let path = std::env::args().nth(1).expect("provide server socket path");
let raw = unsafe { libc::memfd_create(c"local-request".as_ptr(),
libc::MFD_CLOEXEC | libc::MFD_ALLOW_SEALING) };
if raw < 0 { return Err(io::Error::last_os_error()); }
let mut file = unsafe { File::from_raw_fd(raw) };
file.write_all(b"hello")?;
let seals = libc::F_SEAL_WRITE | libc::F_SEAL_GROW | libc::F_SEAL_SHRINK | libc::F_SEAL_SEAL;
if unsafe { libc::fcntl(file.as_raw_fd(), libc::F_ADD_SEALS, seals) } < 0 {
return Err(io::Error::last_os_error());
}
let mut stream = UnixStream::connect(path)?;
stream.set_read_timeout(Some(Duration::from_secs(4)))?;
stream.set_write_timeout(Some(Duration::from_secs(4)))?;
rights::send_fd(&stream, file.as_raw_fd())?;
drop(file);
let mut length = [0_u8; 4];
stream.read_exact(&mut length)?;
assert_eq!(u32::from_be_bytes(length), 5);
let mut response = [0_u8; 5];
stream.read_exact(&mut response)?;
assert_eq!(&response, b"HELLO");
Ok(())
}
כדי לבדוק חתימות חסרות, השמיטו את קריאת הוספת החתימות בלקוח. השרת צריך לסגור את הבקשה עם שגיאה ולא לקרוא תוכן משתנה כנתון מהימן. כדי לבדוק יותר ממתאר אחד, הגדילו את רשומת הבקרה לשני מתארים; פונקציית הקבלה תעטוף את שניהם ותדחה את ההודעה, כך ששניהם ייסגרו במסלול השגיאה.
עבור גוף קטוע, שלחו כותרת שמצהירה על חמישה בתים ואז רק שניים וסגרו את כיוון הכתיבה. עבור גוף גדול מדי, שלחו אורך של ארבעת אלפים תשעים ושבע בלי גוף. בשני המקרים השרת לא ימסור תשובה מוצלחת. במקרה השני הדחייה מתרחשת לפני הקצאת הגוף.
גבולות הבסיס וההרחבה¶
השרת מטפל בבקשה אחת לכל חיבור, ולכן ביטול באמצעות ניתוק אינו דורש שימור מצב פענוח לשימוש חוזר באותו חיבור. בהרחבת ריבוי הבקשות צריך לשמור מפענח מתמשך, טבלת מזהים ותור תשובות בעל תקציב. אין להעתיק את read_exact לתוך בחירה שמבטלת אותו וממשיכה לקרוא מסגרת חדשה בלי לשמור את ההתקדמות.
הגבלת מספר הבקשות יחד עם מגבלת גודל הגוף נותנת גבול ברור לזיכרון הגופים שבאחריות השירות, אך אינה כוללת את כל תקורת המנוע והקרנל. במדידה מוסיפים את התקורות ומציגים גם דחיות. התיקייה הפרטית וזהות העמית הן מדיניות מעבדה בסיסית; בפרק הבא נרחיב אותה להרשאות שירות, יכולות ובידוד.