power_play/src/base/win32/base_win32_job.h

////////////////////////////////
//~ Win32 headers

#pragma warning(push, 0)
#  define UNICODE
#  define WIN32_LEAN_AND_MEAN
#  include <Windows.h>
#  include <WinSock2.h>
#  include <TlHelp32.h>
#  include <WS2tcpip.h>
#  include <windowsx.h>
#  include <ShlObj_core.h>
#  include <fileapi.h>
#  include <dwmapi.h>
#  include <bcrypt.h>
#  include <avrt.h>
#  include <shellapi.h>
#pragma warning(pop)

////////////////////////////////
//~ Thread types

#define W32_ThreadStackSize Kibi(64)
#define W32_ThreadDef(name, arg_name) void name(void *arg_name)
typedef W32_ThreadDef(W32_ThreadFunc, data);

Struct(W32_Thread)
{
    W32_ThreadFunc *entry_point;
    void *thread_data;
    char thread_name_cstr[256];
    wchar_t thread_name_wstr[256];
    i32 profiler_group;

    W32_Thread *next;
    W32_Thread *prev;

    HANDLE handle;
};

////////////////////////////////
//~ Wait list types

AlignedStruct(W32_WaitList, 64)
{
    u64 value;
    i16 first_waiter;
    i16 last_waiter;
    i32 num_waiters;
    W32_WaitList *next_in_bin;
    W32_WaitList *prev_in_bin;
};
StaticAssert(alignof(W32_WaitList) == 64);  /* Avoid false sharing */

AlignedStruct(W32_WaitBin, 64)
{
    W32_WaitList *first_wait_list;
    W32_WaitList *last_wait_list;
    W32_WaitList *first_free_wait_list;
    TicketMutex lock;
};
StaticAssert(alignof(W32_WaitBin) == 64);  /* Avoid false sharing */

////////////////////////////////
//~ Fiber types

#define W32_FiberStackSize Mebi(4)
#define W32_FiberNamePrefixCstr "Fiber ["
#define W32_FiberNameSuffixCstr "]"
#define W32_FiberNameMaxSize 64

//- Yield param
typedef i32 W32_YieldKind; enum
{
    W32_YieldKind_None,
    W32_YieldKind_Done,
    W32_YieldKind_Wait,

    W32_YieldKind_Count
};

Struct(W32_YieldParam)
{
    W32_YieldKind kind;
    union
    {
        struct
        {
            volatile void *addr;
            void *cmp;
            u32 size;
            i64 timeout_ns;
        } wait;
    };
};

//- Fiber
AlignedStruct(W32_Fiber, 64)
{
    /* ---------------------------------------------------- */
    void *addr;                                 /* 08 bytes */
    /* ---------------------------------------------------- */
    char *name_cstr;                            /* 08 bytes */
    /* ---------------------------------------------------- */
    Atomic32 wake_lock;                         /* 04 bytes (4 byte alignment) */
    i16 id;                                     /* 02 bytes */
    i16 parent_id;                              /* 02 bytes */
    /* ---------------------------------------------------- */
    u64 wait_addr;                              /* 08 bytes */
    /* ---------------------------------------------------- */
    u64 wait_time;                              /* 08 bytes */
    /* ---------------------------------------------------- */
    i16 next_addr_waiter;                       /* 02 bytes */
    i16 prev_addr_waiter;                       /* 02 bytes */
    i16 next_time_waiter;                       /* 02 bytes */
    i16 prev_time_waiter;                       /* 02 bytes */
    /* ---------------------------------------------------- */
    GenericJobDesc *job_desc;                   /* 08 bytes */
    /* ---------------------------------------------------- */
    /* -------------------- Cache line -------------------- */
    /* ---------------------------------------------------- */
    i32 job_id;                                 /* 04 bytes */
    i16 job_pool;                               /* 02 bytes */
    i16 job_priority;                           /* 02 bytes */
    /* ---------------------------------------------------- */
    W32_YieldParam *yield_param;                /* 08 bytes */
    /* ---------------------------------------------------- */
    u8 _pad0[48];                               /* 48 bytes (padding) */

};
StaticAssert(sizeof(W32_Fiber) == 128);  /* Padding validation (increase if necessary) */
StaticAssert(alignof(W32_Fiber) == 64);  /* Verify alignment to avoid false sharing */
StaticAssert(offsetof(W32_Fiber, wake_lock) % 4 == 0);  /* Atomic must be aligned */

////////////////////////////////
//~ Job queue types

//- Worker ctx
AlignedStruct(W32_WorkerCtx, 64)
{
    JobPool pool_kind;
    i32 id;
};

