使用条件变量的 Pthread 程序死锁

Question

该程序试图完成的任务：

该程序应该同步“访客”和“汽车”的多个线程。游客会随机闲逛一段时间，直到决定乘坐汽车。如果他们排在第一位乘车并且有车可用，他们就会乘车，否则他们必须等到他们排在第一位或有车回来。如果没有游客排队，汽车就会按顺序等待，直到有游客想要乘车。

更多背景信息：

我按照接受的答案中的建议使用条件变量重新制作了线程同步程序此处。我知道我走在正确的轨道上，但我的程序由于某种原因仍然陷入僵局，我一生都无法弄清楚为什么。我不知道你能如何帮助我，除非我给你代码，所以这里是：

问题：

1.) 短暂的死锁

2.) 有时访问者首先排队等待有一辆车，但从来没有上过车。

解决方案：

我的代码中存在太多错误...而且我认为，当我修复一个错误时，我经常（无意中）引入另一个错误。我只是不断删除程序的功能，直到消除了所有错误，然后以一种不会使程序陷入僵局的方式一一重新构建功能。谢谢大家的建议。

Answer 1

您拥有大量代码，因此不太可能有人为您找到所有错误。然而，有一些评论：

互斥体不是信号量。 main() 中的几个 for 循环正在解锁尚未锁定的互斥锁。这几乎肯定是一个错误。从概念上讲，互斥体可以使用信号量来实现，并且您可以使用互斥体和 condvar 来实现信号量，但是您要解锁未锁定的互斥体，这是不正确的。

每个线程都应该锁定一个互斥锁，做一些工作，然后解锁它。线程不应解锁已被另一个线程锁定的互斥体，或抢先解锁它未锁定的互斥体。如果这完全有效，那么这是您当前使用的实现中的一个怪癖，并且不能移植到其他实现。

main 中的第二个 for 循环连续两次锁定同一个互斥锁。您是否已经超越了代码中的这一点？由于您正在循环，因此您锁定它的次数多于解锁它的次数。如果您的代码停在这里，我不会感到惊讶。 （有时互斥体可以是递归的，但默认情况下 pthread 互斥体不是递归的。）

pthread_cond_signal() 实际上只是对 pthread_cond_broadcast() 的优化。使用广播，直到解决了比赛条件。

在启动线程之前，您应该在 main 顶部初始化所有互斥体和条件变量。您可能在这里没有错误，但这不会造成伤害，而且可能会有所帮助。

如果在短期内将所有内容减少到单个互斥体和单个条件变量，您可能会做得更好。看起来您正在尝试对所有内容进行细粒度锁定，但除非您非常小心锁定的顺序，否则您将遇到竞争条件和死锁。

实际上，只有一种模式/模板应该与互斥体和条件变量一起使用：

pthread_mutex_lock(...);

// optionally wait for something to be true
while (!some_condition) {
    pthread_cond_wait(...);
}

// make changes for how things should now be
shared_variable = new_value;

// optionally notify the rest of the world of your change
pthread_cond_broadcast(...);

pthread_mutex_unlock(...);

如果您有一个互斥体和条件变量，则应该尝试使所有同步块看起来像这样。如果您不需要等待，您可以省略 while(...)/wait 内容；如果没有其他线程关心您所做的更改，但如果您的代码看起来不粗略，您可以省略广播就像每个同步块一样，这可能是一个错误。

Answer 2

我认为你对信号量更满意。这是信号量在互斥体和条件变量方面的实现：

typedef struct {
    pthread_mutex_t mutex;
    pthread_cond_t condvar;
    unsigned long count;
} semaphore_t;

void semaphore_init (semaphore_t* sem, unsigned long count) {
    pthread_mutex_init(&sem->mutex, 0);
    pthread_cond_init(&sem->condvar, 0);
    pthread_mutex_lock(&sem->mutex);
    sem->count = count;
    pthread_mutex_unlock(&sem->mutex);
}

