pipelinuxexample.c

pipe的实现的Example

#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <string.h>
#include <errno.h>
#include <limits.h>		// for PIPE_BUF
#include <signal.h>
#include <sys/types.h>
#include <sys/stat.h>

#define BUFFER_SIZE	PIPE_BUF

int global_exit_flag = 0;

void parent_process_main(int fd);
void child_process_main(int fd);
void signal_handler(int s);

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

  //int pipe(int fildes[2]);
  if (pipe(fds) < 0)
  {
    fprintf(stderr, "Create a pipe failed: %s\n", strerror(errno));
    exit(1);
  }

  fprintf(stdout, "write endpoint: %d\n", fds[1]);
  fprintf(stdout, "read endpoint: %d\n", fds[0]);

  pid_t pid;

  //pid_t fork(void);
  if ((pid = fork()) < 0)
  {
    fprintf(stderr, "fork() failed: %s\n", strerror(errno));
    exit(1);
  }
  else if (pid == 0)
  {
    // XXX: child process, close write endpoint
    close(fds[1]);
    child_process_main(fds[0]);
    exit(0);
  }
  else
  {
    // XXX: parent process, close read endpoint
    close(fds[0]);
    parent_process_main(fds[1]);
  }

  close(fds[0]);
  close(fds[1]);

  return 0;
}

void parent_process_main(int fd)
{
  char buffer[BUFFER_SIZE];
  ssize_t written;

  ssize_t offset = 0;
  int length;

  signal(SIGINT, signal_handler);
  signal(SIGTERM, signal_handler);

  //char *fgets(char *s, int size, FILE * stream);
  while (fgets(buffer, BUFFER_SIZE, stdin))
  {
    length = strlen(buffer);
    offset = 0;

    while (offset < length)
    {
    again:
      //ssize_t write(int fd, const void *buf, size_t count);
      if ((written = write(fd, buffer + offset, length - offset)) < 0)
      {
	if (errno == EINTR)
	{
	  fprintf(stdout, "[%d]Got a signal, retry.\n", getpid());
	  goto again;
	}
	else
	{
	  fprintf(stderr, "[%d]write() failed: %s\n", getpid(), strerror(errno));
	  // FIXME: 
	  exit(1);
	}
      }
      else
      {
	offset += written;
      }
    }

    fprintf(stdout, "[%d]written %d bytes to pipe.\n", getpid(), length);

    if (global_exit_flag)
    {
      break;
    }
  }

  fprintf(stdout, "[%d]out of fetch string from stdin.\n", getpid());
}

void child_process_main(int fd)
{
  char buffer[BUFFER_SIZE];
  ssize_t n;

  for (;;)
  {
  again:
    //ssize_t read(int fd, void *buf, size_t count);
    if ((n = read(fd, buffer, BUFFER_SIZE)) < 0)
    {
      if (errno == EINTR)
      {
	goto again;
      }

      fprintf(stderr, "[%d]read() failed: %s\n", getpid(), strerror(errno));
      // FIXME:
      exit(1);
    }
    else if (n == 0)
    {
      struct stat s;

      // FIXME: check return value
      //int fstat(int filedes, struct stat *buf);
      fstat(fd, &s);

      if ((S_ISFIFO(s.st_mode)))
      {
	fprintf(stdout, "[%d]The write endpoint is closed, exiting ...\n", getpid());
	exit(0);
      }
    }
    else
    {
      fprintf(stdout, "%s", buffer);
    }
  }
}

void signal_handler(int s)
{
  fprintf(stdout, "Caught signal %d, exiting...\n", s);
  global_exit_flag = 1;
}