在 C 中父子之间使用 Pipes IPC 进行多次测试

Question

我正在编写一个使用管道进行父进程通信的程序，关键是我需要从终端发送不同数量的数据（大数据）但我不知道如何订购 'split'将消息分成较小的消息（每个消息 4096 字节）并按顺序将它们发送到读取端。我已经尝试过一个进行（消息大小/4096）次迭代的循环，但它没有用，所以我放弃了这个想法。

这是我的代码：

#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <sys/types.h>
#include <string.h>
#include <fcntl.h>
#include <time.h>

#define SIZES 6
#define KB 1024

//1KB, 10KB, 100KB, 1MB, 10MB, 100MB
int PACK_SIZES[SIZES] = {1*KB, 10*KB, 100*KB, 1*KB*KB, 10*KB*KB, 100*KB*KB};
//int PACK_TESTS[SIZES] = {1000,1000,100,100,10,5};
int PACK_TESTS[SIZES] = {6,5,4,3,2,1};

void fill(char * bufptr, size_t size);

int main() {
    int tests_amount;
    size_t size;
    
    //pipefd[0] -> lectura
    //pipefd[1] -> escritura
    int pipefd[2];

    int r;  //Variable de retorno.
    pid_t pid;

    r = pipe(pipefd);
    if (r < 0){
        perror("Error al crear la tubería");
        exit(-1);
    }

    printf("Medidas de envío de paquetes con comunicación mediante Tuberías.\n\n");

    pid = fork();
    if (pid == (pid_t) -1){
        perror("Error en fork");
        exit(-1);
    }

    for (int i = 0; i < SIZES; i++){
        int packages_amount;
        tests_amount = PACK_TESTS[i];
        size = PACK_SIZES[i];

        float time0, time1;
        float total_time = 0;

        char *write_buffer[size];
        char *read_buffer[size];
        fill(write_buffer, size);

        for (int j = 1; j <= tests_amount; j++){
            time0 = clock();

            if (pid == 0){
                //Consumidor

                close(pipefd[1]);
                read(pipefd[0], read_buffer, size);
                close(pipefd[0]);
                //printf("\n Mensaje recibido de tamaño %d: %d \n", size, strlen(read_buffer));

            }else {
                //Productor
                close(pipefd[0]);
                write(pipefd[1], write_buffer, size);
                close(pipefd[1]);
            }
            time1 = clock();
            double tiempo = (double)(time1) / CLOCKS_PER_SEC-(double)(time0) / CLOCKS_PER_SEC;
            total_time = total_time + tiempo;
        }

        printf("El promedio con %d Bytes y %d pruebas es: %f seg. \n\n",size,tests_amount,total_time/tests_amount);
        total_time = 0;

    }
    return 0;
}

void fill(char * bufptr, size_t size){
    static char ch = 'A';
    int filled_count;

    //printf("size is %d\n", size);
    if (size != 0){
        memset(bufptr, ch, size);
    } else {
        memset(bufptr, 'q', size);
    }
    
    bufptr[size-1] = '[=10=]';
    
    if (ch > 90)
        ch = 65;
    
    filled_count = strlen(bufptr);

    //printf("Bytes escritos: %d\n\n", filled_count);
    //printf("buffer filled is:%s\n", bufptr);
    ch++;
}

我收到类似以下错误的信息（我知道这是竞争条件，但我不知道如何修复它:c）：

Medidas de envío de paquetes con comunicación mediante Tuberías.

El promedio con 1024 Bytes y 6 pruebas es: 0.000002 seg. 


 Mensaje recibido de tamaño 1024: 1023 

 Mensaje recibido de tamaño 1024: 1023 

 Mensaje recibido de tamaño 1024: 1023 

 Mensaje recibido de tamaño 1024: 1023 

 Mensaje recibido de tamaño 1024: 1023 

 Mensaje recibido de tamaño 1024: 1023 
El promedio con 10240 Bytes y 5 pruebas es: 0.000001 seg. 

El promedio con 1024 Bytes y 6 pruebas es: 0.000012 seg. 


 Mensaje recibido de tamaño 10240: 0 

 Mensaje recibido de tamaño 10240: 0 

 Mensaje recibido de tamaño 10240: 0 

 Mensaje recibido de tamaño 10240: 0 

 Mensaje recibido de tamaño 10240: 0 
El promedio con 10240 Bytes y 5 pruebas es: 0.000005 seg. 

El promedio con 102400 Bytes y 4 pruebas es: 0.000001 seg. 


 Mensaje recibido de tamaño 102400: 0 

 Mensaje recibido de tamaño 102400: 0 

 Mensaje recibido de tamaño 102400: 0 

 Mensaje recibido de tamaño 102400: 0 
El promedio con 102400 Bytes y 4 pruebas es: 0.000008 seg. 

Segmentation fault (core dumped)

Answer 1

两个三个问题

char *write_buffer[size];
char *read_buffer[size];

首先，这些是variable-length arrays allocated with automatic storage duration。这也称为 “分配在堆栈上”。大多数操作系统对程序堆栈的大小都有不同的限制。在现代 Linux 中，这通常默认为 8MB（ulimit -s 进行检查）。

使用给定的大小，您可以轻松地在大多数机器上 overflow the stack。

