Threading using pthread

Let say I have an array of 5 threads :

//main thread
pthread_t t[5];
pthread_mutex_t mutex[5];
queue<int> q[5];

for(int i = 0; i < 5; i++){
pthread_create(&pthread_t[i], NULL, worker, NULL);
}

for(int i = 0; i < 5; i++){
   pthread_mutex_lock(&mutex[i]);
      queue[i].push_back(i);
   pthread_mutex_unlock(&mutex[i]);
}

void* worker(void* arg){

    pthread_mutex_lock(&mutex[?]);      

}

I am confused with the mutex_lock here. My question is:

1. How could I let the worker kn开发者_开发问答ow which mutex to lock?
2. When I access the mutex through mutex[i], do I need another lock since the child thread might be accessing the mutex array as well?

Thanks.

You need to be clear which threads are sharing which queues. The code you've written suggests each worker thread works on a specific queue, but the main thread (that spawns the workers) will be pushing back new values onto those queues. If that's what you want, then what you've done is basically correct, and you can let the worker threads know the array index of the mutex they're to lock/unlock by casting it to void* and passing it as the argument to pthread_create, which will in turn be passed as a void* to the worker function. You do not need any additional layer of locking around the mutex array - it is entirely safe to access specific elements independently, though if it were say a vector that was being resized at run-time, then you would need that extra level of locking.

Associate the mutex with the queue creating a new struct;

typedef struct {

  pthread_mutex_t mutex;
  queue<int> q;

} safe_queue;

safe_queue queue_pool [5];


void* worker(safe_queue){
 pthread_mutex_lock(&safe_queue.mutex);      
}

That last argument to the pthread_create is handed over to the thread when it's called, so you can just pass a value to the specific thread.

Since you want both a specific mutex and a specific queue, you're better off passing in the value of i directly.

for(int i = 0; i < 5; i++){
    pthread_create(&pthread_t[i], NULL, worker, (void*)i);
}

void *worker (void *pvI) {
    int idx = (int)pvI;  // Check for cast problems.
    // Use mutex[idx] and q[idx].
}

However, if you want to do it this way, I'd go for a single queue and mutex.

That's because the act of putting something on the queue is almost certainly going to be much faster than processing an item on the queue (otherwise you wouldn't need threads at all).

If you have multiple queues, the main thread has to figure out somehow which are the underutilised threads so it can select the best queue. If you have one queue and one mutex to protect it, the threads will self-organise for efficiency. Those threads that do long jobs won't try to get something from the queue. Those doing short jobs will come back sooner.

I should mention that mutexes on their own are not a good solution for this producer/consumer model. You can't have a thread lock the mutex then wait indefinitely on the queue since that will prevent the main thread putting anything on the queue.

So that means your worker threads will be constantly polling the queues looking for work.

If you use a mutex combined with a condition variable, it will be a lot more efficient. That's because the threads are signalled by the main thread when work is available rather than constantly grabbing the mutex, checking for work, then releasing the mutex.

The basic outline will be, for the main thread:

initialise
while not finished:
    await work
    lock mutex
    put work on queue
    signal condvar
    unlock mutex
terminate

and, for the worker threads:

initialise
while not finished:
    lock mutex
    while queue is empty:
        wait on condvar
    get work from queue
    unlock mutex
    do work
terminate

Don't pass a NULL pointer as arg to the thread. Instead use a pointer to an object that defines what the thread has to do.

How could I let the worker know which mutex to lock?

Pass the number as the last parameter to pthread_create()

for(int i = 0; i < 5; i++)
{
    pthread_create(&pthread_t[i], NULL, worker, reinterpret_cast<void*>(i));
}

Then you can get the value like this:

void* worker(void* arg)
{
    int index = reinterpret_cast<int>(arg);
    pthread_mutex_lock(&mutex[index]);      
}

When I access the mutex through mutex[i], do I need another lock since the child thread might be accessing the mutex array as well?

No. Because the variable mutex itself is never modified. Each member of the array behaves in an atomic fashion via the pthread_mutext_X() methods.

A slightly better design would be:

//main thread
struct ThreadData
{
    pthread_mutex_t mutex;
    queue<int>      queue;
};
pthread_t   t[5];
ThreadData  d[5];

for(int i = 0; i < 5; i++)
{
    pthread_create(&t[i], NULL, worker, &d[i]); // Pass a pointer to ThreadData
}


void* worker(void* arg)
{
    // Retrieve the ThreadData object.
    ThreadData  d = reinterpret_cast<ThreadData*>(arg);

    pthread_mutex_lock(&(d.mutex));
    <STUFF>      
}