GNU C Multi-process handling with pipes

Question

I was taught in the university to the easiest way of handling processes in the background to hang the running of the children/parent processes with the pipe read() function. Honestly I've been working on my homework for 2 weeks already and I can't solve the asynchronous process handling. I minimized my code, to write a single child handling with 2 pipes, and block the parent and child processes with the read() function. You can find the current state of my code below:

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

char pipeMessage[100];

char* readFromPipe(int pipe)
{
    pipeMessage[0] = '\0';
    read(pipe, pipeMessage, sizeof(pipeMessage));
    fflush(NULL);
    return pipeMessage;
}

void writeToPipe(int pipe, char *input)
{
    char text[strlen(input) + 1];
    strncpy(text, input, (int)strlen(input));
    strncat(text, "\0", 1);
    printf("TEXT: %s\n", text);
    write(pipe, text, sizeof(text));
    fflush(NULL);
}

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

    if(pipe(pipes[0]) < 0 || pipe(pipes[1]) < 0)
    {
        printf("[ERROR] create pipes\n");
        return -1;
    }

    printf("[PARENT] Create child1\n");
    if((pid = fork()) < 0)
    {
        printf("[ERROR] Fork error\n");
        return -1;
    }

    if(pid == 0)
    {
        // Child code
        close(pipes[0][0]);
        writeToPipe(pipes[0][1], "TEST MESSAGE");
        printf("[CHILD1] pipe message: %s\n", readFromPipe(pipes[1][0]));
        writeToPipe(pipes[0][1], "-1");
    }
    else if(pid > 0)
    {
        // Parent code
        close(pipes[0][1]);
        close(pipes[1][0]);
        char *message;
        do
        {
            message = readFromPipe(pipes[0][0]);
            printf("[PARENT] pipe message: %s\n", message);
            writeToPipe(pipes[1][1], "-1");
        }
        while(atoi(message) != -1);
    }
    return 0;
}

The error is the following:

[PARENT] Create child1
TEXT: TEST MESSAGE
[PARENT] pipe message: TEST MESSAGE
TEXT: -1
[CHILD1] pipe message: -1
TEXT: -1�po
[PARENT] pipe message: -1�T MESSAGE
TEXT: -1�po

I tried to implement this process handling with signals, but in the final application I will need 3 different child processes, and the asynchronous running caused issues with the signal handling. I also tried to find a tutorial on the net, but every multi processing topic covers a simple single message solution from parent to child and vice versa. But I need a character based menu in the parent process, so the children should wait continuously for the parent signal/message, and the parent also need to wait for a child to finish the actual task. Please help me out, because I'm really stuck. If you have any normal solution for process handling, please let me know, because I know this code is a terrible one. The only reason is the lack of correct tutorials. Thanks in advance.

Answer 1

You're write logic is questionable, and the layout of your pipes is overtly complicated. Below is your code tailored down to as simple a case as I can muster. Comments are included to help yo along. I find it easiest when dealing with chained pipes (which is for all intents exactly what this is) to lay out a single descriptor array indexed by macros that exhibit the chaining:

// some hand macros for access the correct pipes
#define P_READ  0
#define C_WRITE 1
#define C_READ  2
#define P_WRITE 3
#define N_PIPES 4

the above will be in the final source list. The monikers should be self-evident, but in case they aren't, P_XXX notes the pipes the parent process uses, C_XXX notes the pipes the child process uses. Keep that in mind when see the code:

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

// some hand macros for access the correct pipes
#define P_READ  0
#define C_WRITE 1
#define C_READ  2
#define P_WRITE 3
#define N_PIPES 4

// reads a buffer up-to len size.
ssize_t readFromPipe(int pipe, char *buff, size_t len)
{
    buff[0] = 0;
    ssize_t res = read(pipe, buff, len);
    assert(res >= 0 && "Failed to read from pipe");
    return res;
}

ssize_t writeToPipe(int pipe, const char *input)
{
    size_t len = strlen(input)+1;
    ssize_t res = write(pipe, input, len);
    assert(res == len && "Failed to write to pipe");
    return res;
}

