File-Server (I)

A. First Edition

This is comp444 lab3 and it is concentrated on thread synchronization, FIFO etc.

B.The problem

1. file list:
a) source code:
	myhead.h	//general head file for includes
	errHandle.c	//universal error handler
	fprotocol.h	//all protocol-related definition, i.e. structure
	fclient.c	//client source
	fserver.c	//server source
	jobgen.c	//a job generator source
b) executable files:
	fclient.exe	//client, takes no parameter, must be run after fserver.exe already being running
	fserver.exe	//server, takes an optional parameter of length of job queue,default is 10
	jobgen.exe	//an optional automatic job generator which reads all files in current diretory and generate random file 
			//request
c) makefile:
	makefile

d) test script file:
	run.script	//a sample test script file, must run server first. 
			//i.e. fserver.exe [lengthofqueue]& 
			//run.script
2. how to compile:
	make

3. how to run:
a) run server: 
	./fserver.exe [lengthofqueue] & 
b) you have three choices to run client:
i) run client from command line: 
	./fclient.exe filename requesttype
ii) run client from script:
	./run.script
iii) run job generator:
	./jobgen.exe

4. feature of program:
a) I designed two conditional variables so that listening thread and all other processing threads are associated with two different 
	conditional variables. When the queue is empty, processing threads will only signal listening threads since the job queue is 
	already emptied. When the queue is full, the processing thread will only broadcast all processing threads. And another 
	potential usage of two conditional variables is that originally I designed to force the queue to be 80% full before all 
	processing threads can be notified. That is to say, when job queue is less than 80% full, only listening thread is working. By 
	doing this, I can possibly create a queue full scenario. Later, when I found "sleep" call won't block whole process, this 
	design is dropped.
b) I write an automatic testing program, jobgen.exe, which reads all file and directory name in current directory and "fork&exec" a 
	client.exe to run a random request. 
c) All read and write of both client and server are in "blocking" mode which is more efficient than "busy-waiting" in 
	"non-blocking" mode. In order to achieve this, it is necessary to open fifo in mode of "read&write". 

D.The major functions

The seems to me is that you must be very careful for the mode of opening a FIFO. As we know, "block" mode for 

reading has a higher efficiency than "non-block" mode reading since programmer has to polling FIFO in a busy-waiting

model. But in almost all books of Linux programming, it is said very clear: 

	i) Block-mode: If a FIFO is opened with either "read-only" or "write-only", then it is blocked when there 

	is correspondent reader or writer already at any side of FIFO. And later reading and writing is just 

	"blocked". (Do I have to point it out?)

	ii) Non-Block mode: If a FIFO is opened with "read-only" returns immediately and a open with "write-only"

	generates an error when no other reader exists. And the reading and writing is just "non-blocked". (Do I have

	to point it out?)

E.Further improvement

F.File listing

1. myhead.h

2. errHandle.c

3. fprotocol.h

4. fclient.c

5. fserver.c

6. jobgen.c

7. makefile

file name: myhead.h

#ifndef MYHEAD_H
#define MYHEAD_H
#include <sys/resource.h>
#include <sys/wait.h>
#include <sys/time.h>
#include <sys/types.h>
#include <sys/stat.h>
#include <stdio.h>
#include <stdlib.h>
#include <errno.h>
#include <dirent.h>
#include <unistd.h>
#include <string.h>
#include <fcntl.h>
#include <signal.h>

extern int errno;

//define three mode and it is purely like
// DFA=Deterministic Finite Automaton
#define FindNext        0
#define Comparing       1

typedef int bool;

#define TRUE  1
#define FALSE 0

//for a particular file NAME, NOT path
//const int MaxNameLength=20;
#define MaxPathLength 128

#define MaxNameLength  20 

#define MaxIntLength   20

#define BuffSize   128

#define EmptyGridNode   'E' //the empty tic-tac-toe grid

#define PlayerTypeLength 2

#define PlayerPidLength (MaxIntLength+1)

#define BoardLength 9 //the tic-tac-toe length

#define PlayerX_Pos PlayerTypeLength

