Anonymous

// 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 consumerint count=0;            // Number of chars in bufferint in=0;               // Index in buffer of next place to writeint out=0;              // Index in buffer of next place to readbool endOfFile=false;   // true when producer is done// Semaphoressem_t empty;            // Number of empty slots in buffersem_t full;             // Number of filled slots in buffersem_t mutex;            // Gate for critical sectionstatic 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        int k;        k = 0;        while(*s){                if(*s++ == c) k++;        }        return k;}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        char c;        while(1){                c = fgetc(fp);                sem_wait(&empty);                sem_wait(&mutex);                while(count == BufSize) ;                if(c != EOF){                        buffer[in] = c;                }else{                        endOfFile = true;                        sem_post(&mutex);                        sem_post(&full);			fclose(fp);                        return 0;                }                in = (in + 1) % BufSize;                count++;                sem_post(&mutex);                sem_post(&full);        }        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        while(1){                sem_wait(&full);                sem_wait(&mutex);                while (count == 0 && !endOfFile) ;		if(endOfFile && count == 0){                        sem_post(&mutex);                        sem_post(&empty);			fclose(fp);                        return 0;                }                fputc(buffer[out], fp);                out = (out + 1) % BufSize;                count--;                sem_post(&mutex);                sem_post(&empty);        }        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        int sem_err;        sem_err = sem_init(&empty, 0, BufSize);        if(sem_err < 0){                printf("Error initializing semaphore.");                return 83;        }        sem_err = sem_init(&full, 0, 0);        if(sem_err < 0){                printf("Error initializing semaphoe.");                return 83;        }        sem_err = sem_init(&mutex, 0, 1);        if(sem_err < 0){                printf("Error intializing semaphore.");                return 83;        }        // Setup and spawn Producer        int p_pid;        p_pid = fork();        if(p_pid == 0){                freopen(ReadFrom, "w", stdout);                char **p_argv;                p_argv  = prepArgv(argv[1]);                execvp(p_argv[0], p_argv);                fclose(stdout);                _exit(0);        }else if(p_pid < 0){                printf("Couldn't fork a child");                return 80;        }        // Start reader thread        pthread_t reader;        int r_tid;        r_tid = pthread_create(&reader, NULL, readFromProducer, NULL);        if(r_tid != 0){                printf("Error creating reader thread.");                return 84;        }        // Setup and spawn Consumer        int c_pid;        c_pid = fork();        if(c_pid == 0){                freopen(WriteTo, "r", stdin);                char **c_argv;                c_argv = prepArgv(argv[2]);                execvp(c_argv[0], c_argv);                fclose(stdin);                _exit(0);        }else if(c_pid < 0){                printf("Couldn't fork a child");                return 80;        }        // Start writer thread        pthread_t writer;        int w_tid;        w_tid = pthread_create(&writer, NULL, writeToConsumer, NULL);        if(w_tid != 0){                printf("Error creating writer thread.");                return 84;        }        // Wait for threads to exit        pthread_join(reader, NULL);        pthread_join(writer, NULL);        closeFifos();        return 0;}