• Process:
  • An instance of a program execution
• Interprocess Communication:
  • Processes can communicate with each other in a variety of different ways
  • Pipes and FIFOs are examples of IPC mechanisms

• Using the File System:
  • Processes can use the file system to share information (e.g., one can write to a file and another can read from the file) but it is slow
• Some History:
  • People often used to use "RAM disks" to make the process faster
• The Motivation for Pipes/FIFOs:
  • Why not use memory more directly

• Byte Streams:
  • As with files, they involve an undelimited byte stream
• Unidirectional:
  • One end is used for writing and the other end is used for reading
• Destructive Reads:
  • The reader removes bytes from the stream (i.e., buffer)

• Limited Capacity:
  • Since a pipe/FIFO is a buffer in kernel memory it has a limited capacity
• Atomic Writes:
  • Writes (of up to PIP_BUF bytes) are guaranteed to be atomic (hence, if multiple processes are writing to the same pipe/FIFO their bytes won't be "intermingled")
• Blocking:
  • When a pipe/FIFO is "full" (i.e., if there is not enough space for the write), a write will block until enough bytes have been read
  • When a pipe/FIFO is empty, a read will block until at least one byte has been written (or the write end has been closed)

• Pipes:
  • Use file descriptors (hence can only be used by processes that can "share" file descriptors)
  • When the pipe is created, two file descriptors are created (one for the read end and one for the write end)
  • If the reading process doesn't close the write end of a pipe then it will not see an EOF after the writing process closes its end (because the kernel knows there is an open write end)
  • The writing process must close the read end so that when a process tries to write to a pipe that has no read end it can be sent a SIGPIPE signal
• FIFOs:
  • Created using a name within the file system (hence can be used by unrelated processes)
  • The FIFO is created first and then the file descriptors are created by opening the name

• Common Usage:
  • One process writes and another reads
  • Multiple processes write and one other process reads
• Uncommon Usage:
  • One process reads and writes - there are usually better ways to achieve the same result
  • One or more processes write and multiple other processes read - the order of the reads is unknown

• Associating a FILE*:
  • As with other files, fdopen() can be used to associate a FILE* with a valid file descriptor
• Buffering:
  • Output is often written only when the stdio buffer is full (in which case the reader will not receive the bytes immediately) hence one may need to call fflush()

unixexamples/pipes/simple.c

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

int
main(void)
{
  char  msg[20];
  int   fd_pipe[2];
  int   n, status;
  pid_t pid;

  // Construct the pipe
  pipe(fd_pipe);
  
  pid    = fork();

  switch(pid)
    {
    case -1:
      exit(1);

    case 0:
      // The child is "receiving" so must close the write end
      close(fd_pipe[1]);

      // Receive the message
      read(fd_pipe[0], msg, 20);    // Block until the parent is done

      // Close the read end
      close(fd_pipe[0]);

      write(STDOUT_FILENO, msg, 20);
      exit(0);

    default:
      // The parent is "sending" so must close the read end
      close(fd_pipe[0]);

      // Send the message
      write(fd_pipe[1], "Because I said so!\n\0", 20); 

      // Close the write end
      close(fd_pipe[1]);

      wait(&status); // Reap the child

      exit(0);
    }
}
        

unixexamples/pipes/gpa.c

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

#define FIELD_WIDTH 11

float
update_mean(char *value)
{
  static float total = 0.0;
  static int   count = 0;
  
  count++;
  total += atof(value);

  if (count > 0) return total/count;
  else           return -1.0;
}

int
index_of(char *needle, char *haystack)
{
  char *r = strstr(haystack, needle);
  if (r == NULL) return -1;
  else           return r - haystack;
}

int
main(void)
{
  char  grade[FIELD_WIDTH+1]; // 7 for ID, 1 for comma, 3 for grade
  float gpa;
  int   fd_pipe[2], fd_data, n;

  pipe(fd_pipe);
  switch(fork())
    {
    case -1:
      exit(1);

    case 0: // Code to execute in the child process
      // The child is reading from the pipe so must close the write end
      close(fd_pipe[1]);

      // Read from the pipe and update the GPA
      while ((n = read(fd_pipe[0], grade, FIELD_WIDTH)) == FIELD_WIDTH)
        {
          grade[FIELD_WIDTH] = '\0';
          gpa = update_mean(grade+8);
        }
      // Done reading from the the pipe so close it
      close(fd_pipe[0]);

      // Print the CS GPA
      printf("CS GPA:      %4.2f\n", gpa);
      exit(0);

    default: // Code to execute in the parent process
      // The parent is writing to the pipe so must close the read end
      close(fd_pipe[0]);

      // Open the data file
      fd_data = open("grades.dat", O_RDONLY);

      // Read from the data file, write to the pipe, and update the GPA 
      while ((n = read(fd_data, grade, FIELD_WIDTH)) == FIELD_WIDTH)
        {
          grade[FIELD_WIDTH] = '\0';

          // Write CS courses (only) to the pipe
          if (index_of("CS", grade) == 0)
            {
              write(fd_pipe[1], grade, FIELD_WIDTH);
            }
          
          // Update the GPA for all courses
          gpa = update_mean(grade+8);
        }
      // Done reading from the file so close it.
      close(fd_data);

      // Done writing to the pipe so close it
      close(fd_pipe[1]);

      // Reap the child process
      wait(NULL);

      // Print the overall GPA
      printf("Overall GPA: %4.2f\n", gpa);
      exit(0);
    }
}
        

Since you need to have at least one writing process on one end and one writing process on the other, by default a call to open() one end (e.g., O_WRONLY will block until another process calls open() for the other end (e.g., O_RDONLY). This behavior can be modified with the O_NONBLOCK flag (though there are many details to consider).

Note: There is nothing FIFO-specific about unlink().

• Remember:
  • FIFOs have process persistence (so it remains in existence as long as one process has it open)
• Implications:
  • You MUST remember to unlink FIFOs
• Getting a List of FIFOs:
  • Depending on your naming conventions you may need to use ls -A

unixexamples/pipes/averager.c

        #include <fcntl.h>
#include <stdio.h>
#include <stdlib.h>
#include <sys/stat.h>
#include <unistd.h>

#define FIELD_WIDTH 11

int
main(int argc, char **argv)
{
  char  grade[FIELD_WIDTH+1]; // 7 for ID, 1 for comma, 3 for grade
  int   count, fd, n;
  float total;

  // Make the FIFO for the requested department 
  mkfifo(argv[1], S_IRUSR | S_IWUSR);

  // Open the read end of the FIFO (blocking by default so that this program
  // will wait until the splitter writes to the FIFO)
  fd = open(argv[1], O_RDONLY);

  count = 0;
  total = 0.0;
  
  // Read from the FIFO and update the GPA
  while ((n = read(fd, grade, FIELD_WIDTH)) == FIELD_WIDTH)
    {
      grade[FIELD_WIDTH] = '\0';
      count++;
      total += atof(grade+8);
    }

  printf("\nGPA in %s: %4.2f\n", argv[1], total/count);
  close(fd);
  unlink(argv[1]);
  
  return 0;
}
        

unixexamples/pipes/splitter.c

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


#define DEPARTMENTS  3
#define FIELD_WIDTH     11

int
index_of(char *needle, char *haystack)
{
  char *r = strstr(haystack, needle);
  if (r == NULL) return -1;
  else           return r - haystack;
}


int
main(int argc, char **argv)
{
  char  grade[FIELD_WIDTH+1]; // 7 for ID, 1 for comma, 3 for grade
  int   fd_data, i, n;
  int   fd_fifo[DEPARTMENTS];
  char  *depts[] = {"CS", "ENG", "MUS"};

  printf("Remember: First run averager for each department.\n");
  
  for (i=0; i<DEPARTMENTS; i++)
    {
      // Create the FIFO for department i
      mkfifo(depts[i], S_IRUSR | S_IWUSR);

      // Open the write end of the FIFO in non-blocking mode
      // so that the user can be get information only on those
      // departments she/he is interested in
      fd_fifo[i] = open(depts[i], O_WRONLY | O_NONBLOCK);
    }

  // Open the data file
  fd_data = open("grades.dat", O_RDONLY);

  // Read from the data file and write to the FIFOs
  while ((n = read(fd_data, grade, FIELD_WIDTH)) == FIELD_WIDTH)
    {
      grade[FIELD_WIDTH] = '\0';

      // Write to the appropriate FIFO
      for (i=0; i<DEPARTMENTS; i++)
        {
          if (index_of(depts[i], grade) == 0)
            {
              write(fd_fifo[i], grade, FIELD_WIDTH);
              break;
            }
        }
    }

  // Done reading from the file so close it.
  close(fd_data);
  
  // Close all of the file descriptors
  for (i=0; i<DEPARTMENTS; i++)
    {
      close(fd_fifo[i]);
      unlink(depts[i]);
    }

  return 0;
}
        

• Pipes:
  1. Parent builds a pipe
  2. Parent constructs a child process (which "inherits" the file descriptors for the pipe)
  3. The "receiving party" reads from the pipe (and, hence, blocks) when it wants to be "notified"
  4. The "sending party" closes the write-end to when it wants to "notify"
• FIFOs:
  • Can also make use of the fact that (by default) the calls to open() will block until both ends are open.

unixexamples/pipes/sequencef.c

        #include <fcntl.h>
#include <stdio.h>
#include <stdlib.h>
#include <sys/stat.h>
#include <sys/types.h>
#include <sys/wait.h>
#include <unistd.h>

int
main(void)
{
  char  c;
  int   fd, n, status;
  pid_t pid;

  
  pid    = fork();

  switch(pid)
    {
    case -1:
      exit(1);

    case 0:
      mkfifo(".continue", S_IRUSR);
      fd = open(".continue", O_RDONLY);  // Block until the parent is done
      close(fd);
      printf("Task 2\n");
      unlink(".continue");  // Unlink the FIFO
      exit(0);

    default:
      printf("Task 1\n");
      mkfifo(".continue", S_IWUSR);
      fd = open(".continue", O_WRONLY); // Let the child know the parent is done
      close(fd);
      wait(&status);        // Reap the child
      printf("Task 3\n");
      unlink(".continue");  // Unlink the FIFO
      exit(0);
    }
}
        

unixexamples/pipes/sequence.c

        #include <stdio.h>
#include <stdlib.h>
#include <sys/types.h>
#include <sys/wait.h>
#include <unistd.h>
#include "../signals/tax_lib.h"

int
main(void)
{
  char  c;
  int   fd_pipe[2];
  int   n, status;
  pid_t pid;

  // Construct the pipe
  pipe(fd_pipe);
  
  pid    = fork();

  switch(pid)
    {
    case -1:
      exit(1);

    case 0:
      // The child is "receiving" so must close the write end
      close(fd_pipe[1]);
      
      read(fd_pipe[0], &c, 1);    // Block until the parent is done
      printf("Task 2\n");

      // Close the read end
      close(fd_pipe[0]);
      exit(0);

    default:
      // The parent is "sending" so must close the read end
      close(fd_pipe[0]);

      printf("Task 1\n");
      close(fd_pipe[1]); // Let the child know I'm done

      wait(&status); // Reap the child
      printf("Task 3\n");

      exit(0);
    }
}
        

unixexamples/pipes/taxp1.c

        #include <stdio.h>
#include <stdlib.h>
#include <sys/types.h>
#include <sys/wait.h>
#include <unistd.h>
#include "../signals/tax_lib.h"

int
main(void)
{
  char  should;
  float rate, sales, tax;
  int   fd_pipe[2];
  int   n, status;
  pid_t pid;

  // Construct the pipe to be used for synchronization
  pipe(fd_pipe);
  
  rate   = 0.05;
  pid    = fork();

  switch(pid)
    {
    case -1:
      exit(1);

    case 0:
      // The child is writing to the pipe so must close the read end
      close(fd_pipe[0]);
      
      if (should_tax()) write(fd_pipe[1], "?", 1); // "Signal"
      close(fd_pipe[1]);                           // Don't "Signal"

      record_status(); // This takes a long time

      exit(0);

    default:
      // The parent is reading from the pipe so must close the write end
      close(fd_pipe[1]);

      sales = total_sales();
      n = read(fd_pipe[0], &should, 1); // Blocks until the child is done
      if (n == 1) tax = sales * rate;   // The child "signaled"
      else        tax = 0.0;            // The child didn't "signal"

      // Close the read end
      close(fd_pipe[0]);

      printf("Tax: %5.2f\n", tax);
      wait(&status); // Reap the child
      exit(0);
    }
}
        

Note: The "sender" transmits one bit of information (as in the similar example that used signals) by either writing or not. This approach is more subtle than necessary since the pipe can be used to transmit information more directly.

unixexamples/pipes/taxp2.c

        #include <stdio.h>
#include <stdlib.h>
#include <sys/types.h>
#include <sys/wait.h>
#include <unistd.h>
#include "../signals/tax_lib.h"

int
main(void)
{
  char  should;
  float rate, sales, tax;
  int   fd_pipe[2];
  int   n, status;
  pid_t pid;

  // Construct the pipe to be used for communication and synchronization
  pipe(fd_pipe);
  
  rate   = 0.05;
  pid    = fork();

  switch(pid)
    {
    case -1:
      exit(1);

    case 0:
      // The child is writing to the pipe so must close the read end
      close(fd_pipe[0]);
      
      if (should_tax()) write(fd_pipe[1], "1", 1); // Send "should tax"
      else              write(fd_pipe[1], "0", 1); // Send "shouldn't tax"

      // Close the write end
      close(fd_pipe[1]);

      record_status(); // This takes a long time

      exit(0);

    default:
      // The parent is reading from the pipe so must close the write end
      close(fd_pipe[1]);

      sales = total_sales();
      n = read(fd_pipe[0], &should, 1); // Blocks until the child is done

      if (should == '1') tax = sales * rate;
      else               tax = 0.0;

      // Close the read end
      close(fd_pipe[0]);

      printf("Tax: %5.2f\n", tax);
      wait(&status); // Reap the child
      exit(0);
    }
}
        

• Recall:
  • There are two parties, Andrea and Bill, each of whom performs two tasks in order, \((A1, A2)\) for Andrea and \((B1, B2)\) for Bill
  • The protocol must ensure that \(A1\) happens before \(B2\) and \(B1\) happens before \(A2\)
• Using Pipes/FIFOs:
  • Use one pipe/FIFO for each sequencing that must be ensured

unixexamples/pipes/rendezvous.c

        #include <stdio.h>
#include <stdlib.h>
#include <sys/types.h>
#include <sys/wait.h>
#include <unistd.h>
#include "../signals/tax_lib.h"

int
main(void)
{
  char  c;
  int   c2p_pipe[2], p2c_pipe[2];
  int   n, status;
  pid_t pid;

  // Construct the pipes
  pipe(c2p_pipe);
  pipe(p2c_pipe);
  
  pid    = fork();

  switch(pid)
    {
    case -1:
      exit(1);

    case 0:
      // The child is writing to c2p_pipe so must close the read end
      close(c2p_pipe[0]);
      // The child is reading from p2c_pipe so must close the write end
      close(p2c_pipe[1]);
      
      printf("A1\n");
      write(c2p_pipe[1], "?", 1); // Let the parent know A1 is done

      read(p2c_pipe[0], &c, 1);   // Block until B1 is done
      printf("A2\n");

      // Close the other ends
      close(c2p_pipe[1]);
      close(p2c_pipe[0]);

      exit(0);

    default:
      // The parent is reading from c2p_pipe so must close the write end
      close(c2p_pipe[1]);
      // The parent is writing to p2c_pipe so must close the read end
      close(p2c_pipe[0]);

      printf("B1\n");
      write(p2c_pipe[1], "?", 1); // Let the child know B1 is done

      read(c2p_pipe[0], &c, 1);   // Block until A1 is done
      printf("B2\n");

      // Close the other ends
      close(c2p_pipe[0]);
      close(p2c_pipe[1]);

      wait(&status); // Reap the child
      exit(0);
    }
}
        

unixexamples/pipes/rendezvous_toostrict.c

        #include <stdio.h>
#include <stdlib.h>
#include <sys/types.h>
#include <sys/wait.h>
#include <unistd.h>
#include "../signals/tax_lib.h"

int
main(void)
{
  char  c;
  int   c2p_pipe[2], p2c_pipe[2];
  int   n, status;
  pid_t pid;

  // Construct the pipes
  pipe(c2p_pipe);
  pipe(p2c_pipe);
  
  pid    = fork();

  switch(pid)
    {
    case -1:
      exit(1);

    case 0:
      // The child is writing to c2p_pipe so must close the read end
      close(c2p_pipe[0]);
      // The child is reading from p2c_pipe so must close the write end
      close(p2c_pipe[1]);
      
      read(p2c_pipe[0], &c, 1);   // Block until B1 is done
      printf("A1\n");
      write(c2p_pipe[1], "?", 1); // Let the parent know A1 is done
      printf("A2\n");

      // Close the other ends
      close(c2p_pipe[1]);
      close(p2c_pipe[0]);

      exit(0);

    default:
      // The parent is reading from c2p_pipe so must close the write end
      close(c2p_pipe[1]);
      // The parent is writing to p2c_pipe so must close the read end
      close(p2c_pipe[0]);

      printf("B1\n");
      write(p2c_pipe[1], "?", 1); // Let the child know B1 is done

      read(c2p_pipe[0], &c, 1);   // Block until the A1 is done
      printf("B2\n");

      // Close the other ends
      close(c2p_pipe[0]);
      close(p2c_pipe[1]);

      wait(&status); // Reap the child
      exit(0);
    }
}
        

Note: With this protocol B1 will always be completed before A1 (which is not required).