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

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

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

מאגדת עבודה עם ניקוי משותף

התוכנית הבאה יוצרת מאגר קטן, אובייקט work ו־timer. שניהם משתמשים באותו context ומצורפים ל־cleanup group.

#define WIN32_LEAN_AND_MEAN
#include <windows.h>
#include <atomic>
#include <cstdio>

struct Context {
    std::atomic<unsigned> work_calls{0};
    std::atomic<unsigned> timer_calls{0};
};

void CALLBACK on_work(PTP_CALLBACK_INSTANCE, void* raw, PTP_WORK) {
    auto& context = *static_cast<Context*>(raw);
    context.work_calls.fetch_add(1, std::memory_order_relaxed);
}

void CALLBACK on_timer(PTP_CALLBACK_INSTANCE, void* raw, PTP_TIMER) {
    auto& context = *static_cast<Context*>(raw);
    context.timer_calls.fetch_add(1, std::memory_order_relaxed);
}

int main() {
    Context context;
    PTP_POOL pool = CreateThreadpool(nullptr);
    if (!pool) return 1;
    SetThreadpoolThreadMaximum(pool, 2);
    if (!SetThreadpoolThreadMinimum(pool, 1)) {
        CloseThreadpool(pool);
        return 2;
    }
    PTP_CLEANUP_GROUP group = CreateThreadpoolCleanupGroup();
    if (!group) {
        CloseThreadpool(pool);
        return 3;
    }
    TP_CALLBACK_ENVIRON environment;
    InitializeThreadpoolEnvironment(&environment);
    SetThreadpoolCallbackPool(&environment, pool);
    SetThreadpoolCallbackCleanupGroup(&environment, group, nullptr);

    PTP_WORK work = CreateThreadpoolWork(on_work, &context, &environment);
    PTP_TIMER timer = work ?
        CreateThreadpoolTimer(on_timer, &context, &environment) : nullptr;
    if (!work || !timer) {
        CloseThreadpoolCleanupGroupMembers(group, TRUE, nullptr);
        DestroyThreadpoolEnvironment(&environment);
        CloseThreadpoolCleanupGroup(group);
        CloseThreadpool(pool);
        return 4;
    }

    for (unsigned i = 0; i < 5; ++i) SubmitThreadpoolWork(work);
    LARGE_INTEGER due{};
    due.QuadPart = -50LL * 10000;
    FILETIME time{due.LowPart, static_cast<DWORD>(due.HighPart)};
    SetThreadpoolTimer(timer, &time, 20, 0);
    Sleep(120);

    SetThreadpoolTimer(timer, nullptr, 0, 0);
    CloseThreadpoolCleanupGroupMembers(group, FALSE, nullptr);
    DestroyThreadpoolEnvironment(&environment);
    CloseThreadpoolCleanupGroup(group);
    CloseThreadpool(pool);

    std::printf("work=%u timer=%u\n",
        context.work_calls.load(), context.timer_calls.load());
}

cl /nologo /std:c++20 /EHsc /W4 pool.cpp /Fepool.exe

המונים atomic מפני שיותר מ־callback אחד יכול לרוץ במקביל. הסדר relaxed מספיק כאן לפעולת ספירה בלבד; הוא אינו מפרסם מבנה נתונים אחר. את ההבחנה הזו נעמיק בפרק מודל הזיכרון.

ביטול מקור האירועים אינו סוף callback

הקריאה עם זמן ריק ל־SetThreadpoolTimer מפסיקה את לוח הזמנים העתידי. היא אינה הוכחה ש־callback שכבר נבחר או התחיל הפסיק להשתמש ב־context. לכן אנחנו ממשיכים ל־cleanup group שממתינה לסיום.

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

אחרי CloseThreadpoolCleanupGroupMembers האובייקטים שהיו חברי הקבוצה כבר שוחררו. אין לקרוא גם CloseThreadpoolWork או CloseThreadpoolTimer על אותם מצביעים. זו דוגמה לממשק שבו פעולת group מעבירה כמה מצבי בעלות יחד.

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

ה־Context נבנה לפני אובייקטי המאגר ונשאר חי עד אחרי ה־drain. אם היה אובייקט מקומי בפונקציה שחוזרת מיד אחרי SubmitThreadpoolWork, המצביע שה־callback מחזיק היה הופך לתלוי.

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

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

לא ממתינים לעצמנו

המתנה לסיום callbacks מתוך callback של אותה קבוצה יכולה ליצור תלות עצמית או מעגל עם משאבים אחרים. גם החזקת mutex בזמן drain מסוכנת אם callback צריך אותו כדי לסיים.

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

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

אין להניח ש־callbacks של אותו work תמיד רצים על אותו thread. שימוש ב־TLS כדי לשמור state של בקשה יכול להשאיר נתונים מבקשה קודמת או לפצל אותם בין עובדים.

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

סביבת session אינה אובייקט של המאגר

המאגר פועל בתוך סביבת התהליך. הוא אינו משנה session, window station או desktop. שירות ב־session 0 אינו מקבל גישה לשולחן העבודה האינטראקטיבי רק מפני שה־callback שלו רץ על thread אחר.

ה־window station וה־desktop הם חלק מארגון סביבת הממשק והאבטחה. פונקציות שמחזירות handle לסביבה הנוכחית יכולות להיות borrowing ולא רכישה חדשה; צריך לבדוק אם מותר לסגור אותן. בפרק שולחן העבודה נבנה חלון מתוך ההקשר המתאים ונחבר את התמונה הזאת למערכת ההודעות.

לקריאה נוספת: שימוש במאגרי threads וחוזה ניקוי חברי קבוצה.