Unix Prog: IPC Message Queue

1. Message Queue Overview

A message queue is a linked list of messages stored within the kernel and identified by a message queue identifier. 

Using msgget can create a new queue or refer to one existing queue.

Using msgsnd can add new messages to the end of a queue.

Every message has a positive long integer type field, a non-negative length, and the actual data bytes.

Using msgrcv can fetch a message from the a queue, and we don't have to fetch the message in a first-in, first-out order, instead, we can fetch messages based on their type field.

Each message queue has the following msqid_ds structure associated with it:

 ubuntu@ip-172-31-23-227:~$ less /usr/include/x86_64-linux-gnu/bits/msq.h  
 ......  
 struct msqid_ds  
 {  
  struct ipc_perm msg_perm;   /* structure describing operation permission */  
  __time_t msg_stime;      /* time of last msgsnd command */  
 #ifndef __x86_64__  
  unsigned long int __glibc_reserved1;  
 #endif  
  __time_t msg_rtime;      /* time of last msgrcv command */  
 #ifndef __x86_64__  
  unsigned long int __glibc_reserved2;  
 #endif  
  __time_t msg_ctime;      /* time of last change */  
 #ifndef __x86_64__  
  unsigned long int __glibc_reserved3;  
 #endif  
  __syscall_ulong_t __msg_cbytes; /* current number of bytes on queue */  
  msgqnum_t msg_qnum;      /* number of messages currently on queue */  
  msglen_t msg_qbytes;     /* max number of bytes allowed on queue */  
  __pid_t msg_lspid;      /* pid of last msgsnd() */  
  __pid_t msg_lrpid;      /* pid of last msgrcv() */  
  __syscall_ulong_t __glibc_reserved4;  
  __syscall_ulong_t __glibc_reserved5;  
 };  
 ......  

Above structure defined the current status of the queue

2. Message Queue System Calls
1) msgget

 ubuntu@ip-172-31-23-227:~$ less /usr/include/x86_64-linux-gnu/sys/msg.h  
 ......  
 /* Get messages queue. */  
 extern int msgget (key_t __key, int __msgflg) __THROW;  
 ......  

msgget can be used to create a new queue or refer to one existing queue.
When a new queue is created, following members of the msqid_ds structure are initialized:
ipc_perm structure is initialized. msg_qnum, msg_lspid, msg_lrpid, msg_stime, and msg_rtime are all set to 0, msg_ctime is set to the current time, msg_qbytes is set to the system limit.

On success, msgget returns the non-negative queue ID.

2) msgctl

 ubuntu@ip-172-31-23-227:~$ less /usr/include/x86_64-linux-gnu/sys/msg.h  
 ......  
 /* Message queue control operation. */  
 extern int msgctl (int __msqid, int __cmd, struct msqid_ds *__buf) __THROW;  
 ......  

msgctl function performs various operations on a queue.
The cmd argument specifies the command to be performed:

 ubuntu@ip-172-31-23-227:~$ less /usr/include/x86_64-linux-gnu/bits/ipc.h  
 ......  
 /* Control commands for `msgctl', `semctl', and `shmctl'. */  
 #define IPC_RMID    0        /* Remove identifier. */  
 #define IPC_SET     1        /* Set `ipc_perm' options. */  
 #define IPC_STAT    2        /* Get `ipc_perm' options. */  
 #ifdef __USE_GNU  
 # define IPC_INFO    3        /* See ipcs. */  
 #endif  
 ......  

IPC_STAT: Fetch the msgid_ds structure for this queue, storing it in the structure pointed to by buf.
IPC_SET: Copy the following fields from the structure pointed to by buf to the msqid_ds structure associated with this queue: msg_perm.uid, msg_perm.gid, msg_perm.mode, and msg_qbytes. This command can only be executed by a process whose effective user id equals msg_perm.cuid or msg_perm.uid or by a process with superuser privilege.
IPC_RMID: rmeove the message queue from the system and any data still on the queue.The removal is immediate. And it can only be executed by a process whose effective user id equals msg_perm.cuid or msg_perm.uid or by a process with super user privilege.

3) msgsnd
msgsnd can be used to place a message into the message queue. And messages are always placed in the end of the queue.

 ubuntu@ip-172-31-23-227:~$ less /usr/include/x86_64-linux-gnu/sys/msg.h  
 ......  
 /* Send message to message queue.  
   
   This function is a cancellation point and therefore not marked with  
   __THROW. */  
 extern int msgsnd (int __msqid, const void *__msgp, size_t __msgsz,  
           int __msgflg);  
 ......  

