基于pthread多线程的多视频文件管道传输故障排查求助

需求说明 #

发送端（sender.c）需求 #

• 能够打开并读取多个视频文件内容
• 将文件分块并通过共享管道发送
• 向接收端标识文件结束（EOF）
• 文件发送完成后关闭管道

接收端（receiver.c）需求 #

• 能够创建新的视频文件
• 通过不同线程接收发送端的分块数据并写入新文件
• 能够识别EOF以终止单个文件传输
• 平滑终止线程，关闭文件及管道

当前问题 #

新文件可正常创建，但视频无法正确接收——三个视频中仅一个被传输，且该视频已损坏。

现有核心代码 #

sender.c #

char *videoFiles[NUM_THREADS] = {"KungFuPanda.mp4", "flowerVideo.mp4","Flower2.mp4"}; // 根据需要调整文件名

// 为每个视频文件创建线程
pthread_t threads[NUM_THREADS];
for (int i = 0; i < NUM_THREADS; i++) {
    pthread_create(&threads[i], NULL, sendVideo, (void *)videoFiles[i]);
}

// 等待所有线程完成
for (int i = 0; i < NUM_THREADS; i++) {
    pthread_join(threads[i], NULL);
}

printf("所有视频文件发送成功！\n");

void *sendVideo(void *filename) {
    // 变量定义
    int p_fd, n;
    char *fileName = (char *)filename;
    FILE *videoFile;

    // 初始化文件指针
    videoFile = fopen(fileName, "r");
    if (videoFile == NULL) {
        printf("未找到文件: %s\n", fileName);
        pthread_exit(NULL);
    }

    // 打开管道用于写入
    p_fd = open("videoPipe", O_WRONLY);
    if (p_fd == -1) {
        perror("打开管道失败");
        fclose(videoFile);
        pthread_exit(NULL);
    }

    // 分配缓冲区读取文件内容
    char buffer[CHUNK_SIZE];

    // 分块读取并发送文件内容
    size_t bytesRead;
    while ((bytesRead = fread(buffer, 1, CHUNK_SIZE, videoFile)) > 0) {
        // 将文件块写入管道
        n = write(p_fd, buffer, bytesRead);
        if (n == -1) {
            perror("写入管道失败");
            fclose(videoFile);
            close(p_fd);
            pthread_exit(NULL);
        }
    }

    // 通过写入0字节发送EOF信号
    n = write(p_fd, "", 0);
    if (n == -1) {
        perror("发送EOF信号失败");
        fclose(videoFile);
        close(p_fd);
        pthread_exit(NULL);
    }

    // 关闭文件和管道
    fclose(videoFile);
    close(p_fd);

    printf("视频文件 %s 发送成功！\n", fileName);

    pthread_exit(NULL);
}

receiver.c #

for (int i = 0; i < NUM_THREADS; i++) {
    // 创建新文件用于写入接收的数据
    char filename[20];
    snprintf(filename, sizeof(filename), "received_file_%d.mp4", i);
    receivedFiles[i] = fopen(filename, "wb");
    if (receivedFiles[i] == NULL) {
        perror("创建文件失败");
        close(p_fd);
        return 1;
    }
}

while (1) {
    // 从管道读取数据
    n = read(p_fd, buffer, CHUNK_SIZE);
    if (n == -1) {
        perror("读取管道失败");
        break;
    } else if (n == 0) { // 到达文件末尾
        break;
    }

    // 将接收的数据写入所有文件
    for (int i = 0; i < NUM_THREADS; i++) {
        fwrite(buffer, 1, n, receivedFiles[i]);
    }
}

// 关闭管道和文件
close(p_fd);
for (int i = 0; i < NUM_THREADS; i++) {
    fclose(receivedFiles[i]);
}

问题根源分析 #

1. 发送端：多线程写管道数据混乱
   三个线程同时往同一个管道写数据，不同文件的字节块会被随机混杂，接收端根本分不清哪些数据属于哪个文件。
2. 发送端：EOF信号无效
   写入0字节的操作无法标识单个文件结束——管道的EOF状态只有在所有写端都关闭时才会触发，单个线程关闭管道后，其他线程还在写，管道不会停止。
3. 接收端：逻辑完全错误
   接收端没有启动多线程处理，反而把读到的所有数据同时写入所有文件，导致每个输出文件都是三个原文件的混合体，必然损坏；同时接收端只能等所有发送端关闭管道才停止，完全没处理单文件结束的逻辑。

修复方案 #

1. 发送端：给每个文件加头部标识 #

