רשת אסינכרונית ועומס

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

כבר למדנו IOCP עבור פעולות אסינכרוניות. עכשיו נחבר אליו socketים. המבנה נשאר מוכר: אנחנו מגישים פעולה עם OVERLAPPED, שומרים את המצב ואת הbuffer בחיים, ומקבלים completion שמאפשר להתקדם. הcompletion אינו מבטיח שהודעת היישום שלמה; framing נשאר באחריותנו.

קבלה אסינכרונית של חיבור

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

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

שמרו את התוכנית בaccept_completion.cpp. היא מאזינה בloopback, מקבלת חיבור, ואז קוראת פעם אחת עד 64 bytes. היא אינה שרת הecho מההרצאה הקודמת ואינה מפרשת frame.

#define WIN32_LEAN_AND_MEAN
#include <winsock2.h>
#include <mswsock.h>
#include <windows.h>
#include <array>
#include <exception>
#include <iostream>
#include <stdexcept>
#include <string>

#pragma comment(lib, "ws2_32.lib")

[[noreturn]] void fail(const char* call, DWORD code) {
    throw std::runtime_error(std::string(call) + ": " + std::to_string(code));
}
struct winsock_scope {
    winsock_scope() {
        WSADATA data{};
        int error = WSAStartup(MAKEWORD(2, 2), &data);
        if (error) fail("startup", error);
    }
    winsock_scope(const winsock_scope&) = delete;
    winsock_scope& operator=(const winsock_scope&) = delete;
    ~winsock_scope() { WSACleanup(); }
};
struct socket_owner {
    SOCKET value = WSASocketW(AF_INET, SOCK_STREAM, IPPROTO_TCP,
                              nullptr, 0, WSA_FLAG_OVERLAPPED);
    socket_owner() { if (value == INVALID_SOCKET) fail("socket", WSAGetLastError()); }
    socket_owner(const socket_owner&) = delete;
    socket_owner& operator=(const socket_owner&) = delete;
    ~socket_owner() { closesocket(value); }
};
struct port_owner {
    HANDLE value = CreateIoCompletionPort(INVALID_HANDLE_VALUE, nullptr, 0, 1);
    port_owner() { if (!value) fail("port", GetLastError()); }
    port_owner(const port_owner&) = delete;
    port_owner& operator=(const port_owner&) = delete;
    ~port_owner() { CloseHandle(value); }
};
struct completion { DWORD bytes; DWORD error; bool timeout_requested; };

completion wait_one(HANDLE port, SOCKET socket, OVERLAPPED& operation) {
    DWORD bytes = 0;
    ULONG_PTR key = 0;
    OVERLAPPED* returned = nullptr;
    BOOL ok = GetQueuedCompletionStatus(port, &bytes, &key, &returned, 5000);
    DWORD error = ok ? NO_ERROR : GetLastError();
    bool timed_out = false;
    if (!ok && !returned && error == WAIT_TIMEOUT) {
        timed_out = true;
        if (!CancelIoEx(reinterpret_cast<HANDLE>(socket), &operation)) {
            DWORD cancel_error = GetLastError();
            if (cancel_error != ERROR_NOT_FOUND)
                std::cerr << "cancel request error=" << cancel_error << '\n';
        }
        ok = GetQueuedCompletionStatus(port, &bytes, &key, &returned, INFINITE);
        error = ok ? NO_ERROR : GetLastError();
    }
    if (returned != &operation) std::terminate();
    return {bytes, error, timed_out};
}

void associate(HANDLE port, SOCKET socket, ULONG_PTR key) {
    if (!CreateIoCompletionPort(reinterpret_cast<HANDLE>(socket), port, key, 0))
        fail("associate", GetLastError());
}