msgp argument points to the long integer that contains the positive integer message type, and it is immediately followed by the message data.

 ubuntu@ip-172-31-23-227:~$ less /usr/include/x86_64-linux-gnu/sys/msg.h  
 ......  
 /* Template for struct to be used as argument for `msgsnd' and `msgrcv'. */  
 struct msgbuf  
  {  
   __syscall_slong_t mtype;  /* type of received/sent message */  
   char mtext[1];       /* text of the message */  
  };  
 ......  

msgp points to a msgbuf structure. The message type can be used by the receiver to fetch messages in an order other than first in , first out.

Note: length of mtext should be same as third argument msgsz in real world.

A flag value of IPC_NOWAIT can be specified. If the message queue is full, specifying the IPC_NOWAIT causes msgsnd to return immediately with an error of EAGAIN. If IPC_NOWAIT is not specified, we are blocked until there is room for the message, the queue is removed from the system, or a signal is caught and signal handler returns.

After msgsnd returns successfully, the msqid_ds structure associated with the message queue is updated to indicate the process ID that made the call

4) msgrcv

 ubuntu@ip-172-31-23-227:~$ less /usr/include/x86_64-linux-gnu/sys/msg.h  
 ......  
 /* Receive message from message queue.  
   
   This function is a cancellation point and therefore not marked with  
   __THROW. */  
 extern ssize_t msgrcv (int __msqid, void *__msgp, size_t __msgsz,  
             long int __msgtyp, int __msgflg);  
 ......  

As with msgsnd, the ptr argument points to a long integer(where the message type of the returned message is stored) followed by a data buffer for the actual message data. msgp can point to the struct msgbuf as indicated above.

If the returned message buffer is larger than msgsz, and the MSG_NOERROR bit in flag is set, and the message is truncated, and message is totally removed from the queue. If MSG_NOERROR bit is not set in flag, an error of E2BIG is returned, and message still stays on the queue.

For fourth argument, if msgtyp == 0, the first message on the queue is returned. If msgtyp > 0, the first message on the queue whose message type equals type is returned. If msgtyp < 0, the first message on the queue whose message type is the lowest value less than or equal to the absolute value of type is returned.

In client server model, message from clients normally has the type which is the process id of client. And server can based on process id as type to retrieve message from specific client.

If IPC_NOWAIT is specified, msgrc will return -1 with errno set to ENOMSG if a message of the specified type is not available. If IPC_NOWAIT is not specified, the operation blocks until a message of the specified type is available, the queue is removed from the ssytem, or a signal is caught and the signal handler returns.

Unix Prog: XSI IPC Overview

1. XSI IPC
Three types of XSI IPC: message queue, semaphores, and shared memory.
All three objects share some similarities as following.

2. Identifiers and Keys
Each IPC structure(message queue, semaphores, and shared memory) is identified in the kernel by a non-negative integer. To send or fetch a message to or from a message queue, we need to know the identifier for the message queu.

Each IPC structure is associated with a key that acts as an external name, compared to identifier, which is internal name. Whenever an IPC structure is being created(msgget, semget, or shmget), a key must be specified.

Several ways for clients and server to rendezvous at the same IPC structure:
1) The server can create a new IPC structure by specifying a key of IPC_PRIVATE and store the returned identifier somewhere, like a file, for the client to obtain
2) The server create a new IPC structure using IPC_PRIVATE Key, then the resulting identifier is available to the child after the fork.
3) The client and the server can agree on a key by defining the key in a common header. The server then creates a new IPC structure specifying this key. If the key is already associated with one existing IPC structure, the server need to delete that structure and re-create it.
4) The client and server can agree on a pathname and project id(between 0 and 255), and call the function ftok to convert these two values into a key.

System Definition:

 ubuntu@ip-172-31-23-227:~$ less /usr/include/x86_64-linux-gnu/sys/ipc.h  
 ......  
 /* Generates key for System V style IPC. */  
 extern key_t ftok (const char *__pathname, int __proj_id) __THROW;  
 ......  

3. Create and Get the IPC Structure
For 3 get functions: msgget, semget, and shmget, all need to specify a key and an integer flag.

Create the IPC Structure with key IPC_PRIVATE:
1) After server create the IPC structure with key IPC_PRIVATE, then it is not possible to be conflict with any existing structure, since it would create a new structure.
2) When client want to refer to this structure, for example, sending a message to the message queue, the client must know the identifier of this structure.

