homework.c

这是研究生的并行计算的作业

/*Coding by wang yue ru.
*/
#include<stdio.h> 
#include<pthread.h>
#include <sys/time.h>
#include <errno.h>
int data_list[10000]; /*源数据数组*/
/*总处理数据个数--计算机个数--分组数据个数*/
int Ntotal,pcomputer,Numgroup;
/*采样数据数组*/
int sample_data[900];
/*主元数据数组*/
int key_data[30];
/*主元选好标志*/
int key_ok=0;
/*全局交换可以进行标志.各个计算机都主元选好*/
int exchange_ok=0;
/*数据由计算机回送主机*/
int databack=0;
/*回送好的数据长度*/
int databacklen=0;
/*this struct is used for the exchange of data*/
struct segdata
{
	int datalen;
	int segmentdata[30];/*段数据*/
}SegData[30][30] ; /*计算机序号*//*段序号*/

/**************************************************
*function: 
*           waiting signal for some time.
*input:
*	cond is the poiter of the struct of pthread_cond_t
*	milseconds is intervals of time in ms
*relative function :
*		pthread_cond_signal(&mycond);
*return:   
*          0 is ok    1 is timeout    else  is  -1
***************************************************/
int thread_wait(pthread_cond_t* cond,int milseconds)
{
	int retval=-1;
	int rv; 
	pthread_mutex_t mymutex = PTHREAD_MUTEX_INITIALIZER; 
	struct timespec ts; 
	struct timeval now;
	//get the current time point
	gettimeofday(&now,NULL);
	ts.tv_sec = (milseconds/1000); 
	milseconds=milseconds-1000*ts.tv_sec;
	ts.tv_nsec = milseconds*1000; /* 500,000 nanoseconds = 500 ms */ 

	ts.tv_sec= now.tv_sec + ts.tv_sec;
	ts.tv_nsec = now.tv_usec +ts.tv_nsec;

        pthread_mutex_lock(&mymutex); 
        rv = pthread_cond_timedwait(cond, &mymutex, &ts); 
        pthread_mutex_unlock(&mymutex); 

	return retval;
} 
void fast_sort(int startindex,int endindex,int *data){ 
	int a_mid=data[(startindex+endindex)/2]; /*中间元素作支撑点*/
	int low=startindex,high=endindex;
	/*向中间收缩, 交换数据*/
	do{ 
		while (data[low]<a_mid) low++; 
		while (a_mid<data[high]) high--; 
		if (low<=high){ 
			int temp; 
			temp=data[low]; 
			data[low]=data[high]; 
			data[high]=temp; 
			low++; 
			high--; 
		} 
	}while (low<=high); 
	/*分成左右两块和支撑点数据. 再递归调用*/
	if (low<endindex)
		fast_sort(low,endindex,data); 
	if (startindex<high)
		fast_sort(startindex,high,data); 
} 
/*归并排序*/
void Merge(int* indata, int leftindex, int midindex, int rightindex)
{/*把indata[leftindex:midindex]] 和indata[midindex:rightindex] 归并到outdata[ leftindex :rightindex ]*/

	int 	leftsegcursor = leftindex, /*第一段的游标*/
		rightsegsursor = midindex+1,/*第二段的游标*/
		resultsursor = leftindex, /*结果的游标*/
		index;
	int outdata[1000];
	/*只要在段中存在leftsegcursor 和rightsegsursor，则不断进行归并
		两个数组中的元素谁小先排谁*/
	while ((leftsegcursor <= midindex) && (rightsegsursor <= rightindex)){
		if (indata[leftsegcursor] <= indata[rightsegsursor]) 
			outdata[resultsursor++] = indata[leftsegcursor++];
		else outdata[resultsursor++] = indata[rightsegsursor++];
		}
	/* 考虑余下的部分*/
	if (leftsegcursor > midindex){ 
		for (index = rightsegsursor; index <= rightindex; index++)
			outdata[resultsursor++] = indata[index];
		}
	else {
		for (index =leftsegcursor; index <= midindex; index++)
			outdata[resultsursor++] = indata[index];
		}
	for(index=leftindex;index<=rightindex;index++)
		indata[index]=outdata[index];
}