//- Job info
Struct(W32_JobInfo)
{
    GenericJobDesc *desc;

    i32 count;
    i32 num_dispatched;
    i16 fiber_id;  /* If the job is being resumed from a yield */

    W32_JobInfo *next;
};

//- Job queue
AlignedStruct(W32_JobQueue, 64)
{
    TicketMutex lock;
    Arena *arena;

    W32_JobInfo *first;
    W32_JobInfo *last;

    W32_JobInfo *first_free;
};

//- Job pool
AlignedStruct(W32_JobPool, 64)
{
    /* Jobs */
    W32_JobQueue job_queues[JobPriority_Count];

    TicketMutex free_fibers_tm;
    i16 first_free_fiber_id;

    /* Workers */
    Atomic32Padded workers_shutdown;
    Atomic64Padded num_jobs_in_queue;
    TicketMutex workers_wake_tm;

    i32 num_worker_threads;
    i32 thread_priority;
    u64 thread_affinity_mask;
    b32 thread_is_audio;
    Arena *worker_threads_arena;
    W32_Thread **worker_threads;
    W32_WorkerCtx *worker_contexts;
};

////////////////////////////////
//~ Shared state

/* Assume scheduler cycle is 20hz at start to be conservative */
#define W32_DefaultSchedulerPeriodNs 50000000
#define W32_NumRollingSchedulerPeriods 1000

#define W32_NumWaitAddrBins 16384
#define W32_NumWaitTimeBins 1024

/* Arbitrary threshold for determining when to fall back from a looped WakeByAddressSingle to WakeByAddressAll */
#define W32_WakeAllThreshold 16

Struct(W32_SharedJobCtx)
{
    i64 timer_start_qpc;
    i64 ns_per_qpc;
    Atomic32 shutdown;

    //- Worker thread pool
    TicketMutex threads_tm;
    Arena *threads_arena;
    W32_Thread *first_thread;
    W32_Thread *last_thread;
    W32_Thread *first_free_thread;

    //- Scheduler
    Atomic64Padded current_scheduler_cycle;
    Atomic64Padded current_scheduler_cycle_period_ns;

    //- Fibers
    TicketMutex fibers_tm;
    i16 num_fibers;
    Arena *fiber_names_arena;
    W32_Fiber fibers[MaxFibers];

    //- Job descs
    Arena *job_descs_arena;
    TicketMutex job_descs_tm;
    GenericJobDesc *first_free_job_desc;

    //- Wait lists
    Atomic64Padded waiter_wake_gen;
    TicketMutex wait_lists_arena_tm;
    Arena *wait_lists_arena;

    //- Wait tables
    W32_WaitBin wait_addr_bins[W32_NumWaitAddrBins];
    W32_WaitBin wait_time_bins[W32_NumWaitTimeBins];

    //- Job pools
    W32_JobPool job_pools[JobPool_Count];
};

extern W32_SharedJobCtx W32_shared_job_ctx;

////////////////////////////////
//~ Startup

void StartupJobs(void);

////////////////////////////////
//~ Shutdown

void ShutdownJobs(void);

////////////////////////////////
//~ Thread operations

DWORD WINAPI W32_Win32ThreadProc(LPVOID vt);
W32_Thread *W32_AllocThread(W32_ThreadFunc *entry_point, void *thread_data, String thread_name, i32 profiler_group);
b32 W32_TryReleaseThread(W32_Thread *thread, f32 timeout_seconds);
void W32_WaitReleaseThread(W32_Thread *thread);

////////////////////////////////
//~ Wait list operations

void W32_WakeLockedFibers(i32 num_fibers, W32_Fiber **fibers);
void W32_WakeByAddress(void *addr, i32 count);
void W32_WakeByTime(u64 time);

////////////////////////////////
//~ Fiber operations

W32_Fiber *W32_AllocFiber(W32_JobPool *pool);
void W32_ReleaseFiber(W32_JobPool *pool, W32_Fiber *fiber);
ForceInline W32_Fiber *W32_FiberFromId(i16 id);
ForceNoInline void W32_FiberResume(W32_Fiber *fiber);
void W32_YieldFiber(W32_Fiber *fiber, W32_Fiber *parent_fiber);

////////////////////////////////
//~ Fiber entry

void W32_FiberEntryPoint(void *id_ptr);

////////////////////////////////
//~ Job worker entry

W32_ThreadDef(W32_JobWorkerEntryFunc, worker_ctx_arg);

////////////////////////////////
//~ Job scheduler entry

W32_ThreadDef(W32_JobSchedulerEntryFunc, _);