SONET 中的 3:1 多路复用器模拟：使用管道和进程

多个 STS-1 数据流的复用在 SONET 中扮演了非常重要的角色，这些 STS-1 数据流称为支流 (tributary)。一个 3:1 多路复用器将 3 个输入的 STS1 支流复用到一个 STS3 输出流中。复用过程按字节进行，也就是说，前 3 字节分别是支流 1、2 和 3 的第一字节，接下去的 3 字节分别是支流 1、2 和 3 的第二字节，以此类推。编写一个程序模拟这样的 3:1 多路复用器。程序应该包含 5 个进程：主进程创建 4 个进程，其中 3 个进程分别对应 3 个 STS-1 支流，第 4 个对应多路复用器。每个支流进程从一个输入文件中读取连续的 810B 作为一个 STS-1 顿，它们将这些 (逐字节) 发送给多路复用器进程。多路复用器进程接收这些字节，然后逐字节输出到标准输出设备上。进程之间的通信采用管道形式进行。

代码实现

#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <sys/wait.h>

#define BUFFER_SIZE 810

int main()
{
    // 创建管道
    int tributary1_to_mux[2], tributary2_to_mux[2], tributary3_to_mux[2], mux_to_output[2];
    if (pipe(tributary1_to_mux) < 0 || pipe(tributary2_to_mux) < 0 || pipe(tributary3_to_mux) < 0 || pipe(mux_to_output) < 0) {
        perror("pipe error");
        exit(EXIT_FAILURE);
    }

    // 创建 3 个支流进程
    int tributary1_pid = fork();
    if (tributary1_pid < 0) {
        perror("fork error");
        exit(EXIT_FAILURE);
    }
    if (tributary1_pid == 0) {
        // 支流 1 进程
        close(tributary1_to_mux[0]); // 关闭读端
        FILE *input = fopen("input1", "rb"); // 打开输入文件
        if (input == NULL) {
            perror("fopen error");
            exit(EXIT_FAILURE);
        }
        unsigned char buffer[BUFFER_SIZE];
        while (fread(buffer, BUFFER_SIZE, 1, input) > 0) { // 逐个 STS-1 顿读取并发送
            for (int i = 0; i < BUFFER_SIZE; i++) {
                write(tributary1_to_mux[1], &buffer[i], 1); // 逐字节发送到多路复用器
            }
        }
        fclose(input);
        close(tributary1_to_mux[1]); // 关闭写端
        exit(EXIT_SUCCESS);
    }

    // 创建支流 2 进程
    int tributary2_pid = fork();
    if (tributary2_pid < 0) {
        perror("fork error");
        exit(EXIT_FAILURE);
    }
    if (tributary2_pid == 0) {
        // 支流 2 进程
        close(tributary2_to_mux[0]); // 关闭读端
        FILE *input = fopen("input2", "rb"); // 打开输入文件
        if (input == NULL) {
            perror("fopen error");
            exit(EXIT_FAILURE);
        }
        unsigned char buffer[BUFFER_SIZE];
        while (fread(buffer, BUFFER_SIZE, 1, input) > 0) { // 逐个 STS-1 顿读取并发送
            for (int i = 0; i < BUFFER_SIZE; i++) {
                write(tributary2_to_mux[1], &buffer[i], 1); // 逐字节发送到多路复用器
            }
        }
        fclose(input);
        close(tributary2_to_mux[1]); // 关闭写端
        exit(EXIT_SUCCESS);
    }

    // 创建支流 3 进程
    int tributary3_pid = fork();
    if (tributary3_pid < 0) {
        perror("fork error");
        exit(EXIT_FAILURE);
    }
    if (tributary3_pid == 0) {
        // 支流 3 进程
        close(tributary3_to_mux[0]); // 关闭读端
        FILE *input = fopen("input3", "rb"); // 打开输入文件
        if (input == NULL) {
            perror("fopen error");
            exit(EXIT_FAILURE);
        }
        unsigned char buffer[BUFFER_SIZE];
        while (fread(buffer, BUFFER_SIZE, 1, input) > 0) { // 逐个 STS-1 顿读取并发送
            for (int i = 0; i < BUFFER_SIZE; i++) {
                write(tributary3_to_mux[1], &buffer[i], 1); // 逐字节发送到多路复用器
            }
        }
        fclose(input);
        close(tributary3_to_mux[1]); // 关闭写端
        exit(EXIT_SUCCESS);
    }

    // 创建多路复用器进程
    int mux_pid = fork();
    if (mux_pid < 0) {
        perror("fork error");
        exit(EXIT_FAILURE);
    }
    if (mux_pid == 0) {
        // 多路复用器进程
        close(tributary1_to_mux[1]); // 关闭支流 1 的写端
        close(tributary2_to_mux[1]); // 关闭支流 2 的写端
        close(tributary3_to_mux[1]); // 关闭支流 3 的写端
        close(mux_to_output[0]); // 关闭输出的读端
        unsigned char buffer[3][BUFFER_SIZE]; // 用于存储 3 个支流当前的字节
        int current_byte[3] = {0}; // 用于记录 3 个支流当前的字节位置
        int finished[3] = {0}; // 用于记录 3 个支流是否已经完成
        while (1) {
            // 从 3 个支流中读取下一个字节
            for (int i = 0; i < 3; i++) {
                if (!finished[i]) {
                    int n = read(i == 0 ? tributary1_to_mux[0] : i == 1 ? tributary2_to_mux[0] : tributary3_to_mux[0], &buffer[i][current_byte[i]], 1);
                    if (n < 0) {
                        perror("read error");
                        exit(EXIT_FAILURE);
                    }
                    if (n == 0) { // 如果支流已经完成，则标记为已完成
                        finished[i] = 1;
                    }
                    current_byte[i] += n;
                    if (current_byte[i] == BUFFER_SIZE) { // 如果当前顿已经读完，则重置字节位置
                        current_byte[i] = 0;
                    }
                }
            }

            // 将 3 个字节输出到标准输出设备上
            for (int i = 0; i < 3; i++) {
                if (!finished[i]) {
                    write(mux_to_output[1], &buffer[i][current_byte[i]], 1);
                    current_byte[i]++;
                    if (current_byte[i] == BUFFER_SIZE) { // 如果当前顿已经读完，则重置字节位置
                        current_byte[i] = 0;
                    }
                }
            }

            // 如果 3 个支流都已经完成，则退出循环
            if (finished[0] && finished[1] && finished[2]) {
                break;
            }
        }
        close(tributary1_to_mux[0]);
        close(tributary2_to_mux[0]);
        close(tributary3_to_mux[0]);
        close(mux_to_output[1]);
        exit(EXIT_SUCCESS);
    }

    // 关闭管道的未使用端口
    close(tributary1_to_mux[0]);
    close(tributary1_to_mux[1]);
    close(tributary2_to_mux[0]);
    close(tributary2_to_mux[1]);
    close(tributary3_to_mux[0]);
    close(tributary3_to_mux[1]);
    close(mux_to_output[1]);

    // 等待所有进程结束
    waitpid(tributary1_pid, NULL, 0);
    waitpid(tributary2_pid, NULL, 0);
    waitpid(tributary3_pid, NULL, 0);
    waitpid(mux_pid, NULL, 0);

    return 0;
}

程序解释

程序的核心部分是多路复用器进程。它使用一个循环不断从 3 个支流中读取下一个字节，并将这 3 个字节依次输出到标准输出设备上。如果某个支流已经完成，则不再从它读取数据，直到所有支流都完成为止。支流进程的代码比较简单，只是从输入文件中读取数据并逐字节发送到多路复用器进程。

需要注意的是，程序中使用了 4 个管道：3 个管道用于支流进程向多路复用器进程发送数据，1 个管道用于多路复用器进程向标准输出设备输出数据。在创建进程时，需要在子进程中关闭不需要的管道端口，以免造成死锁或其他问题。

总结

本程序成功地模拟了 SONET 中的 3:1 多路复用器，展示了如何使用管道和进程进行通信。该程序还体现了进程间同步和数据流控制的机制。