taskqueue.c File Reference

#include "quakedef.h"
#include "taskqueue.h"

Include dependency graph for taskqueue.c:

Go to the source code of this file.

Data Structures
struct	taskqueue_state_t
struct	taskqueue_state_thread_t

Macros
#define	MAXTHREADS   1024
#define	RECENTFRAMES   64
#define	THREADBATCH   64
#define	THREADSLEEPCOUNT   1000
#define	THREADTASKS   256

Functions
static void	TaskQueue_DistributeTasks (void)
void	TaskQueue_Enqueue (int numtasks, taskqueue_task_t *tasks)
static void	TaskQueue_ExecuteTask (taskqueue_task_t *t)
void	TaskQueue_Frame (qbool shutdown)
void	TaskQueue_Init (void)
qbool	TaskQueue_IsDone (taskqueue_task_t *t)
void	TaskQueue_Setup (taskqueue_task_t *t, taskqueue_task_t *preceding, void(*func)(taskqueue_task_t *), size_t i0, size_t i1, void *p0, void *p1)
void	TaskQueue_Shutdown (void)
void	TaskQueue_Task_CheckTasksDone (taskqueue_task_t *t)
static int	TaskQueue_ThreadFunc (void *d)
void	TaskQueue_WaitForTaskDone (taskqueue_task_t *t)
void	TaskQueue_Yield (taskqueue_task_t *t)

Variables
cvar_t	taskqueue_maxthreads = {CF_CLIENT | CF_SERVER | CF_ARCHIVE, "taskqueue_maxthreads", "32", "maximum number of threads to start up as needed based on task count"}
cvar_t	taskqueue_minthreads = {CF_CLIENT | CF_SERVER | CF_ARCHIVE, "taskqueue_minthreads", "0", "minimum number of threads to keep active for executing tasks"}
static taskqueue_state_t	taskqueue_state
cvar_t	taskqueue_tasksperthread = {CF_CLIENT | CF_SERVER | CF_ARCHIVE, "taskqueue_tasksperthread", "4000", "expected amount of work that a single thread can do in a frame - the number of threads being used depends on the average workload in recent frames"}

Macro Definition Documentation

MAXTHREADS

#define MAXTHREADS   1024

Definition at line 8 of file taskqueue.c.

Referenced by TaskQueue_Frame().

RECENTFRAMES

#define RECENTFRAMES   64

Definition at line 9 of file taskqueue.c.

Referenced by TaskQueue_Frame().

THREADBATCH

#define THREADBATCH   64

Definition at line 11 of file taskqueue.c.

THREADSLEEPCOUNT

#define THREADSLEEPCOUNT   1000

Definition at line 12 of file taskqueue.c.

Referenced by TaskQueue_ThreadFunc().

THREADTASKS

#define THREADTASKS   256

Definition at line 10 of file taskqueue.c.

Referenced by TaskQueue_DistributeTasks(), and TaskQueue_ThreadFunc().

Function Documentation

TaskQueue_DistributeTasks()

static void TaskQueue_DistributeTasks ( void )

static

Definition at line 144 of file taskqueue.c.

