Unable to exit while loop after reading information written to pipe

Question

TLDR: You have to close the write end of all pipes in all children. The read will detect EOF only if no process has the write end still open. Credits to @Bodo

As part of an assignment for an operating systems course, I'm trying to read lines from a file which is in the format of x operand y and distribute the lines to different child processes so that each one can take those lines as input and conduct calculations and write it to one output file.

I feel like I'm almost there by getting the right results, but my code seems to lead to an endless while loop after reading all of the written lines to the read end of a pipe.

Here's the relevant code snippet

int child_work(int pipes[][2], int proc, int procid, FILE * out)
{
    int i;
    pid_t mypid;
    Expression exp;
    float result;
    int procidx = procid;
    char expression[MAIN_BUF_LEN];
    int r_val;
    printf("entered while loop for child process %d\n", mypid);
    while(1)
    {
        if ( (r_val = read(pipes[procid][0], expression, MAIN_BUF_LEN)) > 0)
        {
            printf("return values of read: %d\n", r_val);
            exp_readln(&exp, expression);
            result = exp_cal(&exp);
            printf("[#%d]: %d %0.3f\n", procidx, mypid, result);
            fprintf(out, "#%d: %d %0.3f\n", procidx, mypid, result);
            fflush(out);
            procidx += proc;
        }
        else
        {
            break;
        }
    }
    printf("exited while loop and reached end of child process %d\n", mypid);
    return 0;

int main(int argc, char **argv)
{
    if (argc != 4)
    {
        printf("not enough arguments");
        return 0;
    }

    const char *infile;  // Name of infile
    const char *outfile; // Name of outfile
    int proc;            // Number of child process to fork

    // Save arguments to variables
    infile = argv[1];
    outfile = argv[2];
    sscanf(argv[3], "%u", &proc);

    int pipes[proc][2]; // Pipes to be created
    pid_t child_pids[proc]; // store all the pids of children created

    int i; // Loop counter

    char buf[MAIN_BUF_LEN];
    Expression exp;

    FILE * in_ptr, *out_ptr;
    // Open infile with read-only, outfile with write and append.
    if ((in_ptr = fopen(infile, "r")) == NULL)
    {
        printf("Error in opening file. Ending program. \n");
        return 1;
    }
    out_ptr = fopen(outfile, "a+");

    // Get parent pid and print to outfile
    int ppid = getpid();
    fprintf(out_ptr, "%d\n", ppid);
    fflush(out_ptr);

    // $proc pipes should be created and saved to pipes[proc][2]
    for (i = 0; i < proc; ++i)
    {
        // TODO
        if (pipe(pipes[i]) == -1 )
        {
            printf("Pipe failed for pipe %d\n", i);
            return 1;
        }
    }

    // $proc child processes should be created.
    // Call child_work() immediately for each child.
    for (i = 0; i < proc; ++i)
    {
        int pid;
        // create child only if in parent process
        if (getpid() == ppid)
        {
            pid = fork();
            if (pid != 0)
                printf("created child with child pid %d\n", pid);
                child_pids[i] = pid;
        }

        if (pid == 0) // in child process
        {
            child_work(pipes, proc, i, out_ptr);
            break;
        }
        else if (pid < 0) // error in forking
        {
            printf("Fork failed.\n");
        }
    }

    // Close reading end of pipes for parent
    for (i = 0; i < proc; ++i)
    {
        // TODO
        if (getpid() == ppid)
            close(pipes[i][0]);
    }

    // Read lines and distribute the calculations to children in round-robin
    // style.
    // Stop when a empty line is read.

    char* line = NULL;
    size_t len = 0;
    ssize_t read = 0;
    int j = 0;
    while ((read = getline(&line, &len, in_ptr)) != -1) {
        //printf("Retrieved line of length %zu:\n", read);
        //printf("%s", line);
        j = j % proc;
        write(pipes[j++][1], line, strlen(line)+1);
    }

    // Close all the pipes when the task ends
    for (i = 0; i < proc; ++i)
    {
    //   close(pipes[i][READ]);
       close(pipes[i][WRITE]);
    }
    printf("Task 6 complete!");

    for (i = 0; i < proc; ++i)
    {
        waitpid(child_pids[i], NULL, 0);
    }

    fprintf(out_ptr, "\n");
    fflush(out_ptr);

    return 0;
}

This is the output that I am getting, which seemingly gets stuck in an infinite while loop as the process won't terminate. Also, the value of return values of read: should either be 22 or 23 based on the particular input file that I am using, but I don't know why it is incrementing for particular subsequent child processes. None of the child processes seem to be able to exit the while loop as this printf("exited while loop and reached end of child process %d\n", mypid); doesn't seem to be executed. My understanding is that if a pipe has been read, the return value will be the byte size of the line read, and if it reaches EOF or an error, the return value is 0 or -1, respectively.

entered while loop for child process 16016
entered while loop for child process 16017
entered while loop for child process 16018
entered while loop for child process 16020
return values of read: 22
entered while loop for child process 16019
[#0]: 16016 1.783
return values of read: 22
return values of read: 22
[#2]: 16018 0.061
[#1]: 16017 0.195
return values of read: 22
return values of read: 22
[#5]: 16016 0.269
return values of read: 46
[#10]: 16016 1.231
return values of read: 22
return values of read: 22
[#6]: 16017 0.333
return values of read: 22
return values of read: 46
[#11]: 16017 1.684
[#7]: 16018 -0.734
return values of read: 46
[#12]: 16018 0.134
[#3]: 16019 0.778
return values of read: 68
[#4]: 16020 -0.362
return values of read: 68
[#9]: 16020 0.506
[#8]: 16019 -0.450

I would appreciate any insight for a silly mistake I might be making. Thanks!

Answer 1

There are several problems in the code.

Unfortunately I cannot compile it and fix the errors because it is incomplete.

1. You cannot define arrays with a size that is not constant like this.

   int pipes[proc][2]; // Pipes to be created
   

   I would expect the compiler to show a warning at this line.
   You should either use dynamic allocation (malloc) or statically allocate the arrays with a maximum size and check that proc is not greater than the maximum.

2. You have to close the write end of all pipes in all children. The read will detect EOF only if no process has the write end still open.

3. Instead of

   while(1)
   {
       if ( (r_val = read(pipes[procid][0], expression, MAIN_BUF_LEN)) > 0)
       {
           /*...*/
       }
       else
       {
           break;
       }
   }
   

   I suggest

   while((r_val = read(pipes[procid][0], expression, MAIN_BUF_LEN)) > 0)
   {
           /*...*/
   }
   

4. Instead of

       pid = fork();
       if (pid != 0)
           printf("created child with child pid %d\n", pid);
   

   it should be

       pid = fork();
       if (pid > 0)
           printf("created child with child pid %d\n", pid);
   

   because pid < 0 is an error.

5. Instead of

   if (pid == 0) // in child process
   {
       child_work(pipes, proc, i, out_ptr);
       break;
   }
   

   use

   if (pid == 0) // in child process
   {
       child_work(pipes, proc, i, out_ptr);
       return 0;
   }
   

   With break; the child would continue with the code after the for loop that would read the file and write to the pipes when child_work returns.

6. It is not guaranteed that every child will get its turn to read from the pipe before the parent writes the next data, so it may get two or more messages in a single read. In real applications you should also be prepared to handle incomplete read or write calls and to continue writing/reading the remaining data with additional read or write calls.

   I think the easiest way to handle partial read or write would be to use buffered IO. You can use fdopen with the wrtite file descriptor or the read file descriptor of the pipe and write/read the data as a line of text terminated with a newline using e.g. fprintf or fgets respectively.