int main(void)
{
    int pipes[N_PIPES];
    char msg[128] = {0};
    pid_t pid;

    if(pipe(pipes) < 0 || pipe(pipes+2) < 0)
    {
        printf("[ERROR] create pipes\n");
        return EXIT_FAILURE;
    }

    if((pid = fork()) < 0)
    {
        printf("[ERROR] Fork error\n");
        return EXIT_FAILURE;
    }

    // parent code
    if(pid > 0)
    {
        // Parent code. close down the child pipes; don't need them
        printf("parent(%d) create: child(%d)\n", getpid(), pid);
        close(pipes[C_WRITE]);
        close(pipes[C_READ]);

        do
        {
            if (readFromPipe(pipes[P_READ], msg, sizeof(msg)) > 0)
            {
                printf("parent(%d) read : %s\n", getpid(), msg);
                writeToPipe(pipes[P_WRITE], "-1");
            }
            else break;
        }
        while(atoi(msg) != -1);

        // close remaining pipes. no longer needed
        close(pipes[P_READ]);
        close(pipes[P_WRITE]);
    }

    else if(pid == 0)
    {
        // Child code. don't need parent write or child-read lines
        close(pipes[P_READ]);
        close(pipes[P_WRITE]);

        // write message
        writeToPipe(pipes[C_WRITE],"test message");

        // read test message
        if (readFromPipe(pipes[C_READ], msg, sizeof(msg)) > 0)
            printf("child(%d) read : %s\n", getpid(), msg);

        // write another message
        writeToPipe(pipes[C_WRITE], "-1");

        // close remaining pipes. no longer needed
        close(pipes[C_READ]);
        close(pipes[C_WRITE]);
    }

    return EXIT_SUCCESS;
}

The pipe descriptor array not withstanding, the biggest change in this is the simplified writeToPipe logic, which simply writes a terminator C string to the pipe, including the terminating null char.

ssize_t writeToPipe(int pipe, const char *input)
{
    size_t len = strlen(input)+1;
    ssize_t res = write(pipe, input, len);
    assert(res == len && "Failed to write to pipe");
    return res;
}

The caller checks the result to ensure it wrote all the requested data, and an embedded assert() macro will trip your debugger on failure. Similar logic exists for the read functionality.

Output (varies by process id)

parent(2067) create: child(2068)
parent(2067) read : test message
child(2068) read : -1
parent(2067) read : -1

I hope this helps. When dealing with pipes, and in particular with redirection of stdio which you will likely encounter in the not-too-distant future (see spoiler here), it helps tremendously to have meaningful mnemonics for your code, such as the macros I used above for indexing the pipes array.

Best of luck.

Answer 2

Okay, I have a template which I use professionally for embedded solutions. Using this, I have created a working solution, which contains the client loop but the server loop is only declared. It should be clear though how a solution could be made.

It should be clear how multiple readers / writers can be added to the select call. Note have SOCK_DGRAM is used. In addition to socketpairs, pipes can be added (which is preferred for streaming raw data from process to process). I hope that you can learn something from this. We should delete it (sometime). It requires some more work to be complete (the server loop)

#include <stdlib.h>
#include <stdio.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <sys/wait.h>
#include <errno.h>
#include <cstring>  // strerror
#include <unistd.h> // pipe
#include <fcntl.h>

#include <stdint.h>
#include <algorithm>

#define CallErrExit(fun, arg, retval)  {    \
    if ((fun arg)<0) {              \
      FailErr(#fun);                \
      return retval;                \
    }                       \
  }

#define FailErr(msg) {              \
    (void)fprintf(stderr, "%s, %s(%d)\n",   \
          msg, strerror(errno), errno); \
    (void)fflush(stderr);}

const int parentsocket = 0;
const int childsocket = 1;

int disk_main(int comm_server[]);
int server_main(int comm_disk[]);