在发送文件内容前，先发送包含文件名长度、文件名、文件总大小的头部，让接收端能区分不同文件的边界，还能通过文件大小判断何时接收完成。

修改sendVideo函数：

// 定义文件头部结构体
typedef struct {
    int name_len;
    char name[256];
    long long file_size;
} FileHeader;

void *sendVideo(void *filename) {
    int p_fd, n;
    char *fileName = (char *)filename;
    FILE *videoFile = fopen(fileName, "rb"); // 用二进制模式打开，避免视频文件损坏
    if (videoFile == NULL) {
        printf("未找到文件: %s\n", fileName);
        pthread_exit(NULL);
    }

    // 获取文件总大小
    fseek(videoFile, 0, SEEK_END);
    long long file_size = ftell(videoFile);
    fseek(videoFile, 0, SEEK_SET);

    p_fd = open("videoPipe", O_WRONLY);
    if (p_fd == -1) {
        perror("打开管道失败");
        fclose(videoFile);
        pthread_exit(NULL);
    }

    // 构造并发送头部
    FileHeader header;
    header.name_len = strlen(fileName);
    strncpy(header.name, fileName, sizeof(header.name)-1);
    header.file_size = file_size;
    n = write(p_fd, &header, sizeof(FileHeader));
    if (n != sizeof(FileHeader)) {
        perror("发送文件头部失败");
        fclose(videoFile);
        close(p_fd);
        pthread_exit(NULL);
    }

    // 分块发送文件内容
    char buffer[CHUNK_SIZE];
    size_t bytesRead;
    while ((bytesRead = fread(buffer, 1, CHUNK_SIZE, videoFile)) > 0) {
        n = write(p_fd, buffer, bytesRead);
        if (n == -1) {
            perror("写入管道失败");
            fclose(videoFile);
            close(p_fd);
            pthread_exit(NULL);
        }
    }

    fclose(videoFile);
    close(p_fd);
    printf("视频文件 %s 发送成功！\n", fileName);
    pthread_exit(NULL);
}

2. 接收端：多线程对应接收文件 #

主线程先读取每个文件的头部，然后创建独立线程负责该文件的接收，直到读取完头部指定的文件大小为止。

修改接收端核心逻辑：

typedef struct {
    int p_fd;
    long long file_size;
    FILE *out_file;
} ReceiverArgs;

// 单个文件的接收线程函数
void *receiveFile(void *args) {
    ReceiverArgs *recv_args = (ReceiverArgs*)args;
    char buffer[CHUNK_SIZE];
    long long received = 0;

    while (received < recv_args->file_size) {
        // 计算本次需要读取的字节数（避免最后一次读超）
        size_t to_read = recv_args->file_size - received;
        if (to_read > CHUNK_SIZE) to_read = CHUNK_SIZE;

        int n = read(recv_args->p_fd, buffer, to_read);
        if (n == -1) {
            perror("读取管道失败");
            fclose(recv_args->out_file);
            free(args);
            pthread_exit(NULL);
        }

        fwrite(buffer, 1, n, recv_args->out_file);
        received += n;
    }

    fclose(recv_args->out_file);
    free(args);
    pthread_exit(NULL);
}

int main() {
    int p_fd = open("videoPipe", O_RDONLY);
    if (p_fd == -1) {
        perror("打开管道失败");
        return 1;
    }

    pthread_t threads[NUM_THREADS];
    for (int i = 0; i < NUM_THREADS; i++) {
        FileHeader header;
        // 读取文件头部
        int n = read(p_fd, &header, sizeof(FileHeader));
        if (n != sizeof(FileHeader)) {
            perror("读取文件头部失败");
            close(p_fd);
            return 1;
        }

        // 创建对应输出文件
        char out_name[256];
        snprintf(out_name, sizeof(out_name), "received_%s", header.name);
        FILE *out_file = fopen(out_name, "wb");
        if (!out_file) {
            perror("创建输出文件失败");
            close(p_fd);
            return 1;
        }

        // 准备线程参数
        ReceiverArgs *args = malloc(sizeof(ReceiverArgs));
        args->p_fd = p_fd;
        args->file_size = header.file_size;
        args->out_file = out_file;

        // 创建接收线程
        pthread_create(&threads[i], NULL, receiveFile, args);
    }

    // 等待所有接收线程完成
    for (int i = 0; i < NUM_THREADS; i++) {
        pthread_join(threads[i], NULL);
    }

    close(p_fd);
    printf("所有视频文件接收完成！\n");
    return 0;
}

3. 前置准备 #

在运行程序前，先通过命令创建管道：

mkfifo videoPipe

内容的提问来源于stack exchange，提问作者Rabbiya Riaz