void semaphore_incr (semaphore_t* sem) {
    pthread_mutex_lock(&sem->mutex);
    sem->count++;
    pthread_cond_broadcast(&sem->condvar);
    pthread_mutex_unlock(&sem->mutex);
}

void semaphore_decr (semaphore_t* sem) {
    pthread_mutex_lock(&sem->mutex);
    while (sem->count == 0) {
        pthread_cond_wait(&sem->condvar, &sem->mutex);
    }
    sem->count--;
    pthread_mutex_unlock(&sem->mutex);
}

也许如果您根据信号量实现您的解决方案，您将得到您想要的结果。

Answer 3

首先，xscott 是正确的，您错误地使用了互斥体。即使它看起来有效一段时间也没关系，它仍然是错误的，并且可能只是由于纯粹的偶然而看起来有效。

我认为最好的方法是从第一原则构建设计，而不是逐行检查代码。我将描述我认为您所追求的基本算法：

visitor {
    sleep
    join end of visitor queue
    wait until at head of visitor queue
    wait until there is a car free
    remove car from car queue
    remove self from visitor queue
    occupy car
    wait until not in car anymore
}

car {
    join end of car queue
    wait until occupied
    sleep
    eject visitor from car
}

请注意，我没有明确标记唤醒点 - 只是等待。这是最好的方法 - 弄清楚你需要在哪里等待状态改变，然后你只需要在状态改变时进行唤醒（向条件变量发出信号）。

下一步是确定需要互斥体保护的主要共享数据。我明白了：

- The visitor queue;
- The car queue;
- The status of each car.

所以最细粒度的方法是为访客队列设置一个互斥锁（我们可以使用您的 v_mutex），为汽车队列设置一个互斥锁（c_mutex），还有一个用于车辆队列（c_mutex）每辆车 (sc[CARS])。或者，您可以使用c_mutex来保护汽车队列和每辆车的状态；或者只是使用 v_mutex 来保护一切。但为了学习，我们将采用更复杂的方法。

下一步是确定条件变量有用的等待点。对于等到访问者队列的头部，您的每个访问者条件变量（v_cond）就可以了；对于等到有车空闲，最简单的方法是添加另一个条件变量 (v_car_cond)。对于等待占用，每辆车的条件变量c_cond 是合适的。对于等到不再在车里，可以使用v_cond或c_cond，因为汽车和访客之间存在一对一的联系那一点。不需要v_cond2。

我们现在可以根据 pthreads 原语重写上面的伪代码。在大多数情况下，这非常接近您已有的 - 所以您绝对走在正确的轨道上。首先是访客：

    /* join end of visitor queue */
    pthread_mutex_lock(&v_mutex);
    v_line[v_id] = set_visitor_place_in_line();
    printf("Visitor %d is %d in line for a ride\n", v_id, v_line[v_id]);
    pthread_mutex_unlock(&v_mutex);

    /* wait until first in line */
    pthread_mutex_lock(&v_mutex);
    while (v_line[v_id] != 1) {
        pthread_cond_wait(&v_cond[v_id], &v_mutex);
    }
    pthread_mutex_unlock(&v_mutex);

    /* wait until there is a car free */
    pthread_mutex_lock(&c_mutex);
    while ((car = get_next_car()) == CARS) {
        pthread_cond_wait(&v_car_cond, &c_mutex);
    }
    pthread_mutex_unlock(&c_mutex);

    /* remove car from car queue */
    pthread_mutex_lock(&c_mutex);
    move_car_line();
    /* NOTE: We do not need to signal v_car_cond here, because only the first
     * visitor in line can be waiting there, and we are still the first visitor
     * in line at this point. */
    pthread_mutex_unlock(&c_mutex);

    /* remove self from visitor queue */
    pthread_mutex_lock(&v_mutex);
    move_passenger_line();
    next_v = get_next_passenger();
    if (next_v < VISITORS)
        pthread_cond_signal(&v_cond[next_v]);
    pthread_mutex_unlock(&v_mutex);

    /* occupy car */
    pthread_mutex_lock(&sc[car]);
    c_state[car] = v_id;
    pthread_cond_signal(&c_cond[car]);
    pthread_mutex_unlock(&sc[car]);

    /* wait until not in car anymore */
    pthread_mutex_lock(&sc[car]);
    while(c_state[car] == v_id) {
        pthread_cond_wait(&v_cond[v_id], &sc[car]);
    }
    pthread_mutex_unlock(&sc[car]);

