c - 模拟为火车站的 pthread 同步

Question

我正在使用 pthreads 为火车站建模。每列火车都有自己的线程和自己的条件变量，用于监控其对主轨道的访问。列车信息从文件中读取，格式为:

(方向):(加载时间):(穿越时间)

主轨道上一次只能有一列火车。除非装载并准备就绪，否则火车不能在主轨道上行驶。

有一个单独的调度线程负责协调所有列车之间对主轨道的访问。调度程序线程还根据各种规则(例如火车行驶的方向)决定哪列火车可以进入。现在，如果我能让火车按准备就绪的顺序到达主轨道，我会很高兴。

到目前为止，这是我的代码:

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

struct Train{
    pthread_t thread;
    pthread_cond_t granted;

    int train_number;
    int loading_time;
    int crossing_time;

    int priority;
    char direction;
    char state;

}*newTrain;

struct Train *trains[3];

pthread_mutex_t track       = PTHREAD_MUTEX_INITIALIZER;
pthread_cond_t  dispatcher  = PTHREAD_COND_INITIALIZER;

char *loading = "L";
char *ready = "R";
char *granted_t = "T";
char *gone = "G";
char *acknowledged_gone = "X";

void *dispatcher_function(void *train_count) {
    int count = *((int *) train_count);
    int trains_remaining = count; 

    /* Check for ready trains until all trains have left the station */
    while (trains_remaining > 0) {
        pthread_mutex_lock(&track);

        int t_granted = 0;
        int next = -1;

        for(int i = 0; i < count; i++){
            if (strcmp(&trains[i]->state, "T") == 0)
                t_granted = 1;
            if (strcmp(&trains[i]->state, "R") == 0)
                next = i;
            if (strcmp(&trains[i]->state, "G") == 0){
                trains_remaining--;
                trains[i]->state = *acknowledged_gone;
            }
        }

        /* Train was granted access to station wait for it to leave */
        if (t_granted) {
            pthread_cond_wait(&dispatcher, &track);
        }

        /* No trains in station. Wait for train */
        if (next == -1) {
            pthread_cond_wait(&dispatcher, &track);
        }

        /* Train ready in station grant next train track permission*/
        else{
            trains[next] -> state = *granted_t;
            pthread_cond_signal(&(trains[next] -> granted));
        }
        pthread_mutex_unlock(&track);
    }
    pthread_exit(0);
}

void *train_function(void* train) {
    struct Train *self = (struct Train*)train;

    /* Each train has its own cond var */
    pthread_cond_init(&self->granted, NULL);

    /* Load train */
    usleep(self -> loading_time);

    /* Lock track */
    pthread_mutex_lock(&track);

    /* Train ready */
    self -> state = *ready;
    printf("Train %d is ready to go %c\n", self -> train_number, self -> direction);

    /* Signal dispatcher */
    pthread_cond_signal(&dispatcher);

    while(strcmp(&self->state, "T") != 0)
        pthread_cond_wait(&(self->granted), &track);

    /* Use the track */
    printf("Train %d is ON the main track going %c\n", self -> train_number, self -> direction);
    usleep(self -> crossing_time);
    self -> state = *gone;
    printf("Train %d is OFF the main track after going %c\n", self -> train_number, self -> direction);

    pthread_cond_signal(&dispatcher);
    pthread_mutex_unlock(&track);

    pthread_exit(0);
}

int main() {

    FILE *ptr_file;
    char buff[10];
    int train_count = 0;
    char *train;
    char line[15];
    pthread_t train_threads[3];
    pthread_t dispatcher_thread;

    ptr_file = fopen("./trains.txt", "r");
    if (!ptr_file) 
    {
        perror("fopen for trains.txt failed");
        exit(EXIT_FAILURE);
    }

    /* Create train for each line of file */
    while (fgets(buff,10, ptr_file)!=NULL) {
        train = (char*)malloc(10 * sizeof(char));

        /* Build train string */
      sprintf(line, "%d:", train_count);
        strcat(line, buff);
        strcpy(train, line);

        /* Parse train information */
        int line_number = atoi(strtok(train, ":,"));
        char *direction = strtok(NULL,":,");
        int loading_time = atoi(strtok(NULL, ":,"));
        int crossing_time = atoi(strtok(NULL, ":,"));

        /* Create trains */
        newTrain = (struct Train *) malloc(sizeof(struct Train));
        newTrain -> train_number = line_number;
        newTrain -> crossing_time = crossing_time;
        newTrain -> loading_time = loading_time;
        newTrain -> direction = *direction;
        newTrain -> state = *loading;

        if(pthread_create(&train_threads[train_count], NULL, &train_function, (void *) newTrain))
        {
            perror("pthread create failed");
            exit(EXIT_FAILURE);
        }

        trains[line_number] = newTrain;
        train_count++;
    }
    fclose(ptr_file);

    /* Create dispatcher */
    if(pthread_create(&dispatcher_thread, NULL, &dispatcher_function, (void *) &train_count))
    {
        perror("pthread create failed");
        exit(EXIT_FAILURE);
    }

    /* Wait for dispatcher to finish */
    pthread_join(dispatcher_thread, NULL);
    printf("all done");

    free(train);
    for (int i = 0; i < train_count; i++) {
        free(trains[i]);
    }
    exit(EXIT_SUCCESS);
}

