out_sort.c

外部排序程序

	  * outstu为缓冲区，结构合seg相同
	  *seg为内存段
	  k为那个段
	  flag=0是正常输出，=1是直接末尾全部输出。
  */
void output_ex(pSORT_TEMP_FILE newfiles,int newf,pSEG_MEM outbuf,pSEG_MEM seg,\
			   int k,int flag){

    if(flag==1){
        fwrite((*outbuf).buf,dbfHead.record_length,(*outbuf).inrows,newfiles[newf].f);
		(*outbuf).inrows=0;
		(*outbuf).allrows=0;
    }else{
    	if((*outbuf).inrows==(*outbuf).allrows){ //输出文件
            fwrite((*outbuf).buf,dbfHead.record_length,(*outbuf).inrows,newfiles[newf].f);
	    	(*outbuf).inrows=0;
	    	//(*outbuf).allrows=0;
		}
    	//从第k个归并段，输入到缓冲区。
        memcpy((*outbuf).buf+(*outbuf).inrows*dbfHead.record_length,\
		       seg[k].buf+(seg[k].inrows-1)*dbfHead.record_length,\
               dbfHead.record_length);

    	(*outbuf).inrows++;
	}
}

/*
 *功能：K路归并排序算法,字符型
 *K路归并程序,利用败者树ls将编号从0到k-1的k个输入
 *归并段中的记录归并到输出归并段 b[0]到b[k-1]为败者
 *树上的k个叶子结点，分别存放k个输入归并段中当前记录
 *的关键码。
 *参数：
 *      * infiles是合并文件参数
 *      * newfiles是合并以后文件结构
 *      newf是输出文件索引
 *      *ls为失败树
 *      *b为合并排序segment各段关键字
 *      k为路数
 */

void K_Merge(pSORT_TEMP_FILE infiles,pSORT_TEMP_FILE  newfiles,int newf,\
			 pSEG_MEM seg,pSEG_MEM outbuf,pLoserTree ls,pExNode b,int k)
{
   int q=0; 
   int i;
   for(i=0;i<k;i++) input_ex(infiles,seg,b,i);    //初始化从0 ---k-1           
   CreateLoserTree(ls,b,k);                                         
   while( strcmp(b[ls[0].element].key,MAXKEY)!=0 ){ 
       q=ls[0].element;                                     
       output_ex(newfiles,newf,outbuf,seg,q,0);   
       input_ex(infiles,seg,b,q); 
       Adjust(ls,b,k,q);     
   }
   output_ex(newfiles,newf,outbuf,seg,q,1);                                  
}

/*
功能：输入文件名字进行k路合并排序，输出到指定文件
      合并排序是递归过程。数个文件进，一个文件出，
	  文件名依次后缀。在unix系统中使用并行进程。
	  返回-1失败,0成功。此函数是一个递归函数
      合并方法是先将快速排序文件k路合并存放在第1次
	  合并排序结构文件中全部完成以后，然后合并第一次
	  文件生成第二次的文件，依次。

参数：infiles是当前排序文件
      outfiles输出文件
	  merge_k 是当前路数
  */
