sort-parallel-uniq.c

/* pipe/fork/dup2/execve exercise.

   The results are similar to: `sort -u --parallel 2 < input_file`

   http://stackoverflow.com/questions/19876980/create-2-child-processes-to-sort-words-of-a-file-using-pipe

   Usage:

   $ gcc -o sort-parallel-uniq *.c && ./sort-parallel-uniq < input_file
*/
/* for fdopen(), getline() */
#ifndef _POSIX_C_SOURCE
#define _POSIX_C_SOURCE 200809L
#endif

#include <errno.h>
#include <stdarg.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>

#include <sys/types.h> /* pid_t */
#include <sys/wait.h>
#include <unistd.h>

#define PROGNAME "sort-parallel-uniq"

#define Close(FD) do {                                  \
    int Close_fd = (FD);                                \
    if (close(Close_fd) == -1) {                        \
      perror("close");                                  \
      fprintf(stderr, "%s:%d: close(" #FD ") %d\n",     \
              __FILE__, __LINE__, Close_fd);            \
    }                                                   \
  }while(0)

#define perror_exit err_exit
#define _perror_exit _err_exit

enum {
  nchildren = 2, /* number of child processes that perform sorting */
  npipes = nchildren*2
};

static int err_print(const char* format, ...) {
  va_list ap;
  int n = -1;
  fputs(PROGNAME ": error: ", stderr);

  va_start(ap, format);
  n = vfprintf(stderr, format, ap);
  va_end(ap);

  if (errno != 0)
    fprintf(stderr, ": %s\n", strerror(errno));
  else
    fputs("\n", stderr);

  return n;
}

static void err_exit(const char* format, ...) {
  va_list ap;

  va_start(ap, format);
  err_print(format, ap);
  va_end(ap);
  exit(EXIT_FAILURE);
}

static void _err_exit(const char* format, ...) {
  va_list ap;

  va_start(ap, format);
  err_print(format, ap);
  va_end(ap);
  _exit(EXIT_FAILURE);
}

static void redirect(int oldfd, int newfd) {
  if (oldfd != newfd) {
    if (dup2(oldfd, newfd) != -1)
      Close(oldfd);
    else
      _perror_exit("dup2");
  }
}

/* return `malloc`ed "/proc/self/fd/`fd`" or NULL on error */
static char* make_proc_filename(int fd) {
  const char* format = "/proc/self/fd/%d";
  const int n = snprintf(NULL, 0, format, fd);
  if (n > 0) {
    char *buf = (char*)malloc(n+1);
    if (buf) {
      int c = snprintf(buf, n+1, format, fd);
      if (buf[n] == '\0' && c == n)
        return buf; /* success */
      free(buf); /* error */
    }
  }
  return NULL; /* error */
}

int main(void) {
  int fd[npipes][2]; /* even indexes: parent -> child, odd: child -> parent */
  size_t i, j, k, P2C, C2P; /* loop indexes */
  pid_t pid;

  /* open pipes */
  for (i = 0; i < npipes; ++i)
    if (pipe(fd[i]) == -1) /* error */
      perror_exit("pipe");

  /* start child processes */
  for (i = 0; i < nchildren; ++i) {
    P2C = 2*i; C2P = 2*i + 1;
    if ((pid = fork()) == -1) /* error */
      perror_exit("fork");
    else if (pid == 0) { /* child */
      Close(fd[P2C][1]); /* close the unused write end of P2C pipe */
      Close(fd[C2P][0]); /* close the unused read end of C2P pipe */

      /* close other pipes */
      for (j = 0; j < npipes; ++j)
        if (j != P2C && j != C2P)
          for (k = 0; k < 2; ++k)
            Close(fd[j][k]);

      /* redirect stdin, stdout */
      redirect(fd[P2C][0], STDIN_FILENO);
      redirect(fd[C2P][1], STDOUT_FILENO);

      /* run sort */
      execlp("sort", "sort", NULL);
      _perror_exit("execlp"); /* error */
    }
  }

  /* parent */
  /* close unused pipes */
  for (i = 0; i < nchildren; ++i) {
    P2C = 2*i; C2P = 2*i + 1;
    Close(fd[P2C][0]);   /* close the unused read end of P2C pipe */
    Close(fd[C2P][1]);   /* close the unused write end of C2P pipe */
  }

  { /* read input line by line, and distribute it in a round robin
       fashion among children
    */
    /*NOTE: line may grow unlimited, never shrinks, see getline */
    char *line = NULL;
    size_t len = 0;
    FILE* fp[nchildren];

    /* open pipes as files to enable buffering */
    for (i = P2C = 0; i < nchildren; ++i, P2C+=2)
      if ((fp[i] = fdopen(fd[P2C][1], "wb")) == NULL)
        err_exit("fdopen");

    while(1)
      for (i = 0; i < nchildren; ++i)
        if (getline(&line, &len, stdin) == -1 || /* read line from stdin, */
      fputs(line, fp[i]) == EOF)           /* write it to a child */
          goto eof_or_error;

  eof_or_error:
    for (i = 0; i < nchildren; ++i)
      fclose(fp[i]); /* no more input for children */

    if (!feof(stdin)) /* error */
      err_exit("stdin: expected EOF");

    fclose(stdin); /* no more input */

    free(line);
  }

  /* run "sort -um" */
  if ((pid = fork()) < 0) { /* error */
    perror_exit("fork 'sort -um'");
  }
  else if (pid == 0) { /* child */
    /* read sorted output from children, merge it, remove duplicates */
    /* call sort -um /proc/self/fd/{[C2P][0]} */
    char* args[2+nchildren+1];
    args[0] = (char*)"sort";
    args[1] = (char*)"-um"; /* merge, remove duplicates (uniq) */
    for (i = 0; i < nchildren; ++i) {
      C2P = 2*i + 1;
      if ((args[2+i] = make_proc_filename(fd[C2P][0])) == NULL)
       _err_exit("make_proc_filename");
    }
    args[2+nchildren] = NULL;

    execvp("sort", args);
    _perror_exit("execvp");
  }
  else if (pid > 0) { /* parent */
    fclose(stdout);
    /* close unused pipes */
    for (C2P = 1; C2P < npipes; C2P+=2)
      Close(fd[C2P][0]);

    /* fail if any of the children failed */
    for (i = 0; i < (nchildren + 1); ) {
      int status = 0;
      if ((pid = waitpid(-1, &status, 0)) == -1) {
        if (errno == EINTR)
          continue; /* try again */
        else
          perror_exit("waitpid");
      }
      else if (WIFEXITED(status)) {
        if (WEXITSTATUS(status))
          err_exit("%d child returned non-zero status %d",
                   pid, WEXITSTATUS(status));
      }
      else if (WIFSIGNALED(status)) {
        err_exit("%d child killed by signal %d",  pid, WTERMSIG(status));
      }
      else {
        err_exit("%d child returned unexpected status %d", pid, status);
      }
      ++i; /* next child */
    }
  }
  return 0;
}