#define PlayerO_Pos (PlayerTypeLength*2+PlayerPidLength)

#define BoardPos ((PlayerTypeLength+PlayerPidLength)*2+1)

#define WinnerPos (BoardPos+BoardLength+1)

#define WinnerTypeLength 1 //simply the X or O

#define SleepSeconds 1

#define FIFOMode S_IRUSR|S_IWUSR|S_IXUSR|S_IRGRP|S_IWGRP|S_IXGRP

void errHandle(char* msg);
	
#endif

file name: errHandle.c

#include "myhead.h"

extern int errno;

void errHandle(char* msg)
{
	if (msg!=NULL)
	{
		perror(msg);
	}
	else
	{
		strerror(errno);
	}	
	exit(errno);	
}
	

file name: fprotocol.h

#ifndef FPROTOCOL_H
#define FPROTOCOL_H
#include <sys/types.h>

typedef enum
{
	FILE_TYPE      =   1,
	FILE_INODE     =   2,
	FILE_SIZE      =   3,
	FILE_NOT_FOUND =  -1,
	UNKNOWN_ERROR  =  -2,
	QUEUE_FULL     =  -3
}RequestType;		


/*
#define FILE_TYPE       1
#define FILE_INODE      2
#define FILE_SIZE       3
#define FILE_NOT_FOUND -1
#define UNKNOWN_ERROR  -2
#define QUEUE_FULL     -3
*/

typedef enum
{
	Regular    =    1,
	Directory  =    2,
	SoftLink   =    3,
	Character  =    4,
	Block      =    5, 
	FIFO       =    6, 
	Socket     =    7
}FileType; 


typedef struct
{
	pid_t pid;
	RequestType fileOp;
	int result;
}ResponseRec;

/*
//file type
#define Regular        1
#define Directory      2
#define SoftLink       3
#define Character      4
#define Block          5
#define FIFO           6
#define Socket         7
*/

#define MaxRequestFileNameLength 64


/*
typedef struct 
{
	pid_t pid;
	RequestType fileOp;
	char fileName[MaxRequestFileNameLength];
}RequestRec;
*/



/*
#define ProcessIDLength          6
#define FileOperationLength      2
#define RequestRecordLength      MaxRequestFileNameLength+ProcessIDLength+FileOperationLength
*/
                        
typedef struct 
{
        pid_t pid;
        int fileOp;
        char fileName[MaxRequestFileNameLength];
}Job_t;

//typedef struct Job Job_t;

#define RequestRecordLength sizeof(Job_t)

char* serverFifoName="fserver.fifo";

#define DefaultThreadNumber 10


#endif

	

		
file name: fclient.c

#include "myhead.h"
#include "fprotocol.h"

extern char* serverFifoName;

int main(int argc, char* argv[])
{
	int fdwrite, fdread, result;
	char pidBuf[MaxIntLength+7];
	Job_t requestRec;
	char* queueFullMsg="The job queue is full\n";
	char readBuf[MaxIntLength];
	if (argc!=3)
	{
		errHandle("usage fclient.exe [filename] [fileop]\n");
	}
	if (!(argv[2][0]>='0'&&argv[2][0]<='9'))
	{
		errHandle("the file operation must be a number between 0-9");
	}
	requestRec.pid=getpid();
	requestRec.fileOp=atoi(argv[2]);
	strcpy(requestRec.fileName, argv[1]);
	sprintf(pidBuf, "%d.fifo", getpid());
	if (mkfifo(pidBuf, FIFOMode)<0)
	{
		errHandle("create fifo at client fails");
	}
	if ((fdwrite=open(serverFifoName, O_RDWR))<0)
	{
		errHandle("open fifo error");
	}
	if (write(fdwrite, &requestRec, RequestRecordLength)!=RequestRecordLength)
	{
		errHandle("write error for fifo");
	}
	if ((fdread=open(pidBuf, O_RDWR))<0)
	{
		errHandle("open fifo for read error");
	}
	//open the result file first
	sprintf(pidBuf, "%d.result", getpid());
	if ((fdwrite=open(pidBuf, O_WRONLY|O_CREAT, FIFOMode))<0)
	{
		errHandle("create result file error");
	}

	while (1)
	{
		//printf("client %d begin reading\n", getpid());
		if (read(fdread, &result, sizeof(int))!=sizeof(int))
		{
			errHandle("error of read fifo");
		}
		else
		{
			if (result!=QUEUE_FULL)//if it is, then need to be patient
			{
				break;
			}
			else
			{
				if (write(fdwrite, queueFullMsg, strlen(queueFullMsg))!=strlen(queueFullMsg))
				{
					errHandle("write error of client\n");
				}
			}
		}
	}
	//read successfuland close fifo
	if (close(fdread)<0)
	{
		errHandle("close fifo error");
	}
	sprintf(readBuf, "%d", result);
	if (write(fdwrite, readBuf, strlen(readBuf))!=strlen(readBuf))
	{
		errHandle("write result error");
	}
	return 0;
}
			

