c - 如何生成遵循特定顺序的三个子进程？

Question

我被分配编写一个程序，该程序创建三个子进程，如下所示:A、B 和 C。

• 进程 A 从文件中读取数据并将 4KB 长的片段发送到消息队列。
• 进程 B 从队列中读取这些片段，对其进行转换并写入队列中的新数据。
• 最后，进程C从消息队列中读取转换后的字符串，并将其写入另一个文件中。

我使用 fork() 函数来创建它们，但无法按顺序生成三个子进程。当我运行程序时，通常会在A之前创建进程B或C，并且它们无法从消息队列中正确读取，因为进程A尚未生成。

我该如何解决这个问题？

<小时/>

这是我一直在编写的代码:

(注意:运行程序前必须添加两个参数:./program.exe source.txt destination.txt)

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/msg.h>
#include <unistd.h>

#define TAM 4096
#define FILEKEY "/bin/cat"
#define KEY 4242
#define N_PROCESSES 3

typedef struct msgbuf{
    long type;
    char text[TAM];
}message;

int main (int argc, char *argv[]){

key_t key;
pid_t pid;
int msqid;
message env, enda, endb;
message rcv;
message conv;
message msg;
int buf_length;
FILE *f1=NULL;
FILE *f2=NULL;
char string[TAM], *receive;
int rc;
int i;
int status;
int p;

if(argc < 2){
    printf("Incorrect command line arguments");
    exit(-1);
}

p = getpid();

/*Opens files*/
f1 = fopen(argv[1], "r");
if(f1 == NULL){
    exit(-1);
}

f2 = fopen(argv[2], "w");
if(f2 == NULL){
    fclose(f1);
    exit(-1);
}

/*Obtains key for message queue*/
key = ftok(FILEKEY, KEY);
if(key == -1){
    fclose(f1);
    fclose(f2);
    exit(-1);
}

/*Creates message queue*/
msqid = msgget(key, IPC_CREAT | 0600);
if(msqid == -1){
    fclose(f1);
    fclose(f2);
    exit(-1);
}

/*Message types*/
env.type = 1; /*Message from process A to process B*/
enda.type = 2; /*Process A has finished reading data from f1*/
conv.type = 3; /*Message from process B to process C*/
endb.type = 4; /*Process B has finished converting the string*/

/*Message size (4KB)*/
buf_length = sizeof(message)-sizeof(long);

/*Creates processes A, B and C*/
for ( i = 0; i < N_PROCESSES; i++){
    pid = fork();
    if(pid == -1){ /*Error*/
        msgctl(msqid, IPC_RMID, (struct msqid_ds *)NULL);
        fclose(f1);
        fclose(f2);
        exit(-1);
    }else if (pid == 0 && i == 0){/*Process A*/

        /*Reads from f1 while end of file is not reached*/
        while (fgets(string, TAM, f1) !=NULL){

            /*Copies string to env.text*/
            strcpy(env.text, cadena);

            /*Sends text fragments (4KB) to message queue*/
            if(msgsnd(msqid, &env, buf_length, IPC_NOWAIT)<0){
                msgctl(msqid, IPC_RMID, (struct msqid_ds *)NULL);
                fclose(f1);
                fclose(f2);
                exit(-1);
            }

        }
        /*Process A sends this message when there's no more data to read*/
        if(msgsnd(msqid, &enda, buf_length, IPC_NOWAIT)<0){
            msgctl(msqid, IPC_RMID, (struct msqid_ds *)NULL);
            fclose(f1);
            fclose(f2);
            exit(-1);
        }
        exit(EXIT_SUCCESS);

    }else if(pid == 0 && i == 1){/*Process B*/

        /*Reads text fragments (4KB) from message queue*/
        while (msgrcv(msqid, &rcv, buf_length, 1, IPC_NOWAIT)>0) {
            /*Converts string*/
            strcpy(receive, rcv.text);

            for(i = 0; i < TAM; i++){
                receive[i] = toupper(receive[i]);
            }

            strcpy(conv.text, receive);

            /*Sends fragments of converted string to message queue*/
            if(msgsnd(msqid, &conv, buf_length, IPC_NOWAIT)<0){
                msgctl(msqid, IPC_RMID, (struct msqid_ds *)NULL);
                fclose(f1);
                fclose(f2);
                exit(-1);
            }
        }
        /*The loop finishes when there's an error or when there are no more type 1 messages to read*/
        /*Reads type 2 message from process A*/
        rc = msgrcv(msqid, &rcv, buf_length, 2, 0);
        if( rc == -1){
            msgctl(msqid, IPC_RMID, (struct msqid_ds *)NULL);
            fclose(f1);
            fclose(f2);
            exit(-1);
        }
        /*Process B sends this message indicating that it has finished sending string fragments*/
        if(msgsnd(msqid, &endb, buf_length, IPC_NOWAIT)<0){
            msgctl(msqid, IPC_RMID, (struct msqid_ds *)NULL);
            fclose(f1);
            fclose(f2);
            exit(-1);
        }
        exit(EXIT_SUCCESS);

    }else if(pid == 0 && i == 2){/*Process C*/

        /*Reads converted string fragments from message queue*/
        while (msgrcv(msqid, &msg, buf_length, 3, IPC_NOWAIT)>0) {
        /*Writes fragments on another file*/
            if(fputs(msg.text, f2)<0){
                exit(-1);
            }

        }
        /*The loop finishes when there are no more fragments to write on f2*/
        /*Then process C reads the last message sent from B*/
        rc = msgrcv(msqid, &rcv, buf_length, 4, 0);
        if( rc == -1){
            msgctl(msqid, IPC_RMID, (struct msqid_ds *)NULL);
            fclose(f1);
            fclose(f2);
            exit(-1);
        }
        exit (EXIT_SUCCESS);
    }

}

/*Parent process waits for processes A, B and C*/
for (i=0; i<N_PROCESSES; i++) {
    pid=wait(&status);
    printf("Process %d with PPID = %d terminated\n", pid, p);
}
msgctl(msqid, IPC_RMID, (struct msqid_ds *)NULL);
fclose(f1);
fclose(f2);
return (EXIT_SUCCESS);
}

Answer 1

进程是按照您指定的顺序创建的，但是调度程序可以按照它喜欢的任何顺序来调度它们，因此您会发现进程 B 的代码在进程 A 完成之前执行，等等。

为了设置您想要的情况，我要做的是有 2 个队列，一个从 A 到 B，另一个从 B 到 C。为了简单起见，我将它们全部创建在父级中。

然后我会让读取子进程阻塞读取(没有 IPC_NOWAIT 的 msgrcv())，或者我会循环直到除了没有收到数据之外出现故障。

发送进程(A 和 B)在完成后都会将“仅此而已”指示符放入队列中，因此接收者会知道他们已经拥有了所有内容。