Workers Pool WinApi

[Jeffs]

В общем. Нужно было реализовать пул воркеров, я как обычно пошёл в гугл, в поисках костылей. Но костыли писать не пришлось, как оказалось, винда их реализовала за меня (https://docs.microsoft.com/en-us/windows/win32/procthread/using-the-thread-pool-functions). Почитал доки, реализовал:

#include <windows.h>
#include <stdio.h>

#define WORK_ITEMS_COUNT 10
#define WORKERS_COUNT 2

struct WorkItem
{
    SLIST_ENTRY entry;
    ULONG data;
};

void CALLBACK ProcessItem(PTP_CALLBACK_INSTANCE Instance, PVOID Context, PTP_WORK Work)
{
    PSLIST_ENTRY listEntry = nullptr;
    WorkItem* item = nullptr;

    item = (WorkItem*)InterlockedPopEntrySList((PSLIST_HEADER)Context);
    wprintf(L"worker: tid = %d, data = %d\n", GetCurrentThreadId(), item->data);
    _aligned_free(listEntry);
}

PSLIST_HEADER InitWorkItems()
{
    PSLIST_HEADER listHead;
    WorkItem* item;

    listHead = (PSLIST_HEADER)_aligned_malloc(sizeof(SLIST_HEADER), MEMORY_ALLOCATION_ALIGNMENT);
    InitializeSListHead(listHead);

    for (int i = 0; i < WORK_ITEMS_COUNT; i++)
    {
        item = (WorkItem*)_aligned_malloc(sizeof(WorkItem), MEMORY_ALLOCATION_ALIGNMENT);
        item->data = i;
        InterlockedPushEntrySList(listHead, &(item->entry));
    }

    return listHead;
}

int main()
{
    PSLIST_HEADER workItems = nullptr;
    TP_CALLBACK_ENVIRON callbackEnv;
    PTP_POOL pool = nullptr;
    PTP_WORK work = nullptr;

    workItems = InitWorkItems();

    InitializeThreadpoolEnvironment(&callbackEnv);
    pool = CreateThreadpool(nullptr);

    SetThreadpoolThreadMaximum(pool, WORKERS_COUNT);

    SetThreadpoolCallbackPool(&callbackEnv, pool);

    work = CreateThreadpoolWork(ProcessItem, (LPVOID)workItems, &callbackEnv);
   
    for (int i = 0; i < WORK_ITEMS_COUNT; i++)
        SubmitThreadpoolWork(work);

    WaitForThreadpoolWorkCallbacks(work, FALSE);

    _aligned_free(workItems);

    CloseThreadpool(pool);
}

Надеюсь эта заметка кому-то скоротает вечерок, как и мне

 

[premiumcat]

Как я понял код нужен для того что бы запустить нескалько функций ProcessItem, что бы они отработали десять раз, но работало одновременно две? Я просто учусь только

 

[Jeffs]

Как я понял код нужен для того что бы запустить нескалько функций ProcessItem, что бы они отработали десять раз, но работало одновременно две? Я просто учусь только

Да, воркер - отдельный поток. Все задачи делятся между воркерами (в данном случае воркера только 2). Можно убрать ограничение на кол-во воркеров - под каждую задачу запустится новый поток. Все задачи передаются в виде односвязного списка, но можно и просто создать несколько задач (CreateThreadpoolWork()) и ожидать их завершения.

 

[DildoFagins]

Ну если придираться, то использование атомарного SList'а кажется по сути избыточным. Во-первых, когда ты заполняешь список, то делать атомарно еще не нужно (это происходит в одном потоке), удаление элемента из списка, это да, но это можно было бы сделать и любым другим примитивом для синхронизации потоков, тебе же интерлоками флаг в структуре выставлять. Во-вторых элементы списка в описанной реализации должны быть выровнены, что тоже кажется избыточным.

 

[Jeffs]

Ну если придираться, то использование атомарного SList'а кажется по сути избыточным. Во-первых, когда ты заполняешь список, то делать атомарно еще не нужно (это происходит в одном потоке), удаление элемента из списка, это да, но это можно было бы сделать и любым другим примитивом для синхронизации потоков, тебе же интерлоками флаг в структуре выставлять. Во-вторых элементы списка в описанной реализации должны быть выровнены, что тоже кажется избыточным.

У меня в коде все данные прилетают в виде односвязного списка, поэтому в этом примере решил тоже его использовать. Спасибо, учту на будущее.