CODEWITHSUNDEEP
cconcurrencymutex
Ian M
Ian M

Reputation: 355

C uninitialized mutex works and initialized mutex fails?

My C program creates a producer thread, saving data as fast as possible. The main thread consumes and prints these.

After days of bug finding, I noticed that if the mutex was initialized, then the program stops within 30 seconds (deadlock?).

However if the mutex is left uninitialized it works perfectly.

Can anyone explain this?? To avoid undefined behavior, I would prefer to initialize them if possible.

Further Info: Specifically it's locking up if "pthread_mutex_t signalM" (the signaling mutex) is initialized

Initialized

#include <stdlib.h>                     // exit_failure, exit_success
#include <stdio.h>                      // stdin, stdout, printf
#include <pthread.h>                    // threads
#include <string.h>                     // string
#include <unistd.h>                     // sleep
#include <stdbool.h>                    // bool
#include <fcntl.h>                      // open



struct event {
    pthread_mutex_t critical;
    pthread_mutex_t signalM;
    pthread_cond_t signalC;
    int eventCount;
};

struct allVars {
    struct event inEvents;
    struct event outEvents;
    int bufferSize;
    char buffer[10][128];
};




/**
 * Advance the EventCount
 */
void advance(struct event *event) {
    // increment the event counter
    pthread_mutex_lock(&event->critical);
    event->eventCount++;
    pthread_mutex_unlock(&event->critical);

    // signal await to continue
    pthread_mutex_lock(&event->signalM);
    pthread_cond_signal(&event->signalC);
    pthread_mutex_unlock(&event->signalM);
}



/**
 * Wait for ticket and buffer availability
 */
void await(struct event *event, int ticket) {

    int eventCount;

    // get the counter
    pthread_mutex_lock(&event->critical);
    eventCount = event->eventCount;
    pthread_mutex_unlock(&event->critical);

    // lock signaling mutex
    pthread_mutex_lock(&event->signalM);

    // loop until the ticket machine shows your number
    while (ticket > eventCount) {
        // wait until a ticket is called
        pthread_cond_wait(&event->signalC, &event->signalM);

        // get the counter
        pthread_mutex_lock(&event->critical);
        eventCount = event->eventCount;
        pthread_mutex_unlock(&event->critical);
    }

    // unlock signaling mutex
    pthread_mutex_unlock(&event->signalM);
}



/**
 * Add to buffer
 */
void putBuffer(struct allVars *allVars, char data[]) {
    // get the current write position
    pthread_mutex_lock(&allVars->inEvents.critical);
    int in = allVars->inEvents.eventCount;
    pthread_mutex_unlock(&allVars->inEvents.critical);

    // wait until theres a space free in the buffer
    await(&allVars->outEvents, in - allVars->bufferSize + 1);   // set to 2 to keep 1 index distance

    // add data to buffer
    strcpy(allVars->buffer[in % allVars->bufferSize], data);

    // increment the eventCounter and signal
    advance(&allVars->inEvents);
}



/**
 * Get from buffer
 */
char *getBuffer(struct allVars *allVars) {
    // get the current read position
    pthread_mutex_lock(&allVars->outEvents.critical);
    int out = allVars->outEvents.eventCount;
    pthread_mutex_unlock(&allVars->outEvents.critical);

    // wait until theres something in the buffer
    await(&allVars->inEvents, out + 1);

    // allocate memory for returned string
    char *str = malloc(128);

    // get the buffer data
    strcpy(str, allVars->buffer[out % allVars->bufferSize]);

    // increment the eventCounter and signal
    advance(&allVars->outEvents);

    return str;
}



/** child thread (producer) */
void *childThread(void *allVars) {
    char str[10];
    int count = 0;

    while (true) {
        sprintf(str, "%d", count++);
        putBuffer(allVars, str);
    }

    pthread_exit(EXIT_SUCCESS);
}



