c - 从c中的管道读取

Question

场景:假设我有 8 个文件，我想按从小到大的顺序对其中的所有数字数据进行排序。只有叶进程可以对文件包含的所有数字进行排序。这些叶进程必须通过管道将排序后的数据发送到父进程。该父进程将比较接收到的数据并将较小的数据发送给下一个进程。它将执行此操作，直到管道中的所有数字都为空。

所以把它想象成一棵树。我们有一个主流程。有 8 个文件需要排序，主进程将从中产生 2 个进程(左进程和右进程)。这两个新进程将产生它们自己的进程。这会一直发生，直到底部有 8 个叶进程为止。内部节点只能保留一个数字。它们将沿着一系列管道传递它们的编号，直到到达主进程。主进程会将其管道内容输出到文件中。

我已添加代码 here (因为它有点长但很简单)。

如果我有 2 个文件要排序，则此方法有效。所以我们有 1 个主进程，然后是两个子进程。两个 child 将他们的文件编号排序，然后将它们传递出去。然后主进程从管道中按顺序打印出数据。但是，如果我添加一些复杂性(4个文件)，叶进程仍然向上发送数据，但是当主进程开始从内部节点管道读取数据时，它认为它是空的并在没有任何数据的情况下完成程序。

知道为什么主进程认为它的左右管道是空的吗？就像我说的，当有一名 parent 和两个 child 时效果很好。继续处理就会失败。 (假设处理将以 2 的幂次发生)。

注意:perror 用于调试目的。