其次，这些是pointer-to-char的数组，意思是space乘以sizeof (char *)，这些是不正确的存储 null-terminated 个字符串的类型。

调高编译器的警告级别应该会提醒您注意第二个问题，警告如

main.c:55:14: warning: passing argument 1 of ‘fill’ from incompatible pointer type [-Wincompatible-pointer-types]
   55 |         fill(write_buffer, size);
      |              ^~~~~~~~~~~~
      |              |
      |              char **
main.c:16:17: note: expected ‘char *’ but argument is of type ‘char **’
   16 | void fill(char *bufptr, size_t size);
      |           ~~~~~~^~~~~~

strlen.

也类似

第三，两个缓冲区都存在于两个进程中，其中每个进程只使用一个。

虽然这是一个玩具程序，但使用这么大的缓冲区是值得怀疑的。您正在处理非常简单的 fixed-length 消息。编写器正在写入相同的字节 size - 1 次，然后是零字节。 reader 只能消耗 size 字节。

下一个问题是您在每个进程的内部循环的每次迭代中重复关闭管道的两端。

读取进程应该在外循环之前关闭管道的写入端，仅一次。写入过程应该在外循环之前关闭管道的读取端，仅一次。

循环后，两个进程都应关闭管道的一侧。

您没有检查 read 或 write 的 return 值来确认没有发生错误，或者未完成预期处理的数据量.

fill 是有缺陷的，尽管它的使用方式并不如此。

if (size != 0){
    memset(bufptr, ch, size);
} else {
    memset(bufptr, 'q', size);
}

bufptr[size-1] = '[=12=]';

如果 size 为零：

memset(bufptr, 'q', size); 实际上是一个 NOP，没有效果。
bufptr[size-1] = '[=24=]';会造成无符号整数溢出，会索引SIZE_MAX.
数组中的偏移量

请注意，if (ch > 90) 仅在 使用了 ch 之后才会被选中。对于 ASCII，第 27 次使用此函数将生成一个包含 [.

的字符串

filled_count 毫无意义。字符串长度将为 size - 1.

这里是一个粗略的示例程序，供您试用。

如果您有兴趣分析一次处理单个字节与处理更大缓冲区之间的区别，或者想实际保留您处理的数据，并想重新引入字符串缓冲区，请知道 char foo[size]; 实际上只适用于相对较小的 size 值。您可以尝试使用 dynamic memory。

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

#define SIZES 6
#define KB 1024

/* 1KB, 10KB, 100KB, 1MB, 10MB, 100MB */
size_t PACK_SIZES[SIZES] = {1*KB, 10*KB, 100*KB, 1*KB*KB, 10*KB*KB, 100*KB*KB};
int PACK_TESTS[SIZES] = {6,5,4,3,2,1};

static void pdie(const char *msg) {
    perror(msg);
    exit(EXIT_FAILURE);
}

static char get_next_byte(void) {
    static char byte = 'A';

    if (byte > 'Z')
        byte = 'A';

    return byte++;
}

static int read_exactly(int fd, size_t size) {
    unsigned char b;

    while (size > 0 && 1 == read(fd, &b, 1))
        size--;

    return 0 == size;
}

static int write_exact_string(int fd, size_t size, char b) {
    unsigned char zero = 0;

    while (size > 1 && 1 == write(fd, &b, 1))
        size--;

    return (1 == size) && (1 == write(fd, &zero, 1));
}

int main(void) {
    pid_t pid;
    int pipefd[2];

    if (-1 == pipe(pipefd))
        pdie("pipe");

    if (-1 == (pid = fork()))
        pdie("fork");

    if (0 == pid)
        close(pipefd[1]);
    else
        close(pipefd[0]);

    for (int i = 1; i <= SIZES; i++) {
        size_t expected_size = PACK_SIZES[i];
        int tests_amount = PACK_TESTS[i];

        double total_time = 0;

        for (int j = 1; j <= tests_amount; j++) {
            clock_t start = clock();

            if (pid == 0) {
                if (!read_exactly(pipefd[0], expected_size)) {
                    fprintf(stderr, "Failed to read %zu bytes.\n", expected_size);
                    exit(EXIT_FAILURE);
                }
            } else {
                char byte = get_next_byte();

                if (!write_exact_string(pipefd[1], expected_size, byte)) {
                    fprintf(stderr, "Failed to write an exact string of [%zu*'%c'+'\0'].\n", expected_size - 1, byte);
                    exit(EXIT_FAILURE);
                }
            }

            double duration = ((double) (clock() - start)) / CLOCKS_PER_SEC;

            printf("Test<%d-%d>: %s %zu bytes in %f.\n", i, j, (pid == 0) ? "Read" : "Wrote", expected_size, duration);
            total_time += duration;
        }

        printf("Time taken in <%s> to process %d * %zu bytes: %f (%f avg.)\n",
                (pid == 0) ? "CHILD" : "PARENT",
                tests_amount,
                expected_size,
                total_time,
                total_time / tests_amount);
    }

    if (0 == pid) {
        close(pipefd[0]);
    } else {
        close(pipefd[1]);

        if (-1 == waitpid(pid, NULL, WUNTRACED))
            pdie("waitpid");
    }
}

在 C 中父子之间使用 Pipes IPC 进行多次测试

Multiple tests with Pipes IPC between parent and child in C

c

linux

pipe

ipc