The following is the producer-consumer algorithm for a piece of software that I'm upgrading to take advantage of multi-core processing. The intended platform is some flavor of Linux running on an EC2 HPC cc4x.large cluster, which feature 2 x Intel Xeon X5570, quad-core “Nehalem” architecture processors (2*4=8 cores). The software runs a genetic algorithm for optimizing artificial neural networks.
My dominating concern is performance. RAM and HD capacity are not an issue, but CPU time and anything else that delays processing is. Right now I've made a few (noted below), hopefully trivial, compromises to make the program portable to Mac OS X, which is on my home computer that I use for coding/debugging. I note a few other minor issues in the comments, most notably an uncertainty about thread-safeness in the consumer function. This is my first time working with threads. Advice/criticism of any kind is much appreciated.
#include <iostream>
#include <pthread.h>
#include <semaphore.h>
#include <vector>
#define NUM_THREADS 3 //will be >= to # of cores
#define N 30
//globals
sem_t* producer_gate;
sem_t* consumer_gate;
pthread_mutex_t access_queued =PTHREAD_MUTEX_INITIALIZER;
int queued;
int completed;
//a dummy class for testing thread-safeness
class the_class
{
public:
void find_num()
{
//make sure completion is non-instant and variable
double num = rand();
for(double q=0; 1; q++)
if(num == q)
return;
}
};
//the consumer function for handling the parallelizable code
void* Consumer(void* argument)
{
std::vector <the_class>* p_v_work = (std::vector <the_class>*) argument ;
while(1)
{
sem_wait(consumer_gate);
pthread_mutex_lock (&access_queued);
int index = queued;
queued--;
pthread_mutex_unlock (&access_queued);
(*p_v_work)[index-1].find_num();
completed++; //<-- thread safe??
std::cout << "\n" << index;
if(completed == N)
sem_post(producer_gate);
}
return NULL;
}
int main ()
{
srand(5);
//holds data to be processed and the methods for processing it
std::vector <the_class> work(N);
std::vector <the_class>* p_work = &work;
sem_unlink("producer_gate");
sem_unlink("consumer_gate");
//can't use `sem_init();` under Mac OS X so use sem_open instead
producer_gate = sem_open("producer_gate", O_CREAT, 0700, 0);
consumer_gate = sem_open("consumer_gate", O_CREAT, 0700, N);
completed = 0;
queued = N;
pthread_attr_t attr;
pthread_attr_init (&attr);
pthread_attr_setdetachstate (&attr, PTHREAD_CREATE_DETACHED); //no joining necessary
pthread_t threads[NUM_THREADS];
for(int q=0; q<NUM_THREADS; q++)
//on OS X, a pthread_t must be provided or a bus error occurs. relevant to performance?
pthread_create(&threads[q], &attr, Consumer, (void*) p_work );
for( int q=0; q < 4; q++)
{
sem_wait(producer_gate);
std::cout << "\nDONE\n";
////////////////////////////////////////////////////////
// Summate work done and layout work for next iteration
////////////////////////////////////////////////////////
completed = 0;
queued = N;
for(int q=0;q<N;q++)
sem_post(consumer_gate); //some way to just set this instead of incrementing?
}
sem_wait(producer_gate);
std::cout << "\n\nCompleted (: !!\n";
}