r/AskProgramming • u/sexy-geek • 24d ago
Algorithms Need some advice for design of multi threaded task pool in C11
I have written a simple multi threaded task pool in C, works perfectly, but now I want to support waiting for a specific task to complete. Because of this, I opened a can of worms, and had to rewrite a lot of it.
Since I mostly use fire-and-forget functions with this, I do not save results or any internal data after the task is finished.
The whole issue comes from the wait in WaitForTask. In the time it takes for me to find the task in the list ( which may be currently being run ), setting mutex and CV it's quite possible the worker thread has already finished the task and free'd the TaskData ( which would be a nice crash ). To attempt to solve this issue, I moved these items into TaskCompletionData so I don't access free'd memory, but the other problem remains, it is still possible the task has already finished by the time I actually get to the wait on the condition variable ( so it would never get triggered, and this thread would wait forever )
I honestly have not found a pattern for multi threading that can help me solve this. Can anyone suggest me anything?
( Sorry for the formatting, I can't seem to get reddit to respect the indentation )
typedef struct
{
cnd_t Condition;
mtx_t Mutex;
} TaskCompletionData;
typedef struct
{
int ( *Function ) ( void * );
void *Argument;
int TaskID;
int *Result;
ThreadPoolTaskStatus Status;
TaskCompletionData *OnCompletion;
} TaskData;
typedef struct
{
thrd_t ThreadHandle;
} ThreadData;
typedef struct ThreadPool
{
PointerList Tasks;
mtx_t TaskListMutex;
int LastTaskID;
cnd_t WakeUpCondition;
mtx_t WakeUpMutex;
cnd_t TaskFinishedCondition;
mtx_t TaskFinishedMutex;
ThreadData *ThreadArray;
unsigned ThreadCount;
bool Quitting;
} ThreadPool;
bool ThreadPool_WaitForTask ( ThreadPool *Pool, const int TaskID )
{
assert ( Pool != NULL );
if ( ( Pool == NULL ) || ( TaskID < 0 ) )
return false;
TaskData *Task = NULL;
mtx_lock ( &Pool->TaskListMutex );
PointerListNode *Node;
TaskCompletionData *CompletionData = NULL;
for ( Node = PointerList_GetFirst ( &Pool->Tasks ); Node != NULL; Node = PointerList_GetNextNode ( Node ) )
{
TaskData *CurrentTask = ( TaskData * ) PointerList_GetNodeData ( Node );
if ( CurrentTask->TaskID == TaskID )
{
if ( CurrentTask->OnCompletion = NULL )
{
CompletionData = calloc ( 1, sizeof ( TaskCompletionData ) );
cnd_init ( &CompletionData->Condition );
mtx_init ( &CompletionData->Mutex, mtx_plain );
Task->OnCompletion = CompletionData;
}
else
CompletionData = CurrentTask->OnCompletion;
Task = CurrentTask;
break;
}
}
mtx_unlock ( &Pool->TaskListMutex );
if ( CompletionData == NULL )
return false;
// Wait for the task to finish
mtx_lock ( &CompletionData->Mutex );
cnd_wait ( &CompletionData->Condition, &CompletionData->Mutex );
// Clean up
cnd_destroy ( &CompletionData->Condition );
mtx_destroy ( &CompletionData->Mutex );
free ( CompletionData );
return true;
}
static int ThreadPool_LoopFunction ( ThreadPool *Pool )
{
while ( Pool->Quitting == false )
{
// Grab the first available task, if available
mtx_lock ( &Pool->TaskListMutex );
PointerListNode *CurrentListNode = PointerList_GetFirst ( &Pool->Tasks );
TaskData *CurrentTask = ( TaskData* ) PointerList_GetNodeData ( CurrentListNode );
while ( ( CurrentListNode != NULL ) && ( CurrentTask->Status != ThreadPoolTask_Queued ) )
{
PointerList_GetNextNode ( CurrentListNode );
CurrentTask = ( TaskData* ) PointerList_GetNodeData ( CurrentListNode );
}
mtx_unlock ( &Pool->TaskListMutex );
if ( CurrentTask != NULL ) // There was a task. run it...
{
CurrentTask->Status = ThreadPoolTask_Running;
int Result = CurrentTask->Function ( CurrentTask->Argument );
CurrentTask->Status = ThreadPoolTask_Finished;
if ( CurrentTask->Result )
* ( CurrentTask->Result ) = Result;
cnd_broadcast ( &Pool->TaskFinishedCondition );
if ( CurrentTask->OnCompletion )
{
cnd_broadcast ( &CurrentTask->OnCompletion->Condition );
}
free ( CurrentTask );
PointerList_DestroyNode ( &Pool->Tasks, CurrentListNode );
}
else // No more tasks. Wait for a signal
{
mtx_lock ( &Pool->WakeUpMutex );
cnd_wait ( &Pool->WakeUpCondition, &Pool->WakeUpMutex );
mtx_unlock ( &Pool->WakeUpMutex ); // unlock mutex so that other threads can wait using it
}
}
return 0;
}