pipe

// 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;	for(int i = 0; i < sizeof(s); i++){		if(c == s[i]) 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{			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) ;		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){		wait(0);	}else{		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){		wait(0);	}else{		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;}