{
    Thread_AtomicLock(&taskqueue_state.command_lock);
    if (taskqueue_state.numthreads > 0)
    {
        unsigned int attempts = taskqueue_state.numthreads;
        while (attempts-- > 0 && taskqueue_state.queue_enqueueposition != taskqueue_state.queue_dequeueposition)
        {
            taskqueue_task_t *t = taskqueue_state.queue_data[taskqueue_state.queue_dequeueposition];
            if (t->preceding && t->preceding->done == 0)
            {
                // task is waiting on something
                // first dequeue it properly
                taskqueue_state.queue_data[taskqueue_state.queue_dequeueposition] = NULL;
                taskqueue_state.queue_dequeueposition++;
                if (taskqueue_state.queue_dequeueposition >= taskqueue_state.queue_size)
                    taskqueue_state.queue_dequeueposition = 0;
                // now put it back in the distributor queue - we know there is room because we just made room
                taskqueue_state.queue_data[taskqueue_state.queue_enqueueposition] = t;
                taskqueue_state.queue_enqueueposition++;
                if (taskqueue_state.queue_enqueueposition >= taskqueue_state.queue_size)
                    taskqueue_state.queue_enqueueposition = 0;
                // we do not refresh the attempt counter here to avoid deadlock - quite often the only things sitting in the distributor queue are waiting on other tasks
            }
            else
            {
                taskqueue_state_thread_t *s = &taskqueue_state.threads[taskqueue_state.enqueuethread];
                if (s->enqueueposition - s->dequeueposition < THREADTASKS)
                {
                    // add the task to the thread's queue
                    s->queue[(s->enqueueposition++) % THREADTASKS] = t;
                    // since we succeeded in assigning the task, advance the distributor queue
                    taskqueue_state.queue_data[taskqueue_state.queue_dequeueposition] = NULL;
                    taskqueue_state.queue_dequeueposition++;
                    if (taskqueue_state.queue_dequeueposition >= taskqueue_state.queue_size)
                        taskqueue_state.queue_dequeueposition = 0;
                    // refresh our attempt counter because we did manage to assign something to a thread
                    attempts = taskqueue_state.numthreads;
                }
            }
        }
    }
    Thread_AtomicUnlock(&taskqueue_state.command_lock);
    // execute one pending task on the distributor queue, this matters if numthreads is 0
    if (taskqueue_state.queue_dequeueposition != taskqueue_state.queue_enqueueposition)
    {
        taskqueue_task_t *t = taskqueue_state.queue_data[taskqueue_state.queue_dequeueposition];
        taskqueue_state.queue_dequeueposition++;
        if (taskqueue_state.queue_dequeueposition >= taskqueue_state.queue_size)
            taskqueue_state.queue_dequeueposition = 0;
        if (t)
            TaskQueue_ExecuteTask(t);
    }
}

References taskqueue_state_t::command_lock, taskqueue_state_thread_t::dequeueposition, taskqueue_state_thread_t::enqueueposition, taskqueue_state_t::enqueuethread, NULL, taskqueue_state_t::numthreads, taskqueue_task_t::preceding, taskqueue_state_thread_t::queue, taskqueue_state_t::queue_data, taskqueue_state_t::queue_dequeueposition, taskqueue_state_t::queue_enqueueposition, taskqueue_state_t::queue_size, TaskQueue_ExecuteTask(), taskqueue_state, Thread_AtomicLock, Thread_AtomicUnlock, taskqueue_state_t::threads, and THREADTASKS.

Referenced by TaskQueue_Frame(), and TaskQueue_WaitForTaskDone().

TaskQueue_Enqueue()

void TaskQueue_Enqueue	(	int	numtasks,
		taskqueue_task_t *	tasks )

Definition at line 105 of file taskqueue.c.

{
    int i;
    Thread_AtomicLock(&taskqueue_state.command_lock);
    if (taskqueue_state.queue_size <
        (taskqueue_state.queue_enqueueposition < taskqueue_state.queue_dequeueposition ? taskqueue_state.queue_size : 0) +
        taskqueue_state.queue_enqueueposition - taskqueue_state.queue_dequeueposition + numtasks)
    {
        // we have to grow the queue...
        unsigned int newsize = (taskqueue_state.queue_size + numtasks) * 2;
        if (newsize < 1024)
            newsize = 1024;
        taskqueue_state.queue_data = (taskqueue_task_t **)Mem_Realloc(zonemempool, taskqueue_state.queue_data, sizeof(*taskqueue_state.queue_data) * newsize);
        taskqueue_state.queue_size = newsize;
    }
    for (i = 0; i < numtasks; i++)
    {
        if (tasks[i].yieldcount == 0)
            taskqueue_state.tasks_thisframe++;
        taskqueue_state.queue_data[taskqueue_state.queue_enqueueposition] = &tasks[i];
        taskqueue_state.queue_enqueueposition++;
        if (taskqueue_state.queue_enqueueposition >= taskqueue_state.queue_size)
            taskqueue_state.queue_enqueueposition = 0;
    }
    Thread_AtomicUnlock(&taskqueue_state.command_lock);
}

References taskqueue_state_t::command_lock, i, Mem_Realloc, taskqueue_state_t::queue_data, taskqueue_state_t::queue_dequeueposition, taskqueue_state_t::queue_enqueueposition, taskqueue_state_t::queue_size, taskqueue_state, taskqueue_state_t::tasks_thisframe, Thread_AtomicLock, Thread_AtomicUnlock, and zonemempool.