int merge_sort(SORT_TEMP_FILE * infiles,char * outfiles,int merge_k,\
			   DBF_FILE_ALL_FIELD allf){
    SEG_MEM    seg[MERGE_K_MAX];            //段的内存指针
	SEG_MEM    outbuf;                      //输出缓冲结构
	int        part=0;                      //几个大段，每个包含MERGE_K_MAX个segment
	ExNode     exc[MERGE_K_MAX];            //字符型
	LoserTree  ls[MERGE_K_MAX];             //字符型败者树是完全二叉树且不含叶子，
	                                        //可采用顺序存储结构
    SORT_TEMP_FILE  newfiles[SORT_FILE_MAX];  //生成的新的文件
#ifdef UNIX
    int        nowpa=0;                     //当前并行度
	pid_t      pid;                         //进程号
    int        process=0;
#endif
	int pa=0;
	int k=0,v=0,l=0;
	char tmppix1[10]="";
	char tmppix2[10]="";
	int  mergek=0;
	unsigned long rec_num=0;

    if(infiles==NULL){
		printf("infile or outfile in marge is null");
		return -1;
	}

	//最终已经生成了文件了
	if(infiles[0].max==1){
		strcpy(outfiles,infiles[0].file);
		return 0;
	}

	if(infiles[0].max%merge_k==0){
        part=(int)(infiles[0].max/merge_k);
	}else{
        part=(int)(infiles[0].max/merge_k)+1;
	}

	for(pa=0;pa<part;pa++){    //pa值可以当作输出newfiles文件数。
        //输出文件，递归调用的输入文件，文件分配，
		//完了以后关闭，递归调用的时候再打开。
#ifdef UNIX
		now_process++;
		process++;

		if(process>=MAX_PROCESS){
			printf("open too many process than %d and exit \n",MAX_PROCESS);
			exit(1);
		}
        if(pid=fork()<0){
              printf("fork process error");
			  if(pa!=0){pa--;now_process--;} //不成功重新开进程。
	    	  continue;
		}else if(pid==0){   //父进程
			  nowpa++;
			 // parallel_all++;
			  if(parallel_num==nowpa){
				  for(v=0;v<nowpa;v++){
					  wait(NULL);
				  }
				  nowpa=0;  //当前并行度重新置0
			  }
			  continue;  //父进程跳转到for开头。

		}else{
#endif
	    //并行系统中是子进程。
        itoa(golfileId,tmppix1,10);
		strcpy(newfiles[pa].file,argsg);
		strcat(newfiles[pa].file,tmppix1);
        itoa(pa,tmppix2,10);
		strcat(newfiles[pa].file,tmppix2);
		strcat(newfiles[pa].file+strlen(argsg)+strlen(tmppix1)+strlen(tmppix2),".dbf");  
		
		if((newfiles[pa].f=openAFile(newfiles[pa].file,"w+b"))==NULL){
			printf("open file %s failed",newfiles[k].file);
			return -1;
		}
        newfiles[pa].fileid=pa;
        newfiles[pa].max=part;  

		if((outbuf.buf=malloc(OUT_BUFF*dbfHead.record_length))==NULL){
			printf("malloc in merge sort error");
			return -1;
		}
        outbuf.allrows=OUT_BUFF;
		outbuf.inrows=0;

		//这里调整merge_k，前面的mergek传入是建议值。
		if((infiles[0].max-(pa+1)*merge_k)>=0) {mergek=merge_k;}
		else mergek=infiles[0].max-pa*merge_k;

		if(mergek==1){
			free(outbuf.buf);
			fclose(newfiles[pa].f);
			remove(newfiles[pa].file);
			rename(infiles[pa*merge_k].file,newfiles[pa].file);
			continue;  
		}//只有一个文件不用合并

	    //分配段内存,读入文件
		for(k=0;k<mergek;k++){
			if((seg[k].buf=(char *)malloc(SORT_AREA_ROWS*dbfHead.record_length))==NULL){
				printf("malloc in merge sort error");
				return -1;
			}
			if((infiles[k].f=openAFile(infiles[pa*merge_k+k].file,"rb"))==NULL){
			    printf("open file %s failed",infiles[k].file);
				return -1;
			}
			//验证原文件的正确性
		    if(verifyDbfFile(infiles[k].f)==-1){
			    printf("file %s is destoryed",infiles[pa*merge_k+k].file);
			    return -1;
			}
			infiles[k].used=1;
            infiles[k].curr_read_row=0;

			fseek(infiles[k].f,0,SEEK_SET);//移动到开始位置
    		fread(&(infiles[k].dff),sizeof(DBF_FILE_HEAD),1,infiles[k].f);
#ifdef HP_UX
    		revert_unsigned_short(&(infiles[k].dff.head_length));		/* 用于定位到第一条记录*/
    		revert_unsigned_long(&(infiles[k].dff.record_num));			/* 总的记录数*/
    		revert_unsigned_short(&(infiles[k].dff.record_length));		/* 每条记录长度*/
#endif

#ifdef PRINT_MSG
    		printf("head_length:%d\n",infiles[k].dff.head_length);
    		printf("record_num:%ld\n",infiles[k].dff.record_num);
    		printf("record_length:%d\n",infiles[k].dff.record_length);
#endif

		    fseek(infiles[k].f,dbfHead.head_length,SEEK_SET);//移动到第一条记录
	

			read_data(infiles,k,seg);  //读取记录到内存段

			rec_num+=infiles[k].dff.record_num;

		}
        //写文件头。
		writeAFile(NULL,0,newfiles[pa].f,allf,2,rec_num);
		rec_num=0;

	    //k路合并
		K_Merge(infiles,newfiles,pa,seg,&outbuf,ls,exc,mergek);

		//取消内存，关闭文件或删除
    	for(k=0;k<mergek;k++){
			fclose(infiles[k].f);
			free(seg[k].buf);
			remove(infiles[pa*merge_k+k].file);
		}

		free(outbuf.buf);
		fclose(newfiles[pa].f);
#ifdef UNIX
		}          //并行系统结束
#endif
	}//for

    //递归调用本函数
	golfileId++;

    return merge_sort(newfiles,outfiles,merge_k,allf);

	return 0;
   
}

/*
功能：提取参数的值,返回1失败，返回0成功
  */
