phread_cond_broadcast состояние гонки

Question

Я не могу найти способ остановить это состояние гонки. Основной поток вызывает программу вещания, чтобы разбудить все потоки, затем вызывает cond_wait, чтобы дождаться завершения всех потоков. Последняя нить заканчивается, сигнализируя о главном потоке. Проблема в том, что иногда не все рабочие потоки ждут переменную условия, когда основной поток выполняет широковещательную передачу. Код немного грязный, потому что я пытался исправить различные исправления.

unsigned nProcs, curProc; 
    pthread_mutex_t WORKlock = PTHREAD_MUTEX_INITIALIZER; 
    pthread_mutex_t MAINlock = PTHREAD_MUTEX_INITIALIZER; 
    pthread_mute_t wj_varlock = PTHREAD_MUTEX_INITIALIZER; 
    pthread_cond_t WORKsig = PTHREAD_COND_INITIALIZER; 
    pthread_cond_t MAINsig = PTHREAD_COND_INITIALIZER; 

    void *do_work(void *t) 
    { 
    static unsigned offset = 0; 
    unsigned myoff = offset++, locked = 0; 
    volatile int nc; 

    while(1) 
      { 
      if(locked == 0) 
        pthread_mutex_lock(&WORKlock); 
      else 
        locked = 0; 
      pthread_cond_wait(&WORKsig, &WORKlock); 
      pthread_mutex_unlock(&WORKlock); 
      // this is where the work gets done 
      //sleep(1); 
      printf("thread %d done!\n", myoff); 

      pthread_mutex_lock(&wj_varlock); 
      nc = --curProc; 
      pthread_mutex_unlock(&wj_varlock); 
      if(nc == 0) 
        { 
        pthread_mutex_lock(&WORKlock); 
        locked = 1; 
        pthread_cond_signal(&MAINsig); 
        } 

      } 
    } 

    void main(int argc, char **argv) 
    { 
    unsigned i, k; 
    pthread_t pth; 

    nProcs = get_nprocs(); // get number of core from system 
    for(i = 0; i < nProcs; ++i) 
      { 
      k = pthread_create(&pth, NULL, do_work, NULL); 
      if(k != 0) 
        { 
        perror("pthread_create"); 
        exit(k); 
        } 
      } 

    pthread_mutex_lock(&MAINlock); 
    while(1) 
      { 
      //prepare work to be done 

      puts("work prep"); 
      //sleep(1); 

      curProc = nProcs; // use global var to track active threads 
      pthread_cond_broadcast(&WORKsig); 
      pthread_cond_wait(&MAINsig, &MAINlock); 
      } 
    } 

Answer 1

Переменные условия должны быть сопряжены с условием в течение некоторого общего состояния (часто называемой «предикат») - вот почему они называются переменные условия. Так, например, для запуска рабочих можно использовать простую глобальную переменную флаг:

int start_work = 0; /* Protected by WORKlock */ 

Тогда в рабочие потоки вы могли бы сделать:

pthread_mutex_lock(&WORKlock); 
while (!start_work) 
    pthread_cond_wait(&WORKsig, &WORKlock); 
pthread_mutex_unlock(&WORKlock); 

/* this is where the work gets done */ 

И в главном потоке вы могли бы сделать:

pthread_mutex_lock(&WORKlock); 
start_work = 1; 
pthread_mutex_unlock(&WORKlock); 
pthread_cond_broadcast(&WORKsig); 

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

Чтобы заблокировать основную нить до тех пор, пока рабочие не будут закончены, вы можете использовать curProc > 0 в качестве предиката. Обратите внимание, что вам не нужны как wj_varlock, так и MAINlock - вам просто нужно защитить переменную curProc.

(Для того, чтобы сделать ваш дизайн правильно, что вам нужно сделать некоторые тщательное чередование условий на curProc и start_work)

#include <stdio.h> 
#include <stdlib.h> 
#include <pthread.h> 
#include <unistd.h> 

int start_work = 0; /* Protected by WORKlock */ 
unsigned curProc;  /* Protected by MAINlock */ 

pthread_mutex_t WORKlock = PTHREAD_MUTEX_INITIALIZER; 
pthread_mutex_t MAINlock = PTHREAD_MUTEX_INITIALIZER; 
pthread_cond_t WORKsig = PTHREAD_COND_INITIALIZER; 
pthread_cond_t MAINsig = PTHREAD_COND_INITIALIZER; 


