唤醒单个线程而不是 pthread 中的忙等待

Question

我不确定标题是否反映了我在这里的要求，但如果没有一个很长的标题，我能做到的最好。我正在尝试在pthreads 中实现worker thread 模型。我想从main函数中生成一组线程，然后main线程将作业委托给worker并等待所有线程完成后再分配下一个作业（实际上，要求是安排线程在一个类似于 CUDA 编程模型但在 CPU 上的块中。虽然它与当前问题无关）。 job 数组用于向每个线程指示作业的类型。目前，我已经使用信号量实现了这一点，这会导致繁忙的等待。我正在寻找方法使线程只在需要时进入睡眠状态并唤醒，而不是连续轮询。

每个线程执行的函数

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);
}

主要功能如下

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();
}

这个实现的问题是main 线程忙于等待所有工作线程完成并且工作线程也在不断地轮询作业数组以检查新作业。当线程进入睡眠并在需要时按照信号处理程序的行并使用pthread_kill() 唤醒时，是否有更好的方法来执行此操作，但是使用单独的信号处理程序有点混乱。

Answer 1

您可以使用conditional variable 使线程进入休眠状态，直到发出信号。

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);
}

然后在主目录中：

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();
}