調用pthread_cond_wait在「produce_numbers」給我分割故障

Question

我的問題是，我不能在main()初始化circular_buffer和基本的問題是，該方案在produce_numbers()函數調用pthread_cond_wait()的時候，我採取了分段故障。

我的代碼如下。

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

int seed; 

typedef struct circular_buffer { 
    void *buffer;    // data buffer 
    void *buffer_end;   // end of data buffer 
    size_t capacity;   // maximum number of items in the buffer 
    size_t count;    // number of items in the buffer 
    size_t sz;     // size of each item in the buffer 
    void *head;     // pointer to head 
    void *tail;     // pointer to tail 
    pthread_mutex_t mutex;  // needed to add/remove data from the buffer 
    pthread_cond_t can_produce; // signaled when items are removed 
    pthread_cond_t can_consume; // signaled when items are added 
} circular_buffer; 

struct thread_data { 
    int thread_id; 
    int thread_id2; 
    int seed; 
    int numbersofprod; 
    int talenum; 
}; 

struct thread_data thread_data_array[40]; 

void cb_init(circular_buffer *cb, size_t capacity, size_t sz) 
{ 
    cb->buffer = malloc(capacity * sz); 
    if (cb->buffer == NULL) { 
     printf("Could not allocate memory..Exiting! \n"); 
     exit(1); 
    } 
    // handle error 

    cb->buffer_end = (char *)cb->buffer + capacity * sz; 
    cb->capacity = capacity; 
    cb->count = 0; 
    cb->sz = sz; 
    cb->head = cb->buffer; 
    cb->tail = cb->buffer; 
} 
// remove first item from circular item void 
cb_pop_front(circular_buffer *cb, void *item) 
{ 
    printf("pop_front"); 
    if (cb->count == 0) { 
     printf("Access violation. Buffer is empty\n"); 
     exit(1); 
    } 
    memcpy(item, cb->tail, cb->sz); 
    cb->tail = (char *)cb->tail + cb->sz; 
    if (cb->tail == cb->buffer_end) 
     cb->tail = cb->buffer; 
    cb->count--; 
} 
// add item to circular buffer 
void cb_push_back(circular_buffer *cb, const void *item) 
{ 
    if (cb->count == cb->capacity) { 
     printf("Access violation. Buffer is full\n"); 
     exit(1); 
    } 
    memcpy(cb->head, item, cb->sz); 
    cb->head = (char *)cb->head + cb->sz; 
    if (cb->head == cb->buffer_end) 
     cb->head = cb->buffer; 
    cb->count++; 
} 
// destroy circular buffer 
void cb_free(circular_buffer *cb) 
{ 
    free(cb->buffer); 
    // clear out other fields too, just to be safe 
} 

void *consume_numbers(void *arg) 
{ 
    struct thread_data *my_data; 
    int threadId = my_data->thread_id2; 
    int rc; 
    circular_buffer *cb = 
     (circular_buffer *)arg; // δήλωση δείκτη σε δομή circular_buffer 
    printf("to capacity einai %d!\n", cb->capacity); 
    while (1) { 
     pthread_mutex_lock(&cb->mutex); 

     // elegxos an to buffer einai adeio 
     if (cb->count == 0) { // empty 
      // wait for new items to be appended to the buffer 
      printf("eeeeeee"); 
      pthread_cond_wait(&cb->can_consume, &cb->mutex); 
     } // if 
     int *tmp_read = (int *)malloc(4); 
     cb_pop_front(cb, (void *)tmp_read); 
     free(tmp_read); 
     printf("Thread: consuming item %d!\n", cb->count); 

     // signal the fact that new items may be produced 
     rc = pthread_cond_signal(&cb->can_produce); 
     if (rc != 0) { 
      printf("ERROR: return code from pthread_cond_signal() is %d\n", rc); 
      pthread_exit(&rc); 
     } // if 
     rc = pthread_mutex_unlock(&cb->mutex); 
     if (rc != 0) { 
      printf("ERROR: return code from pthread_mutex_unlock() is %d\n", rc); 
      pthread_exit(&rc); 
     } // if 
    }  // while 

    // de ftanei pote edw 
    return NULL; 

} // consume_numbers 