Referenced by R_Shadow_BounceGrid_EnqueuePhotons_Task(), R_Shadow_BounceGrid_EnqueueSlices_Task(), R_Shadow_UpdateBounceGridTexture(), and TaskQueue_Yield().

TaskQueue_ExecuteTask()

static void TaskQueue_ExecuteTask ( taskqueue_task_t * t )

static

Definition at line 66 of file taskqueue.c.

{
    // see if t is waiting on something
    if (t->preceding && t->preceding->done == 0)
        TaskQueue_Yield(t);
    else
        t->func(t);
}

References taskqueue_task_t::func, taskqueue_task_t::preceding, and TaskQueue_Yield().

Referenced by TaskQueue_DistributeTasks(), and TaskQueue_ThreadFunc().

TaskQueue_Frame()

void TaskQueue_Frame

(

qbool

shutdown

)

Definition at line 213 of file taskqueue.c.

{
    int i;
    unsigned long long int avg;
    int maxthreads = bound(0, taskqueue_maxthreads.integer, MAXTHREADS);
    int numthreads = maxthreads;
    int tasksperthread = bound(10, taskqueue_tasksperthread.integer, 100000);
#ifdef THREADDISABLE
    numthreads = 0;
#endif
 
    Thread_AtomicLock(&taskqueue_state.command_lock);
    taskqueue_state.tasks_recentframesindex = (taskqueue_state.tasks_recentframesindex + 1) % RECENTFRAMES;
    taskqueue_state.tasks_recentframes[taskqueue_state.tasks_recentframesindex] = taskqueue_state.tasks_thisframe;
    taskqueue_state.tasks_thisframe = 0;
    avg = 0;
    for (i = 0; i < RECENTFRAMES; i++)
        avg += taskqueue_state.tasks_recentframes[i];
    taskqueue_state.tasks_averageperframe = avg / RECENTFRAMES;
    Thread_AtomicUnlock(&taskqueue_state.command_lock);
 
    numthreads = taskqueue_state.tasks_averageperframe / tasksperthread;
    numthreads = bound(taskqueue_minthreads.integer, numthreads, taskqueue_maxthreads.integer);
 
    if (shutdown)
        numthreads = 0;
 
    // check if we need to close some threads
    if (taskqueue_state.numthreads > numthreads)
    {
        // tell extra threads to quit
        Thread_AtomicLock(&taskqueue_state.command_lock);
        for (i = numthreads; i < taskqueue_state.numthreads; i++)
            taskqueue_state.threads[i].quit = 1;
        Thread_AtomicUnlock(&taskqueue_state.command_lock);
        for (i = numthreads; i < taskqueue_state.numthreads; i++)
        {
            if (taskqueue_state.threads[i].handle)
                Thread_WaitThread(taskqueue_state.threads[i].handle, 0);
            taskqueue_state.threads[i].handle = NULL;
        }
        // okay we're at the new state now
        taskqueue_state.numthreads = numthreads;
    }
 
    // check if we need to start more threads
    if (taskqueue_state.numthreads < numthreads)
    {
        // make sure we're not telling new threads to just quit on startup
        Thread_AtomicLock(&taskqueue_state.command_lock);
        for (i = taskqueue_state.numthreads; i < numthreads; i++)
            taskqueue_state.threads[i].quit = 0;
        Thread_AtomicUnlock(&taskqueue_state.command_lock);
 
        // start new threads
        for (i = taskqueue_state.numthreads; i < numthreads; i++)
        {
            taskqueue_state.threads[i].thread_index = i;
            taskqueue_state.threads[i].handle = Thread_CreateThread(TaskQueue_ThreadFunc, &taskqueue_state.threads[i]);
        }
 
        // okay we're at the new state now
        taskqueue_state.numthreads = numthreads;
    }
 
    // just for good measure, distribute any pending tasks that span across frames
    TaskQueue_DistributeTasks();
}

