העברת ממשקים בין גבולות

נעתיק מצביע לworker, נוסיף AddRef, ונריץ את אותה מתודה. שמרנו על הזיכרון, אז לכאורה הכול אמור לעבוד. הבעיה היא שחיי האובייקט הם רק חצי מהחוזה. ייתכן שהמצביע מותר לשימוש רק בתוך apartment מסוים. העתקת כתובת אינה מעבירה איתה את ההרשאה לקרוא מכל thread.

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

מי באמת מקבל את הקריאה

הproxy נמצא בצד הלקוח ומציג את הממשק שהלקוח מכיר. הstub נמצא בצד שמריץ את האובייקט, מפענח את הפרמטרים ומבצע את הקריאה. ביניהם יש channel. עבור ממשקים תקניים רבים מערכת ההפעלה מספקת את המימוש הנדרש. עבור IID פרטי צריך לספק תמיכת marshalling מתאימה, למשל קוד שנוצר מIDL או type library מתאימה.

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

שולחים קריאה לSTA אמיתי

נשתמש בIClassFactory, ממשק תקני שיש עבורו תמיכת marshalling. המימוש במעבדה הוא probe: הוא מדפיס באיזה thread הופעלה CreateInstance ומחזיר E_NOINTERFACE. הוא אינו factory שנרשם לצורך activation ואינו מתיימר לנהל server locks. כך אפשר לבדוק את מסלול הקריאה בלי להתקין רכיב או לכתוב proxy פרטי.

שמרו את התוכנית כmarshal.cpp. הthread הראשי הוא STA והworker הוא MTA. מצב block מוסיף תחילה המתנה שאינה מעבדת הודעות, מוגבלת לשנייה, כדי לראות מה חסר לפני שמתחילים pumping.

#include <windows.h>
#include <objbase.h>
#include <atomic>
#include <cstdio>
#include <cstring>
#include <new>

class Probe final : public IClassFactory {
    std::atomic<ULONG> refs_{1};
public:
    HRESULT STDMETHODCALLTYPE QueryInterface(REFIID iid, void** out) override {
        if (!out) return E_POINTER;
        *out = nullptr;
        if (!IsEqualIID(iid, IID_IUnknown) && !IsEqualIID(iid, IID_IClassFactory))
            return E_NOINTERFACE;
        *out = static_cast<IClassFactory*>(this);
        AddRef();
        return S_OK;
    }
    ULONG STDMETHODCALLTYPE AddRef() override { return ++refs_; }
    ULONG STDMETHODCALLTYPE Release() override {
        const ULONG n = --refs_;
        if (!n) delete this;
        return n;
    }
    HRESULT STDMETHODCALLTYPE CreateInstance(IUnknown*, REFIID, void** out) override {
        if (!out) return E_POINTER;
        *out = nullptr;
        std::printf("object method tid=%lu\n", GetCurrentThreadId());
        return E_NOINTERFACE;
    }
    HRESULT STDMETHODCALLTYPE LockServer(BOOL) override { return E_NOTIMPL; }
};

struct Transfer {
    IStream* packet = nullptr;
    bool unmarshaled = false;
    HRESULT result = E_PENDING;
};

DWORD WINAPI worker(void* raw) {
    auto& transfer = *static_cast<Transfer*>(raw);
    const HRESULT init = CoInitializeEx(nullptr, COINIT_MULTITHREADED);
    if (FAILED(init)) { transfer.result = init; return 1; }
    std::printf("worker tid=%lu\n", GetCurrentThreadId());
    IClassFactory* proxy = nullptr;
    transfer.result = CoUnmarshalInterface(transfer.packet, IID_IClassFactory,
                                          reinterpret_cast<void**>(&proxy));
    if (SUCCEEDED(transfer.result)) {
        transfer.unmarshaled = true;
        IUnknown* unused = nullptr;
        transfer.result = proxy->CreateInstance(nullptr, IID_IUnknown,
                                                 reinterpret_cast<void**>(&unused));
        if (unused) unused->Release();
        proxy->Release();
    }
    CoUninitialize();
    return 0;
}