void *do_work(void *t) 
{ 
    static int off; /* Protected by WORKlock */ 
    int myoff; 

    pthread_mutex_lock(&WORKlock); 
    myoff = ++off; 
    pthread_mutex_unlock(&WORKlock); 

    while (1) 
    { 
     /* Wait to start work */ 
     pthread_mutex_lock(&WORKlock); 
     while (start_work == 0) 
      pthread_cond_wait(&WORKsig, &WORKlock); 
     pthread_mutex_unlock(&WORKlock); 

     /* Increase number of active processes */ 
     pthread_mutex_lock(&MAINlock); 
     ++curProc; 
     pthread_mutex_unlock(&MAINlock); 
     pthread_cond_signal(&MAINsig); 

     /* this is where the work gets done */ 
     printf("Working (%d)...\n", myoff); 
     sleep(1); 

     /* Wait for all work to be done */ 
     pthread_mutex_lock(&WORKlock); 
     while (start_work == 1) 
      pthread_cond_wait(&WORKsig, &WORKlock); 
     pthread_mutex_unlock(&WORKlock); 

     /* Reduce number of active processes */ 
     pthread_mutex_lock(&MAINlock); 
     --curProc; 
     if (curProc == 0) 
      pthread_cond_signal(&MAINsig); 
     pthread_mutex_unlock(&MAINlock); 
    } 
} 

int get_nprocs(void) 
{ 
    return 8; 
} 

int main(int argc, char **argv) 
{ 
    unsigned i, k; 
    pthread_t pth; 

    unsigned nProcs = get_nprocs(); // get number of core from system 
    for (i = 0; i < nProcs; ++i) 
    { 
     k = pthread_create(&pth, NULL, do_work, NULL); 
     if (k != 0) 
     { 
      perror("pthread_create"); 
      exit(k); 
     } 
    } 

    curProc = 0; 
    while (1) 
    { 
     //prepare work to be done 

     puts("work prep"); 
     //sleep(1); 

     /* Tell threads to start work */ 
     pthread_mutex_lock(&WORKlock); 
     start_work = 1; 
     pthread_mutex_unlock(&WORKlock); 
     pthread_cond_broadcast(&WORKsig); 

     /* Wait for threads to start */ 
     pthread_mutex_lock(&MAINlock); 
     while (curProc < nProcs) 
      pthread_cond_wait(&MAINsig, &MAINlock); 
     pthread_mutex_unlock(&MAINlock); 

     /* Tell threads not to start next lot of work */ 
     pthread_mutex_lock(&WORKlock); 
     start_work = 0; 
     pthread_mutex_unlock(&WORKlock); 
     pthread_cond_broadcast(&WORKsig); 

     /* Wait for threads to finish */ 
     pthread_mutex_lock(&MAINlock); 
     while (curProc > 0) 
      pthread_cond_wait(&MAINsig, &MAINlock); 
     pthread_mutex_unlock(&MAINlock); 
    } 

    return 0; 
} 

Следует отметить, что этот вид закрываемой работы более просто сделан с барьерами :

#include <stdio.h> 
#include <stdlib.h> 
#include <pthread.h> 
#include <unistd.h> 

pthread_mutex_t WORKlock = PTHREAD_MUTEX_INITIALIZER; 
pthread_barrier_t WORKbarrier; 

void *do_work(void *t) 
{ 
    static int off; /* Protected by WORKlock */ 
    int myoff; 

    pthread_mutex_lock(&WORKlock); 
    myoff = ++off; 
    pthread_mutex_unlock(&WORKlock); 

    while (1) 
    { 
     /* Wait to start work */ 
     pthread_barrier_wait(&WORKbarrier); 

     /* this is where the work gets done */ 
     printf("Working (%d)...\n", myoff); 
     sleep(1); 

     /* Wait for all work to be done */ 
     pthread_barrier_wait(&WORKbarrier); 
    } 
} 

int get_nprocs(void) 
{ 
    return 8; 
} 

int main(int argc, char **argv) 
{ 
    unsigned i, k; 
    pthread_t pth; 

    unsigned nProcs = get_nprocs(); // get number of core from system 
    for (i = 0; i < nProcs; ++i) 
    { 
     k = pthread_create(&pth, NULL, do_work, NULL); 
     if (k != 0) 
     { 
      perror("pthread_create"); 
      exit(k); 
     } 
    } 

    pthread_barrier_init(&WORKbarrier, NULL, nProcs + 1); 
    while (1) 
    { 
     //prepare work to be done 

     puts("work prep"); 
     //sleep(1); 

     /* Tell threads to start work */ 
     pthread_barrier_wait(&WORKbarrier); 

     /* Wait for threads to finish */ 
     pthread_barrier_wait(&WORKbarrier); 
    } 

    return 0; 
}