其次是汽车：

    /* join end of car queue */
    pthread_mutex_lock(&c_mutex);
    c_line[c_id] = set_car_place_in_line();
    if (c_line[c_id] == 1)
        pthread_cond_signal(&v_car_cond);
    pthread_mutex_unlock(&c_mutex);

    /* wait until occupied */
    pthread_mutex_lock(&sc[c_id]);
    while ((v_id = c_state[c_id]) == VISITORS) {
        pthread_cond_wait(&c_cond[c_id], &sc[c_id]);
    }
    pthread_mutex_unlock(&sc[c_id]);

    /* visitor is on car ride for random amount of time */
    sleep(rand()%5);

    /* eject visitor from car */
    pthread_mutex_lock(&sc[c_id]);
    c_state[c_id] = VISITORS;
    pthread_cond_signal(&v_cond[v_id]);
    pthread_mutex_unlock(&sc[c_id]);

上面的情况可以简化 - 只要有 pthread_mutex_unlock() 紧随其后的是同一个互斥体的 pthread_mutex_lock()，解锁/lock 对可以删除。

PS：

您不必担心您的汽车以“错误的顺序”加入汽车队列 - 无论如何，它们在公园里行驶时都会失去秩序。如果您确实关心这一点，请在启动任何汽车线程之前将汽车放入主线程中的队列中，并更改汽车代码，以便它在主循环结束时将自身重新添加到队列中。

---

使用这种方法的整体代码，省略了全局变量和辅助函数，这些都很好，看起来像这样：

pthread_mutex_t c_mutex = PTHREAD_MUTEX_INITIALIZER; /* mutex protecting c_line and cars_out */
pthread_mutex_t v_mutex = PTHREAD_MUTEX_INITIALIZER; /* mutex protecting v_line */
pthread_cond_t c_cond[CARS]; /* condition variables for waiting for c_state[i] to change from VISITORS to another value */
pthread_cond_t v_cond[VISITORS]; /* condition variables for visitor waiting to be first in line or ejected from a car */
pthread_cond_t v_car_cond = PTHREAD_COND_INITIALIZER; /* Condition variable for a visitor first in line, but waiting for a car. */
pthread_mutex_t sc[CARS]; /* one mutex per car, sc[i] protects c_state[i] */

int main(){
    int i;
    void *status;
    pthread_t c[CARS];
    pthread_t v[VISITORS];

    srand(time(NULL));

    printf("Jurassic Park is open, cars are being prepped for passengers\n");

    for (i = 0; i < CARS; i++){
        pthread_mutex_init(&sc[i], NULL);
        pthread_cond_init(&c_cond[i], NULL);
    }

    for (i = 0; i < VISITORS; i++){
        pthread_cond_init(&v_cond[i], NULL);
    }

    for (i = 0; i < CARS; i++){
        c_state[i] = VISITORS;
        pthread_create(&c[i], NULL, car, (void *)i);
    }

    for (i = 0; i < VISITORS; i++){
        pthread_create(&v[i], NULL, visitor, (void *)i);
    }

    for (i = 0; i < VISITORS; i++){
        pthread_join(v[i], &status);
    }

    museum_empty++;

    printf("All visitors have left Jurassic Park\n");

    for (i = 0; i < CARS; i++){
        pthread_mutex_lock(&sc[i]);
        c_state[i] = -1;
        pthread_cond_signal(&c_cond[i]);
        pthread_mutex_unlock(&sc[i]);
    }

    for (i = 0; i < CARS; i++){
        pthread_join(c[i], &status);
    }

    printf("Jurrasic Park is closed, all cars have been parked\n");

    pthread_exit(NULL);

    return 0;
}