Create the IPC Structure with a key who is not currently associated with an IPC structure of the particular type and the IPC_CREAT bit of flag is specified.
1) In this case, when client want to refer to this IPC structure, it need to specify the same key type without specifying IPC_CREAT bit flag.

Create the IPC Structure with both IPC_CREAT and IPC_EXCL bits set:
If the specified key refer to some existing IPC structure, , it will return an error of EEXIST.

4. Permission Structure
XSI IPC associates ipc_perm structure with each IPC structure:
System Definition:

 ubuntu@ip-172-31-23-227:~$ less /usr/include/x86_64-linux-gnu/bits/ipc.h  
 ......  
 /* Data structure used to pass permission information to IPC operations. */  
 struct ipc_perm  
  {  
   __key_t __key;           /* Key. */  
   __uid_t uid;            /* Owner's user ID. */  
   __gid_t gid;            /* Owner's group ID. */  
   __uid_t cuid;            /* Creator's user ID. */  
   __gid_t cgid;            /* Creator's group ID. */  
   unsigned short int mode;      /* Read/write permission. */  
   unsigned short int __pad1;  
   unsigned short int __seq;      /* Sequence number. */  
   unsigned short int __pad2;  
   __syscall_ulong_t __glibc_reserved1;  
   __syscall_ulong_t __glibc_reserved2;  
  };  
 ......  

All above fields are initialized when IPC structure is created, at a later time, we can modify the uid, gid, and mode fields by calling msgctl, semctl, or shmctl.  And the caller must be either the creator of the IPC structure or the super user.

mode member represents the permission bits. In IPC permission structure, only 6 bits are specified:
1) user read 2) user write 3) group read 4) group write 5) other read 6) other write

5. Advantages and Disadvantages:
1) IPC structure is system wide and do not have one reference count. If we create a message queue, place some messages on the queue, and then terminate, the message queue and its contents are not deleted.
For pipe, when the last process to reference it terminates, the data in there is completely removed.
For FIFO, although the file stays in the file system. But any data left in a FIFO is removed when the last process to reference FIFO terminates.

2) IPC structure is not known by names in the file system, we can't access them and modify their properties with the standard I/O functions

3) These forms of IPC don't use file descriptors, we can't use the multiplexed I/O functions(select and poll) with them.

Unix Prog: IPC: FIFO

1. FIFO File
Pipes can be used only between related processes when a common ancestor has created the pipe.
With FIFO, unrelated processes can exchange data.

FIFO is a type of file, st_mode member of the stat structure indicates that a file is FIFO. http://dreamchaser1986unixdev.blogspot.com/2014/08/unix-prog-files-types_19.html

System Call used to create the FIFO file:

 ubuntu@ip-172-31-23-227:~$ less /usr/include/x86_64-linux-gnu/sys/stat.h  
 ......  
 /* Create a new FIFO named PATH, with permission bits MODE. */  
 extern int mkfifo (const char *__path, __mode_t __mode)  
    __THROW __nonnull ((1));  
 ......  

Once the FIFO file is created, we can open, read, write, close, unlink it like treating regular files.

If the O_NONBLOCK bit is not set in FIFO file descriptor:
1) An open for read will block until some other process open it for writing. And a write will block until some other process open it for reading.
Otherwise:
2) An open for read will return immediately. open for write-only returns -1 with errno set to ENXIO if no process has the FIFO for reading.

If there are multiple writers writing to the same FIFO, we need to make sure each write is atomic to avoid interleaving different writes. As long as the data amount get written doesn't exceed the PIPE_BUF, each write is atomic.

Usage of FIFO:
1) Used by shell commands to pass data from one shell pipelines to another without creating intermediate temp files
2) Used as rendezvous point in client-server applications to pass data between clients and server.

2. Example:
tr.c:

 #include<stdio.h>  
 #include<stdlib.h>  
 #include<unistd.h>  
   
 int main(int argc, char* argv[])  
 {  
  char line[20];  
  fgets(line, 20, stdin);  
  fputs(line, stdout);  
   
  exit(0);  
 }  

tr2.c:

 #include<stdio.h>  
 #include<stdlib.h>  
 #include<unistd.h>  
   
 int main(int argc, char* argv[])  
 {  
  char line[20];  
  fgets(line, 20, stdin);  
  fputs("tr2.out:", stdout);  
  fputs(line, stdout);  
   
  exit(0);  
 }  

