// 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;		// Number 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
	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;
}