int main() {
    try {
        winsock_scope winsock;
        port_owner port;
        socket_owner listener;
        socket_owner accepted;
        int exclusive = 1;
        if (setsockopt(listener.value, SOL_SOCKET, SO_EXCLUSIVEADDRUSE,
                       reinterpret_cast<char*>(&exclusive), sizeof(exclusive)) == SOCKET_ERROR)
            fail("exclusive", WSAGetLastError());
        sockaddr_in address{};
        address.sin_family = AF_INET;
        address.sin_port = htons(39092);
        address.sin_addr.s_addr = htonl(INADDR_LOOPBACK);
        if (bind(listener.value, reinterpret_cast<sockaddr*>(&address), sizeof(address)) == SOCKET_ERROR)
            fail("bind", WSAGetLastError());
        if (listen(listener.value, SOMAXCONN) == SOCKET_ERROR) fail("listen", WSAGetLastError());
        associate(port.value, listener.value, 1);
        GUID extension = WSAID_ACCEPTEX;
        LPFN_ACCEPTEX accept_ex = nullptr;
        DWORD returned = 0;
        if (WSAIoctl(listener.value, SIO_GET_EXTENSION_FUNCTION_POINTER,
                     &extension, sizeof(extension), &accept_ex, sizeof(accept_ex),
                     &returned, nullptr, nullptr) == SOCKET_ERROR)
            fail("extension", WSAGetLastError());

        OVERLAPPED accept_operation{};
        constexpr DWORD address_bytes = sizeof(sockaddr_in) + 16;
        std::array<char, address_bytes * 2> addresses{};
        DWORD immediate = 0;
        BOOL started = accept_ex(listener.value, accepted.value, addresses.data(),
                                 0, address_bytes, address_bytes, &immediate, &accept_operation);
        if (!started) {
            DWORD error = WSAGetLastError();
            if (error != ERROR_IO_PENDING) fail("submit accept", error);
        }
        auto accepted_result = wait_one(port.value, listener.value, accept_operation);
        std::cout << "accept error=" << accepted_result.error
                  << " timed_out=" << accepted_result.timeout_requested << '\n';
        if (accepted_result.error != NO_ERROR || accepted_result.timeout_requested) return 0;
        if (setsockopt(accepted.value, SOL_SOCKET, SO_UPDATE_ACCEPT_CONTEXT,
                reinterpret_cast<const char*>(&listener.value), sizeof(listener.value)) == SOCKET_ERROR)
            fail("accept context", WSAGetLastError());
        associate(port.value, accepted.value, 2);

        OVERLAPPED read_operation{};
        std::array<char, 64> data{};
        WSABUF buffer{static_cast<ULONG>(data.size()), data.data()};
        DWORD flags = 0;
        int submitted = WSARecv(accepted.value, &buffer, 1, nullptr,
                                &flags, &read_operation, nullptr);
        if (submitted == SOCKET_ERROR) {
            DWORD error = WSAGetLastError();
            if (error != WSA_IO_PENDING) fail("submit read", error);
        }
        auto read_result = wait_one(port.value, accepted.value, read_operation);
        std::cout << "read bytes=" << read_result.bytes
                  << " error=" << read_result.error
                  << " timed_out=" << read_result.timeout_requested << '\n';
    } catch (const std::exception& error) {
        std::cerr << error.what() << '\n';
        return 1;
    }
}

בנו והריצו, ואז התחברו בתוך חמש שניות מטרמינל אחר.

cl /std:c++20 /EHsc /W4 accept_completion.cpp
.\accept_completion.exe

$client = [Net.Sockets.TcpClient]::new()
try {
    $client.Connect('127.0.0.1', 39092)
    $bytes = [Text.Encoding]::ASCII.GetBytes('abc')
    $client.GetStream().Write($bytes, 0, $bytes.Length)
} finally {
    $client.Dispose()
}

הצלחה מיידית והשלמה אחת

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

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

פקיעת הזמן אינה מוות של פעולה

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

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

הניסוי משתמש בהמתנה ללא deadline לאחר בקשת הביטול כדי להמחיש ניקוז. במוצר נמשיך את הניקוז בתוך event loop ונפעיל מדיניות סיום רחבה יותר; לא נחסום worker יחיד שמשרת את כל המערכת. אם הפורט עצמו נכשל בלי להחזיר את הפעולה הצפויה, הדוגמה מבצעת fail-fast ולא משחררת מחסנית שעשויה עדיין להיות בשימוש.

עכשיו נכפיל לחיבורים רבים

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

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

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

ויסות ותקציב - backpressure

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

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