CODEWITHSUNDEEP
cmultithreadingpthreadssemaphore
user1135552
user1135552

Reputation: 73

Waking up individual threads instead of busy wait in pthreads

I'm not sure if the title reflects what I'm asking here but thats best I can do without a very loong title. I'm trying to implement a worker thread model in pthreads. I want to spawn a set of threads from the main function and thereafter the main thread delegates the job to the worker and wait for all threads to complete before assigning them the next job (Actually, the requirement is to arrange the thread in a block much like CUDA programming model but on the CPU. Although its not relevant to the current question). The job array is used to indicate the type of job to each thread. Currently, I've implemented this using semaphores which imposes a busy wait. I'm looking for ways to make it so that the threads go to sleep and wake up only when they are required rather than polling continuously.

The function executed by each thread

volatile int jobs[MAX_THREADS]; // global job indicator array
sem_t semaphore;                // semaphore to indicate completion
thread_execute(void *args)
{
  tid = get_id(args);
  while(jobs[tid] != -1)
  {
    if(jobs[tid] == 0) continue; // no job
    if(jobs[tid] == JOBS_1)
    {
      jobs1();
      jobs[tid] = 0; // go back to idle state
      sem_post(&semapahore);
    }
    if(jobs[tid] == JOBS_2)
    {
      jobs2();
      jobs[tid] = 0; // go back to idle state
      sem_post(&semapahore);
    }
  }

  pthread_exit(NULL);
}

The main function is as follows

int main()
{
  sem_init(&semaphore, 0, 0);
  jobs[0...MAX_THREADS] = 0;
  spawn_threads();

  // Dispatch first job
  jobs[0...MAX_THREADS] = JOBS_1;
  int semvalue = 0;
  while (semvalue < MAX_THREADS) // Wait till all threads increment the semaphore
    sem_getvalue(&sempaphore, &semvalue);

  sem_init(&semaphore, 0, 0); // Init semaphore back to 0 for the next job
                              // I'm actually using diff. semaphores for diff. jobs
  jobs[0...MAX_THREADS] = JOBS_2;
  while (semvalue < MAX_THREADS)
    sem_getvalue(&sempaphore, &semvalue);

  jobs[0...MAX_THREADS] = -1; // No more jobs
  pthread_join();
}

The problem with this implementation is that the main thread is busy waiting for all worker threads to finish and worker threads are also constantly polling the jobs array to check for a new job. Is there a better way to do this when threads go to sleep and wake up when needed along the lines of a singal handler and using pthread_kill() but it is kind of messy with a separate signal handler.

Upvotes: 4

Views: 8477

Answers (1)

Tudor
Tudor

Reputation: 62439

You can use a conditional variable to make the threads go to sleep until signaled. 

volatile int jobs[MAX_THREADS]; // global job indicator array
pthread_cond_t th_cond;     // threads wait on this
pthread_mutex_t th_mutex;   // mutex to protect the signal
int busyThreads = MAX_THREADS;

pthread_cond_t m_cond;      // main thread waits on this
pthread_mutex_t m_mutex;    // mutex to protect main signal

thread_execute(void *args)
{
  tid = get_id(args);
  while(jobs[tid] != -1)
  {
    if(jobs[tid] == 0) continue; // no job
    if(jobs[tid] == JOBS_1)
    {
      jobs1();
      jobs[tid] = 0; // go back to idle state
      pthread_mutex_lock(&th_mutex);      
          pthread_mutex_lock(&m_mutex);   
          --busyThreads;                       // one less worker
          pthread_cond_signal(&m_cond);        // signal main to check progress
          pthread_mutex_unlock(&m_mutex);
      pthread_cond_wait(&th_cond, &th_mutex);   // wait for next job
      pthread_mutex_unlock(&th_mutex);      
    }
    if(jobs[tid] == JOBS_2)
    {
      jobs2();
      jobs[tid] = 0; // go back to idle state
      pthread_mutex_lock(&th_mutex);
      --busyThreads;
      pthread_cond_wait(&th_cond, &th_mutex);
      pthread_mutex_unlock(&th_mutex);
    }
  }

  pthread_exit(NULL);
}

then in main:

int main()
{
  sem_init(&semaphore, 0, 0);
  jobs[0...MAX_THREADS] = 0;
  spawn_threads();

  // Dispatch first job
  jobs[0...MAX_THREADS] = JOBS_1;
  int semvalue = 0;

  pthread_mutex_lock(&m_mutex);
  while(busyThreads > 0)        // check number of active workers
      pthread_cond_wait(&m_cond, &m_mutex);   
  pthread_mutex_unlock(&m_mutex);

  busyThreads = MAX_THREADS;
  pthread_mutex_lock(&th_mutex);
  pthread_cond_broadcast(&th_cond);   // signal all workers to resume
  pthread_mutex_unlock(&th_mutex);

  // same for JOBS_2;

  jobs[0...MAX_THREADS] = -1; // No more jobs
  pthread_join();
}

Upvotes: 5

Related Questions