pthread_cond_wait() based multithreaded Linux daemon skeleton

Question

I'm trying to design a multithreaded daemon for an industrial automation related project.

The daemon will be using a number of 3rd party libs like MQTT, mysql, etc.. My idea is to have worker threads (or callbacks registered with these 3rd party libs) relay events to the main thread using a global mutex and conditional variable based synchronisation mechanism.

The main thread then performs some global state management and some lite processing that I'm willing to restrict to the main thread (like MySQL stuff). After that the main thread defers the processing to a pool of worker threads.

I haven't posted the logging and much of the error checking code here. I'm planning on using the start-stop-daemon facility for daemonizing.

I wanted to know if I'm headed in the right direction. Constructive criticism is welcome.

#include <stdio.h>
#include <pthread.h>
#include <errno.h>

#define RESPONSE_MAX_LENGTH 256
/**
 * Internal data structures and enums
 */
typedef struct myThread_     {
    pthread_t id;
} myThreadT;

typedef enum myEventType_ {
    MY_EVENT_TYPE_NONE,
    MY_EVENT_TYPE_STOP,
    MY_EVENT_TYPE_RELOAD
} myEventTypeT;

typedef struct myEvent_ {
    int is_triggered;
    myEventTypeT type;
    void *data;        /* may include a callback */
} myEventT;

const myEventT myEventInitializer = {
        .is_triggered = 0,
        .type = MY_EVENT_TYPE_NONE,
        .data = NULL
};

/**
 * Global variables
 */
static myEventT myEvent;
static myThreadT worker;
/* Mutex and cond_var */
static pthread_mutex_t my_event_mutex = PTHREAD_MUTEX_INITIALIZER;
static pthread_cond_t  my_event_trigger_cv = PTHREAD_COND_INITIALIZER;

/*
 * Triggers an internal myEvent
 * Blocks calling thread till event acknowledgement
*/
int trigger_event(myEventTypeT type, void *data) {
    pthread_mutex_lock(&my_event_mutex);
    myEvent.type = type;
    myEvent.data = data;
    // Trigger event!
    myEvent.is_triggered = 1;
    pthread_cond_signal(&my_event_trigger_cv);
    pthread_mutex_unlock(&my_event_mutex);
    // Block till current trigger has been acknowledged
    while(myEvent.is_triggered);
    return 0;
}

void *worker_loop() {
    while(1) {
        sleep(5);       /* Do something worthwhile */
        trigger_event(MY_EVENT_TYPE_NONE, NULL);
    }
    pthread_exit(NULL);
}

/*
 * Called from main thread, mainly for state management & db related processing,
 * i.e. stuff we want to restrict to the main thread.
*/
int handle_event(myEventT *event, char *response) {
    // handle the event, fill response if required
    return 0;
}

int main(int argc, char* argv[]) {
    // Initialize globals
    myEvent = myEventInitializer;

    // Thread management
    pthread_attr_t attr;
    pthread_attr_init(&attr);
    pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED);

    int thread_error;
    thread_error = pthread_create(&worker.id, &attr, worker_loop, NULL);
    if (thread_error != 0) {
        return -1;
    }

    pthread_attr_destroy(&attr);
    pthread_mutex_lock(&my_event_mutex);

    // Main loop: wait for event
    while(1) {

        // Wait for signal from worker threads/ callbacks
        int res_wait = pthread_cond_wait(&my_event_trigger_cv, &my_event_mutex);
        if (res_wait == EINVAL) {   // TODO: Consolidated error handling
            return -1;
        }

        /**
         * Event handling construct
         *
         */
        if (myEvent.is_triggered) {
            pthread_mutex_unlock(&my_event_mutex);

            char response_string[RESPONSE_MAX_LENGTH];  //TODO macro for size!
            if(handle_event(&myEvent, response_string) >= 0) {
                // Initiate deferred processing on another thread (maybe callback)
            }

            // Clear event triggered flag and reinit event data
            myEvent = myEventInitializer;

            pthread_mutex_lock(&my_event_mutex);
        }
    }

    // Clean up pthread resources
    pthread_mutex_destroy(&my_event_mutex);
    pthread_cond_destroy(&my_event_trigger_cv);

    pthread_exit(NULL);
}

Answer 1

