פתרון פרויקט שירות עבודות

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

חלוקת רכיבים

נשתמש במעטפת השירות מהרצאת העצירה, ונחליף את ההמתנה הריקה שלה במנוע. הcontrol handler יסגור קבלה ויסמן בקשת עצירה; הthread שמנהל את הסיום ימתין לסיום המנוע. נקודת הדיווח stopped נשארת אחרי ניקוי כל הqueues והcallbacks.

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

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

הגדרת הודעה

אפשר לבחור header בן 32 bytes: ארבעה bytes של magic, שניים לגרסה, שניים לפקודה, ארבעה לאורך, ארבעה לשדה שמור שצריך להיות אפס, ושישה עשר למזהה בקשה. שדות המספר מיוצגים בסדר little endian שהגדרנו בפרוטוקול. הbody של יצירה כולל מספר בן 64 ביט, ובשאילתה ובביטול הוא כולל מזהה עבודה בן 64 ביט.

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

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

החלטת סיום אחת

הקוד הבא מממש את ההכרעה בזיכרון בין בקשת ביטול לבין תוצאת חישוב. זאת יחידת בדיקה עצמאית שאפשר לבנות לפני חיבור הjournal. היא אינה טוענת שהתוצאה durable עד שהשכבה החיצונית שמרה record ואישרה אותו.

#include <cassert>
#include <cstdint>
#include <mutex>
#include <optional>

enum class phase { queued, running, completing, completed, cancelled, failed };
enum class cancel_result { requested, too_late, already_terminal };
struct finish_record {
    phase outcome;
    std::uint64_t result;
};

class job_state {
    mutable std::mutex lock_;
    phase phase_ = phase::queued;
    bool cancel_requested_ = false;
    std::optional<finish_record> selected_;
public:
    bool start() {
        std::lock_guard guard(lock_);
        if (phase_ != phase::queued) return false;
        phase_ = phase::running;
        return true;
    }
    cancel_result request_cancel() {
        std::lock_guard guard(lock_);
        if (phase_ == phase::completing) return cancel_result::too_late;
        if (phase_ == phase::completed || phase_ == phase::cancelled ||
            phase_ == phase::failed) return cancel_result::already_terminal;
        cancel_requested_ = true;
        return cancel_result::requested;
    }
    bool should_cancel() const {
        std::lock_guard guard(lock_);
        return cancel_requested_;
    }
    std::optional<finish_record> choose_finish(std::uint64_t result) {
        std::lock_guard guard(lock_);
        if (phase_ != phase::running) return std::nullopt;
        selected_ = finish_record{
            cancel_requested_ ? phase::cancelled : phase::completed,
            cancel_requested_ ? 0 : result
        };
        phase_ = phase::completing;
        return selected_;
    }
    bool confirm_durable() {
        std::lock_guard guard(lock_);
        if (phase_ != phase::completing || !selected_) return false;
        phase_ = selected_->outcome;
        return true;
    }
    phase state() const {
        std::lock_guard guard(lock_);
        return phase_;
    }
};

int main() {
    job_state first;
    bool started = first.start();
    auto cancellation = first.request_cancel();
    auto record = first.choose_finish(123);
    assert(started);
    assert(cancellation == cancel_result::requested);
    assert(record && record->outcome == phase::cancelled);
    assert(first.state() == phase::completing);
    bool committed = first.confirm_durable();
    assert(committed && first.state() == phase::cancelled);
    assert(!first.choose_finish(456));

    job_state second;
    started = second.start();
    record = second.choose_finish(789);
    cancellation = second.request_cancel();
    assert(started && record && record->outcome == phase::completed);
    assert(cancellation == cancel_result::too_late);
    committed = second.confirm_durable();
    assert(committed && second.state() == phase::completed);
}

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

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

החישוב ונקודות הביטול

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

std::uint64_t compute(std::uint64_t count, const job_state& state) {
    std::uint64_t result = 0;
    for (std::uint64_t i = 1; i <= count; ++i) {
        if ((i % 4096) == 0 && state.should_cancel()) break;
        result += i * i;
    }
    return result;
}

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

סדר קבלה והתמדה

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

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

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

קישור לעצירה ולתצורה

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

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

בדיקות שמכריעות את החוזה

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

בבדיקת משתמש זר נוודא שאין שינוי בrecord הסיום, במונה הביטולים ובתוכן הjournal העסקי. אירוע audit של דחייה הוא אפקט מותר שהגדרנו בנפרד. בבדיקת client ישן נוודא שגרסה לא מוכרת נדחית לפני קריאת שדות body לפי layout חדש.

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

הוספת התקדמות

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

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