c - C eventcounter + sequencer 中的死锁并发问题

Question

我试图让这段代码工作，但出于某种原因它在 30 秒内死锁。

死锁似乎发生在 Putting 或 Getting 上，不管缓冲区是否已满。

我是不是遗漏了一些明显的东西或者没有使用正确的东西？我是 C 的新手!

编译:gcc -Wall -o test test.c -lpthread

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



struct sequence {
    pthread_mutex_t mutex;
    int count;
};

struct event {
    pthread_mutex_t critical;
    pthread_mutex_t critical2;
    pthread_mutex_t signalM;
    pthread_cond_t signalC;
    int eventCount;
};

struct allVars {
    struct sequence putSeq;
    struct sequence getSeq;
    struct event inEvents;
    struct event outEvents;
    int bufferSize;
    char buffer[10][128];
};



/**
 * Issue tickets in sequence. Used in conjunction with an Event Counter
 */
int getTicket(struct sequence *seq) {
    // begin critical section
    if (pthread_mutex_lock(&seq->mutex) != 0) {
        printf("mutex_lock in getticket error\n");
    }

    // remember current count
    int oldCount = seq->count;

    // increment count
    seq->count++;

    // end critical section
    if (pthread_mutex_unlock(&seq->mutex) != 0) {
        printf("mutex_unlock in getticket error\n");
    }

    return oldCount;
}



/**
 * Advance the EventCount (ticket)
 */
void advance(struct event *event) {
    // begin critical section
    if (pthread_mutex_lock(&event->critical) != 0) {
        fprintf(stderr, "mutex_lock in advance error\n");
        exit(EXIT_FAILURE);
    }

    // increment the event counter
    event->eventCount++;

    // end critical section
    if (pthread_mutex_unlock(&event->critical) != 0) {
        fprintf(stderr, "mutex_unlock in advance error\n");
        exit(EXIT_FAILURE);
    }

    // signal await to continue
    if (pthread_cond_signal(&event->signalC) != 0) {
        fprintf(stderr, "cond_signal in advance error\n");
        exit(EXIT_FAILURE);
    }
}



/**
 * Wait for ticket and buffer availability
 */
void await(struct event *event, int ticket) {

    int eventCount;

    // begin critical section
    if (pthread_mutex_lock(&event->critical) != 0) {
        fprintf(stderr, "mutex_lock in advance error\n");
        exit(EXIT_FAILURE);
    }

    eventCount = event->eventCount;

    // end critical section
    if (pthread_mutex_unlock(&event->critical) != 0) {
        fprintf(stderr, "mutex_unlock in advance error\n");
        exit(EXIT_FAILURE);
    }


    // loop until the ticket machine shows your number
    while (ticket > eventCount) {
        // wait until a ticket is called
        pthread_cond_wait(&event->signalC, &event->signalM);

        // begin critical section
        if (pthread_mutex_lock(&event->critical) != 0) {
            fprintf(stderr, "mutex_lock in advance error\n");
            exit(EXIT_FAILURE);
        }

        eventCount = event->eventCount;

        // end critical section
        if (pthread_mutex_unlock(&event->critical) != 0) {
            fprintf(stderr, "mutex_unlock in advance error\n");
            exit(EXIT_FAILURE);
        }
    }
}



/**
 * Add to buffer
 */
void putBuffer(struct allVars *allVars, char data[]) {
    // get a ticket
    int ticket = getTicket(&allVars->putSeq);

    // get the current write position
    int in;

    // begin critical section
    if (pthread_mutex_lock(&allVars->inEvents.critical) != 0) {
        fprintf(stderr, "mutex_lock in put error\n");
        exit(EXIT_FAILURE);
    }

    in = allVars->inEvents.eventCount;

    // end critical section
    if (pthread_mutex_unlock(&allVars->inEvents.critical) != 0) {
        fprintf(stderr, "mutex_unlock in put error\n");
        exit(EXIT_FAILURE);
    }

    // wait for ticket to be called (sequential writing)
    await(&allVars->inEvents, ticket);

    // wait until theres a space free in the buffer
    await(&allVars->outEvents, in - allVars->bufferSize + 1);   // set to 2 to keep 1 index distance

    // begin critical section
    if (pthread_mutex_lock(&allVars->inEvents.critical2) != 0) {
        fprintf(stderr, "mutex_lock in put error\n");
        exit(EXIT_FAILURE);
    }

    // add data to buffer
    strcpy(allVars->buffer[ticket % allVars->bufferSize], data);

    // end critical section
    if (pthread_mutex_unlock(&allVars->inEvents.critical2) != 0) {
        fprintf(stderr, "mutex_unlock in put error\n");
        exit(EXIT_FAILURE);
    }

    // increment the ticket display
    advance(&allVars->inEvents);
}