void *produce_numbers(void *threadarg) 
{ 

    struct thread_data *my_data; 
    my_data = (struct thread_data *)threadarg; 
    int seeed = my_data->seed; 
    int numberofprod = my_data->numbersofprod; 
    int threadId = my_data->thread_id; 
    int talenumber = my_data->talenum; 
    int j, r; 
    int *tid; 
    int rc; 
    circular_buffer *cb;            

    // open file 
    FILE *f = fopen("producers_in.txt", "w"); 
    if (f == NULL) { 
     printf("Error opening file!\n"); 
     exit(1); 
    } // telos anoigmatos arxeiou 
    srand(seeed); // paragwgi sporou 

    for (j = 0; j < numberofprod; j++) { 

     pthread_mutex_lock(&cb->mutex); 

     // elegxos an o buffer einai full 
     if (cb->count == cb->capacity) { 
      // wait until some elements are consumed 
      printf("mpika"); 
      // pthread_cond_wait(&cb->can_produce, &cb->mutex); 
      if (rc != 0) { 
       printf("ERROR: return code from pthread_cond_wait() is %d\n", 
         rc); 
       pthread_exit(&rc); 
      } // if 
     }  // while 

     else { 
      r = rand(); 
      printf("tyxaios arithmos %d\n", r); 
      cb_push_back(
       cb, (void *)&r); // grapse sto arxeio txt ton tyxaio arithmo r 
      const char *text = "the random number of producer with id"; 
      fprintf(f, "%s ", text); 
      tid = (int *)threadarg; 
      fprintf(f, "%d\n", tid); 

      const char *text2 = "is"; 
      fprintf(f, " %s ", text2); 
      fprintf(f, "%d\n", r); 
      printf("to count einai %d\n", cb->count); 

      // signal the fact that new items may be consumed 
      rc = pthread_cond_signal(&cb->can_consume); 
      if (rc != 0) { 
       printf("ERROR: return code from pthread_cond_signal() is %d\n", 
         rc); 
       pthread_exit(&rc); 
      } // if 

      rc = pthread_mutex_unlock(&cb->mutex); 
      if (rc != 0) { 
       printf("ERROR: return code from pthread_mutex_unlock() is %d\n", 
         rc); 
       pthread_exit(&rc); 
      } // if 
     }  // else 

    } // for 
    fclose(f); 
    // NEVER REACHED 
    tid = (int *)threadarg; 
    pthread_exit(tid); 

} // produce_numbers 

