שירותי תקשורת מקומיים והתאוששות מחיבור שנקטע

שם מוכר אינו מופע קבוע

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

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

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

מונה עם זיהוי בקשות

נבנה מעבדה באוטובוס פרטי בעזרת dbus-run-session -- bash. כל הפקודות של הניסוי, השרת והלקוחות ירוצו בתוך אותה מעטפת או בתהליכים היורשים את משתני הסביבה שלה. כך איננו מוסיפים מדיניות למערכת השירותים הכללית של המחשב. נשתמש בפרויקט עם התלויות הבאות ונשמור את קובץ הנעילה שנוצר בבנייה.

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

[dependencies]
zbus = { version = "=5.19.0", default-features = false, features = ["tokio"] }
tokio = { version = "1", features = ["macros", "rt-multi-thread", "time", "signal"] }

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

use std::collections::HashMap;
use std::time::Duration;
use zbus::{connection, interface, object_server::SignalEmitter};

struct Counter {
    epoch: String,
    value: u64,
    seen: HashMap<String, (u64, u64)>,
}

#[interface(name = "org.amit.Counter1")]
impl Counter {
    fn info(&self) -> (String, u64) { (self.epoch.clone(), self.value) }

    async fn add(&mut self, expected_epoch: &str, request_id: &str,
        amount: u64, delay_reply_ms: u64,
        #[zbus(signal_emitter)] emitter: SignalEmitter<'_>,
    ) -> zbus::fdo::Result<(String, u64)> {
        if expected_epoch != self.epoch {
            return Err(zbus::fdo::Error::Failed("service epoch changed".into()));
        }
        if request_id.is_empty() || request_id.len() > 64 || delay_reply_ms > 1000 {
            return Err(zbus::fdo::Error::InvalidArgs("invalid request".into()));
        }
        if let Some(&(old_amount, result)) = self.seen.get(request_id) {
            if old_amount != amount {
                return Err(zbus::fdo::Error::InvalidArgs("request ID reused".into()));
            }
            return Ok((self.epoch.clone(), result));
        }
        if self.seen.len() >= 128 {
            return Err(zbus::fdo::Error::LimitsExceeded("request table full".into()));
        }
        let result = self.value.checked_add(amount)
            .ok_or_else(|| zbus::fdo::Error::InvalidArgs("counter overflow".into()))?;
        self.value = result;
        self.seen.insert(request_id.to_owned(), (amount, result));
        if let Err(error) = Self::changed(&emitter, &self.epoch, result).await {
            eprintln!("signal failed after commit: {error}");
        }
        tokio::time::sleep(Duration::from_millis(delay_reply_ms)).await;
        Ok((self.epoch.clone(), result))
    }

    #[zbus(signal)]
    async fn changed(emitter: &SignalEmitter<'_>, epoch: &str, value: u64)
        -> zbus::Result<()>;
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let epoch = std::fs::read_to_string("/proc/sys/kernel/random/uuid")?
        .trim().to_owned();
    let service = Counter { epoch, value: 0, seen: HashMap::new() };
    let connection = connection::Builder::session()?
        .name("org.amit.Counter")?
        .serve_at("/org/amit/Counter", service)?
        .build().await?;
    println!("owner={:?}", connection.unique_name());
    tokio::signal::ctrl_c().await?;
    drop(connection);
    Ok(())
}

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

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

לקוח עם תוצאה לא ידועה

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

use std::time::Duration;
#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let connection = zbus::Connection::session().await?;
    let proxy = zbus::Proxy::new(&connection, "org.amit.Counter",
        "/org/amit/Counter", "org.amit.Counter1").await?;
    let (epoch, before): (String, u64) = proxy.call("Info", &()).await?;
    let request = std::fs::read_to_string("/proc/sys/kernel/random/uuid")?
        .trim().to_owned();
    let first: Result<zbus::Result<(String, u64)>, _> = tokio::time::timeout(
        Duration::from_millis(100),
        proxy.call("Add", &(epoch.as_str(), request.as_str(), 5_u64, 500_u64)),
    ).await;
    println!("first={first:?}");
    let result: (String, u64) = tokio::time::timeout(Duration::from_secs(2),
        proxy.call("Add", &(epoch.as_str(), request.as_str(), 5_u64, 0_u64)),
    ).await??;
    assert_eq!(result.0, epoch);
    assert_eq!(result.1, before + 5);
    println!("confirmed={}", result.1);
    Ok(())
}

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

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

אותות, הפעלה ומדיניות

נוכל לצפות באותות באמצעות dbus-monitor בתוך אותו אוטובוס, עם מסנן לממשק org.amit.Counter1. מאזין שהתחבר מאוחר אינו מקבל בהכרח היסטוריה של שינויים. לאחר חיבור מחדש עליו לקבל תמונת מצב ולהתייחס לדור המופע, ולא להניח שאותות בלבד משחזרים את כל מה שהחמיץ.

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

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

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