8.10 - ביטול עבודה חיצונית וסגירת callbacks - הרצאה

כשהקוד הזר ממשיך בלעדינו

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

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

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

שלוש נקודות שונות בסגירה

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

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

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

מודל ספק עם רישום כניסה מוגן

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

use std::sync::{Arc, Condvar, Mutex, mpsc};
use std::sync::atomic::{AtomicUsize, Ordering};

struct Context {
    active: Mutex<usize>,
    quiet: Condvar,
    calls: Arc<AtomicUsize>,
}

struct Entry(Arc<Context>);

impl Drop for Entry {
    fn drop(&mut self) {
        let mut active = self.0.active.lock().unwrap();
        *active -= 1;
        if *active == 0 { self.0.quiet.notify_all(); }
    }
}

struct Provider {
    registration: Mutex<Option<Arc<Context>>>,
}

impl Provider {
    fn enter(&self) -> Option<Entry> {
        let registration = self.registration.lock().unwrap();
        let context = registration.as_ref()?;
        *context.active.lock().unwrap() += 1;
        Some(Entry(context.clone()))
    }

    fn close(&self, removed: mpsc::Sender<()>) {
        let context = self.registration.lock().unwrap().take();
        removed.send(()).unwrap();
        if let Some(context) = context {
            let mut active = context.active.lock().unwrap();
            while *active != 0 {
                active = context.quiet.wait(active).unwrap();
            }
        }
    }
}

fn main() {
    let calls = Arc::new(AtomicUsize::new(0));
    let context = Arc::new(Context {
        active: Mutex::new(0),
        quiet: Condvar::new(),
        calls: calls.clone(),
    });
    let weak = Arc::downgrade(&context);
    let provider = Arc::new(Provider {
        registration: Mutex::new(Some(context)),
    });
    let (entered, entrance) = mpsc::channel();
    let (release, released) = mpsc::channel();
    let worker_provider = provider.clone();
    let worker = std::thread::spawn(move || {
        let entry = worker_provider.enter().unwrap();
        entered.send(()).unwrap();
        released.recv().unwrap();
        entry.0.calls.fetch_add(1, Ordering::SeqCst);
        drop(entry);
    });
    entrance.recv().unwrap();
    let (removed, removal) = mpsc::channel();
    let (finished, finish) = mpsc::channel();
    let closing_provider = provider.clone();
    let closer = std::thread::spawn(move || {
        closing_provider.close(removed);
        finished.send(()).unwrap();
    });
    removal.recv().unwrap();
    assert!(provider.enter().is_none());
    assert!(matches!(finish.try_recv(), Err(mpsc::TryRecvError::Empty)));
    assert!(weak.upgrade().is_some());
    release.send(()).unwrap();
    worker.join().unwrap();
    finish.recv().unwrap();
    closer.join().unwrap();
    assert!(weak.upgrade().is_none());
    assert_eq!(calls.load(Ordering::SeqCst), 1);
    println!("closed after active callback returned");
}

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

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

מה המודל מוכיח ומה עדיין חסר

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

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

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

גבול הפאניקה וכניסה חוזרת

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

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

השלמה מאוחרת והחזרת תוצאה

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

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

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