Пробуждение отдельных streamов вместо занятости в pthreads

Я не уверен, что название отражает то, что я прошу здесь, но это лучше всего я могу обойтись без очень короткого титула. Я пытаюсь реализовать модель worker thread в pthreads . Я хочу создать набор streamов из main функции, после чего main stream делегирует задание работнику и дождитесь завершения всех streamов, прежде чем назначить им следующее задание (на самом деле, требование состоит в том, чтобы упорядочить stream в блоке как модель программирования CUDA, но на процессоре. Хотя это не относится к текущему вопросу). Массив job используется для указания типа задания для каждого streamа. В настоящее время я реализовал это с помощью семафоров, которые накладывают оживленное ожидание. Я ищу способы сделать это так, чтобы streamи спали и просыпались только тогда, когда они требуются, а не опроса непрерывно.

Функция, выполняемая каждым streamом

 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 stream занят, ожидая завершения всех рабочих streamов, а рабочие streamи также постоянно опросают массив заданий для проверки нового задания. Есть ли лучший способ сделать это, когда streamи будут спать и просыпаться, когда это необходимо, по линиям одиночного обработчика и используя pthread_kill() но это довольно грязно с отдельным обработчиком сигнала.

    Вы можете использовать условную переменную, чтобы streamи спали до тех пор, пока они не будут сигнализированы.

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