References bound, taskqueue_state_t::command_lock, taskqueue_state_thread_t::handle, i, cvar_t::integer, MAXTHREADS, NULL, taskqueue_state_t::numthreads, taskqueue_state_thread_t::quit, RECENTFRAMES, TaskQueue_DistributeTasks(), taskqueue_maxthreads, taskqueue_minthreads, taskqueue_state, taskqueue_tasksperthread, TaskQueue_ThreadFunc(), taskqueue_state_t::tasks_averageperframe, taskqueue_state_t::tasks_recentframes, taskqueue_state_t::tasks_recentframesindex, taskqueue_state_t::tasks_thisframe, Thread_AtomicLock, Thread_AtomicUnlock, Thread_CreateThread, taskqueue_state_thread_t::thread_index, Thread_WaitThread, and taskqueue_state_t::threads.

Referenced by Host_Frame(), and TaskQueue_Shutdown().

TaskQueue_Init()

void TaskQueue_Init

(

void

)

Definition at line 53 of file taskqueue.c.

{
    Cvar_RegisterVariable(&taskqueue_minthreads);
    Cvar_RegisterVariable(&taskqueue_maxthreads);
    Cvar_RegisterVariable(&taskqueue_tasksperthread);
}

References Cvar_RegisterVariable(), taskqueue_maxthreads, taskqueue_minthreads, and taskqueue_tasksperthread.

Referenced by Host_Init().

TaskQueue_IsDone()

qbool TaskQueue_IsDone

(

taskqueue_task_t *

t

)

Definition at line 139 of file taskqueue.c.

{
    return !!t->done;
}

References taskqueue_task_t::done.

TaskQueue_Setup()

void TaskQueue_Setup	(	taskqueue_task_t *	t,
		taskqueue_task_t *	preceding,
		void(*	func )(taskqueue_task_t *),
		size_t	i0,
		size_t	i1,
		void *	p0,
		void *	p1 )

Definition at line 282 of file taskqueue.c.

{
    memset(t, 0, sizeof(*t));
    t->preceding = preceding;
    t->func = func;
    t->i[0] = i0;
    t->i[1] = i1;
    t->p[0] = p0;
    t->p[1] = p1;
}

References taskqueue_task_t::func, taskqueue_task_t::i, taskqueue_task_t::p, and taskqueue_task_t::preceding.

Referenced by R_Shadow_BounceGrid_EnqueuePhotons_Task(), R_Shadow_BounceGrid_EnqueueSlices_Task(), and R_Shadow_UpdateBounceGridTexture().

TaskQueue_Shutdown()

void TaskQueue_Shutdown

(

void

)

Definition at line 60 of file taskqueue.c.

{
    if (taskqueue_state.numthreads)
        TaskQueue_Frame(true);
}

References taskqueue_state_t::numthreads, TaskQueue_Frame(), and taskqueue_state.

Referenced by Host_Shutdown().

TaskQueue_Task_CheckTasksDone()

void TaskQueue_Task_CheckTasksDone

(

taskqueue_task_t *

t

)

Definition at line 293 of file taskqueue.c.

{
    size_t numtasks = t->i[0];
    taskqueue_task_t *tasks = (taskqueue_task_t *)t->p[0];
    while (numtasks > 0)
    {
        // check the last task first as it's usually going to be the last to finish, so we do the least work by checking it first
        if (!tasks[numtasks - 1].done)
        {
            // update our partial progress, then yield to another pending task.
            t->i[0] = numtasks;
            // set our preceding task to one of the ones we are watching for
            t->preceding = &tasks[numtasks - 1];
            TaskQueue_Yield(t);
            return;
        }
        numtasks--;
    }
    t->done = 1;
}

References taskqueue_task_t::done, taskqueue_task_t::i, taskqueue_task_t::p, taskqueue_task_t::preceding, and TaskQueue_Yield().

Referenced by R_Shadow_BounceGrid_EnqueuePhotons_Task(), and R_Shadow_BounceGrid_EnqueueSlices_Task().

TaskQueue_ThreadFunc()

static int TaskQueue_ThreadFunc ( void * d )

static

Definition at line 77 of file taskqueue.c.

