How can I use pthread to manipulate data concurrently?

Question

I want to use one thread to generate node and the other thread manipulate it. I have the following code. It stuck half way in the producer's for loop and never executes consumer. I'm hoping those two for loops can be executed in parallel.

typedef struct _Net {
    int id;
    struct _Net * next;
    unsigned long net;
} Net;
typedef Net * NetPtr;

pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;
pthread_cond_t cond = PTHREAD_COND_INITIALIZER;

NetPtr prevQueue[11];
int total = 0;

NetPtr queue;

void *producer(){
    int i;
    NetPtr prev, netNode;
    prev = (Net *)malloc(sizeof(Net));
    pthread_mutex_lock(&mutex);
    pthread_mutex_unlock(&mutex);

    for (i = 0; i<10 ; i++) {
        netNode = (Net *)malloc(sizeof(Net));
        (*netNode).id = i;
        pthread_mutex_lock(&mutex);
        prevQueue[i] = netNode;
        total += 1;
        pthread_cond_signal(&cond);
        pthread_mutex_unlock(&mutex);
    }
}

void *consumer(){
    int i;
    while(total == 0)
    {
        pthread_cond_wait(&cond,&mutex);
    }
    for(i = 0; i<10; i++){
        pthread_mutex_lock(&mutex);
        printf("%d\n", prevQueue[i]->id);
        pthread_mutex_unlock(&mutex);
    }
}

int main(int argc, char **argv) {

    int tid1, tid2;
    pthread_t thread[2];
    pthread_attr_t attributes;


    pthread_attr_init(&attributes);
    if((tid1 = pthread_create(&thread[0],&attributes,producer,NULL)))
    {
        printf("\nError in the producer thread\n");
        printf("\n");
    }

    if((tid2 = pthread_create(&thread[1],&attributes,consumer,NULL)))
    {
        printf("\nERror in the consumer thread\n");
    }

    pthread_join(thread[0],NULL);
    pthread_join(thread[1],NULL);


    return 0;
}

`netNode = (Net *)malloc(sizeof(Net))` -> `netNode = malloc(sizeof(*netNode))`! [is a better way](http://stackoverflow.com/questions/605845/do-i-cast-the-result-of-malloc). — Iharob Al Asimi, Nov 17 '15 at 17:44

score 1 · Accepted Answer · answered Nov 17 '15 at 19:38

As a preliminary matter, it is not much useful for the producer to start by locking the mutex and then immediately unlocking it again. It's not harmful, but not useful, either, at least not in your case.

You do have several problems with your program, however. First and foremost, pthread_cond_wait() must be called only by a thread that holds the specified mutex locked. Your consumer thread fails to lock the mutex before calling.

Also, your code appears to be written with the assumption that if one thread releases a mutex while another is blocked on locking that mutex, then the blocked thread will immediately acquire the mutex and proceed. That is by no means guranteed. Either your producer thread or your consumer thread could and quite likely often would relock the mutex immediately after unlocking it, without giving the other a chance to run. This sort of problem is what semaphores and condition variables are for; you could use the condition variable you already have to serve this purpose.

When using a condition variable, however, you do need to beware of spurrious wakeups, and you also need to watch out for missing notifications sent when no thread is waiting on the condition variable. The normal idiom is to test the condition that the condition variable protects after waking from a wait, and to wait again if it is not satisfied. I see that you in fact do that -- the problem is that the condition you are waiting on is not quite the one you want to wait on.

Let's analyze your problem. You need to synchronize access to shared state embodied by array prevQueue, and you provide a shared variable total to communicate which elements of that array have been populated by the producer. The producer modifies that shared state only while holding the mutex, which is well and good, and each time it adds an element it signals the condition variable (still holding the mutex) which is also good.

Now consider the consumer. In addition to the shared state described above, it has local state recording which is the next element for it to consume. What, then, is the condition under which it can make progress? It can progress if the next element for it to consume is one of those already produced. That is the condition it needs to wait on, and it probably needs to do so multiple times, as the first time it is able make progress may be before the producer produces all the elements it intends to do. On the other hand, when the consumer determines that it can make progress, how much progress can it make? Clearly it can consume up to as many elements as it can be sure have been produced, which may permit consuming more than one before waiting again.

This modified version of your consumer thread function can do all that:

void *consumer() {
    int i = 0;
    int max = 0;

    do {
        /* we need to protect access to variable 'total' */
        pthread_mutex_lock(&mutex);
        /* wait, if necessary, for at least one more element to be produced */
        while (max >= total) {
            pthread_cond_wait(&cond,&mutex);
        }
        /* update how many elements are now available */
        max = total;
        /* allow the producer to continue while we consume available items */
        pthread_mutex_unlock(&mutex);
        while (i < max) {
            /*
             * Although prevQueue is shared, we know we can safely access
             * elements up to index 'max', last-observed limit.
             */
            printf("%d\n", prevQueue[i]->id);
            i += 1;
        }
    } while (max < 10);
}

Note in particular that when you perform a pthread_cond_wait(), the specified mutex (currently held by the caller) is released, allowing other threads to lock it, which is important for a thread to be able to signal the condition variable. It will be re-locked prior to the call returning.

How can I use pthread to manipulate data concurrently?

1 Answers1