8 - פרויקט מסכם - פתרון פרויקט

פתרון הפרויקט - חוזה אחד מעל שני תורים

הפתרון הבא מממש ליבה מלאה של ארבע פעולות. הוא משתמש בביטול לוגי: פעולה שבוטלה ממשיכה להיצרך עד סוף הזרם, אבל הנתונים שלה אינם נמסרים כתוצאה. המדיניות הזאת מתאימה לניסוי מוגבל שבו המפיקים ידועים ומסיימים. בשירות פתוח נדרש גם מועד סיום או ביטול פיזי, כדי שלקוח שאינו סוגר לא יחזיק משאבים לנצח.

הגדרת החבילה

צרו חבילת לינוקס עם ההגדרה הבאה. מנגנון הטבעת עשוי להיות חסום בסביבת הרצה מסוימת, ולכן בדקו יצירה לפני פירוש תוצאות הניסוי.

[package]
name = "dual_reactor"
version = "0.1.0"
edition = "2024"

[dependencies]
libc = "0.2"
io-uring = "=0.7.15"

מימוש הליבה

הקוד הבא הוא קובץ מקור מלא. מימוש הטבעת מחכה להשלמת פעולת מוכנות, ולא מבצע קריאת נתונים ישירה דרך הטבעת. לכן הבאפרים המקומיים אינם נמסרים לקרנל מעבר למשך הקריאה הסינכרונית.

use std::io::{self, Read, Write};
use std::os::fd::{AsRawFd, FromRawFd, OwnedFd};
use std::os::unix::net::UnixStream;
use io_uring::{opcode, types, IoUring};

enum Backend { Epoll(OwnedFd), Uring(IoUring) }

impl Backend {
    fn new(uring: bool) -> io::Result<Self> {
        if uring { return Ok(Self::Uring(IoUring::new(16)?)); }
        let raw = unsafe { libc::epoll_create1(libc::EPOLL_CLOEXEC) };
        if raw < 0 { return Err(io::Error::last_os_error()); }
        Ok(Self::Epoll(unsafe { OwnedFd::from_raw_fd(raw) }))
    }

    fn register(&mut self, fd: i32, token: u64) -> io::Result<()> {
        match self {
            Self::Epoll(epoll) => {
                let mut event = libc::epoll_event { events: libc::EPOLLIN as u32, u64: token };
                if unsafe { libc::epoll_ctl(epoll.as_raw_fd(), libc::EPOLL_CTL_ADD,
                                           fd, &mut event) } < 0 {
                    return Err(io::Error::last_os_error());
                }
            }
            Self::Uring(ring) => {
                let entry = opcode::PollAdd::new(types::Fd(fd), libc::POLLIN as u32)
                    .build().user_data(token);
                unsafe {
                    ring.submission().push(&entry)
                        .map_err(|_| io::Error::other("submission queue full"))?;
                }
            }
        }
        Ok(())
    }

    fn wait(&mut self) -> io::Result<Vec<u64>> {
        match self {
            Self::Epoll(epoll) => loop {
                let mut events = [libc::epoll_event { events: 0, u64: 0 }; 8];
                let count = unsafe { libc::epoll_wait(epoll.as_raw_fd(),
                                                     events.as_mut_ptr(), 8, -1) };
                if count < 0 {
                    let error = io::Error::last_os_error();
                    if error.kind() == io::ErrorKind::Interrupted { continue; }
                    return Err(error);
                }
                return Ok(events[..count as usize].iter().map(|event| event.u64).collect());
            },
            Self::Uring(ring) => loop {
                let mut ready = Vec::new();
                for completion in ring.completion() {
                    if completion.result() < 0 {
                        return Err(io::Error::from_raw_os_error(-completion.result()));
                    }
                    ready.push(completion.user_data());
                }
                if !ready.is_empty() { return Ok(ready); }
                match ring.submit_and_wait(1) {
                    Ok(_) => {}
                    Err(error) if error.kind() == io::ErrorKind::Interrupted => {}
                    Err(error) => return Err(error),
                }
            },
        }
    }

    fn rearm(&mut self, fd: i32, token: u64) -> io::Result<()> {
        if matches!(self, Self::Uring(_)) { self.register(fd, token)?; }
        Ok(())
    }

