8.5 - מכונת קלט ופלט עם זמן וביטול - הרצאה

זמן וביטול הם מקורות אירועים

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

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

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

בחירת שעון היא החלטת משמעות

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

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

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

שני מתארים ולולאת המתנה אחת

הדוגמה הבאה מיועדת ללינוקס. היא מחכה גם להודעה מתהליכון עובד וגם לתפוגת שעון, בלי לולאת בדיקה פעילה. השתמשו בחבילת מעבדה בעלת מהדורת ראסט 2024 ותלות libc = "0.2", ושמרו את קובץ הנעילה של התלויות.

use std::fs::File;
use std::io::{self, Read, Write};
use std::os::fd::{AsRawFd, FromRawFd};
use std::thread;
use std::time::Duration;

fn file(raw: i32) -> io::Result<File> {
    if raw < 0 { Err(io::Error::last_os_error()) }
    else { Ok(unsafe { File::from_raw_fd(raw) }) }
}

fn main() -> io::Result<()> {
    let notification = file(unsafe {
        libc::eventfd(0, libc::EFD_CLOEXEC | libc::EFD_NONBLOCK)
    })?;
    let timer = file(unsafe {
        libc::timerfd_create(libc::CLOCK_MONOTONIC, libc::TFD_CLOEXEC | libc::TFD_NONBLOCK)
    })?;
    let specification = libc::itimerspec {
        it_interval: libc::timespec { tv_sec: 0, tv_nsec: 0 },
        it_value: libc::timespec { tv_sec: 0, tv_nsec: 50_000_000 },
    };
    if unsafe { libc::timerfd_settime(timer.as_raw_fd(), 0, &specification,
                                    std::ptr::null_mut()) } < 0 {
        return Err(io::Error::last_os_error());
    }
    let mut sender = notification.try_clone()?;
    let worker = thread::spawn(move || -> io::Result<()> {
        thread::sleep(Duration::from_millis(10));
        sender.write_all(&1_u64.to_ne_bytes())
    });
    let mut sources = [notification, timer];
    let mut seen = [false; 2];
    while seen.iter().any(|done| !done) {
        let mut interests: Vec<_> = sources.iter().enumerate().map(|(index, source)| {
            libc::pollfd {
                fd: if seen[index] { -1 } else { source.as_raw_fd() },
                events: libc::POLLIN,
                revents: 0,
            }
        }).collect();
        let result = unsafe { libc::poll(interests.as_mut_ptr(), interests.len() as _, -1) };
        if result < 0 {
            let error = io::Error::last_os_error();
            if error.kind() == io::ErrorKind::Interrupted { continue; }
            return Err(error);
        }
        for index in 0..sources.len() {
            let flags = interests[index].revents;
            if flags & (libc::POLLERR | libc::POLLNVAL | libc::POLLHUP) != 0 {
                return Err(io::Error::other("unexpected source state"));
            }
            if flags & libc::POLLIN == 0 { continue; }
            let mut bytes = [0_u8; 8];
            match sources[index].read(&mut bytes) {
                Ok(8) => {
                    println!("source={index}, count={}", u64::from_ne_bytes(bytes));
                    seen[index] = true;
                }
                Ok(_) => return Err(io::Error::other("unexpected counter length")),
                Err(error) if matches!(error.kind(), io::ErrorKind::WouldBlock |
                                                    io::ErrorKind::Interrupted) => {}
                Err(error) => return Err(error),
            }
        }
    }
    worker.join().expect("worker panicked")?;
    Ok(())
}

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

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

אותות בלי לבצע עבודה מתוך מטפל

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

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

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

ההתקדמות שייכת לפעולה

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

Sending { buffer, offset: 0 }
  write returns 200
Sending { buffer, offset: 200 }
  write returns EAGAIN
WaitingWritable { buffer, offset: 200 }
  writable event
Sending { buffer, offset: 200 }

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

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

תרחיש חקירה של סגירה איטית

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

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