{
    taskqueue_state_thread_t *s = (taskqueue_state_thread_t *)d;
    unsigned int sleepcounter = 0;
    for (;;)
    {
        qbool quit;
        while (s->dequeueposition != s->enqueueposition)
        {
            taskqueue_task_t *t = s->queue[s->dequeueposition % THREADTASKS];
            TaskQueue_ExecuteTask(t);
            // when we advance, also clear the pointer for good measure
            s->queue[s->dequeueposition++ % THREADTASKS] = NULL;
            sleepcounter = 0;
        }
        Thread_AtomicLock(&taskqueue_state.command_lock);
        quit = s->quit != 0;
        Thread_AtomicUnlock(&taskqueue_state.command_lock);
        if (quit)
            break;
        sleepcounter++;
        if (sleepcounter >= THREADSLEEPCOUNT)
            Sys_Sleep(0.001);
        sleepcounter = 0;
    }
    return 0;
}

References taskqueue_state_t::command_lock, taskqueue_state_thread_t::dequeueposition, taskqueue_state_thread_t::enqueueposition, NULL, taskqueue_state_thread_t::queue, taskqueue_state_thread_t::quit, Sys_Sleep(), TaskQueue_ExecuteTask(), taskqueue_state, Thread_AtomicLock, Thread_AtomicUnlock, THREADSLEEPCOUNT, and THREADTASKS.

Referenced by TaskQueue_Frame().

TaskQueue_WaitForTaskDone()

void TaskQueue_WaitForTaskDone

(

taskqueue_task_t *

t

)

Definition at line 199 of file taskqueue.c.

{
    qbool done = false;
    for (;;)
    {
        Thread_AtomicLock(&taskqueue_state.command_lock);
        done = t->done != 0;
        Thread_AtomicUnlock(&taskqueue_state.command_lock);
        if (done)
            break;
        TaskQueue_DistributeTasks();
    }
}

References taskqueue_state_t::command_lock, taskqueue_task_t::done, TaskQueue_DistributeTasks(), taskqueue_state, Thread_AtomicLock, and Thread_AtomicUnlock.

Referenced by R_Shadow_UpdateBounceGridTexture().

TaskQueue_Yield()

void TaskQueue_Yield

(

taskqueue_task_t *

t

)

Definition at line 133 of file taskqueue.c.

{
    t->yieldcount++;
    TaskQueue_Enqueue(1, t);
}

References TaskQueue_Enqueue(), and taskqueue_task_t::yieldcount.

Referenced by R_Shadow_BounceGrid_EnqueueSlices_Task(), TaskQueue_ExecuteTask(), and TaskQueue_Task_CheckTasksDone().

Variable Documentation

taskqueue_maxthreads

cvar_t taskqueue_maxthreads = {CF_CLIENT | CF_SERVER | CF_ARCHIVE, "taskqueue_maxthreads", "32", "maximum number of threads to start up as needed based on task count"}

Definition at line 5 of file taskqueue.c.

{CF_CLIENT | CF_SERVER | CF_ARCHIVE, "taskqueue_maxthreads", "32", "maximum number of threads to start up as needed based on task count"};

Referenced by TaskQueue_Frame(), and TaskQueue_Init().

taskqueue_minthreads

cvar_t taskqueue_minthreads = {CF_CLIENT | CF_SERVER | CF_ARCHIVE, "taskqueue_minthreads", "0", "minimum number of threads to keep active for executing tasks"}

Definition at line 4 of file taskqueue.c.

{CF_CLIENT | CF_SERVER | CF_ARCHIVE, "taskqueue_minthreads", "0", "minimum number of threads to keep active for executing tasks"};

Referenced by TaskQueue_Frame(), and TaskQueue_Init().

taskqueue_state

taskqueue_state_t taskqueue_state

static

Definition at line 51 of file taskqueue.c.

Referenced by TaskQueue_DistributeTasks(), TaskQueue_Enqueue(), TaskQueue_Frame(), TaskQueue_Shutdown(), TaskQueue_ThreadFunc(), and TaskQueue_WaitForTaskDone().

taskqueue_tasksperthread

cvar_t taskqueue_tasksperthread = {CF_CLIENT | CF_SERVER | CF_ARCHIVE, "taskqueue_tasksperthread", "4000", "expected amount of work that a single thread can do in a frame - the number of threads being used depends on the average workload in recent frames"}

Definition at line 6 of file taskqueue.c.

{CF_CLIENT | CF_SERVER | CF_ARCHIVE, "taskqueue_tasksperthread", "4000", "expected amount of work that a single thread can do in a frame - the number of threads being used depends on the average workload in recent frames"};

Referenced by TaskQueue_Frame(), and TaskQueue_Init().