int main(int argc, char* argv[]) {

  int status; // Status parameter used waitpid

  // Communication sockets
  int comm_server_disk[2];

  // Process Id's
  pid_t proc_server;
  pid_t proc_disk;

  if (argc < 2) {
    fprintf(stderr,"ERROR, no port provided\n");
    return EXIT_FAILURE;
  }

  proc_server = getpid();

  int socket;

  // SOCK_DGRAM are connectionless as opposed to SOCK_STREAM or SOCK_SEQPACKET
  CallErrExit(socketpair, (AF_UNIX, SOCK_DGRAM, 0,
                   comm_server_disk), EXIT_FAILURE);

  CallErrExit(proc_disk = fork,(),EXIT_FAILURE);

  if (proc_disk == 0) {
    // Disk process
    proc_disk = getpid();

    // TODO: Try the alternative, use the names '/proc/<process id>/fd/
    printf("disk_main started with sockets:\n"
           "\tserver->disk: /proc/%d/fd/%d\n"
           "\tdaup->disk: /proc/%d/fd/%d\n",
           proc_disk, socket+1,
           proc_disk, socket+3);

    /*
     * Closing comm_server_disk[1] before function entry
     */
    close(comm_server_disk[childsocket]); // Write to server

    return disk_main(comm_server_disk);
  }
  else {

    // Server process, closes comm_server_disk[0]
    server_main(comm_server_disk);

    // Never reached

    // calling process sets SIGCHLD to SIG_IGN, ECHILD is set
    CallErrExit(waitpid,  (proc_disk,&status,0), EXIT_FAILURE);
    return EXIT_SUCCESS;
  }
}

typedef struct internal_cmd {
  char cmd[32];
} internal_cmd_t;

enum e_cmd {
  eExit = 0x01,
};

// TODO: Some map between enums and cmd[32]

int disk_main(int comm_server[]) {

  char buf[1024];

  int select_width;
  fd_set rfds, wfds, efds;

  int next_select_width;
  fd_set next_rfds, next_wfds, next_efds;

  int n_fds;

  struct timeval timeout = { 10, 0 };

  FD_ZERO(&next_rfds);
  FD_ZERO(&next_wfds);
  FD_ZERO(&next_efds);

  FD_SET(comm_server[0], &next_rfds);

  // Default: is blocking, but be sure
  fcntl(comm_server[0],
        F_SETFL, fcntl(comm_server[0], F_GETFL, 0) & ~O_NONBLOCK);

  // Add other file descriptors
  int ofd = comm_server[0];
  next_select_width = std::max(comm_server[0],ofd) + 1;

  do {
    // Update read/write state
    select_width = next_select_width;
    rfds = next_rfds;
    wfds = next_wfds;
    efds = next_efds;

    // Wait for interrupt
    n_fds = select(select_width,&rfds, &wfds, &efds, &timeout);

    if (n_fds < 0) {
      fprintf(stderr,"ERROR\n");
      exit(1);
    }

    if (FD_ISSET(comm_server[0], &wfds))
      printf("Disk process can write to server\n");

    if (FD_ISSET(comm_server[0], &rfds)) {
      printf("Disk process received message from server\n");
      int rv = recv(comm_server[0], buf, sizeof(buf), MSG_DONTWAIT);
      if (rv < 0) {
        printf("Disk process - %s(%d)\n", strerror(errno), errno);
        exit (1);
      }
      printf("Disk process received %d bytes:\n", rv);
      // Interpret command
      if (rv == sizeof(internal_cmd_t)) {

        // Here interpret command
        e_cmd cmd;
        if (cmd == eExit) {
          printf("Exiting\n");
          close(comm_server[0]);
          return EXIT_SUCCESS;
        }
      }
    }

    FD_ZERO(&next_rfds); FD_ZERO(&next_wfds); FD_ZERO(&next_efds);
    FD_SET(comm_server[0], &next_rfds);

    fcntl(comm_server[0], F_SETFL, fcntl(comm_server[0], F_GETFL, 0) & ~O_NONBLOCK);

    int ofd = comm_server[0];
    next_select_width = std::max(comm_server[0],ofd) + 1;
  }
  while (true);

  close(comm_server[0]);

  return EXIT_SUCCESS;
}