מנוע השלמות עם IOCP

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

המנגנון נקרא I/O Completion Port. הוא אינו מבטל את חובת הבעלות על OVERLAPPED ועל buffer. הוא משנה את המקום שממנו אנחנו מקבלים את הראיה לסיום. נבנה תחילה dispatcher יחיד כדי שהפרוטוקול יהיה גלוי, ורק אחר כך נדון במספר עובדים.

שתי בקשות ותור אחד

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

#define WIN32_LEAN_AND_MEAN
#include <Windows.h>
#include <array>
#include <cstddef>
#include <exception>
#include <iostream>

struct Handle {
    HANDLE value;
    explicit Handle(HANDLE v) : value(v) {}
    ~Handle() {
        if (value && value != INVALID_HANDLE_VALUE) CloseHandle(value);
    }
    Handle(const Handle&) = delete;
    Handle& operator=(const Handle&) = delete;
};

struct Request {
    OVERLAPPED operation{};
    std::array<std::byte, 4096> buffer{};
    bool active = false;
    DWORD error = ERROR_SUCCESS;
    DWORD bytes = 0;
};

int wmain(int argc, wchar_t** argv) {
    if (argc != 2) return 2;
    Handle port(CreateIoCompletionPort(INVALID_HANDLE_VALUE, nullptr, 0, 1));
    if (!port.value) return 1;
    Handle file(CreateFileW(argv[1], GENERIC_READ, FILE_SHARE_READ,
        nullptr, OPEN_EXISTING, FILE_FLAG_OVERLAPPED, nullptr));
    if (file.value == INVALID_HANDLE_VALUE) return 1;
    constexpr ULONG_PTR key = 17;
    if (!CreateIoCompletionPort(file.value, port.value, key, 0)) return 1;

    std::array<Request, 2> requests{};
    unsigned active = 0;
    for (std::size_t i = 0; i < requests.size(); ++i) {
        auto& request = requests[i];
        request.operation.Offset = static_cast<DWORD>(i * 4096);
        request.active = true;
        ++active;
        DWORD ignored = 0;
        const BOOL ok = ReadFile(file.value, request.buffer.data(),
            static_cast<DWORD>(request.buffer.size()),
            &ignored, &request.operation);
        if (!ok) {
            const DWORD error = GetLastError();
            if (error != ERROR_IO_PENDING) {
                request.active = false;
                request.error = error;
                --active;
            }
        }
    }

    while (active != 0) {
        DWORD bytes = 0;
        ULONG_PTR received_key = 0;
        OVERLAPPED* operation = nullptr;
        const BOOL ok = GetQueuedCompletionStatus(
            port.value, &bytes, &received_key, &operation, INFINITE);
        const DWORD error = ok ? ERROR_SUCCESS : GetLastError();
        if (!operation || received_key != key) std::terminate();

        Request* completed = nullptr;
        for (auto& request : requests)
            if (&request.operation == operation) completed = &request;
        if (!completed || !completed->active) std::terminate();
        completed->bytes = bytes;
        completed->error = error;
        completed->active = false;
        --active;
    }

    for (std::size_t i = 0; i < requests.size(); ++i) {
        std::cout << i << ": bytes=" << requests[i].bytes
                  << " error=" << requests[i].error << '\n';
    }
}

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

שני מזהים לשני תפקידים

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

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

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

כשל שיש בו בקשה וכשל שאין בו בקשה

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

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

השלמה חלקית אינה תמיד סיום משימה

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

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

מאגר תהליכונים של המערכת

אפשר להשתמש ב־thread-pool I/O כדי שהמערכת תנהל את העובדים והקריאות ל־callback. לפני כל הגשה נדרשת StartThreadpoolIo. אם ההגשה נכשלת באופן שלא ייצור completion, מאזנים את ההודעה באמצעות CancelThreadpoolIo. הפעולה הזו מאזנת רישום למאגר; היא אינה ביטול של I/O בקרנל.

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

קריאה להעמקה

התיעוד של תורי השלמות, של איסוף השלמה ושל רישום פעולה למאגר מגדיר את ההבדלים בין המסלולים.