file name: fserver.c

#include "myhead.h"
#include "fprotocol.h"
#include <pthread.h>

Job_t* jobs;

extern char* serverFifoName;

int fdServer;
int jobCount=0;
int workerCount=0;
int currentJob=0;
int threadNumber;
Job_t requestRec;
ResponseRec responseRec;
//int fifoMode=S_IRUSR|S_IWUSR|S_IXUSR|S_IRGRP|S_IWGRP|S_IXGRP;
pthread_t* threads;
//char* serverFifoName="fserver.fifo";
//DynaArrayPtr arrayPtr;

pthread_cond_t jobCond=PTHREAD_COND_INITIALIZER;
pthread_cond_t controlCond=PTHREAD_COND_INITIALIZER;
pthread_mutex_t mutex=PTHREAD_MUTEX_INITIALIZER;

void createQueue();
void initialize(int argc, char* argv[]);
void uninitialize();
void createThreads();
void createFifo(char* fifoName);
void writeFifo();

//the child do job
void doJob();
//the control thread is main thread
void doControl();
void broadcast();

//the thread function
void* jobFunc(void* arg);
void controlFunc();

int main(int argc, char* argv[])
{
	initialize(argc, argv);
	controlFunc();
	return 0;
}

void controlFunc()
{
	int n;
	while (1)
	{
		//reading here are BLOCKING
		if ((n=read(fdServer, &requestRec, RequestRecordLength))!=RequestRecordLength)
		{
			if (n==0)
			{
				//sleep(SleepSeconds);
			}
			else
			{
				//printf("\ncontrol thread read %d\n", n);
				errHandle("read pipe in control thread error");
			}
		}
		//printf("control read pid=%d\n", requestRec.pid);
		if (pthread_mutex_lock(&mutex))
		{
			errHandle("mutex lock error");
		}
		//if the queue is full
		while (jobCount==threadNumber-1)
		{
			//printf("job queue is full now\n");
			//sprintf(queueFullBuf, "%d", QUEUE_FULL);
			//buf always has processID at beginning and terminated with NULL
			//sprintf(pidBuf,"%d.fifo",requestRec.pid);
			//strcat(pidBuf, ".fifo");
			responseRec.pid=requestRec.pid;
			responseRec.fileOp=QUEUE_FULL;
			responseRec.result=QUEUE_FULL;
			//printf("prepare to write to fifo of pid=%d\n", responseRec.pid);
			writeFifo();
			//printf("write job queue full msg\n");
			if (pthread_cond_broadcast(&jobCond))
			{
				errHandle("broadcast error\n");
			}
			if (pthread_cond_wait(&controlCond, &mutex))
			{
				errHandle("condition wait error");
			}
		}
		doControl();
		broadcast();
		if (pthread_mutex_unlock(&mutex))
		{
			errHandle("mutex unlock error");
		}
		//broadcast();
	}
}

