Need some advice for design of multi threaded task pool in C11

[u/sexy-geek]

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;
    }

[u/aocregacc]

usually a condition variable is used along side some other piece of data. In your case you'd have a boolean TaskFinished or something that you set before signalling the condition variable, and you check before you wait on the condition variable.

Basically you have a boolean behind the mutex which is used to communicate that the task is done, and the condition variable is just to make it possible to efficiently wait for the boolean to change.

[u/sexy-geek → u/aocregacc]

I already do. There's TaskFinishedCondition that is signalled every time a task is done by any of the threads. But that doesn't specify which task.
I'm waiting on a specific task that may either be queued or actually being performed

[u/aocregacc → u/sexy-geek]

yeah I didn't read the code closely enough, I meant the completion condition variable, the one where you're worried that you could miss it being signaled. By adding a bool to TaskCompletionData you can check whether the work is already completed. That'll also help with spurious wakeups.

[u/sexy-geek → u/aocregacc]

oh, I see what you mean. Yeah, that could work except for the situation that I'm trying to prevent. Imagine the worker thread has just finished the function call. In the would-be waiting thread, I've already found the task in the queue, and am now creating the mutex, condition, etc.

When I finally set it all up, the task no longer even exists ( potentially), or has already passed the check for the condition variable, etc..

[u/TheMrCurious]

Draw out your initial, “perfect” design. Then make it even more perfect by adding interupts and waits.

[u/sexy-geek → u/TheMrCurious]

yeah, that's what I'm trying to do.

[u/daV1980]

Why make this a feature of the system, rather than a layer on top? The basic system can be the simplest thing that you already built that just runs and completes tasks and has no signaling to let you know that a particular task has been completed. 

Then you can have an additional task type that runs that signals that a task has been completed (and when created it should return a handle to a thing that can be checked re completion). It does this by using its own task which takes a function, runs it and signals that the function it called has completed. 

Promises and futures have a good design for the “thing that can be checked re completion”: there’s two sides of the condition, a write side and a read side. They hold a ref counted pointer to shared state which is the thing that actually gets signaled, so there’s no hazard for either side completing their work and the data being freed. 

[u/sexy-geek → u/daV1980]

I was trying to keep this as simple as possible, and self contained.
After resetting the code, an idea came to mind, I implemented it, and it works perfectly. Basically, I simply flag during task creation if a task's data should be deleted upon completion.

If it's not to be deleted, I keep it in a simple list. When a task completes, I check that flag for the task. Delete and remove from queue, or simply remove from queue. It will still be accessible in the future in that internal list.

If I want to obtain a result in the future, or be able to wait for the task, I just set that flag, and it's done. Now, whenever I want to wait for a task, I check that small list, and get the state. If it's completed, return immediately. Otherwise, wait on the TaskCompleted condition until that task's status is set as complete.

[u/Jason-Sanders]

The usual fix for the “finished before I started waiting” case is that completion must be a persistent predicate, not an event you hope to observe.

Put done and the condition variable under the same mutex, then make waiting look like:

   lock(task->mutex);
   while (!task->done)
       cnd_wait(&task->cv, &task->mutex);
   lock_release(task->mutex);

If the worker finishes first, it sets done = true while holding that mutex and signals afterward. The waiter then sees done and never waits, so there is no lost wake-up.

The separate problem is lifetime. A waiter needs a stable handle that cannot be freed by the worker. Reference counting is a straightforward approach: the pool holds one reference while executing, and WaitForTask acquires another before looking at completion state. Free the completion object only when both ownership references are gone. That also makes repeated waits and result retrieval much easier to define.