// MAIN 
int main(int argc, char *argv[]) 
{ 

    if (argc != 6) { 
     printf("ERROR: the program should take 5 argument!\n"); 
     exit(-1); 
    } 
    int producers = atoi(argv[1]); 
    int consumers = atoi(argv[2]); 
    int numbersofprod = atoi(argv[3]); 
    int talenum = atoi(argv[4]); 
    int seed = atoi(argv[5]); 
    /*elegxoume oti ta dedomena pou dothikan einai swsta.*/ 
    if (producers < 1) { 
     printf("ERROR: the number of producers to run should be a positive number.Current number given %d.\n", 
       producers); 
     exit(-1); 
    } 
    if (consumers < 1) { 
     printf(
      "ERROR: the number of consumers to run should be a positive number.Current number given %d.\n", consumers); 
     exit(-1); 
    } 
    if (numbersofprod < 
     1) { 
     printf("ERROR: the number of numbersofprod to run should be a positive number. Current number given %d.\n", numbersofprod); 
     exit(-1); 
    } 
    if (talenum < 1) { 
     printf("ERROR: the number of tale to run should be a positive number. Current number given %d.\n", talenum); 
     exit(-1); 
    } 
    if (seed < 1) { 
     printf("ERROR: the number of the seed to run should be a positive number. Current number given %d.\n", seed); 
     exit(-1); 
    } 
    printf("Main: We will create %d producer threads", producers); 
    printf(" and %d consumers threads.\n", consumers); 
    pthread_t *prod; 
    pthread_t *cons; 

    prod = malloc(producers * sizeof(pthread_t)); 
    if (prod == NULL) { 
     printf("NOT ENOUGH MEMORY!\n"); 
     return -1; 
    } 
    cons = malloc(consumers * sizeof(pthread_t)); 
    if (cons == NULL) { 
     printf("NOT ENOUGH MEMORY!\n"); 
     return -1; 
    } 

    int rc; 
    int rc2; 
    int threadCount; 
    int threadCount2; 
    int countArray[producers]; 
    int countArray2[consumers]; 

    //INITIALIZE MUTEX AND CONDITION VARIABLES 
    circular_buffer cb = { 
    .count=0, 
    .mutex = PTHREAD_MUTEX_INITIALIZER, 
    .can_produce = PTHREAD_COND_INITIALIZER, 
    .can_consume = PTHREAD_COND_INITIALIZER 
}; 
    cb_init(cb, talenum, 1); //end_of initializing 

    for (threadCount = 0; threadCount < producers; threadCount++) { 
     printf("Main: creating thread %d for producers\n ", 
       threadCount); 
     countArray[threadCount] = threadCount + 1; 
     thread_data_array[threadCount].thread_id = threadCount; 
     thread_data_array[threadCount].talenum = talenum; 
     thread_data_array[threadCount].seed = seed; 
     thread_data_array[threadCount].numbersofprod = 
      numbersofprod; /*dimiourgia tou thread*/ 
     rc = pthread_create(&prod[threadCount], NULL, produce_numbers, 
          (void *)&thread_data_array[threadCount]); 

     /*elegxos oti to thread dimiourgithike swsta.*/ 
     if (rc != 0) { 
      printf("ERROR: return code from pthread_create() is %d\n", 
        rc); 
      exit(-1); 
     } 

     for (threadCount2 = 0; threadCount2 < consumers; 
      threadCount2++) { 
      countArray2[threadCount2] = threadCount2 + 1; 
      thread_data_array[threadCount2].thread_id2 = threadCount2; 

      printf("Main: creating thread %d for consumers\n", 
        threadCount); 
      /*dimiourgia tou thread*/ 
      rc2 = pthread_create(cons, NULL, consume_numbers, 
           (void *)&cb); 
      /*elegxos oti to thread dimiourgithike swsta.*/ 
      if (rc2 != 0) { 
       printf(
        "ERROR: return code from pthread_create() is %d\n", 
        rc2); 
       exit(-1); 
      } 
     } 
    } 

    void *status; 
    for (threadCount = 0; threadCount < producers; 
     threadCount++) { 
     rc = pthread_join(prod[threadCount], &status); 
     if (rc != 0) { 
      printf("ERROR: return code from pthread_join() is %d\n", 
        rc); 
      exit(-1); 
     } 

     printf("Main: Thread %d returned %d as status code.\n", 
       countArray[threadCount], (*(int *)status)); 
    } 

    free(prod); 
    `` return 1; 
    void *status2; 
    for (threadCount2 = 0; threadCount2 < consumers; threadCount2++) { 
     rc2 = pthread_join(cons[threadCount2], &status2); 
     if (rc2 != 0) { 
      printf("ERROR: return code from pthread_join() is %d\n", 
        rc); 
      exit(-1); 
     } 
     printf("Main: Thread % d returned % d as status code.\n ", countArray2[threadCount2], 
       (*(int *)status2)); 
    } 
    free(cons); 
    return 1; 
} 

Answer 1

你的問題是，你永遠不會呼籲cb.can_produce和cb.can_consume條件變量是struct circular_buffer成員pthread_cond_init()。

您需要更改cb_init()功能，使得它初始化新circular_buffer的mutex，can_produce和can_consume成員pthread_mutex_init()和pthread_cond_init()。

您還在produce_numbers()函數中使用未初始化的指針變量cb - 聲明爲struct circular_buffer *cb;，但從未分配給該函數。您需要使cb成爲struct thread_arg的成員，並將其設置爲main()，以便produce_numbers()與其他線程具有相同的struct circular_buffer對象的引用。