//this is a call that both listening thread and processing thread would call
//it favours notifying processing thread until queue is 80% full
void broadcast()
{
	//always favour control thread
	if (pthread_cond_broadcast(&controlCond))
	{
		errHandle("condition broadcast error\n");
	}
	//printf("jobcontrol broadcasted\n");
	//to make job count become 80% of capacity
	//if (jobCount>8*threadNumber/10)
	{		
		if (pthread_cond_broadcast(&jobCond))
		{
			errHandle("condition broadcast error\n");
		}
	}
}

					
void doJob()
{
	//char pidBuf[MaxIntLength];//20
	//char resultBuf[MaxIntLength];
	int result;
	struct stat statBuf;
	//workerCount++;
	if (stat(jobs[currentJob].fileName, &statBuf)<0)
	{
		if (errno==ENOENT)
		{
			result=FILE_NOT_FOUND;
		}
		else
		{
			errHandle("stat error");
		}
	}
	switch(jobs[currentJob].fileOp)
	{
	case FILE_TYPE:
		result=UNKNOWN_ERROR;
		if (S_ISREG(statBuf.st_mode))
		{
			result=Regular;
		}
		if (S_ISDIR(statBuf.st_mode))
		{
			result=Directory;
		}
		if (S_ISLNK(statBuf.st_mode))
		{
			result=SoftLink;
		}
		if (S_ISCHR(statBuf.st_mode))
		{
			result=Character;
		}
		if (S_ISBLK(statBuf.st_mode))
		{
			result=Block;
		}
		if (S_ISFIFO(statBuf.st_mode))
		{
			result=FIFO;
		}
		if (S_ISSOCK(statBuf.st_mode))
		{
			result=Socket;
		}
		break;
	case FILE_INODE:
		result=statBuf.st_ino;
		break;
	case FILE_SIZE:
		result=statBuf.st_size;
		break;
	default:
		result=UNKNOWN_ERROR;
		break;
	}
	responseRec.pid=jobs[currentJob].pid;
	responseRec.fileOp=jobs[currentJob].fileOp;
	responseRec.result=result;
	//printf("\nhere child read job index=%d", currentJob);
	//printf("\npid=%d, fileop=%d, filename=%s\n", jobs[currentJob].pid, jobs[currentJob].fileOp, jobs[currentJob].fileName);
	//sprintf(pidBuf, "%d.fifo", jobs[currentJob].pid);
	//sprintf(resultBuf, "%d", result);
	//it is client's responsibility to create fifo
	writeFifo();
	currentJob=(currentJob+1)%threadNumber;
	sleep(SleepSeconds);
	jobCount--;					
}

void writeFifo()
{
	int fd;
	//char intBuf[MaxIntLength];
	char fifoName[MaxIntLength+5];
	sprintf(fifoName, "%d.fifo", responseRec.pid);
	if ((fd=open(fifoName, O_WRONLY))<0)
	{
		//printf("\nthe fifonoame is  %s\n", fifoName);
		//write(STDOUT_FILENO, fifoName, strlen(fifoName));
		errHandle("open fifo for write error");
	}
	//sprintf(intBuf, "%d", responseRec.result);
	if (write(fd, &responseRec.result, sizeof(int))!=sizeof(int))
	{
		errHandle("write fifo error");
	}
	close(fd);
}

//assume processID is only 5 digits terminated with NULL
//assume fileoperation is only 1digit terminated with NULL
void doControl()
{
	int index;
	//int pid, requestType;
	//pid=atoi(buf);
	//requestType=atoi(buf+ProcessIDLength+1);
	index=(currentJob+jobCount)%threadNumber;
	jobs[index]=requestRec;//should be struct copy??
	//strcpy(jobs[index].fileName, buf+ProcessIDLength+FileOperationLength+1);
	jobCount++;
	//printf("get a job with pid=%d, fileop=%d, filename=%s\n", pid, requestType, jobs[index].fileName);
}

void* jobFunc(void* arg)
{
	while (1)
	{
		if (pthread_mutex_lock(&mutex))
		{
			errHandle("mutex locking error\n");
		}
		//waiting for condition variable
		while (jobCount==0)
		{
			if (pthread_cond_broadcast(&controlCond))
			{
				errHandle("broadcast error\n");
			}
			if (pthread_cond_wait(&jobCond, &mutex))
			{
				errHandle("condition wait error\n");
			}
		}
		doJob();
		//printf("finished one job\n");
		//workerCount--;
		broadcast();
		if (pthread_mutex_unlock(&mutex))
		{
			errHandle("mutex unlock error\n");
		}
		//broadcast();
	}
}