Bugs

I can see three problems with your program:

1. Your main thread releases the mutex and then uses myEvent without holding the mutex. If another thread grabs the mutex and modifies myEvent, for example by calling trigger_event(), the main thread will be reading/writing the same structure and unpredictable things will happen.
2. The way you use your condition variable, there is a possibility that you will get stuck forever because another thread signalled the condition when you weren't waiting on it.
3. If two threads call trigger_event(), the first event will get overwritten by the second event.

I will elaborate more on each problem.

Using shared variables properly

One thing you should always make sure is that a shared variable is never read/written without holding the mutex. In your case, the shared variable is myEvent. Keeping that in mind, I can see two possible fixes for your main loop. The first fix is simple: just keep holding the mutex while you use myEvent. However, perhaps handle_event() is lengthy and you don't want to hold the mutex for that long. If so, then the second fix is to make a copy of myEvent while you still hold the mutex. Here is an example of how that would look. Notice that I also moved the part where you clear myEvent to where the mutex is still being held:

    if (myEvent.is_triggered) {
        myEventT eventCopy = myEvent;
        myEvent = myEventInitializer;
        pthread_mutex_unlock(&my_event_mutex);

        char response_string[RESPONSE_MAX_LENGTH];  //TODO macro for size!
        if(handle_event(&myEventCopy, response_string) >= 0) {
            // Initiate deferred processing on another thread (maybe callback)
        }

        pthread_mutex_lock(&my_event_mutex);
    }

Avoiding deadlock

When using a mutex with a condition variable, your code should always look like this:

pthread_mutex_lock(&mutex);
while (someVariable) {
    pthread_cond_wait(&cond, &mutex);        
}

This kind of loop guarantees that you will never miss another thread signalling the condition variable. Notice in particular how someVariable is checked while the mutex is being held.

Now let's look at your code and see where it could fail:

    if (myEvent.is_triggered) {
        pthread_mutex_unlock(&my_event_mutex);

        char response_string[RESPONSE_MAX_LENGTH];  //TODO macro for size!
        if(handle_event(&myEvent, response_string) >= 0) {
            // Initiate deferred processing on another thread (maybe callback)
        }

        // Clear event triggered flag and reinit event data
        myEvent = myEventInitializer;

        //
        // What if another thread signals your condition right here?
        //

        pthread_mutex_lock(&my_event_mutex);
    }

I've marked above where the danger lies. If another thread calls trigger_event() when the main thread is at the comment, that other thread will set myEvent.is_triggered and signal the condition, but the main thread won't notice.

Here is how I would structure your main event loop instead:

    pthread_attr_destroy(&attr);

    // Main loop: wait for event
    while(1) {
        pthread_mutex_lock(&my_event_mutex);
        while (!myEvent.is_triggered) {
            // Wait for signal from worker threads/ callbacks
            int res_wait = pthread_cond_wait(&my_event_trigger_cv,
                                                &my_event_mutex);
            if (res_wait == EINVAL) {
                return -1;
            }
        }

        // Mutex is held at this point, so it is safe to modify myEvent.
        myEventT eventCopy = myEvent;
        myEvent = myEventInitializer;
        pthread_mutex_unlock(&my_event_mutex);

        /**
         * Event handling construct
         */
        char response_string[RESPONSE_MAX_LENGTH];
        if(handle_event(&eventCopy, response_string) >= 0) {
            // Initiate deferred processing on another thread.
        }
    }

Handling multiple events

Right now, you only have one variable myEvent, which means that if two or more threads trigger events at the same time, myEvent will get overwritten with the latest event and you will lose the earlier event.

The fix for this is simple: use a queue of events instead of a single event. In trigger_event(), while holding the mutex you push an event onto the queue. In the main loop, while holding the mutex, you pop an event off the queue. You can do this where my code above currently copies myEvent into eventCopy.

Answer 2

MySQL and the main thread

and some lite processing that I'm willing to restrict to the main thread (like MySQL stuff)

Don't consider MySQL stuff a lite processing. Don't consider anything that goes out to the network or external processes a lite processing. Consider using some asynchronous interface to DB or a separate thread to connect to DB, or connecting to DB directly in the threads (if you use separate connections and are a bit careful you'll be OK. That's what most larger web applications do.