/*计算机线程:代表一台计算机处理分来的数据.*/
void computerThread(const int * groupindex)
{
	int i=0,j=0;
	int index=0;
	int local_result[1000];/*交换后的数据*/
	int merge[30];/*归并前各段的长度*/
	int local_result_len=0;/*交换后的数据长度*/
	int computerindex=*groupindex;
	int startindex=computerindex*Numgroup;
	int endindex=(computerindex+1)*Numgroup-1;
	int samplestep=Numgroup/pcomputer;/*抽样步长*/
	int step=0;/*处理步骤*/
	for(i=0;i<30;i++)
		merge[i]=0;
	printf("\nSorted data of the %dth computer :  ",computerindex);
	while(1){
		if((key_ok==1)&&(step==5))
			step=2;
		if(step==0){/*局部排序*/
			fast_sort(startindex,endindex,data_list);
			for(index=startindex;index<=endindex;index++)
				printf("%d ",data_list[index]); 
			step++;
		}
		if(step==1){/*采样*/
			for(index=startindex,i=computerindex*pcomputer;
					((index<endindex)&&(i<(computerindex+1)*pcomputer));
					index+=samplestep,i++){
				sample_data[i]=data_list[index];
				}
			step=5;
		}
		if(step==2){/*用主元划分数据*/
			int temp=startindex;
			for(i=0;i<pcomputer;i++){
				if(i==pcomputer-1)
					SegData[computerindex][i].datalen=endindex-temp+1;				
					j=0;				
					for(index=temp;index<=endindex;index++){
					if(i==pcomputer-1){
						SegData[computerindex][i].segmentdata[j]=data_list[index];
						j++;
						}
					else
						if(data_list[index]<=key_data[i]){
							SegData[computerindex][i].segmentdata[j]=data_list[index];
							j++;
						}
						else{
							SegData[computerindex][i].datalen=j;
							temp=index;
							break;
						}
				}
			}
			step++;
			exchange_ok++;
		}
		if((step==3)&&(exchange_ok==pcomputer)){
			/*全局交换数据取所有计算机上和本机号相同的段*/
			j=0;
			for(index=0;index<pcomputer;index++){
				for(i=0;i<SegData[index][computerindex].datalen;i++){
					local_result[j]=SegData[index][computerindex].segmentdata[i]; 
					j++;
				}
				merge[index]=SegData[index][computerindex].datalen;
			}
			local_result_len=j;
			printf("\nAfter global exchange %dth computer:",computerindex);
			for(i=0;i<local_result_len;i++)printf("%d ",local_result[i]);
			step=4;
		}
		if(step==4){/*归并排序*/
			int mid=0,right=0;
			for(index=0;index<pcomputer;index++){
			if(merge[index]!=0){
				mid=right;
				right+=merge[index];
				if(mid>0)
					Merge(local_result,0,mid-1,right-1);
				}
			}
			printf("\nMerged data %dth computer:",computerindex);
			for(i=0;i<local_result_len;i++)printf("%d ",local_result[i]);
			printf(" \n");
			step=6;
		}
		if(step==6){/*将数据回送出计算机*/
			if(databack==computerindex){
				for(i=0;i<local_result_len;i++)
					data_list[databacklen+i]=local_result[i];
				databack++;
				databacklen+=local_result_len;
			}
		}
	}
}

int init_data()
{
	int ret;
	pthread_t idthread[30];
	pthread_cond_t   mycond = PTHREAD_COND_INITIALIZER;
	int index; 
	for(index=0;index<10000;index++)data_list[index]=0;
	
	printf("\nPlease input the number of sorted data N(<100).\nN=");
	scanf("%d",&Ntotal);
	printf("Please input the number of computers p which can divide N exactly.\np=");
	scanf("%d",&pcomputer);
	Numgroup=Ntotal/pcomputer;
	if(Numgroup>=100||Numgroup<pcomputer||pcomputer>30)return 1;	
	/*读数据源文件*/
	FILE *fin=fopen("sourcedata.txt","r"); 
	for(index=0;index<Ntotal;index++)
		fscanf(fin,"%d",&data_list[index]); 
	fclose(fin); 	
	thread_wait(&mycond,1000);	
	
	for(index=0;index<pcomputer;index++){
		/*启动p 个线程*/
		ret=pthread_create(&idthread[index],NULL,(void *)computerThread,(void*)&index);
		if(ret!=0)
		{
			printf("Create computer  Thread error\n");
			return ret;
		}	
		thread_wait(&mycond,1000);/*让一个线程启动好后再起另一个*/	
	}
	return 0;
}

int main() { 
	pthread_cond_t   mycond = PTHREAD_COND_INITIALIZER;
	int index,i; 
 
	if(init_data())
		return 0;
	
	thread_wait(&mycond,2000);
	/*打印各种数据*/
	printf("\nThe sampling data :  ");
	for(index=0;index<pcomputer*pcomputer;index++)
		printf("%d ",sample_data[index]); 
	
	printf("\nThe sorted sampling data : ");
	fast_sort(0,pcomputer*pcomputer-1,sample_data);
	for(index=0;index<pcomputer*pcomputer;index++){
		if((index%pcomputer==0)&&(index/pcomputer!=0))
			key_data[index/pcomputer-1]=sample_data[index];
		printf("%d ",sample_data[index]); 
		}
	
	printf("\nThe key data : ");	
	for(index=0;index<pcomputer-1;index++)
		printf("%d ",key_data[index]); 
	
	key_ok=1;
	printf("\n");

	thread_wait(&mycond,6000);
	printf("\nGet from all computers data :\n");
	for(index=0;index<Ntotal;index++)
		printf("%d ",data_list[index]); 
	printf("\n");
		
}