void *car(void *i)
{
    int c_id = (int) i;
    int v_id;

    while (museum_empty != 1) {

        /* join end of car queue */
        pthread_mutex_lock(&c_mutex);
        c_line[c_id] = set_car_place_in_line();
        if (c_line[c_id] == 1)
            pthread_cond_signal(&v_car_cond);
        printf("Car %d is ready for a passenger and is %d in line                        %d of %d cars are out\n", c_id, c_line[c_id], cars_out, CARS);
        pthread_mutex_unlock(&c_mutex);

        /* wait until occupied */
        pthread_mutex_lock(&sc[c_id]);
        while ((v_id = c_state[c_id]) == VISITORS) {
            pthread_cond_wait(&c_cond[c_id], &sc[c_id]);
        }
        pthread_mutex_unlock(&sc[c_id]);

        if(museum_empty == 1){
            break;
        }

        pthread_mutex_lock(&c_mutex);
        cars_out++;
        printf("Visitor %d is riding in car %d                                           %d of %d cars are out --\n", v_id, c_id, cars_out, CARS);
        pthread_mutex_unlock(&c_mutex);

        /* visitor is on car ride for random amount of time */
        sleep(rand()%5);

        printf("Visitor %d is  done riding in car %d\n", v_id, c_id);

        /* eject visitor from car */
        pthread_mutex_lock(&sc[c_id]);
        c_state[c_id] = VISITORS;
        pthread_cond_signal(&v_cond[v_id]);
        pthread_mutex_unlock(&sc[c_id]);

        pthread_mutex_lock(&c_mutex);
        cars_out--;
        pthread_mutex_unlock(&c_mutex);
    }

    return NULL;
}

void *visitor(void *i)
{
    int v_id = (int) i;
    int next_v;
    int j = 0;
    int car;

    while (j < RIDES) {
        if (j == 0) {
            printf("Visitor %d entered museum and is wandering around\n", v_id);
        } else {
            printf("Visitor %d got back from ride and is wandering around\n", v_id);
        }

        sleep(rand()%3); /* visitor is wandering */

        /* join end of visitor queue */
        pthread_mutex_lock(&v_mutex);
        v_line[v_id] = set_visitor_place_in_line();
        printf("Visitor %d is %d in line for a ride\n", v_id, v_line[v_id]);

        /* wait until first in line */
        while (v_line[v_id] != 1) {
            pthread_cond_wait(&v_cond[v_id], &v_mutex);
        }
        pthread_mutex_unlock(&v_mutex);

        /* wait until there is a car free */
        pthread_mutex_lock(&c_mutex);
        while ((car = get_next_car()) == CARS) {
            pthread_cond_wait(&v_car_cond, &c_mutex);
        }

        /* remove car from car queue */
        move_car_line();
        pthread_mutex_unlock(&c_mutex);

        /* remove self from visitor queue */
        pthread_mutex_lock(&v_mutex);
        move_passenger_line();
        next_v = get_next_passenger();
        if (next_v < VISITORS)
            pthread_cond_signal(&v_cond[next_v]);
        pthread_mutex_unlock(&v_mutex);

        /* occupy car */
        pthread_mutex_lock(&sc[car]);
        c_state[car] = v_id;
        pthread_cond_signal(&c_cond[car]);

        /* wait until not in car anymore */
        /* NOTE: This test must be against v_id and *not* VISITORS, because the car could have been
         * subsequently occupied by another visitor before we are woken. */
        while(c_state[car] == v_id) {
            pthread_cond_wait(&v_cond[v_id], &sc[car]);
        }
        pthread_mutex_unlock(&sc[car]);

        j++;
    }

    printf("Visitor %d leaving museum.\n", v_id);
    return NULL;
}

我希望这有帮助......