void initialize(int argc, char* argv[])
{
	if (argc==1)
	{
		threadNumber=DefaultThreadNumber;
	}
	else
	{
		if (argc==2)
		{
			threadNumber=atoi(argv[1]);
			if (threadNumber<=0)
			{
				errHandle("usage: fserver.exe [threadnumber]\n");
			}
	
		}
		else
		{
			errHandle("usage: fserver.exe [threadnumber]\n");
		}
	}
	createThreads();
	createQueue();
	createFifo(serverFifoName);
	//arrayPtr=createArray();//create the dynamic array for integer;
	if ((fdServer=open(serverFifoName, O_RDWR))<0)
	{
		errHandle("open server fifo error");
	}
	
}

void createQueue()
{
	if ((jobs=(Job_t*)malloc(threadNumber*sizeof(Job_t)))==NULL)
	{
		errHandle("create job queue error\n");
	}
	jobCount=0;//initialize to 0;
}


void createFifo(char* fifoName)
{
	if (mkfifo(fifoName, FIFOMode)<0)
	{
		if (errno!=EEXIST)	
		{
			errHandle("create fifo error\n");
		}
	}
}


void createThreads()
{
	int i;
	if ((threads=(pthread_t*)malloc(threadNumber*sizeof(pthread_t)))==NULL)
	{
		errHandle("cannot malloc pthreads array\n");
	}
	for (i=0; i<threadNumber; i++)
	{
		if (pthread_create(&threads[i], NULL, jobFunc,(void*)i))
		{
			errHandle("create threads error\n");
		}
	}
}

file name: jobgen.c

#include "myhead.h"

extern int errno;

void doJob(char* path);

int main(int argc, char* argv[])
{
	DIR* dp;
	struct dirent* dnode;
	srand(time(0));
	if ((dp=opendir("."))==NULL)
	{
		errHandle("open dir error\n");
	}
	errno=0;	
	while ((dnode=readdir(dp))!=NULL)
	{
		doJob(dnode->d_name);		
	}
	if (errno!=0)
	{
		errHandle("errno!=0 means read dir error\n");
	}
	
	return 0;
}
	

void doJob(char* path)
{
	char option[2];
	option[0]='1'+rand()%3;
	option[1]='\0';
	if (fork()==0)
	{
		printf("./fclient.exe %s %s\n", path, option);
		if (execl("./fclient.exe", "fclient.exe", path, option, NULL)<0)
		{
			errHandle("exec error");
		}
	}
}
		

file name: makefile

all:    fserver.exe fclient.exe jobgen.exe
	@echo "make complete for fserver.exe, fclient.exe, jobgen.exe"
	cp ./fserver.exe ./rundir/
	cp ./fclient.exe ./rundir/
	cp ./jobgen.exe ./rundir/
	rm -f ./rundir/*.fifo
	rm -f ./rundir/*.result
errHandle.o : errHandle.c myhead.h
	@echo "compiling errHanle module..."
	gcc -g -c errHandle.c -o errHandle.o

jobgen.exe : jobgen.c errHandle.o myhead.h
	@echo "compiling jobgen.exe..."
	gcc -g jobgen.c errHandle.o -o jobgen.exe

fclient.exe : fclient.c errHandle.o myhead.h fprotocol.h
	@echo "compiling fclient.exe..." 
	gcc -g fclient.c errHandle.o -o fclient.exe

fserver.exe : fserver.c myhead.h errHandle.o fprotocol.h
	@echo "compiling fserver.exe..."
	gcc -g -lpthread fserver.c errHandle.o -o fserver.exe
clear :
	@echo "clear up...\n"
	rm *.o *.exe

How to run?

a) run server:
./fserver.exe [lengthofqueue] &
b) you have three choices to run client:
i) run client from command line:
./fclient.exe filename requesttype
ii) run client from script:
./run.script
iii) run job generator:
./jobgen.exe