void abandon_packet(Transfer& transfer) {
    if (!transfer.packet) return;
    if (!transfer.unmarshaled) {
        LARGE_INTEGER start{};
        HRESULT hr = transfer.packet->Seek(start, STREAM_SEEK_SET, nullptr);
        if (SUCCEEDED(hr)) hr = CoReleaseMarshalData(transfer.packet);
        if (FAILED(hr)) std::printf("release marshal data: %08lx\n", (ULONG)hr);
    }
    transfer.packet->Release();
    transfer.packet = nullptr;
}

int main(int argc, char** argv) {
    const HRESULT init = CoInitializeEx(nullptr, COINIT_APARTMENTTHREADED);
    if (FAILED(init)) return 1;
    std::printf("owner STA tid=%lu\n", GetCurrentThreadId());
    auto object = new (std::nothrow) Probe;
    if (!object) { CoUninitialize(); return 1; }
    Transfer transfer;
    const HRESULT hr = CoMarshalInterThreadInterfaceInStream(
        IID_IClassFactory, object, &transfer.packet);
    object->Release();
    if (FAILED(hr)) { CoUninitialize(); return 2; }
    HANDLE thread = CreateThread(nullptr, 0, worker, &transfer, 0, nullptr);
    if (!thread) {
        abandon_packet(transfer);
        CoUninitialize();
        return 3;
    }
    if (argc == 2 && std::strcmp(argv[1], "block") == 0) {
        const DWORD waited = WaitForSingleObject(thread, 1000);
        std::printf("non-pumping wait=%lu\n", waited);
    }
    const ULONGLONG start = GetTickCount64();
    for (;;) {
        if (GetTickCount64() - start > 10000) {
            std::fputs("lab watchdog: process exits, worker not reclaimed\n", stderr);
            ExitProcess(4);
        }
        const DWORD result = MsgWaitForMultipleObjectsEx(
            1, &thread, 100, QS_ALLINPUT, MWMO_INPUTAVAILABLE);
        if (result == WAIT_OBJECT_0) break;
        if (result == WAIT_FAILED) ExitProcess(5);
        if (result == WAIT_OBJECT_0 + 1) {
            MSG message{};
            while (PeekMessageW(&message, nullptr, 0, 0, PM_REMOVE)) {
                TranslateMessage(&message);
                DispatchMessageW(&message);
            }
        }
    }
    CloseHandle(thread);
    std::printf("call result=%08lx\n", (ULONG)transfer.result);
    abandon_packet(transfer);
    CoUninitialize();
    return transfer.unmarshaled && transfer.result == E_NOINTERFACE ? 0 : 6;
}

cl /std:c++20 /EHsc /W4 marshal.cpp ole32.lib user32.lib uuid.lib
marshal.exe
marshal.exe block

נשווה את הthread של הworker לthread שמדפיס מתוך המתודה. במעבר התקין המתודה אמורה להתבצע בSTA המקורי. בחרנו בCreateInstance משום שLockServer של proxy לclass factory יכולה להיות מטופלת בתשתית ואינה בדיקה שקולה להרצת המתודה שלנו בצד השרת. ההמתנה הראשונית במצב block עשויה להראות timeout לפני שהלולאה מאפשרת את הקריאה. זאת אינה בדיקת ביצועים; העיכוב הוכנס בכוונה כדי לחשוף תלות.

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

גם packet מחזיק חיים

אחרי שיצרנו packet אנחנו משחררים את הreference המקומי לאובייקט. הpacket שומר את ההתחייבות הנדרשת כדי שהצד השני יוכל לקבל ממשק. אם הworker לא נוצר, אסור להתייחס לstream כאל סתם מערך בתים ולשכוח אותו. צריך לשחרר את נתוני הmarshalling בapartment שיצר אותם.

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

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

מעבר לתהליך אחר

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

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

לשיתוף חוזר בין apartments אפשר להשתמש בGlobal Interface Table, שמחזיר לכל apartment ממשק מתאים לפי cookie. צריך לבטל את הרישום בסיום. יש גם אובייקטים agile וfree-threaded marshaler, אך הם דורשים שהמימוש באמת עומד בחוזה המקביליות והמרשלינג. הוספת מנגנון כזה לאובייקט שמחזיק state של UI אינה קיצור דרך להפיכתו לבטוח.