pthread_cond_broadcast problem

Question

Использование pthreads в Linux 2.6.30 Я пытаюсь отправить один сигнал, который вызовет запуск нескольких потоков. Передача, по-видимому, принимается только одним потоком. Я пробовал как pthread_cond_signal, так и pthread cond_broadcast, и оба имеют одинаковое поведение. Для мьютекса в pthread_cond_wait я пробовал как общие мьютексы, так и отдельные (локальные) мьютексы без видимой разницы.

worker_thread(void *p) 
{ 
    // setup stuff here 

    printf("Thread %d ready for action \n", p->thread_no); 
    pthread_cond_wait(p->cond_var, p->mutex); 
    printf("Thread %d off to work \n", p->thread_no); 

    // work stuff 
} 

dispatch_thread(void *p) 
{ 
    // setup stuff 

    printf("Wakeup, everyone "); 
    pthread_cond_broadcast(p->cond_var); 
    printf("everyone should be working \n"); 

    // more stuff 
} 

main() 
{ 
    pthread_cond_init(cond_var); 

    for (i=0; i!=num_cores; i++) { 
     pthread_create(worker_thread...); 
    } 

    pthread_create(dispatch_thread...); 
} 

Выход:

    Thread 0 ready for action 
    Thread 1 ready for action 
    Thread 2 ready for action 
    Thread 3 ready for action 
    Wakeup, everyone 
    everyone should be working 
    Thread 0 off to work 

Что такое хороший способ для передачи сигналов всех потоков?

Answer 1

Прежде всего, вы должны заблокировать мьютексы в точке, где вы звоните pthread_cond_wait(). Обычно рекомендуется удерживать мьютекс, когда вы вызываете pthread_cond_broadcast().

Во-вторых, вы должны перезвонить по вызову pthread_cond_wait(), пока условие ожидания является истинным. Возможны побочные пробуждения, и вы должны быть в состоянии справиться с ними.

И, наконец, ваша фактическая проблема: вы сигнализируете все потоки, но некоторые из них еще не ожидают отправки сигнала. Ваша основная нить и потоки отправки участвуют в ваших рабочих потоках: если основной поток может запускать поток отправки, а поток отправки может захватывать мьютексы и транслировать на нем до того, как рабочие потоки могут, тогда эти рабочие потоки никогда не проснутся.

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

В этом случае вы можете использовать мьютексы для защиты количества спальных потоков. Каждый поток захватывает мьютекс и увеличивает счетчик. Если счетчик соответствует количеству рабочих потоков, то последний поток должен увеличивать счетчик, и поэтому сигналы на другой переменной условия разделяют один и тот же мьютекс на поток спящего потока, который все потоки готовы. Затем поток ожидает первоначальное условие, которое вызывает его высвобождение мьютекса.

Если поток отправки еще не спал, когда последний рабочий поток сигнализирует об этом состоянии, он найдет, что счетчик уже соответствует желаемому счету и не беспокоится о ожидании, но сразу же транслируется в общем состоянии, чтобы разбудить работников, которым теперь всем гарантировано спать.

Во всяком случае, вот некоторые из которых работают исходный код, который конкретизирует свой образец кода и включает в себя мое решение:

#include <stdio.h> 
#include <pthread.h> 
#include <err.h> 

static const int num_cores = 8; 

struct sync { 
    pthread_mutex_t *mutex; 
    pthread_cond_t *cond_var; 
    int thread_no; 
}; 

static int sleeping_count = 0; 
static pthread_cond_t all_sleeping_cond = PTHREAD_COND_INITIALIZER; 

void * 
worker_thread(void *p_) 
{ 
    struct sync *p = p_; 
    // setup stuff here 

    pthread_mutex_lock(p->mutex); 
    printf("Thread %d ready for action \n", p->thread_no); 

    sleeping_count += 1; 
    if (sleeping_count >= num_cores) { 
     /* Last worker to go to sleep. */ 
     pthread_cond_signal(&all_sleeping_cond); 
    } 

    int err = pthread_cond_wait(p->cond_var, p->mutex); 
    if (err) warnc(err, "pthread_cond_wait"); 
    printf("Thread %d off to work \n", p->thread_no); 
    pthread_mutex_unlock(p->mutex); 

    // work stuff 
    return NULL; 
} 

void * 
dispatch_thread(void *p_) 
{ 
    struct sync *p = p_; 
    // setup stuff 

    pthread_mutex_lock(p->mutex); 
    while (sleeping_count < num_cores) { 
     pthread_cond_wait(&all_sleeping_cond, p->mutex); 
    } 
    printf("Wakeup, everyone "); 
    int err = pthread_cond_broadcast(p->cond_var); 
    if (err) warnc(err, "pthread_cond_broadcast"); 
    printf("everyone should be working \n"); 
    pthread_mutex_unlock(p->mutex); 

    // more stuff 
    return NULL; 
} 

int 
main(void) 
{ 
    pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER; 
    pthread_cond_t cond_var = PTHREAD_COND_INITIALIZER; 

    pthread_t worker[num_cores]; 
    struct sync info[num_cores]; 
    for (int i = 0; i < num_cores; i++) { 
     struct sync *p = &info[i]; 
     p->mutex = &mutex; 
     p->cond_var = &cond_var; 
     p->thread_no = i; 
     pthread_create(&worker[i], NULL, worker_thread, p); 
    } 

    pthread_t dispatcher; 
    struct sync p = {&mutex, &cond_var, num_cores}; 
    pthread_create(&dispatcher, NULL, dispatch_thread, &p); 

    pthread_exit(NULL); 
    /* not reached */ 
    return 0; 
}