    fn finish(&mut self, fd: i32) -> io::Result<()> {
        if let Self::Epoll(epoll) = self {
            if unsafe { libc::epoll_ctl(epoll.as_raw_fd(), libc::EPOLL_CTL_DEL,
                                       fd, std::ptr::null_mut()) } < 0 {
                return Err(io::Error::last_os_error());
            }
        }
        Ok(())
    }
}

struct Operation {
    stream: UnixStream,
    generation: u32,
    bytes: Vec<u8>,
    cancelled: bool,
}

fn token(slot: usize, generation: u32) -> u64 {
    ((generation as u64) << 32) | slot as u64
}

fn run(uring: bool) -> io::Result<Vec<Option<Vec<u8>>>> {
    let mut backend = Backend::new(uring)?;
    let mut operations = Vec::new();
    let mut workers = Vec::new();
    for slot in 0..4 {
        let (reader, mut writer) = UnixStream::pair()?;
        reader.set_nonblocking(true)?;
        backend.register(reader.as_raw_fd(), token(slot, 1))?;
        operations.push(Some(Operation {
            stream: reader, generation: 1, bytes: Vec::new(), cancelled: slot == 1,
        }));
        workers.push(std::thread::spawn(move || -> io::Result<()> {
            writer.write_all(b"ab")?;
            std::thread::sleep(std::time::Duration::from_millis(5));
            writer.write_all(b"cd")
        }));
    }
    let mut output = vec![None; 4];
    let mut remaining = 4;
    while remaining != 0 {
        for identity in backend.wait()? {
            let slot = (identity & 0xffff_ffff) as usize;
            let generation = (identity >> 32) as u32;
            let Some(Some(operation)) = operations.get_mut(slot) else { continue; };
            if operation.generation != generation { continue; }
            let mut finished = false;
            loop {
                let mut buffer = [0_u8; 8];
                match operation.stream.read(&mut buffer) {
                    Ok(0) => { finished = true; break; }
                    Ok(count) => {
                        if operation.bytes.len() + count > 16 {
                            return Err(io::Error::new(io::ErrorKind::InvalidData, "frame too large"));
                        }
                        operation.bytes.extend_from_slice(&buffer[..count]);
                    }
                    Err(error) if error.kind() == io::ErrorKind::Interrupted => continue,
                    Err(error) if error.kind() == io::ErrorKind::WouldBlock => break,
                    Err(error) => return Err(error),
                }
            }
            if finished {
                backend.finish(operation.stream.as_raw_fd())?;
                let operation = operations[slot].take().unwrap();
                if !operation.cancelled { output[slot] = Some(operation.bytes); }
                remaining -= 1;
            } else {
                backend.rearm(operation.stream.as_raw_fd(), identity)?;
            }
        }
    }
    for worker in workers {
        worker.join().map_err(|_| io::Error::other("producer panicked"))??;
    }
    Ok(output)
}

fn main() -> io::Result<()> {
    let uring = std::env::args().any(|argument| argument == "uring");
    let output = run(uring)?;
    assert_eq!(output, vec![Some(b"abcd".to_vec()), None,
                           Some(b"abcd".to_vec()), Some(b"abcd".to_vec())]);
    println!("four operations drained; one result cancelled");
    Ok(())
}

הפקודות הבאות הן פקודות הרצה מוצעות, ולא דוח על בדיקות שכבר בוצעו.

cargo build --locked
timeout 5s cargo run --locked
timeout 5s cargo run --locked -- uring

ביצירת חבילה חדשה יש ליצור תחילה את קובץ הנעילה באמצעות בנייה רגילה, ורק אחר כך להשתמש באפשרות הנעילה. הפלט הצפוי בשני המסלולים הוא אישור שארבע הפעולות הסתיימו ושאחת התוצאות בוטלה.

הסבר הבעלות והחימוש

כל מתאר קריאה נמצא בבעלות UnixStream יחיד בטבלת הפעולות. במנגנון מבוסס הרמה הרישום נשאר עד סוף הזרם. במנגנון הטבעת כל השלמה צורכת בקשת מוכנות חד־פעמית, ולכן אם הקריאה מגיעה למצב של המתנה יש צורך להגיש בקשה חדשה. הגשה כפולה לפני צריכת הקודמת הייתה יוצרת שתי זהויות פעילות עבור אותו מצב בלי הצדקה.