int main(void) {
    // init structs
    struct event inEvents = {
        PTHREAD_MUTEX_INITIALIZER,
        PTHREAD_MUTEX_INITIALIZER,
        PTHREAD_COND_INITIALIZER,
        0
    };
    struct event outEvents = {
        PTHREAD_MUTEX_INITIALIZER,
        PTHREAD_MUTEX_INITIALIZER,
        PTHREAD_COND_INITIALIZER,
        0
    };
    struct allVars allVars = {
        inEvents,       // events
        outEvents,
        10,             // buffersize
        {"", {""}}      // buffer[][]
    };


    // create child thread (producer)
    pthread_t thread;
    if (pthread_create(&thread, NULL, childThread, &allVars)) {
        fprintf(stderr, "failed to create child thread");
        exit(EXIT_FAILURE);
    }


    // (consumer)
    while (true) {
        char *out = getBuffer(&allVars);
        printf("buf: %s\n", out);
        free(out);
    }


    return (EXIT_SUCCESS);
}

Uninitialized

#include <stdlib.h>                     // exit_failure, exit_success
#include <stdio.h>                      // stdin, stdout, printf
#include <pthread.h>                    // threads
#include <string.h>                     // string
#include <unistd.h>                     // sleep
#include <stdbool.h>                    // bool
#include <fcntl.h>                      // open



struct event {
    pthread_mutex_t critical;
    pthread_mutex_t signalM;
    pthread_cond_t signalC;
    int eventCount;
};

struct allVars {
    struct event inEvents;
    struct event outEvents;
    int bufferSize;
    char buffer[10][128];
};




/**
 * Advance the EventCount
 */
void advance(struct event *event) {
    // increment the event counter
    pthread_mutex_lock(&event->critical);
    event->eventCount++;
    pthread_mutex_unlock(&event->critical);

    // signal await to continue
    pthread_mutex_lock(&event->signalM);
    pthread_cond_signal(&event->signalC);
    pthread_mutex_unlock(&event->signalM);
}



/**
 * Wait for ticket and buffer availability
 */
void await(struct event *event, int ticket) {

    int eventCount;

    // get the counter
    pthread_mutex_lock(&event->critical);
    eventCount = event->eventCount;
    pthread_mutex_unlock(&event->critical);

    // lock signaling mutex
    pthread_mutex_lock(&event->signalM);

    // loop until the ticket machine shows your number
    while (ticket > eventCount) {
        // wait until a ticket is called
        pthread_cond_wait(&event->signalC, &event->signalM);

        // get the counter
        pthread_mutex_lock(&event->critical);
        eventCount = event->eventCount;
        pthread_mutex_unlock(&event->critical);
    }

    // unlock signaling mutex
    pthread_mutex_unlock(&event->signalM);
}



/**
 * Add to buffer
 */
void putBuffer(struct allVars *allVars, char data[]) {
    // get the current write position
    pthread_mutex_lock(&allVars->inEvents.critical);
    int in = allVars->inEvents.eventCount;
    pthread_mutex_unlock(&allVars->inEvents.critical);

    // wait until theres a space free in the buffer
    await(&allVars->outEvents, in - allVars->bufferSize + 1);   // set to 2 to keep 1 index distance

    // add data to buffer
    strcpy(allVars->buffer[in % allVars->bufferSize], data);

    // increment the eventCounter and signal
    advance(&allVars->inEvents);
}



/**
 * Get from buffer
 */
char *getBuffer(struct allVars *allVars) {
    // get the current read position
    pthread_mutex_lock(&allVars->outEvents.critical);
    int out = allVars->outEvents.eventCount;
    pthread_mutex_unlock(&allVars->outEvents.critical);

    // wait until theres something in the buffer
    await(&allVars->inEvents, out + 1);

    // allocate memory for returned string
    char *str = malloc(128);

    // get the buffer data
    strcpy(str, allVars->buffer[out % allVars->bufferSize]);

    // increment the eventCounter and signal
    advance(&allVars->outEvents);

    return str;
}



/** child thread (producer) */
void *childThread(void *allVars) {
    char str[10];
    int count = 0;

    while (true) {
        sprintf(str, "%d", count++);
        putBuffer(allVars, str);
    }

    pthread_exit(EXIT_SUCCESS);
}



