9.5 - סגירה ולחץ חוזר בחיבור רשת - הרצאה

הלקוח האיטי משנה את חיי התשובה

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

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

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

תור בעל גבול אמיתי

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

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

use std::collections::VecDeque;
use std::io::{self, Write};

struct Chunk { bytes: Box<[u8]>, sent: usize }

struct Outbox {
    chunks: VecDeque<Chunk>,
    reserved: usize,
    pending: usize,
    limit: usize,
}

impl Outbox {
    fn new(limit: usize) -> Self {
        Self { chunks: VecDeque::new(), reserved: 0, pending: 0, limit }
    }

    fn enqueue(&mut self, bytes: Box<[u8]>) -> Result<(), Box<[u8]>> {
        let Some(total) = self.reserved.checked_add(bytes.len()) else { return Err(bytes); };
        if total > self.limit { return Err(bytes); }
        if bytes.is_empty() { return Ok(()); }
        self.reserved = total;
        self.pending += bytes.len();
        self.chunks.push_back(Chunk { bytes, sent: 0 });
        Ok(())
    }

    fn flush<W: Write>(&mut self, writer: &mut W, mut budget: usize) -> io::Result<()> {
        while budget != 0 {
            let Some(front) = self.chunks.front_mut() else { break; };
            let end = front.sent + budget.min(front.bytes.len() - front.sent);
            match writer.write(&front.bytes[front.sent..end]) {
                Ok(0) => return Err(io::Error::new(io::ErrorKind::WriteZero, "no progress")),
                Ok(count) => {
                    assert!(count <= end - front.sent);
                    front.sent += count;
                    self.pending -= count;
                    budget -= count;
                    if front.sent == front.bytes.len() {
                        let completed = self.chunks.pop_front().unwrap();
                        self.reserved -= completed.bytes.len();
                    }
                }
                Err(error) if error.kind() == io::ErrorKind::Interrupted => continue,
                Err(error) if error.kind() == io::ErrorKind::WouldBlock => break,
                Err(error) => return Err(error),
            }
        }
        Ok(())
    }
}

struct ScriptedWriter { steps: VecDeque<Option<usize>>, output: Vec<u8> }

impl Write for ScriptedWriter {
    fn write(&mut self, bytes: &[u8]) -> io::Result<usize> {
        match self.steps.pop_front().unwrap_or(Some(bytes.len())) {
            None => Err(io::Error::from(io::ErrorKind::WouldBlock)),
            Some(limit) => {
                let count = bytes.len().min(limit);
                self.output.extend_from_slice(&bytes[..count]);
                Ok(count)
            }
        }
    }
    fn flush(&mut self) -> io::Result<()> { Ok(()) }
}

fn main() -> io::Result<()> {
    let mut outbox = Outbox::new(8);
    outbox.enqueue(b"abcd".to_vec().into_boxed_slice()).unwrap();
    outbox.enqueue(b"EFGH".to_vec().into_boxed_slice()).unwrap();
    let mut writer = ScriptedWriter {
        steps: VecDeque::from([Some(2), None, Some(3), Some(3)]),
        output: Vec::new(),
    };
    outbox.flush(&mut writer, 8)?;
    assert_eq!(outbox.pending, 6);
    assert_eq!(outbox.reserved, 8);
    assert!(outbox.enqueue(b"x".to_vec().into_boxed_slice()).is_err());
    outbox.flush(&mut writer, 8)?;
    assert_eq!(&writer.output, b"abcdEFGH");
    assert_eq!((outbox.pending, outbox.reserved), (0, 0));
    Ok(())
}

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

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

לחץ חוזר צריך להגיע למקום שמייצר עבודה

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

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

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

סגירה של כיוון אחד

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

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

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

סיום מסודר לעומת איפוס

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

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

שירות שמפסיק לקבל עבודה

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

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

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