Thread: Question-please help me

I can't seem to make a pipe circle work.
In the following, process 1's stdout is connected to proc 2's stdin, whose stdout is connected to proc 3's stdin, whose stdout is connected to proc 1's stdin.
But the thing just freezes.
Is something like this possible?
If not, why not?

Code:

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

int main() {
    char line[100];
    int p[3][2];
    pipe(p[0]); pipe(p[1]); pipe(p[2]);

    // proc 1
    if (fork() == 0) {
        dup2(p[0][0], 0); dup2(p[1][1], 1);
        close(p[0][1]); close(p[1][0]);
        close(p[2][0]); close(p[2][1]);

        printf("hello\n");                // write to proc2
        fgets(line, sizeof line, stdin);  // read from proc3
        fprintf(stderr, "1: %s\n", line);
        return 0;
    }

    // proc 2
    if (fork() == 0) {
        dup2(p[1][0], 0); dup2(p[2][1], 1);
        close(p[1][1]); close(p[2][0]);
        close(p[0][0]); close(p[0][1]);

        fgets(line, sizeof line, stdin);  // read from proc1
        printf("%s", line);               // write to proc3
        fprintf(stderr, "2: %s\n", line);
        return 0;
    }

    // proc 3
    if (fork() == 0) {
        dup2(p[2][0], 0); dup2(p[0][1], 1);
        close(p[2][1]); close(p[0][0]);
        close(p[1][0]); close(p[1][1]);

        fgets(line, sizeof line, stdin);  // read from proc2
        printf("%s", line);               // write to proc1
        fprintf(stderr, "3: %s\n", line);
        return 0;
    }

    close(p[0][0]); close(p[0][1]);
    close(p[1][0]); close(p[1][1]);
    close(p[2][0]); close(p[2][1]);
    wait(NULL); wait(NULL); wait(NULL);

    return 0;
}

It's very odd - but if I replace the first printf with a write, then it all works for me.

Code:

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

#define PIPE_RD    0
#define PIPE_WR    1

int main() {
    char    line[100];
    int     p[3][2];
    pid_t   pids[3];
    pipe(p[0]); pipe(p[1]); pipe(p[2]);

    // proc 1
    if ( (pids[0]=fork()) == 0) {
        int in = 0, out = 1, spare = 2;
        dup2(p[in][PIPE_RD], 0); dup2(p[out][PIPE_WR], 1);
        close(p[in][PIPE_WR]); close(p[out][PIPE_RD]);
        close(p[spare][PIPE_RD]); close(p[spare][PIPE_WR]);

        //printf("hello\n");                // write to proc2
        write(1,"world\n",6);
        fgets(line, sizeof line, stdin);  // read from proc3
        fprintf(stderr, "1: %s\n", line);
        return 0;
    }

    // proc 2
    if ( (pids[1]=fork()) == 0) {
        int in = 1, out = 2, spare = 0;
        dup2(p[in][PIPE_RD], 0); dup2(p[out][PIPE_WR], 1);
        close(p[in][PIPE_WR]); close(p[out][PIPE_RD]);
        close(p[spare][PIPE_RD]); close(p[spare][PIPE_WR]);

        fgets(line, sizeof line, stdin);  // read from proc1
        printf("AA:%s", line);               // write to proc3
        fprintf(stderr, "2: %s\n", line);
        return 0;
    }

    // proc 3
    if ( (pids[2]=fork()) == 0) {
        int in = 2, out = 0, spare = 1;
        dup2(p[in][PIPE_RD], 0); dup2(p[out][PIPE_WR], 1);
        close(p[in][PIPE_WR]); close(p[out][PIPE_RD]);
        close(p[spare][PIPE_RD]); close(p[spare][PIPE_WR]);

        fgets(line, sizeof line, stdin);  // read from proc2
        printf("BB:%s", line);               // write to proc1
        fprintf(stderr, "3: %s\n", line);
        return 0;
    }

    close(p[0][PIPE_RD]); close(p[0][PIPE_WR]);
    close(p[1][PIPE_RD]); close(p[1][PIPE_WR]);
    close(p[2][PIPE_RD]); close(p[2][PIPE_WR]);

    int status;
    pid_t q;
    printf("Waiting for children %d %d %d\n", pids[0],pids[1],pids[2]);
    q = waitpid(-1,&status,0);
    printf("Child %d exited with %d\n", q, WEXITSTATUS(status));
    q = waitpid(-1,&status,0);
    printf("Child %d exited with %d\n", q, WEXITSTATUS(status));
    q = waitpid(-1,&status,0);
    printf("Child %d exited with %d\n", q, WEXITSTATUS(status));
    return 0;
}

$ gcc -Wall -g foo.c
$ ./a.out
Waiting for children 4514 4515 4516
2: world

Child 4515 exited with 0
3: AA:world

Child 4516 exited with 0
1: BB:AA:world

Child 4514 exited with 0

It's like the printf doesn't happen?
I mean, even if I run "strace -f ./a.out", it doesn't show up.

Code:

pipe([3, 4])                            = 0
pipe([5, 6])                            = 0
pipe([7, 8])                            = 0
clone(strace: Process 4279 attached
child_stack=0, flags=CLONE_CHILD_CLEARTID|CLONE_CHILD_SETTID|SIGCHLD, child_tidptr=0x7ff17fd1b9d0) = 4279
[pid  4279] dup2(3, 0)                  = 0
[pid  4279] dup2(6, 1)                  = 1
[pid  4279] close(4)                    = 0
[pid  4279] close(5)                    = 0
[pid  4279] close(7)                    = 0
[pid  4279] close(8)                    = 0
[pid  4279] fstat(1, {st_mode=S_IFIFO|0600, st_size=0, ...}) = 0
[pid  4278] clone( <unfinished ...>
[pid  4279] brk(NULL)                   = 0x1a5a000
[pid  4279] brk(0x1a7c000)              = 0x1a7c000
[pid  4279] fstat(0, strace: Process 4280 attached
 <unfinished ...>
[pid  4278] <... clone resumed> child_stack=0, flags=CLONE_CHILD_CLEARTID|CLONE_CHILD_SETTID|SIGCHLD, child_tidptr=0x7ff17fd1b9d0) = 4280
[pid  4279] <... fstat resumed> {st_mode=S_IFIFO|0600, st_size=0, ...}) = 0
[pid  4279] read(0,  <unfinished ...>

I would have expected the printf() to call the underlying write() system call, but it's not there.
So the whole ring locks up waiting on read() calls.


But with write(), it's all as expected!

Code:

pipe([3, 4])                            = 0
pipe([5, 6])                            = 0
pipe([7, 8])                            = 0
clone(strace: Process 4557 attached
child_stack=0, flags=CLONE_CHILD_CLEARTID|CLONE_CHILD_SETTID|SIGCHLD, child_tidptr=0x7fd3909779d0) = 4557
[pid  4557] dup2(3, 0 <unfinished ...>
[pid  4556] clone( <unfinished ...>
[pid  4557] <... dup2 resumed> )        = 0
[pid  4557] dup2(6, 1)                  = 1
[pid  4557] close(4)                    = 0
[pid  4557] close(5)                    = 0
[pid  4557] close(7)                    = 0
[pid  4557] close(8strace: Process 4558 attached
 <unfinished ...>
[pid  4556] <... clone resumed> child_stack=0, flags=CLONE_CHILD_CLEARTID|CLONE_CHILD_SETTID|SIGCHLD, child_tidptr=0x7fd3909779d0) = 4558
[pid  4557] <... close resumed> )       = 0
[pid  4557] write(1, "world\n", 6)      = 6
[pid  4557] fstat(0,  <unfinished ...>
[pid  4558] dup2(5, 0 <unfinished ...>
[pid  4556] clone( <unfinished ...>
[pid  4557] <... fstat resumed> {st_mode=S_IFIFO|0600, st_size=0, ...}) = 0
[pid  4557] brk(NULL)                   = 0x249d000
[pid  4558] <... dup2 resumed> )        = 0
[pid  4557] brk(0x24bf000)              = 0x24bf000
[pid  4556] <... clone resumed> child_stack=0, flags=CLONE_CHILD_CLEARTID|CLONE_CHILD_SETTID|SIGCHLD, child_tidptr=0x7fd3909779d0) = 4559
strace: Process 4559 attached
[pid  4558] dup2(8, 1 <unfinished ...>
[pid  4557] read(0,  <unfinished ...>
///
[pid  4557] <... read resumed> "BB:AA:world\n", 4096) = 12

The bootstrap write() call happens, and the ring runs to completion.

That's a real mystery. Somehow that first printf isn't executing. Replacing it with write(1, "hello\n", 6); primes the pump and then the other printf's apparently work. Changing printf to fputs("hello\n", stdout) or using putchar doesn't help, either. However, changing the roles of stdout and stderr works.

I wonder what could be going on?

EDIT: As soon as I posted this it occurred to me that a fflush(stdout) would fix it ... and it does. I don't know why the other printf's don't seem to need the flush, though.

Code:

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

#define OUTFD 2    // 2 works; 1 doesn't (EDIT: but does with fflush)

#if OUTFD==1
#define OUT stdout
#define LOG stderr
#else
#define OUT stderr
#define LOG stdout
#endif

void setIO(int *pin, int *pout, int *unused) {
    dup2(pin[0], 0); dup2(pout[1], OUTFD);
    close(pin[1]); close(pout[0]);
    close(unused[0]); close(unused[1]);
}

int main() {
    char line[100];
    int p[3][2];
    pipe(p[0]); pipe(p[1]); pipe(p[2]);

    if (fork() == 0) {               // proc 1
        setIO(p[0], p[1], p[2]);
        fprintf(OUT, "hello\n");          // write to proc2
        fflush(OUT);  // EDIT: fixes the problem with stdout
        fgets(line, sizeof line, stdin);  // read from proc3
        fprintf(LOG, "1: %s\n", line);
        return 0;
    }

    if (fork() == 0) {               // proc 2
        setIO(p[1], p[2], p[0]);
        fgets(line, sizeof line, stdin);  // read from proc1
        fprintf(OUT, "%s", line);         // write to proc3
        fprintf(LOG, "2: %s\n", line);
        return 0;
    }

    if (fork() == 0) {               // proc 3
        setIO(p[2], p[0], p[1]);
        fgets(line, sizeof line, stdin);  // read from proc2
        fprintf(OUT, "%s", line);         // write to proc1
        fprintf(LOG, "3: %s\n", line);
        return 0;
    }

    close(p[0][0]); close(p[0][1]);
    close(p[1][0]); close(p[1][1]);
    close(p[2][0]); close(p[2][1]);
    wait(NULL); wait(NULL); wait(NULL);

    return 0;
}

Ah yes, because the pipes will be fully buffered by default, whereas the stdio streams start off line buffered. Until the flush, there would be no system call.

Some fcntl call might fix that.

Now that that's sorted, on to the OP's original question.
Here's one way to control the processes.
proc3 below just reports how many words it sees and, to test the STOP command, stops when it sees a list of more than 80 words.

I don't fully understand what proc3 is supposed to do in the question. It's apparently supposed to start up a bunch of threads (not processes), one thread for each pair of words that are the same length. That sounds like it could be quite a few threads.

Code:

// procmain.c  /////////////////////////////////////////////////
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <errno.h>
#include <sys/wait.h>
#include <sys/types.h>

void forkproc(const char *proc, int pin, int pout, const char *arg) {
    pid_t pid;
    if ((pid = fork()) == (pid_t)-1) {
        fprintf(stderr, "Error: fork %s: %s\n", proc, strerror(errno));
        exit(EXIT_FAILURE);
    }
    if (pid == 0) {
        dup2(pin, 0);
        dup2(pout, 1);
        execl(proc, proc, arg, (const char *)NULL);
        fprintf(stderr, "Error: exec %s: %s\n", proc, strerror(errno));
        exit(EXIT_FAILURE);
    }
}

int main(int argc, char **argv) {
    if (argc != 2) {
        fprintf(stderr, "Usage: procmain MAXREPS\n");
        exit(EXIT_FAILURE);
    }

    int p[3][2];
    if (pipe(p[0]) == -1 || pipe(p[1]) == -1 || pipe(p[2]) == -1) {
        perror("pipe");
        exit(EXIT_FAILURE);
    }

    forkproc("proc1", p[0][0], p[1][1], argv[1]);
    forkproc("proc2", p[1][0], p[2][1], NULL);
    forkproc("proc3", p[2][0], p[0][1], NULL);

    for (int i = 0; i < 3; i++) {
        close(p[i][0]); close(p[i][1]);
    }

    wait(NULL); wait(NULL); wait(NULL);
    return 0;
}


// proc1.c  /////////////////////////////////////////////////////
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <time.h>

int main(int argc, char **argv) {
    if (argc != 2)
        return EXIT_FAILURE;
    int maxreps = atoi(argv[1]);

    char line[1000];
    srand(time(NULL));

    for (int r = 0; r < maxreps; r++) {
        int n = rand() % 90 + 10;
        for (int i = 0; i < n; i++)
            printf("%d ", rand() % 5 + 5);
        putchar('\n');
        fflush(stdout);

        if (fgets(line, sizeof line, stdin) == NULL)
            break;
        // Break if told to STOP by proc3
        if (strcmp(line, "STOP\n") == 0)
            break;
    }

    // Send EXIT command to proc2
    printf("EXIT\n"); fflush(stdout);

    return 0;
}


// proc2.c  /////////////////////////////////////////////////////
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <time.h>

int main() {
    char line[1000];
    srand(time(NULL));

    while (fgets(line, sizeof line, stdin) != NULL) {
        if (strcmp(line, "EXIT\n") == 0) {
            // Pass EXIT command on to proc3
            printf("EXIT\n"); fflush(stdout);
            break;
        }

        char *pline = line;
        int n = 0, delta = 0, word_cnt = 0;
        while (sscanf(pline += delta, "%d%n", &n, &delta) == 1) {
            for (int i = 0; i < n; i++)
                putchar('a' + rand() % 26);
            putchar('\n');
            fflush(stdout);
            word_cnt++;
        }
        // Indicate end of word list.
        printf("LAST\n"); fflush(stdout);
    }

    return 0;
}


// proc3.c  /////////////////////////////////////////////////////
#include <stdio.h>
#include <string.h>

int main() {
    char line[1000];
    int word_cnt = 0;

    while (fgets(line, sizeof line, stdin) != NULL) {
        if (strcmp(line, "EXIT\n") == 0)
            break;

        if (strcmp(line, "LAST\n") == 0) {

            fprintf(stderr, "proc3 saw %d words\n", word_cnt);

            // Arbitrarily stopping when proc3 sees a word
            // list more than 80 words long
            if (word_cnt > 80) {
                // Send STOP command to proc1
                printf("STOP\n"); fflush(stdout);
                break;
            }
            else {
                // Tell proc1 to continue
                printf("\n"); fflush(stdout);
            }
            word_cnt = 0;
        }
        else
            word_cnt++;
    }

    return 0;
}