Posted: Mon Nov 01, 2004 11:39 pm
Program Three: Pipe
Usage: pipe "producer [argument(s)]" "consumer [argument(s)]"
pipe feeds the output of the producer program to the consumer. It creates a buffer between the two programs, copying the producer's output to the buffer and then copying the contents of the buffer to the consumer's input.
Example
$ pipe "who" "grep arf"
arf pts/0 Oct 15 08:09
Submission Instructions
In order for your program to be graded please follow these instructions:
* Submit your program source as a text attachment in an email to the TA. Put the following text in the subject line: "CSU480 Program 3"
* Begin your program source with a comment that includes the following lines:
Name: your name
Compile: the command you used to compile your program
Collaborators: the names of people you worked with, if any
Implementation Details
* Create argv lists for producer and consumer.
* Spawn new processes for the producer and consumer using fork and exec.
* Redirect the standard output of the producer and the standard input of the consumer using freopen.
* Create a reader thread that reads the standard output of the producer and writes to a shared buffer.
* Create a writer thread that reads the shared buffer and writes to the standard input of the consumer.
* Have the reader thread exit when it detects end-of-file.
* Have the writer thread exit after it writes the last character from the buffer.
* Use the pthread_create to create the threads. Use pthread_join to wait for the threads to complete.
* Use sem_wait and sem_post to synchronize the threads. Initialize the semaphores using sem_init.
Please use the following C++ code as a starting point:
// Pipe -- Create a pipe between two programs
// Usage: pipe "producer [args]" "consumer [args]"
// To compile: g++ pipe.cc -lposix4 -lpthread
#include <stdio.h>
#include <string.h>
#include <malloc.h>
#include <pthread.h>
#include <semaphore.h>
#include <sys/stat.h>
#include <sys/wait.h>
#include <unistd.h>
const char *ReadFrom = "readfrom.fifo";
const char *WriteTo = "writeto.fifo";
const int BufSize = 16;
char buffer[BufSize]; // Shared buffer between producer and consumer
int count=0; // Count of chars in buffer
int in=0; // Index in buffer of next place to write
int out=0; // Index in buffer of next place to read
bool endOfFile=false; // true when producer is done
// Semaphores
sem_t empty; // Number of empty slots in buffer
sem_t full; // Number of filled slots in buffer
sem_t mutex; // Gate for critical section
static void closeFifos()
{
unlink(ReadFrom);
unlink(WriteTo);
}
static bool openFifos()
{
closeFifos(); // Remove FIFOs from previous execution
if( mkfifo(ReadFrom,0666)!=0 ) {
perror(ReadFrom);
return false;
}
if( mkfifo(WriteTo,0666)!=0 ) {
perror(WriteTo);
return false;
}
return true;
}
static int countOf(char c, char *s)
// Return number of occurences of c in s
{
// Walk through s counting instances of c
// YOUR CODE GOES HERE
}
static char **prepArgv(char *command)
// Return an argv list by separating the tokens in command
{
char **argv = (char **)calloc(
countOf(' ',command)+2, sizeof(char *) );
if( argv==0 ) return 0;
int i=0;
char *arg = strtok(command," ");
for(; arg!=0; arg=strtok(0," ") ) {
argv[i++] = arg;
}
argv[i] = 0;
return argv;
}
static void *readFromProducer(void *)
{
FILE *fp = fopen(ReadFrom,"r");
if( fp==0 ) {
perror(ReadFrom);
endOfFile = true;
return 0;
}
// Read chars from fp into buffer
// YOUR CODE GOES HERE
fclose(fp);
return 0;
}
static void *writeToConsumer(void *)
{
FILE *fp = fopen(WriteTo,"w");
if( fp==0 ) {
perror(WriteTo);
return 0;
}
// Write chars in buffer to fp
// YOUR CODE GOES HERE
fclose(fp);
return 0;
}
int main(int argc,char **argv)
{
// Check for proper usage
if( argc<3 ) {
fprintf(stderr,"Usage: pipe \"producer [args]\" \"consumer [args]\"\n");
return 1;
}
if( !openFifos() ) return 1;
// Initialize semaphores
// YOUR CODE GOES HERE
// Setup and spawn Producer
// YOUR CODE GOES HERE
// Start reader thread
// YOUR CODE GOES HERE
// Setup and spawn Consumer
// YOUR CODE GOES HERE
// Start writer thread
// YOUR CODE GOES HERE
// Wait for threads to exit
// YOUR CODE GOES HERE
closeFifos();
return 0;
}
DUE: November 8th
:o