Multithreading

I have just started learning multi-threading. I have written a simple application. The application creates three threads. Two threads write and one thread reads. The writer threads write to separate location in a global array. The writer thread after incrementing the value in the array notifies the reader. The reader thread then decrements that value in the array and waits again for the writer threads to update their corresponding value in the array. The code for the application is pasted below.

What I see is that the writer(Producer) threads get more time slice than the reader(Consumer) thread. I think I am doing something wrong. If the output of the application is redirected to a file, then it can be observed that there are more consecutive messages from the Producers and the messages from the Consumer occur infrequently. What I was expecting was that, 开发者_StackOverflowwhen a Producer updates its data, the Consumer immediately processes it i.e. after every Producer message there should be a Consumer message printed.

Thanks and regards,

~Plug

#include <stdio.h>

#include <pthread.h>

const long g_lProducerCount = 2; /*Number of Producers*/

long g_lProducerIds[2]; /*Producer IDs = 0, 1...*/
long g_lDataArray[2]; /*Data[0] for Producer 0, Data[1] for Producer 1...*/

/*Producer ID that updated the Data. -1 = No update*/
long g_lChangedProducerId = -1;

pthread_cond_t g_CondVar = PTHREAD_COND_INITIALIZER;
pthread_mutex_t g_Mutex  = PTHREAD_MUTEX_INITIALIZER;
pthread_t g_iThreadIds[3]; /*3 = 2 Producers + 1 Consumer*/

unsigned char g_bExit = 0; /*Exit application? 0 = No*/

void* Producer(void *pvData)
{
    long lProducerId = *(long*)pvData; /*ID of this Producer*/

    while(0 == g_bExit) {
        pthread_mutex_lock(&g_Mutex);

        /*Tell the Consumer who's Data is updated*/
        g_lChangedProducerId = lProducerId;

        /*Update the Data i.e. Increment*/
        ++g_lDataArray[lProducerId];

        printf("Producer: Data[%ld] = %ld\n",
                lProducerId, g_lDataArray[lProducerId]);

        pthread_cond_signal(&g_CondVar);
        pthread_mutex_unlock(&g_Mutex);
    }

    pthread_exit(NULL);
}

void* Consumer(void *pvData)
{
    while(0 == g_bExit) {
        pthread_mutex_lock(&g_Mutex);

        /*Wait until one of the Producers update it's Data*/
        while(-1 == g_lChangedProducerId) {
            pthread_cond_wait(&g_CondVar, &g_Mutex);
        }

        /*Revert the update done by the Producer*/
        --g_lDataArray[g_lChangedProducerId];

        printf("Consumer: Data[%ld] = %ld\n",
                g_lChangedProducerId, g_lDataArray[g_lChangedProducerId]);

        g_lChangedProducerId = -1; /*Reset for next update*/

        pthread_mutex_unlock(&g_Mutex);
    }

    pthread_exit(NULL);
}

void CreateProducers()
{
    long i;

    pthread_attr_t attr;
    pthread_attr_init(&attr);

    for(i = 0; i < g_lProducerCount; ++i) {
        g_lProducerIds[i] = i;
        pthread_create(&g_iThreadIds[i + 1], &attr,
                        Producer, &g_lProducerIds[i]);
    }

    pthread_attr_destroy(&attr);
}

void CreateConsumer()
{
    pthread_attr_t attr;
    pthread_attr_init(&attr);

    pthread_create(&g_iThreadIds[0], &attr, Consumer, NULL);

    pthread_attr_destroy(&attr);
}

void WaitCompletion()
{
    long i;
    for(i = 0; i < g_lProducerCount + 1; ++i) {
        pthread_join(g_iThreadIds[i], NULL);
    }
}

int main()
{
    CreateProducers();
    CreateConsumer();

    getchar();

    g_bExit = 1;

    WaitCompletion();

    return 0;
}

You would have to clarify what is it exactly that you want to achieve. For now the producers only increment an integer and the consumer decrements the value. This is not a very useful activity ;) I understand that this is only a test app, but still it is not clear enough what's the purpose of this processing, what are the constraints and so on.

The producers produce some 'items'. The outcome of this production is represented as an integer value. 0 means no items, 1 means there is a pending item, that consumer can take. Is that right? Now, is it possible for the producer to produce several items before any of them gets consumed (incrementing the array cell to a value higher than 1)? Or does he have to wait for the last item to be consumed before the next one can be put into the storage? Is the storage limited or unlimited? If it is limited then is the limit shared among all the producers or is it defined per producer?

What I was expecting was that, when a Producer updates its data, the Consumer immediately processes it i.e. after every Producer message there should be a Consumer message printed.

Though it's not really clear what you want to achieve I will hold on to that quote and assume the following: there is a limit of 1 item per producer and the producer has to wait for the consumer to empty the storage before a new item can be put in the cell i.e. the only allowed values in the g_lDataArray are 0 and 1.

To allow maximum concurrency between threads you will need a conditional variable/mutex pair for each cell of g_lDataArray (for each producer). You will also need a queue of updates that is a list of producers that have submitted their work and a conditional variable/mutex pair to guard it, this will replace g_lChangedProducerId which can only hold one value at a time.

Every time a producer wants to put an item into the storage it has to acquire the respective lock, check if the storage is empty (g_lDataArray[lProducerId] == 0), if not wait on the condition variable and then, increment the cell, release the held lock, acquire the consumer lock, add his id to the update queue, notify the consumer, release the consumer lock. Of course if the producer would perform any real computations producing some real item, this work should be performed out of the scope of any lock, before the attempt to put the item in the storage.

In pseudo code this looks like this:

// do some computations
item = compute();

lock (mutexes[producerId]) {
    while (storage[producerId] != 0)
        wait(condVars[producerId]);
    storage[producerId] = item;
}

lock (consumerMutex) {
    queue.push(producerId);
    signal(consumerCondVar);
}        

The consumer should act as follows: acquire his lock, check if there are any pending updates to process, if not wait on the condition variable, take one update out of the queue (that is the number of the updating producer), acquire the lock for producer who's update is going to be processed, decrement the cell, notify the producer, release the producer's lock, release his lock, finally process the update.

lock (consumerMutex) {
    while (queue.isEmpty())
        wait(consumerCondVar);
    producerId = queue.pop();
    lock (mutexex[producerId]) {
        item = storage[producerId];
        storage[producerId] = 0;
        signal(condVars[producerId]);
    }
}

//process the update
process(item);

Hope this answer is what you needed.

The problem may be that all producers change g_lChangedProducerId, so the value written by one producer may be overwritten by another producer before the consumer sees it.

This means that the consumer effectively doesn't see that the first producer has produced some output.

Well,when you producer produced, it may wake up the ProThread or ConThread. And If it waked up the ProThread,the producer produced again,and the ConThread didn't consume immediately after data is produced. That's what you don't want to see. All you need is to make sure that when it produced,it won't wake the ProThread up. Here's one kind of solution for this

    void* Producer(void *pvData)
    {
            ........
            //wait untill consumer consume its number
            while(-1!=g_lChangedProducerId)
                    pthread_cond_wait(&g_CondVar,&g_Mutex);
            //here to inform the consumer it produced the data
            g_lChangedProducerId = lProducerId;
            ........
    }

    void* Consumer(void *pvData)
    {
            g_lChangedProducerId = -1;
            **//wake up the producer when it consume
            pthread_cond_signal(&g_CondVar);**
            pthread_mutex_unlock(&g_Mutex);
    }