int main(void) {
    // init structs
    struct event inEvents; /* = {
        PTHREAD_MUTEX_INITIALIZER,
        PTHREAD_MUTEX_INITIALIZER,
        PTHREAD_COND_INITIALIZER,
        0
    }; */
    struct event outEvents; /* = {
        PTHREAD_MUTEX_INITIALIZER,
        PTHREAD_MUTEX_INITIALIZER,
        PTHREAD_COND_INITIALIZER,
        0
    }; */

    struct allVars allVars = {
        inEvents,       // events
        outEvents,
        10,             // buffersize
        {"", {""}}      // buffer[][]
    };


    // create child thread (producer)
    pthread_t thread;
    if (pthread_create(&thread, NULL, childThread, &allVars)) {
        fprintf(stderr, "failed to create child thread");
        exit(EXIT_FAILURE);
    }


    // (consumer)
    while (true) {
        char *out = getBuffer(&allVars);
        printf("buf: %s\n", out);
        free(out);
    }


    return (EXIT_SUCCESS);
}

Upvotes: 4

Views: 2408

Answers (2)

Michael Burr
Michael Burr

Reputation: 340228

Jonathan explained why the code that didn't initialize mutexes didn't deadlock (essentially because trying to use an uninitialized mutex would never block, it would just immediately return an error).

The problem causing the infinite wait in the version of the program that does properly initialize mutexes is that you aren't using your condition variables properly. The check of the predicate expression and the wait on the condition variable must be done atomically with respect to whatever other thread might be modifying the predicate. You code is checking a predicate that is a local variable that the other thread doesn't even have access to. Your code reads the actual predicate into a local variable within a critical section, but then the mutex for reading the predicate is released and a different mutex is acquired to read the 'false' predicate (which cannot be modified by the other thread anyway) atomically with the condition variable wait.

So you have a situation where the actual predicate, event->eventCount, can be modified and the signal for that modification be issued in between when the waiting thread reads the predicate and blocks on the condition variable.

I think the following will fix your deadlock, but I haven't had a chance to perform much testing. The change is essentially to remove the signalM mutex from struct event and replace any use of it with the critical mutex:

#include <stdlib.h>                     // exit_failure, exit_success
#include <stdio.h>                      // stdin, stdout, printf
#include <pthread.h>                    // threads
#include <string.h>                     // string
#include <unistd.h>                     // sleep
#include <stdbool.h>                    // bool
#include <fcntl.h>                      // open



struct event {
    pthread_mutex_t critical;
    pthread_cond_t signalC;
    int eventCount;
};

struct allVars {
    struct event inEvents;
    struct event outEvents;
    int bufferSize;
    char buffer[10][128];
};




/**
 * Advance the EventCount
 */
void advance(struct event *event) {
    // increment the event counter
    pthread_mutex_lock(&event->critical);
    event->eventCount++;
    pthread_mutex_unlock(&event->critical);

    // signal await to continue
    pthread_cond_signal(&event->signalC);
}



/**
 * Wait for ticket and buffer availability
 */
void await(struct event *event, int ticket) {


    // get the counter
    pthread_mutex_lock(&event->critical);

    // loop until the ticket machine shows your number
    while (ticket > event->eventCount) {
        // wait until a ticket is called
        pthread_cond_wait(&event->signalC, &event->critical);
    }

    // unlock signaling mutex
    pthread_mutex_unlock(&event->critical);
}



/**
 * Add to buffer
 */
void putBuffer(struct allVars *allVars, char data[]) {
    // get the current write position
    pthread_mutex_lock(&allVars->inEvents.critical);
    int in = allVars->inEvents.eventCount;
    pthread_mutex_unlock(&allVars->inEvents.critical);

    // wait until theres a space free in the buffer
    await(&allVars->outEvents, in - allVars->bufferSize + 1);   // set to 2 to keep 1 index distance

    // add data to buffer
    strcpy(allVars->buffer[in % allVars->bufferSize], data);

    // increment the eventCounter and signal
    advance(&allVars->inEvents);
}



/**
 * Get from buffer
 */
