Взаимная блокировка программы Pthread с использованием условных переменных

Question

Чего пытается достичь эта программа:

Эта программа должна синхронизировать несколько потоков "посетителей" и "машин".Посетители бродят в течение случайного количества времени, пока они не решают поехать на машине.Если они первые в очереди на автомобильную прогулку и автомобиль доступен, они отправляются в поездку, иначе они должны ждать, пока они не будут первыми в очереди, или машина вернется.Если в очереди нет посетителей, автомобили ждут, пока посетитель не захочет отправиться в поездку.

Дополнительная справочная информация:

Я переделал свою программу синхронизации потоковиспользуя условные переменные, как предложено в принятом ответе здесь .Я знаю, что на правильном пути, но моя программа по какой-то причине все еще находится в тупике, и я не могу понять почему.Я не знаю, как вы можете мне помочь, если я не дам вам код, поэтому вот он:

Проблемы:

1.) Тупик после короткого бита

2.) Иногда посетитель сначала стоит в очереди за машиной, но никогда не садится в машину.

Решение:

Былислишком много ошибок в моем коде ... и я думаю, что, исправляя одну, я часто (случайно) представлял другую.Я просто продолжал удалять функции программы, пока не устранил все ошибки, а затем собрал функции по очереди таким образом, чтобы не заблокировать мою программу.Спасибо всем за ваши предложения.

Answer 1

Во-первых, 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) и один для каждой машины (sc[CARS]). В качестве альтернативы вы можете использовать c_mutex для защиты очереди и статуса каждого автомобиля; или просто используйте v_mutex, чтобы защитить все. Но ради обучения мы пойдем с более сложным подходом.

Следующим шагом является определение точек ожидания, где переменные условия будут полезны. Для wait until at head of visitor queue ваши переменные условия для каждого посетителя (v_cond) будут в порядке; для wait until there is a car free будет проще всего добавить еще одну условную переменную (v_car_cond). Для wait until occupied подходят переменные состояния для каждого автомобиля c_cond. Для wait until not in car anymore можно использовать либо v_cond, либо c_cond, поскольку в этот момент между автомобилями и посетителями существует связь 1: 1. Нет необходимости в 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() того же мьютекса, пара разблокировки / блокировки может быть удалена.

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;
}

Надеюсь, это полезно ...

Answer 2

У вас много кода, поэтому вряд ли кто-нибудь найдет все ошибки для вас.Однако, несколько комментариев:

Мьютексы не являются семафорами.Некоторые из ваших циклов for в main () разблокируют мьютекс, который вы еще не заблокировали.Это почти наверняка ошибка.Концептуально мьютексы могут быть реализованы с помощью семафоров, и вы можете реализовать семафор с мьютексом и condvar, но вы разблокируете разблокированный мьютекс, и это неверно.

Каждый поток должен заблокировать мьютекс, сделать некоторыеработать, а затем разблокировать его.Поток не должен разблокировать мьютекс, который был заблокирован другим потоком, или приоритетно разблокировать мьютекс, который он не заблокировал.Если это вообще работает, то это странная реализация, которую вы используете в настоящее время, и она не переносима на другие реализации.

Ваш второй цикл for в main блокирует один и тот же мьютекс дважды подряд.Вы проходите этот пункт в коде?Поскольку вы зацикливаетесь, вы блокируете его больше, чем разблокируете.Я не удивлюсь, если ваш код останавливается здесь.(Иногда мьютексы могут быть рекурсивными, но мьютексы pthread не являются по умолчанию.)

pthread_cond_signal () на самом деле является просто оптимизацией по сравнению с pthread_cond_broadcast ().Используйте трансляцию до тех пор, пока не проработаете свои условия гонки.

Вы должны инициализировать все свои мьютексы и condvars вверху main, прежде чем начинать свои потоки.Возможно, что у вас нет ошибки здесь, но это не повредит, и это могло бы помочь.

Вы могли бы сделать лучше, если бы вы сократили все до одного мьютекса и одного condvar для короткогосрок.Похоже, что вы пытаетесь сделать мелкозернистую блокировку со всем, но если вы действительно не будете осторожны с порядком ваших блокировок, вы получите условия гонки и тупик.

Действительно, естьтолько один шаблон / шаблон, который вы должны использовать с мьютексами и condvars:

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(...);

Если у вас есть один мьютекс и condvar, вы должны попытаться сделать так, чтобы все блоки синхронизации выглядели так.Вы можете пропустить функцию while (...) / wait, если вам не нужно ждать, и вы можете пропустить трансляцию, если другие потоки не заботятся о внесенных вами изменениях, но если ваш код примерно не выглядитнапример, для каждого блока синхронизации это, вероятно, ошибка.

Answer 3

Я думаю, вам удобнее с семафорами.Вот реализация семафоров с точки зрения мьютексов и condvars:

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);
}

Может быть, если вы реализуете свое решение с точки зрения семафоров, вы получите те результаты, которые вам нужны.