shell:
mkfifo creates one FIFO file from the shell;
tr2.out reads from fifo , but right now no processes write to fifo, so tr2.out block there.
launch the tr.out to read from text file and then pipe to "tee", whose job is to pass text from standard input to standard output and the fifo file specified after "tee".

So the final output includes 2 lines:
first line is the text from tr.out, output to standard output
second line is the text from tr2.out, tr2.out get the text from fifo, which is then from tee.

 ubuntu@ip-172-31-23-227:~$ mkfifo fifo
 ubuntu@ip-172-31-23-227:~$ ./tr2.out < fifo &  
 [1] 1536  
 ubuntu@ip-172-31-23-227:~$ ./tr.out < text | tee fifo  
 Hello world!  
 tr2.out:Hello world!  

Following diagram can illustrate this:

2. Client Server Communication Example

In client-server communication model, server has one well-known FIFO, which is used to receive different requests sent from clients. If client specified the process id in the request, then after server process the request, server can build the client-specific FIFO to send reply back to client.

For example, if client's process id is 122, then server can create the FIFO called: client_122, then process 122 can use this FIFO specifically to receive the reply from server.

Problem:
If the client crashes, it will leave the client-specific FIFO in file system.
The server also need to catch the SIGPIPE, since it is possible that client send request and then terminate before receiving the reply.

Unix Prog: IPC Coprocesses

1. Coprocesses:

A coprocess normally runs in the background from a shell, and its standard input and standard output are connected to another program using a pipe. Following diagram can illustrate this:

2. Coprocess Example
Here we implement a very simple coprocess add2:

add2.c:

 #include<stdio.h>  
 #include<stdlib.h>  
 #include<unistd.h>  
 #include<string.h>  
   
 #define MAXLINE 20  
   
 int main(void)  
 {  
  int n, int1, int2;  
  char line[MAXLINE];  
   
  while((n = read(STDIN_FILENO, line, MAXLINE)) > 0) {  
   line[n] = 0;  
   if(sscanf(line, "%d%d", &int1, &int2) == 2) {  
    memset(line, 0, MAXLINE);  
    sprintf(line, "%d\n", int1 + int2);  
    n = strlen(line);  
    if(write(STDOUT_FILENO, line, n) != n) {  
     printf("write error!\n");  
     exit(1);  
    }  
   } else {  
    if(write(STDOUT_FILENO, "invalid args\n", 13) != 13) {  
     printf("write error!\n");  
     exit(2);  
    }  
   }  
  }  
   
  exit(0);  
 }  

main.c:

 #include<stdio.h>  
 #include<stdlib.h>  
 #include<signal.h>  
 #include<unistd.h>  
 #include<string.h>  
   
 #define MAXLINE 20  
   
 void sig_pipe(int signo)  
 {  
  printf("SIGPIPE caught\n");  
  exit(1);  
 }  
   
 int main(int argc, char* argv[])  
 {  
  int n, fd1[2], fd2[2];  
  pid_t pid;  
  char line[MAXLINE];  
   
  // Setup the signal for SIGPIPE  
  if(signal(SIGPIPE, sig_pipe) == SIG_ERR) {  
   printf("signal error!\n");  
   exit(1);  
  }  
   
  // Setup the pipe file descriptors  
  if(pipe(fd1) < 0 || pipe(fd2) < 0) {  
   printf("pipe error!\n");  
   exit(2);  
  }  
   
  // Create the child process  
  if((pid = fork()) < 0) {  
   printf("fork error!\n");  
   exit(3);  
  } else if (pid > 0) { // Parent process  
   close(fd1[0]);  
   close(fd2[1]);  
   
   while(fgets(line, MAXLINE, stdin) != NULL) {  
    n = strlen(line);  
    if(write(fd1[1], line, n) != n) {  
     printf("write error to pipe\n");  
     exit(4);  
    }  
    if((n = read(fd2[0], line, MAXLINE)) < 0) {  
     printf("read error from pipe\n");  
     exit(5);  
    }  
    if(n == 0) {  
     printf("child closed pipe");  
     break;  
    }  
   
    // output the result  
    line[n] == 0;  
    if(fputs(line, stdout) == EOF) {  
     printf("fputs error!\n");  
     exit(6);  
    }  
   }  
   
   if(ferror(stdin)) {  
    printf("fgets error on stdin");  
    exit(6);  
   }  
   
   exit(0);  
  } else { // child process  
   close(fd1[1]);  
   close(fd2[0]);  
   if(fd1[0] != STDIN_FILENO) {  
    if(dup2(fd1[0], STDIN_FILENO) != STDIN_FILENO) {  
     printf("dup2 error to stdin\n");  
     exit(1);  
    }  
    close(fd1[0]);  
   }  
   
   if(fd2[1] != STDOUT_FILENO) {  
    if(dup2(fd2[1], STDOUT_FILENO) != STDOUT_FILENO) {  
     printf("dup2 error to stdout\n");  
     exit(1);  
    }  
    close(fd2[1]);  
   }  
   
   if(execl("./add2.out", "add2.out", (char*)0) < 0) {  
    printf("execl error!\n");  
    exit(2);  
   }  
  }  
   
  exit(0);  
 }  

