Pipestream and child processes

Question

I need to write my pipestream. My program should get names of another programs and call them, first program should read from stdin second from output of the first an so on. Last program print the result in stdout. The problem is it doesn't work. When I call this program to two another simple programs (they both are to read x and print 2*x while it can read), and give it some data I suppose that I should get result immediately, but I don't. What is more when I give it end of file it doesn't react. Don't pay attention on safe_* functions, they are the same as standart, but the check for an errors. Help me, please =)

#include <unistd.h>
#include <sys/types.h>
#include <sys/wait.h>
#include <errno.h>
#include <string.h>
#include <stdio.h>
#include <stdlib.h>
#include "error.h"
#include "safe_functions.h"

void call(const char *filename, int in_descr, int out_descr, pid_t *sons, int n)
{
    sons[n] = safe_fork();
    if(sons[n] == 0)
    {
        safe_dup2(in_descr, STDIN_FILENO);
        safe_dup2(out_descr, STDOUT_FILENO);

        safe_execv(filename, (char **)NULL);
    }
}

int find_num(pid_t * sons, int n, pid_t id)
{
    for(int i=0; i<n; i++)
        if(sons[i] == id)
            return i;

    return -1;
}

int main(int argc, const char ** argv)
{

    int **descr;
    descr = (int**)safe_malloc(argc*sizeof(int*));
    for(int i=0; i<argc; i++)
        descr[i] = (int*) safe_malloc(2*sizeof(int));

    for(int i=1; i+1<argc; i++)
        safe_pipe(descr[i]);

    descr[0][0] = 0;
    descr[argc-1][1] = 1;

    pid_t *sons = safe_malloc((argc-1) * sizeof(pid_t));

    for(int i=1; i<argc; i++)
        call(argv[i], descr[i-1][0], descr[i][1], sons, i-1);

    int status;
    pid_t id;
    while(1)
    {
        id = safe_wait(&status);
        if(id == -1)
            break;

        if(WIFEXITED(status))
        {
            int num = find_num(sons, argc-1, id);
            safe_close(descr[num][0]);
            safe_close(descr[num+1][1]);
            continue;
        }

        if(WIFSIGNALED(status))
        {
            int num = find_num(sons, argc-1, id);
            fatal_error("Process was teminated by a signal", argv[num+1], WEXITSTATUS(status));
        }
    }

    free(sons);
    for(int i=0; i<argc; i++)
        free(descr[i]);
    free(descr);
}

Answer 1

You have nowhere near enough calls to close()! The parent must close all its copies of the pipes. After duplicating the relevant pipe descriptors to standard input and output, the children must close every pipe descriptor too. Otherwise, the processes never get EOF because there's a process that could write to the pipes.

An SSCCE (Short, Self-Contained, Correct Example)

This code uses my stderr.h and stderr.c code in place of your error.h (contact me if you want the code — see my profile). It writes out the safe_* functions — or, at least, my implementation of them.

The dump_fds() function reports on which file descriptors are open in the range 0-19, which is sufficient for this and most programs; a more complex version works with sysconf() to determine the number of file descriptors to check, but the number is usually vastly greater (eg 256 or larger) than the number in use. I use it as a simple way to check that all the file descriptors that should be closed are closed.

There is considerable debug output. The call to execv() provides a proper argument list.

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

//#include "error.h"
//#include "safe_functions.h"

#include "stderr.h"

extern pid_t safe_fork(void);
extern void  safe_dup2(int old_fd, int new_fd);
extern void  safe_execv(const char *prog, char **argv);
extern void  safe_pipe(int *pipe_fds);
extern void *safe_malloc(size_t size);
extern int   safe_wait(int *status);
extern void  safe_close(int fd);

/* Report on open file descriptors (0..19) in process */
static void dump_fds(void)
{
    struct stat b;
    char buffer[32];
    sprintf(buffer, "%d: ", getpid());
    char *str = buffer + strlen(buffer);
    for (int i = 0; i < 20; i++)
        *str++ = (fstat(i, &b) == 0) ? 'o' : '-';
    *str++ = '\n';
    *str = '\0';
    fputs(buffer, stderr);
}

static void close_pipes(int **descr, int argc)
{
    for (int i = 0; i < argc; i++)
    {
        for (int j = 0; j < 2; j++)
        {
            if (descr[i][j] > 1)
            {
                err_remark("close %d\n", descr[i][j]);
                safe_close(descr[i][j]);
            }
        }
    }
}

static void call(char *filename, int in_descr, int out_descr, pid_t *sons, int n, int **descr, int argc)
{
    sons[n] = safe_fork();
    if (sons[n] == 0)
    {
        err_remark("call: %s\n", filename);
        char *argv[2] = { filename, NULL };
        err_remark("dup2(%d, %d)\n", in_descr, STDIN_FILENO);
        err_remark("dup2(%d, %d)\n", out_descr, STDOUT_FILENO);
        safe_dup2(in_descr, STDIN_FILENO);
        safe_dup2(out_descr, STDOUT_FILENO);
        close_pipes(descr, argc);
        dump_fds();
        safe_execv(argv[0], argv);
    }
}

static int find_num(pid_t *sons, int n, pid_t id)
{
    for (int i=0; i<n; i++)
    {
        if (sons[i] == id)
            return i;
    }
    return -1;
}