/**
 * Get from buffer
 */
char *getBuffer(struct allVars *allVars) {
    // get a ticket
    int ticket = getTicket(&allVars->getSeq);

    // get the current read position
    int out;

    // begin critical section
    if (pthread_mutex_lock(&allVars->outEvents.critical) != 0) {
        fprintf(stderr, "mutex_lock in get error\n");
        exit(EXIT_FAILURE);
    }

    out = allVars->outEvents.eventCount;

    // end critical section
    if (pthread_mutex_unlock(&allVars->outEvents.critical) != 0) {
        fprintf(stderr, "mutex_unlock in get error\n");
        exit(EXIT_FAILURE);
    }

    // wait for ticket to be called (sequential reading)
    await(&allVars->outEvents, ticket);

    // wait until theres something in the buffer
    await(&allVars->inEvents, out + 1);

    char *str = malloc(128);

    // critical section
    if (pthread_mutex_lock(&allVars->inEvents.critical2) != 0) {
        fprintf(stderr, "mutex_lock in put error\n");
        exit(EXIT_FAILURE);
    }

    // get the buffer data
    strcpy(str, allVars->buffer[ticket % allVars->bufferSize]);

    // end critical section
    if (pthread_mutex_unlock(&allVars->inEvents.critical2) != 0) {
        fprintf(stderr, "mutex_unlock in put error\n");
        exit(EXIT_FAILURE);
    }

    // increment buffer availability
    advance(&allVars->outEvents);

    return str;
}



/** child thread (producer) */
void *childThread(void *allVars) {
    char str[10];
    int count = 0;

    while (true) {
        sprintf(str, "%d", count++);
        putBuffer(allVars, str);
    }

    pthread_exit(EXIT_SUCCESS);
}



int main(void) {
    // init structs
    struct sequence putSeq = {
        PTHREAD_MUTEX_INITIALIZER,
        0
    };
    struct sequence getSeq = {
        PTHREAD_MUTEX_INITIALIZER,
        0
    };
    struct event inEvents = {
        PTHREAD_MUTEX_INITIALIZER,
        PTHREAD_MUTEX_INITIALIZER,
        PTHREAD_MUTEX_INITIALIZER,
        PTHREAD_COND_INITIALIZER,
        0
    };
    struct event outEvents = {
        PTHREAD_MUTEX_INITIALIZER,
        PTHREAD_MUTEX_INITIALIZER,
        PTHREAD_MUTEX_INITIALIZER,
        PTHREAD_COND_INITIALIZER,
        0
    };
    struct allVars allVars = {
        putSeq,         // sequence
        getSeq,
        inEvents,       // events
        outEvents,
        10,             // buffersize
        {"", {""}}      // buffer[][]
    };
    pthread_mutex_lock(&allVars.inEvents.signalM);
    pthread_mutex_lock(&allVars.outEvents.signalM);


    // create child thread (producer)
    pthread_t thread;
    if (pthread_create(&thread, NULL, childThread, &allVars)) {
        fprintf(stderr, "failed to create child thread");
        exit(EXIT_FAILURE);
    }


    // (consumer)
    while (true) {
        char *out = getBuffer(&allVars);
        printf("buf: %s\n", out);
        free(out);
    }


    return (EXIT_SUCCESS);
}

Answer 1

我不确定这是否是您问题的根源，但这是一个肯定会导致锁定问题的错误:

您使用的 pthread_cond_wait 和 pthread_cond_signal 有误。在调用 pthread_cond_wait 或 pthread_cond_signal 之前，您应该始终锁定条件的互斥量。

pthread_cond_wait 在您调用后自动为您释放互斥锁，并在发出信号后重新获取互斥锁。因此，您必须在向线程发出信号后释放锁。

This pthread programming tutorial当我在大学的一个类(class)学习 pthread 编程时，对我帮助很大。我建议看一下它，因为它还涵盖了 pthread 编程的其他一些方面(例如互斥锁)。