char *getBuffer(struct allVars *allVars) {
    // get the current read position
    pthread_mutex_lock(&allVars->outEvents.critical);
    int out = allVars->outEvents.eventCount;
    pthread_mutex_unlock(&allVars->outEvents.critical);

    // wait until theres something in the buffer
    await(&allVars->inEvents, out + 1);

    // allocate memory for returned string
    char *str = malloc(128);

    // get the buffer data
    strcpy(str, allVars->buffer[out % allVars->bufferSize]);

    // increment the eventCounter and signal
    advance(&allVars->outEvents);

    return str;
}



/** child thread (producer) */
void *childThread(void *allVars) {
    char str[10];
    int count = 0;

    while (true) {
        sprintf(str, "%d", count++);
        putBuffer(allVars, str);
    }

    pthread_exit(EXIT_SUCCESS);
}



int main(void) {
    // init structs
    struct event inEvents = {
        PTHREAD_MUTEX_INITIALIZER,
        PTHREAD_COND_INITIALIZER,
        0
    };
    struct event outEvents = {
        PTHREAD_MUTEX_INITIALIZER,
        PTHREAD_COND_INITIALIZER,
        0
    };
    struct allVars allVars = {
        inEvents,       // events
        outEvents,
        10,             // buffersize
        {"", {""}}      // buffer[][]
    };


    // create child thread (producer)
    pthread_t thread;
    if (pthread_create(&thread, NULL, childThread, &allVars)) {
        fprintf(stderr, "failed to create child thread");
        exit(EXIT_FAILURE);
    }


    // (consumer)
    while (true) {
        char *out = getBuffer(&allVars);
        printf("buf: %s\n", out);
        free(out);
    }


    return (EXIT_SUCCESS);
}

Upvotes: 3

Jonathan Leffler
Jonathan Leffler

Reputation: 754130

I modified the getBuffer() and putBuffer() routines as shown (in both the initialized and uninitialized versions of the code):

static
void putBuffer(struct allVars *allVars, char data[])
{
    int lock_ok = pthread_mutex_lock(&allVars->inEvents.critical);
    if (lock_ok != 0)
        printf("%s(): lock error %d (%s)\n", __func__, lock_ok, strerror(lock_ok));
    int in = allVars->inEvents.eventCount;
    int unlock_ok = pthread_mutex_unlock(&allVars->inEvents.critical);
    if (unlock_ok != 0)
        printf("%s(): unlock error %d (%s)\n", __func__, unlock_ok, strerror(unlock_ok));

    await(&allVars->outEvents, in - allVars->bufferSize + 1);

    strcpy(allVars->buffer[in % allVars->bufferSize], data);

    advance(&allVars->inEvents);
}

static
char *getBuffer(struct allVars *allVars)
{
    int lock_ok = pthread_mutex_lock(&allVars->outEvents.critical);
    if (lock_ok != 0)
        printf("%s(): lock error %d (%s)\n", __func__, lock_ok, strerror(lock_ok));
    int out = allVars->outEvents.eventCount;
    int unlock_ok = pthread_mutex_unlock(&allVars->outEvents.critical);
    if (unlock_ok != 0)
        printf("%s(): unlock error %d (%s)\n", __func__, unlock_ok, strerror(unlock_ok));

    await(&allVars->inEvents, out + 1);

    char *str = malloc(128);

    strcpy(str, allVars->buffer[out % allVars->bufferSize]);

    advance(&allVars->outEvents);

    return str;
}

Then running the uninitialized code yields a lot of messages like:

buf: 46566
putBuffer(): lock error 22 (Invalid argument)
getBuffer(): lock error 22 (Invalid argument)
putBuffer(): unlock error 22 (Invalid argument)
getBuffer(): unlock error 22 (Invalid argument)

Basically, it appears to me that your locking and unlocking is being ignored. There are other places in your code that you should check too.

Fundamentally, if you ignore the errors reported, you don't notice that the locking and unlocking is not working at all, and there's no reason for the code to stop running.

Always check the return values from system calls that can fail.

I don't have an immediate explanation for why the initialized code locks up. It does for me, running on Mac OS X 10.10.3 with GCC 5.1.0, after anywhere from about 100,000 to 800,000 iterations.

Upvotes: 2

Related Questions