这是 trains.txt 输入文件:

e:10,6
W:5,7
E:3,10

这是我运行时得到的输出:

Train 0 is ready to go e
Train 0 is ON the main track going e
Train 0 is OFF the main track after going e
Train 2 is ready to go E
Train 1 is ready to go W
Train 2 is ON the main track going E
Train 2 is OFF the main track after going E
Train 1 is ON the main track going W
Train 1 is OFF the main track after going W

程序现在在所有火车离开车站后挂起。这么近，我一定错过了什么。

Answer 1

在你的程序可以推理之前，你有几个错误需要更正:

trains.state是单char object - 它不是字符串，因为它不一定后跟空终止符。这意味着您不能将它的地址传递给 strcmp()就像你在几个地方所做的那样 - 而不是这样:

if (strcmp(&trains[i]->state, "T") == 0)

使用:

if (trains[i]->state == 'T')

(注意字符常量而不是字符串常量的单引号)。

你不能free(train)在 train_function() 的末尾，因为调度员一直在运行并且需要访问所有列车结构。相反，在 main() 中将它们全部释放调度员退出后。

火车条件变量granted从未被初始化。你不能复制 pthread_cond_t变量 - 而不是使用 pthread_cond_init :

void *train_function(void* train)
{
    struct Train *self = (struct Train*)train;

    /* Each train has its own cond var */
    pthread_cond_init(&self->granted, NULL);

train_function() 的开始修改 self->state不持有锁，这意味着它可以与读取相同对象的调度程序竞争。您需要锁定修改:

/* Load train */
pthread_mutex_lock(&track);
self -> state = *loading;
pthread_mutex_unlock(&track);
usleep(self -> loading_time);

/* Lock track */
pthread_mutex_lock(&track);

/* Train ready */
self -> state = *ready;
printf("Train %d is ready to go %c\n", self -> train_number, self -> direction);

(在其他线程启动之前，您可以通过将 state 初始化为“加载” main() 中的所有列车来避免第一次锁定)。

---

此外，您不能仅仅因为pthread_cond_wait()就假定您正在等待的条件为真。已经醒了。 pthread_cond_wait()即使没有发出信号也允许返回；它返回只是意味着它可能已经收到信号。这意味着，例如在 train_function , 你需要绕过 pthread_cond_wait()使用 while而不是 if :

while (self->state != 'T')
    pthread_cond_wait(&self->granted, &track);

对于调度员中发现轨道上有火车的情况，您需要执行类似的操作。

这实际上是您问题的核心 - 当您在此时醒来时:

if (t_granted) {
    pthread_cond_wait(&dispatcher, &track);
    trains_remaining--;
}

您假设这是因为您在铁轨上看到的火车现在已经完工了。但这不一定是真的 - 您可能已经收到下一列火车完成装载的信号。这意味着您将绕过一圈，再次看到轨道上的同一列火车，然后递减 trains_remaining。太多次了。

所以你不能只调整trains_remaining当您在轨道上看到一列火车时，因为您可能会看到同一列火车两次 - 或者根本看不到，如果它设置为非常快地“离开”。

相反，您应该递减 trains_remaining你第一次看到给定的火车处于“消失”状态。您可以通过添加一个新状态来实现此目的，调度员在看到火车“离开”后将其设置为该状态，例如:

    for (int i = 0; i < count; i++) {
        if (trains[i]->state == 'T')
            t_granted = 1;
        if (trains[i]->state == 'R')
            next = i;
        if (trains[i]->state == 'G') {
            trains_remaining--;
            trains[i]->state = 'X'; /* acknowledged gone */
        }
    }