כאשר הפעולה מסתיימת, ההסרה מתבצעת לפני שהאובייקט שמחזיק את המתאר יוצא מהטבלה. ביטול לוגי אינו מוציא את האובייקט מהטבלה מוקדם יותר. לכן נתונים שמגיעים אחרי הביטול עדיין מקבלים בעלים שמסוגל לצרוך אותם ולסיים את הרישום.

הטבעת מקבלת מספרי מתארים ומזהים מספריים, ללא מצביעים לבאפרים של המשתמש. הדבר מצמצם את חוזה הקוד הלא בטוח בדוגמה הזאת. החלפת PollAdd בפעולת Read מחייבת שינוי ממשי של הבעלות, ולא רק החלפת שם הפעולה.

פתרון בדיקות הקצה

עבור קלט ריק אפשר להסיר את שתי הכתיבות של המפיק. הקריאה הראשונה לאחר סגירתו תחזיר סוף זרם, והתוצאה של פעולה שאינה מבוטלת תהיה וקטור ריק. עבור קלט גדול מהמגבלה החליפו את הכתיבה בבאפר בן שבעה עשר בתים; התוצאה הצפויה היא שגיאת קלט גדולה מדי, בלי ניסיון להציג נתונים חלקיים כתוצאה מלאה.

בדיקת קלט מקוטע כבר קיימת בדוגמה, אך אסור להניח שכל כתיבה תגרום לקריאה נפרדת. מערכת ההפעלה יכולה לאחד זמינות של כמה מקטעים. ההוכחה היא שהתוצאה נכונה בלי קשר לאופן החלוקה בין הקריאות. אפשר להחליף את משך ההשהיה באפס ובערכים שונים ולדרוש אותה תוצאה עסקית.

בדיקת דור ישן משתמשת במודל מהרצאה 8.4: לאחר הגדלת דור המשבצת, אירוע עם הדור הקודם נדחה לפני גישה למשאב. במנוע שממחזר משבצות בפועל צריך להשתמש בהגדלה בדוקה ולפרוש משבצת לפני גלישה. הדוגמה הנוכחית אינה ממחזרת משבצות במהלך ריצה, ולכן אינה מציגה בדיקת מודל כראיה שכבר נבדק מחזור אמיתי.

פתרון הרחבת הבאפר בבעלות

ממשק read_at מהרצאה 8.9 מקבל וקטור בבעלות ומחזיר צמד של תוצאה ושל אותו באפר. לכן בזמן שהפעולה ממתינה, המתקשר אינו מסוגל להגדיל את הווקטור או לשחרר אותו. במודל של ביטול שומרים את הבאפר ברשומת הפעולה עד ההשלמה המקורית, ומשאירים רשומת זהות לפי הצורך עד השלמת כל בקשות הבקרה.

במסלול שבו אישור הביטול מגיע ראשון, המצב עובר לביטול ממתין אבל הבאפר נשאר חי. במסלול שבו ההשלמה המקורית מגיעה ראשונה, אפשר להחזיר או לשחרר את הבאפר לפי החוזה, אך עדיין צריך לזהות אישור ביטול מאוחר. בשני המסלולים אין להסיק דבר רק מסדר ההדפסות של שני תהליכונים.

מגבלות מדויקות של הפתרון

המימוש מרוקן כל מקור עד המתנה או סוף זרם, והגבלת הגודל מונעת ממקור יחיד לבצע עבודה בלתי מוגבלת בניסוי. בשירות עם זרמים גדולים נדרש תקציב סבב ותור מוכנות פנימי, כפי שנלמד בהרצאה 8.8. כמו כן, מסלול השגיאה מפסיק את כל הריצה, ולכן שירות ארוך חיים יצטרך מדיניות נפרדת לכשל של פעולה בודדת.

הפתרון אינו מבטיח סיום מול מפיק שאינו סוגר, ואינו מנסה להפסיק בכוח עבודה חיצונית. ההרחבה הנכונה מוסיפה מועדים ובקשות ביטול תוך שמירת זהויות ומשאבים עד סיום הגישה. הגבלת הזמן החיצונית בבדיקות מזהה תקיעה; היא אינה תחליף למנגנון סגירה בתוך המנוע.