int main(int argc, char **argv)
{
    err_setarg0(argv[0]);
    err_setlogopts(ERR_PID);
    dump_fds();

    int **descr;
    descr = (int**)safe_malloc(argc*sizeof(int*));

    for (int i=0; i<argc; i++)
        descr[i] = (int*) safe_malloc(2*sizeof(int));

    for (int i=1; i+1<argc; i++)
        safe_pipe(descr[i]);

    descr[0][0] = 0;
    descr[argc-1][1] = 1;

    pid_t *sons = safe_malloc((argc-1) * sizeof(pid_t));

    for (int i=1; i<argc; i++)
    {
        err_remark("Command: %s\n", argv[i]);
        call(argv[i], descr[i-1][0], descr[i][1], sons, i-1, descr, argc);
    }

    close_pipes(descr, argc);

    while (1)
    {
        int status;
        pid_t id = safe_wait(&status);
        err_remark("wait: pid %d, status 0x%.4X\n", (int)id, status);

        if (id == -1)
            break;

        if (WIFEXITED(status))
        {
            int num = find_num(sons, argc-1, id);
            //safe_close(descr[num][0]);
            //safe_close(descr[num+1][1]);
            continue;
        }

        if (WIFSIGNALED(status))
        {
            int num = find_num(sons, argc-1, id);
            err_remark("Process %s was terminated by signal %d", argv[num+1], WEXITSTATUS(status));
        }
    }

    free(sons);
    for (int i=0; i<argc; i++)
        free(descr[i]);
    free(descr);

    return(0);
}


extern pid_t safe_fork(void)
{
    pid_t pid = fork();
    if (pid < 0)
        err_syserr("Failed to fork() ");
    return pid;
}

extern void safe_dup2(int old_fd, int new_fd)
{
    if (dup2(old_fd, new_fd) < 0)
        err_syserr("Failed to dup2(%d, %d) ", old_fd, new_fd);
}

extern void safe_execv(const char *prog, char **argv)
{
    execv(prog, argv);
    err_syserr("Failed to execv(\"%s\") ", prog);
}

extern void safe_pipe(int *pipe_fds)
{
    assert(pipe_fds != 0);
    if (pipe(pipe_fds) != 0)
        err_syserr("Failed to pipe() ");
    err_remark("pipe: %d, %d\n", pipe_fds[0], pipe_fds[1]);
}

extern void *safe_malloc(size_t size)
{
    void *vp = malloc(size);
    if (vp == 0)
        err_syserr("Out of memory ");
    return vp;
}

extern int safe_wait(int *status)
{
    assert(status != 0);
    return wait(status);
}

extern void safe_close(int fd)
{
    if (close(fd) < 0)
        err_syserr("Failed to close(%d)\n", fd);
}

Example Output

$ ./pipes-15845060 /bin/ps /usr/bin/sort /bin/cat
12096: ooo-----------------
pipes-15845060: pid=12096: pipe: 3, 4
pipes-15845060: pid=12096: pipe: 5, 6
pipes-15845060: pid=12096: Command: /bin/ps
pipes-15845060: pid=12096: Command: /usr/bin/sort
pipes-15845060: pid=12096: Command: /bin/cat
pipes-15845060: pid=12096: close 3
pipes-15845060: pipes-15845060: pid=12098: pid=12096: close 4
pipes-15845060: pid=12096: close 5
pipes-15845060: pid=12096: close 6
call: /bin/ps
pipes-15845060: pid=12098: dup2(0, 0)
pipes-15845060: pid=12098: dup2(4, 1)
pipes-15845060: pid=12099: call: /usr/bin/sort
pipes-15845060: pid=12099: dup2(3, 0)
pipes-15845060: pid=12099: dup2(6, 1)
pipes-15845060: pid=12098: pipes-15845060: pid=12099: close 3
pipes-15845060: pid=12099: close 4
pipes-15845060: pid=12099: close 5
pipes-15845060: pid=12099: close 6
12099: ooo-----------------
close 3
pipes-15845060: pid=12098: close 4
pipes-15845060: pid=12098: close 5
pipes-15845060: pid=12098: close 6
12098: ooo-----------------
pipes-15845060: pid=12100: call: /bin/cat
pipes-15845060: pid=12100: dup2(5, 0)
pipes-15845060: pid=12100: dup2(1, 1)
pipes-15845060: pid=12100: close 3
pipes-15845060: pid=12100: close 4
pipes-15845060: pid=12100: close 5
pipes-15845060: pid=12100: close 6
12100: ooo-----------------
pipes-15845060: pid=12096: wait: pid 12098, status 0x0000
  563 ttys000    0:00.03 -sh
  568 ttys001    0:00.03 -sh
  578 ttys003    0:00.03 -sh
  587 ttys002    0:00.03 -sh
  588 ttys005    0:00.15 -sh
  589 ttys004    0:00.20 -sh
  PID TTY           TIME CMD
12096 ttys004    0:00.00 ./pipes-15845060 /bin/ps /usr/bin/sort /bin/cat
12097 ttys004    0:00.00 sed /./s/^/    /
12099 ttys004    0:00.00 /usr/bin/sort
12100 ttys004    0:00.00 /bin/cat
pipes-15845060: pid=12096: wait: pid 12100, status 0x0000
pipes-15845060: pid=12096: wait: pid 12099, status 0x0000
pipes-15845060: pid=12096: wait: pid -1, status 0x0000
$