完整节目here [非常困惑，因为我已经用它做了很多事情，但它会编译。

Answer 1

Pastebin 中更新的代码不是可编译的函数 - 更不用说完整的程序了。这使得很难说出到底出了什么问题。

但是，问题之一出在代码片段中:

if (pipe(upPipe) < 0 || pipe(leftPipe) < 0 || pipe(rightPipe) < 0)
    ...error exit...
if ((leftPID = fork()) < 0)
    ...error exit...

if(leftPID == 0){
    fMax = ((fMax)/2);
    dup2(leftPipe[WRITE], upPipe[WRITE]);
    pipe(leftPipe);
    pipe(rightPipe);

对dup2()的调用很奇怪；您仔细地将左管道的写入 channel 映射到上管道的写入 channel 。

dup2() 之后的两个 pipe() 调用相当迅速地搞砸了左子级中的所有内容，打开了另外 4 个文件描述符，但丢失了之前存储在leftPipe 和 rightPipe。

您需要使问题陈述更清楚。我无法从你拥有的东西中理解你应该拥有的东西。有一个对 convertToInt() 的调用，它不接受任何参数，也不返回任何值；那到底在做什么？有一个对 freeMem() 的调用；目前还不清楚它的作用。

z.c:42: error: ‘numberChar’ undeclared (first use in this function)
z.c:42: error: ‘sizeNumbers’ undeclared (first use in this function)
z.c:43: warning: implicit declaration of function ‘readFile’
z.c:43: error: ‘fileNames’ undeclared (first use in this function)
z.c:45: warning: implicit declaration of function ‘convertToInt’
z.c:46: error: ‘i’ undeclared (first use in this function)
z.c:46: error: ‘numbs’ undeclared (first use in this function)
z.c:47: error: ‘numbers’ undeclared (first use in this function)
z.c:48: warning: implicit declaration of function ‘freeMem’

抱歉，您的问题无法回答，因为您没有向我们提供:

1. 准确的要求。
2. 您实际编译的代码。

<小时/>

您的代码没有很好地分解函数。您使用 VCS(版本控制系统 - 例如 git)吗？如果没有，你应该这样做。我在 9 个 checkin 中制作了下面的更改版本(本质上是完全重写)，并且可能应该制作比这更小的 checkin 。但使用 VCS 对我来说至关重要；它让我充满信心地做出改变，因为我知道我不会失去任何有值(value)的东西。而且我不必注释掉代码；我删除了我不想要的东西。下面的解决方案是261行；原版总共约687行，其中包括大量注释掉的代码；当我完成删除注释等后，它只剩下 469 行。

当我运行你的代码(并报告每个 child 打开了哪些文件)时，我发现有 2 个进程分别打开文件 2 和 3(并且因为数据文件当时不存在) ，他们在那一点上失败了)。

修改后的代码结构几乎干净；奇数位是“convertToString()”阶段，它从管道读取二进制整数并将它们再次转换为 ASCII 输出。有用;我不相信它很优雅。它不使用硬编码文件名数组，而是从命令行获取任意文件名列表(不一定是 8；它已经使用 0 到 8 进行了测试，我没有理由认为它无法处理 20 个或更多)。我做了相当多的测试:

./piped-merge-sort [1-8]

有大量的诊断输出。我使用了两个对我的工作非常有帮助的函数 - 我将它们与其他一些相关代码打包在一个更复杂的包中，但是 err_error() 和 的简单版本err_remark() 函数确实对我很有帮助。请注意，这些版本报告每个调用的报告进程的 PID。他们还小心地将消息预先格式化为字符串，然后将该字符串写入到标准错误中；否则，我会得到很多交错的输出，这充其量是令人困惑的。

“没什么说的 - 这是代码:

#include <assert.h>
#include <errno.h>
#include <stdarg.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/wait.h>
#include <unistd.h>

/* Error reporting */
static void err_vremark(char *fmt, va_list args)
{
    char buffer[256];
    int errnum = errno;
    int buflen = snprintf(buffer, sizeof(buffer), "%d: ", (int)getpid());
    buflen += vsnprintf(buffer + buflen, sizeof(buffer) - buflen, fmt, args);
    if (errnum != 0)
        buflen += snprintf(buffer + buflen, sizeof(buffer) - buflen,
                           ": errno = %d (%s)", errnum, strerror(errnum));
    fprintf(stderr, "%s\n", buffer);
}

static void err_error(char *fmt, ...)
{
    va_list args;
    va_start(args, fmt);
    err_vremark(fmt, args);
    va_end(args);
    exit(1);
}

static void err_remark(char *fmt, ...)
{
    va_list args;
    va_start(args, fmt);
    err_vremark(fmt, args);
    va_end(args);
}

enum { READ = 0 };
enum { WRITE = 1 };
enum { BUFFER_SIZE = 256 };

static int *a_data = 0;
static int  a_used = 0;
static int  a_size = 0;

void readFile(char const *fileName);
void freeMem(void);
void sortArray(void);
int  intcmp(void const *n1, void const *n2);

static void sortMergeFiles(int fd, int number, char **names);
static void sortOneFile(int fd, const char *file);
static void convertToString(int fd, FILE *fp);

int main(int argc, char **argv)
{
    int m_pipe[2];
    pid_t pid;
    if (pipe(m_pipe) < 0)
        err_error("Failed to create master pipe");
    if ((pid = fork()) < 0)
        err_error("Failed to fork master");
    else if (pid == 0)
    {
        close(m_pipe[READ]);
        sortMergeFiles(m_pipe[WRITE], argc - 1, &argv[1]);
        close(m_pipe[WRITE]);
    }
    else
    {
        close(m_pipe[WRITE]);
        convertToString(m_pipe[READ], stdout);
        close(m_pipe[READ]);
    }
    return 0;
}

static void convertToString(int fd, FILE *fp)
{
    int value;
    while (read(fd, &value, sizeof(int)) == sizeof(int))
        fprintf(fp, "%d\n", value);
}

static int readInteger(int fd, int *value)
{
    if (read(fd, value, sizeof(int)) != sizeof(int))
        return EOF;
    return 0;
}

static void writeInteger(int fd, int value)
{
    if (write(fd, &value, sizeof(int)) != sizeof(int))
        err_error("Failed to write integer to fd %d", fd);
}

static void mergeFiles(int fd_in1, int fd_in2, int fd_out)
{
    int value_1;
    int value_2;
    int eof_1 = readInteger(fd_in1, &value_1);
    int eof_2 = readInteger(fd_in2, &value_2);

    while (eof_1 != EOF && eof_2 != EOF)
    {
        err_remark("v1: %d; v2: %d", value_1, value_2);
        if (value_1 <= value_2)
        {
            writeInteger(fd_out, value_1);
            eof_1 = readInteger(fd_in1, &value_1);
        }
        else
        {
            writeInteger(fd_out, value_2);
            eof_2 = readInteger(fd_in2, &value_2);
        }
    }

    while (eof_1 != EOF)
    {
        err_remark("v1: %d", value_1);
        writeInteger(fd_out, value_1);
        eof_1 = readInteger(fd_in1, &value_1);
    }

    while (eof_2 != EOF)
    {
        err_remark("v2: %d", value_2);
        writeInteger(fd_out, value_2);
        eof_2 = readInteger(fd_in2, &value_2);
    }
}

static void sortMergeFiles(int fd, int number, char **names)
{
    assert(number >= 0);
    if (number == 0)
        return;
    else if (number == 1)
        sortOneFile(fd, names[0]);
    else
    {
        err_remark("Non-Leaf: processing %d files (%s .. %s)", number, names[0], names[number-1]);
        int mid = number / 2;
        int l_pipe[2];
        int r_pipe[2];
        pid_t l_pid;
        pid_t r_pid;
        if (pipe(l_pipe) < 0 || pipe(r_pipe) < 0)
            err_error("Failed to create pipes");
        if ((l_pid = fork()) < 0)
            err_error("Failed to fork left child");
        else if (l_pid == 0)
        {
            close(l_pipe[READ]);
            close(r_pipe[READ]);
            close(r_pipe[WRITE]);
            sortMergeFiles(l_pipe[WRITE], mid, names);
            close(l_pipe[WRITE]);
            exit(0);
        }
        else if ((r_pid = fork()) < 0)
            err_error("Failed to fork right child");
        else if (r_pid == 0)
        {
            close(r_pipe[READ]);
            close(l_pipe[READ]);
            close(l_pipe[WRITE]);
            sortMergeFiles(r_pipe[WRITE], number - mid, names + mid);
            close(r_pipe[WRITE]);
            exit(0);
        }
        else
        {
            close(l_pipe[WRITE]);
            close(r_pipe[WRITE]);
            mergeFiles(l_pipe[READ], r_pipe[READ], fd);
            close(l_pipe[READ]);
            close(r_pipe[READ]);
            err_remark("Non-Leaf: finished %d files (%s .. %s)", number, names[0], names[number-1]);
        }
    }
}

static void addNumberToArray(int number)
{
    assert(a_used >= 0 && a_used <= a_size);
    if (a_used == a_size)
    {
        int  n_size = (a_size + 1) * 2;
        int *n_data = realloc(a_data, sizeof(*n_data) * n_size);
        if (n_data == 0)
            err_error("Failed to allocate space for %d numbers", n_size);
        a_data = n_data;
        a_size = n_size;
    }
    a_data[a_used++] = number;
}

/* Could be compressed to write(fd, a_data, a_used * sizeof(int)); */
/* Arguably should check for write errors - but not SIGPIPE */
static void writeArray(int fd)
{
    for (int i = 0; i < a_used; i++)
    {
        err_remark("Write: %d", a_data[i]);
        write(fd, &a_data[i], sizeof(int));
    }
}

void readFile(char const *fileName)
{
    char buffer[BUFFER_SIZE];
    FILE *fp;
    fp = fopen(fileName, "r");

    if (fp == NULL)
        err_error("Failed to open file %s for reading", fileName);

    while (fgets(buffer, sizeof(buffer), fp) != NULL)
    {
        char *nl = strchr(buffer, '\n');
        if (nl != 0)
            *nl = '\0';
        err_remark("Line: %s", buffer);
        addNumberToArray(atoi(buffer));
    }

    fclose(fp);
}

int intcmp(const void *n1, const void *n2)
{
    const int num1 = *(const int *) n1;
    const int num2 = *(const int *) n2;
    return (num1 < num2) ? -1 : (num1 > num2);
}

void sortArray(void)
{
    qsort(a_data, a_used, sizeof(int), intcmp);
}

void freeMem(void)
{
    free(a_data);
}

static void sortOneFile(int fd, const char *file)
{
    err_remark("Leaf: processing file %s", file);
    readFile(file);
    sortArray();
    writeArray(fd);
    freeMem();
    err_remark("Leaf: finished file %s", file);
}