“produce_numbers”中的 pthread_cond_wait 给了我分段错误

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

您的问题是，您永远不会在属于struct circular_buffer 成员的cb.can_produce 和cb.can_consume 条件变量上调用pthread_cond_init()。

您需要更改cb_init() 函数，以便它使用pthread_mutex_init() 和pthread_cond_init() 初始化新的circular_buffer 的mutex、can_produce 和can_consume 成员。

您还在 produce_numbers() 函数中使用未初始化的指针变量 cb - 它被声明为 struct circular_buffer *cb; 但从未分配给。您需要将cb 改为struct thread_arg 的成员，并将其设置在main() 中，以便produce_numbers() 引用与其他线程相同的struct circular_buffer 对象。