int getParaValue(char * * inpa,int start,int end){
    char tmp[30],p1[3]="-M",p2[3]="-P",* retp;
    int len=end-start+1,i=0;
	for (i=0;i<len;i++){
		if((retp=strstr(inpa[start+i],p1))!=NULL){
            strcpy(tmp,retp+2);
            max_mem=strtoul(tmp,NULL,10);
			if(max_mem<1064960 || max_mem>536870912){
				printf("max memary error it must in 1064960 and 536870912");
				exit(1);
			}
		}
#ifdef UNIX
		else if((retp=strstr(inpa[start+i],p2))!=NULL){
			strcpy(tmp,retp+2);
		    parallel_num=atoi(tmp);
			if(parallel_num>PARALLEL_MAX){
				printf("parallel number you give is too large! \n");
				exit(1);
			}
		}	
#endif
		else
			return 1;
	}
	return 0;
}

/*
 功能：验证dbf文件长度是否正确
       此函数因为有很多文件，所以需要多次
	   调用。
 参数：File	文件路径名
 返回值：0 正确，－1 错误
*/
int verifyDbfFile(FILE * afile)
{
	long fileSize = 0;
	long calSize = 0;
	DBF_FILE_HEAD dbfHead2;

	fseek(afile,0,SEEK_SET);

	fread(&dbfHead2,sizeof(DBF_FILE_HEAD),1,afile);

	fseek(afile,0,SEEK_END);
	fileSize =(long) ftell(afile);
  
	fseek(afile,0,SEEK_SET);
#ifdef HP_UX
	revert_unsigned_short(&dbfHead2.head_length);
	revert_unsigned_short(&dbfHead2.record_length);
	revert_unsigned_long(&dbfHead2.record_num);
#endif
	calSize = (long)dbfHead2.head_length + dbfHead2.record_length*dbfHead2.record_num;
	if(fileSize < calSize)
	{
#ifdef PRINT_MSG
		printf("dbfHead2.head_length:%d\n",dbfHead2.head_length);
		printf("dbfHead2.record_num:%ld\n",dbfHead2.record_num);
		printf("dbfHead2.record_length:%d\n",dbfHead2.record_length);
		printf("\tbut actual length is %ld \n",fileSize);
#endif
		return -1;
	}
	return 0;
}

/*
功能：字符和整数数值比较
      aa 要比较的字符串
	  bb 要比较的字符串
	  aalen aa字符串长度
	  bblen bb字符串长度
	  type 类型"C","N"
	  dec 是小数位数

  当aa<bb则返回-1，相等为0 否则为1
*/
int mycmp(char * aa,int aalen,char * bb,int bblen){
    char tmpaa[MAX_CMP_CHAR],tmpbb[MAX_CMP_CHAR];
	int a=0 ;
    tmpaa[aalen]='\0';
	tmpbb[bblen]='\0';

	memcpy(tmpaa,aa,aalen);
	memcpy(tmpbb,bb,bblen);
	return strcmp(tmpaa,tmpbb);
}

/*
功能：快速排序，字符型
  */
/*	
	快速排序
	传入参数:	    a:需要排序的记录集
					l:支点位置
					r:长度
					排序的字段信息在dbffield结构中
					tyy 为类型："C"为字符
					            "N"为整数
*/
void quickSort(char *sourceArray, long first, long last)
{
	char  tmp[RECORD_MAX];
	char* list_separator;
	long low, high;
	low = first;
	high = last+1;

	//这里记录的结构是不知道的，只知道长度。
	list_separator = sourceArray+first*dbfHead.record_length;

	if(first>=last) return;
	
	while(1){
		do{
			low++;
		}while (mycmp(sourceArray+low*dbfHead.record_length+dbfField.addr, \
			         dbfField.length,list_separator+dbfField.addr,dbfField.length)<0 \
					 && low<=last );
		do{
			high--;
		}while (mycmp(sourceArray+high*dbfHead.record_length+dbfField.addr, \
			         dbfField.length,list_separator+dbfField.addr,dbfField.length)>0\
					 && first<=high );
		if(low>=high) break;

		memcpy(tmp,(char *)(sourceArray+low*dbfHead.record_length),dbfHead.record_length);
		memcpy((char *)(sourceArray+low*dbfHead.record_length),    \
			   (char *)(sourceArray+high*dbfHead.record_length),   \
			   dbfHead.record_length);
		memcpy((char *)(sourceArray+high*dbfHead.record_length),tmp,dbfHead.record_length);
	}

	    //设置pivot
	    memcpy(tmp,(char *)(sourceArray+first*dbfHead.record_length),\
			   dbfHead.record_length);
		memcpy((char *)(sourceArray+first*dbfHead.record_length),\