shell:
The process just establish two pipes with child process, who execute the add2.out. Parent process input string to child process, child process read the input and add numbers and put result into output, which then go back to the parent process.

Note: we can't use standard I/O at child process, since if detecting that child process's output is a pipe, child process will make I/O full buffer, and then we can not see the output immediately.
But we can choose to setup the buffer option of standard I/O with setvbuf

 ubuntu@ip-172-31-23-227:~$ ./main.out  
 90 100  
 190  

Unix Prog: popen, pclose(3)

1. Filter Program

We'll build two process piped up together in following way:

2. Source code:
filter.c:

 #include<stdio.h>  
 #include<stdlib.h>  
 #include<ctype.h>  
   
 int main(int argc, char *argv[])  
 {  
  int c;  
   
  while((c = getchar()) != EOF) {  
   if(isupper(c)) c = tolower(c);  
   if(putchar(c) == EOF) {  
    printf("output error!\n");  
    exit(1);  
   }  
   if(c == '\n')  
    fflush(stdout);  
  }  
   
  exit(0);  
 }  

main.c:

 #include<stdio.h>  
 #include<stdlib.h>  
 #include<sys/wait.h>  
   
 int main(int argc, char* argv[])  
 {  
  char line[20];  
  FILE * fpin;  
   
  // Open the filter.out, fpin is used to read input from filter.out  
  if((fpin = popen("./filter.out", "r")) == NULL) {  
   printf("popen error!\n");  
   exit(1);  
  }  
   
  // Make a loop to read from the filter.out and output to standard output  
  for(;;) {  
   fputs("prompt>", stdout);  
   fflush(stdout);  
   if(fgets(line, 20, fpin) == NULL) break;  
   if(fputs(line, stdout) == EOF) {  
    printf("fputs error to pipe\n");  
    exit(2);  
   }  
  }  
   
  // close the stream  
  if(pclose(fpin) == -1) {  
   printf("pclose error!\n");  
   exit(3);  
  }  
   
  putchar('\n');  
  exit(0);  
 }  

shell:
After running the program, user provide input at terminal. And filter read its input, convert the string into lowercase and then transfer to main.out.
If user type Ctrl D, main.out will read the NULL from filter.out and program exits.

 ubuntu@ip-172-31-23-227:~$ ./main.out  
 prompt>HELLO WORLD!  
 hello world!  
 prompt>  

Unix Prog: popen, pclose(2)

1. popen, pclose implementation
pipe.c:

 #include<stdio.h>  
 #include<stdlib.h>  
 #include<unistd.h>  
 #include<errno.h>  
 #include<fcntl.h>  
 #include<sys/wait.h>  
   
 // pointer to array allocated at run-time  
 static pid_t *childpid = NULL;  
   
 // From our open_max()  
 static int maxfd;  
   
 FILE *popen(const char *cmdstring, const char *type)  
 {  
  int i;  
  int pfd[2];  
  pid_t pid;  
  FILE *fp;  
   
  // only allow "r" or "w"  
  if((type[0] != 'r' && type[0] != 'w') || type[1] != 0) {  
   errno = EINVAL;  
   return NULL;  
  }  
   
  // first time to allocate childpid  
  if(childpid == NULL) {  
   maxfd = 4096;  
   if((childpid = calloc(maxfd, sizeof(pid_t))) == NULL) {  
    return NULL;  
   }  
  }  
   
  // Setup the pipe  
  if(pipe(pfd) < 0) return NULL;  
   
  // Create the child process  
  if((pid = fork()) < 0) {  
   return NULL;  
  } else if(pid == 0) { // child process  
   // Handle the pipe file descriptor  
   if(*type == 'r') {  
    close(pfd[0]);  
    if(pfd[1] != STDOUT_FILENO) {  
     dup2(pfd[1], STDOUT_FILENO);  
     close(pfd[1]);  
    }  
   } else {  
    close(pfd[1]);  
    if(pfd[0] != STDIN_FILENO) {  
     dup2(pfd[0], STDIN_FILENO);  
     close(pfd[0]);  
    }  
   }  
   
   // close all existing descriptors which is opened before  
   // this is required by POSIX.1 standard  
   for(i = 0; i < maxfd; i++)  
    if(childpid[i] > 0)  
     close(i);  
   
   // Execute the descriptors  
   execl("/bin/sh", "sh", "-c", cmdstring, (char*)0);  
   _exit(127);  
  }  
   
  // parent process open the file stream from pipe descriptor  
  if(*type == 'r') {  
   close(pfd[1]);  
   if((fp = fdopen(pfd[0], type)) == NULL) return NULL;  
  } else {  
   close(pfd[0]);  
   if((fp = fdopen(pfd[1], type)) == NULL) return NULL;  
  }  
   
  childpid[fileno(fp)] = pid;  
  return fp;  
 }  
   
 int pclose(FILE* fp)  
 {  
  int fd, stat;  
  pid_t pid;  
   
  // popen has never been called.  
  if(childpid == NULL) {  
   errno = EINVAL;  
   return -1;  
  }  
   
  fd = fileno(fp);  
   
  // fp wasn't opened by popen()  
  if((pid = childpid[fd]) == 0) {  
   errno = EINVAL;  
   return -1;  
  }  
   
  childpid[fd] = 0;  
  if(fclose(fp) == EOF)  
   return -1;  
   
  while(waitpid(pid, &stat, 0) < 0) {  
   if(errno != EINTR)  
    return -1;  
  }  
   
  return stat;  
 }  
   
 int main(int argc, char* argv[])  
 {  
  exit(0);  
 }  

Note: popen should never be called by a set-user-id or set-group-id program. when it executes the command with the environment inherited by the caller.
A malicious user can manipulate the environment so that the shell executes commands other than those intended, with the elevated permissions granted by the set-ID file mode.

If the parent process setup the signal handler for SIGCHLD, waitpid is interrupted when child process exits, at this time, errno == EINTR, then we simply retry waitpid again (while waitpid loop).

Unix Prog: popen, pclose(1)

1. popen pclose
System Definition

 ubuntu@ip-172-31-23-227:~$ less /usr/include/stdio.h  
 ......  
 /* Create a new stream connected to a pipe running the given command.  
   
   This function is a possible cancellation point and therefore not  
   marked with __THROW. */  
 extern FILE *popen (const char *__command, const char *__modes) __wur;  
   
 /* Close a stream opened by popen and return the status of its child.  
   
   This function is a possible cancellation point and therefore not  
   marked with __THROW. */  
 extern int pclose (FILE *__stream);  
 ......  

The function popen does a fork and exec to execute the cmdstring, and returns a standard I/O file pointer. If type is "r", the file pointer is connected to the standard output of cmdstring.

If the type is "w", the file pointer is connected to the standard input of cmd string.

The cmd string will be executed by bourne shell which is:

sh -c cmdstring

2. Example:
pipe.c:

 #include<stdio.h>  
 #include<stdlib.h>  
 #include<unistd.h>  
 #include<sys/wait.h>  
   
 #define PAGER "/bin/more" // environment variable or default  
   
 int main(int argc, char* argv[])  
 {  
  char line[20];  
  FILE *fpin, *fpout;  
   
  // Open the existing file for input  
  if((fpin = fopen(argv[1], "r")) == NULL) {  
   printf("can't open %s", argv[1]);  
   exit(1);  
  }  
   
  // Fork a new process, execute the pager, leave standard input on  
  // So parent process can write to fpout that could pipe to pager input  
  if((fpout = popen(PAGER, "w")) == NULL) {  
   printf("popen error!\n");  
   exit(2);  
  }  
   
  // Write the content to the fpout which is piped to standard input  
  while(fgets(line, 20, fpin) != NULL) {  
   if(fputs(line, fpout) == EOF) {  
    printf("fputs error to pipe");  
    exit(3);  
   }  
  }  
   
  if(ferror(fpin)) {  
   printf("fgets error!\n");  
   exit(4);  
  }  
   
  // Close the file stream  
  if(pclose(fpout) == -1) {  
   printf("pclose error!\n");  
   exit(5);  
  }  
   
  exit(0);  
 }  

shell:
popen will make the pipe much easier.

 ubuntu@ip-172-31-23-227:~$ ./pipe.out test  
 hello world!