out_sort.c

外部排序程序

/*
 *名称：外部排序程序，主要用来排序DBF文件。
 *
 *功能：将一个某字段没有排序的DBF文件排序输出，采用
 *      外部排序算法，K路合并排序算法。排序需要
 *      一定的磁盘空间，否则会出现错误。如果文件很大
 *      时间可能会很长。本程序适合排序非常大的DBF文件
 *      。开始的时候使用快速排序（unix系统并行操作）
 *      ，然后合并K路排序段（Unix系统并行操作），最后合
 *      并最后两个，不使用并行，生成目标文件。比较字符
 *      串最长不超过100字符,而且字符必须是Ascii码
 *      中间有IO操作。
 *
 *编译：程序使用ANSI C 写成，完全采用标准C函数。
 *      程序支持windows,unix平台在Unix上打开UNIX宏支持并行。
 *      支持32 or 64 bit编译注意宏COMPILER_WITH_64。
 *      
 *作者：沈宏 hong Shen (Conserv company)
 *日期：2003年7月28号
 *
 *测试：2003.08.25-2003.09.04测试了在windows系统上的情况
 *      使用64M内存排序200M-2G文件要3-30分钟左右。并行性能
 *      未作测试。
 *
 *建议：排序文件大小在100M-10G左右记录在100万-20亿之间
 *      排序时源文件所在目录要留有源文件两倍大小的空间
 */
#include <stdio.h>
#include <string.h>
#include <stdlib.h>

#undef UNIX
//UNIX 系统，支持fork()子进程
//#define UNIX*/                           

#ifdef  UNIX
#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/shm.h>
#define MAX_PROCESS      200            //最多进程数，超过退出。
#endif

//#define HP_UX                           //HP UNIX则打开这个定义
#define PRINT_MSG                       //打印调试信息
/*#define COMPILER_WITH_64  */              /*定义64编译环境 
                                         *32位编译环境下:
										 *      short 16 bit
										 *      int 32 bit
                                         *      long 64 bit
                                         *64位编译环境下：
										 *      int 32 bit
										 *      long 64 bit
                                         */
#define MERGE_K_MAX      6              //K路合并排序最大的值，此值规定一个上限
int     merge_k;                        //当前合并排序路数值
#define SORT_AREA_ROWS   400000         //外部排序写盘的记录数大小，实际是规定每个文件大小
                                        //这个缓冲区在内存有对应大小

#define DEFAULT_MAX_MEM  33554432       //默认最大使用内存，byte数
#define BUFFERSIZE       32768          //环境缓冲区长度
#define OUT_BUFF         32768          //合并排序缓冲区长度
#define SORT_FILE_MAX    1000           //排序中使用的最大的临时文件个数。

#ifdef  UNIX
#define DEFAULT_PAR      1              //默认的并行度
int     parallel_num  ;                  //并行数量
#define PARALLEL_MAX     20             //最大并行度
#endif

#define MAX_CMP_CHAR     100            //比较最长的字符串长度
#define RECORD_MAX       2000           //记录的最长字符串数

#define MINKEY           "\0"            //失败树关键码最小值
#define MAXKEY           "~~~~~~~~~~~~~~~~~~~~"   //失败树关键码最大值

unsigned int max_mem;                  //内存使用最大数，这是一个总体的数目，包括所有子进程
    
/* dbf头文件结构*/
typedef struct
{
	unsigned char mask ;				// 0位   
	unsigned char date[3] ;				// 1-3   最后改动时间
#ifdef COMPILER_WITH_64
    unsigned int  record_num;
#else
	unsigned long record_num;			// 4-7   记录数
#endif
	unsigned short int head_length;		// 8-9   文件头长度，指第一条记录前的字节数
	unsigned short int record_length ;	// 10-11 每条记录的长度
	char  nouse[20];                    //不用的
} DBF_FILE_HEAD;

/*dbf字段结构*/
typedef struct
{
	unsigned char name[11];             // 名字
	unsigned char type ;                //类型
#ifdef COMPILER_WITH_64
    unsigned int  addr;              
#else
	unsigned long addr;                 //开始地址从1开始计数的
#endif
	unsigned short  length;          //长度
	unsigned char dec ;                 //如果为0表示数值型为整数,其他表示浮点型，
	                                    //注意dbf中间数值型以字符表示
	char nouse[13];                     //没有用的
} DBF_FILE_FIELD,*pDBF_FILE_FIELD;

/*dbf文件字段完全数据*/
typedef struct{
	unsigned int len;                   //长度
	char * pttr;                        //起始地址
} DBF_FILE_ALL_FIELD,*pDBF_FILE_ALL_FIELD;

/*
排序临时文件结构。
*/
typedef struct{
	int  fileid;                        //文件编号，从0开始
	int  max;                           //文件数最大值    
	char file[150];                     //文件路径
	int  used;                          //0表示未使用，1表示已经使用
	DBF_FILE_HEAD dff;                  //此文件的头信息
	FILE * f;                           //打开的文件句柄
#ifdef COMPILER_WITH_64
    unsigned int  curr_read_row;
#else
    unsigned long  curr_read_row;        //当前读到记录数
#endif
}SORT_TEMP_FILE,*pSORT_TEMP_FILE;       

/*外结点c，只存放待归并记录的关键码，字符类型*/
typedef struct{
    char key[MAX_CMP_CHAR];
}ExNode,*pExNode; 

//失败树结构
typedef struct{
    int element;
}LoserTree,*pLoserTree;

typedef struct{
	unsigned long inrows;                          //读取到那条记录数
	unsigned long allrows;                         //总的记录数
    char * buf ;                         //缓存地址
}SEG_MEM,*pSEG_MEM;

SORT_TEMP_FILE  *sort_file;              //初始化时分配内存。
DBF_FILE_HEAD  dbfHead;                  //dbf原文件的头描述
DBF_FILE_FIELD dbfField;                 //要排序的字段描述。

int lowext,koffset;                              //最底层的外部节点数。

int verifyDbfFile(FILE *file);
int golfileId;                           //这个是临时文件嵌套id号,一次嵌套增加一
char argsg[100];                         //存放args[1]
#ifdef UNIX
int  now_process;                        //当前已经开过进程数
#endif

/*
 功能：将短整形数(16位，两个字节)高低位反转
*/
void revert_unsigned_short(unsigned short *a) 
{ 
	unsigned short left,right; 
	left=right=*a; 
	*a=((left&0x00ff)<<8)|((right&0xff00)>>8); 
} 

/*
 功能：将长整形数(32位，四个字节)高低位反转
*/
#ifdef COMPILER_WITH_64
void revert_unsigned_long(unsigned int *a)
#else
void revert_unsigned_long(unsigned long *a)
#endif 
{ 
	unsigned long first,second,third,forth; 
	first=second=third=forth=*a; 
	*a=((first&0x000000ff)<<24)| 
	((second&0x0000ff00)<<8)| 
	((third&0x00ff0000)>>8)| 
	((forth&0xff000000)>>24); 
}

/*
功能：打开一个文件，返回文件句柄
 */
FILE * openAFile(char *FileIn , char * aa)
{
	FILE *fileIn = NULL;
//	char * writeMode[2] = {"rb","wb"};

//#ifdef HP_UX
//	strcpy(writeMode,"wb");
//#endif

	if((fileIn = fopen(FileIn,aa)) == NULL)
	{ 
		    return NULL;
	}
	return fileIn;
}

/*
功能：写一个dbf文件的某段，
参数：buf_sort是缓存段
      sort_area_rows是要写的记录数
	  tmpF文件句柄
	  allf是头信息
	  flag是1表示要写头和文件体，0表示不要写头只写文件,2表示只写头。
失败返回-1，成功返回0
参数：
  */
int writeAFile(char * buf_sort ,unsigned int sort_area_rows, FILE * tmpF,\
			   DBF_FILE_ALL_FIELD allf,int flag,unsigned long rec_num){
	DBF_FILE_HEAD   * write_head;    //写的dbf的头
	DBF_FILE_FIELD  * write_field;   //写的dbf的头

    if((write_head=malloc(sizeof(DBF_FILE_HEAD)))==NULL){
		printf("malloc write head error");
		return -1;
	}
	if((write_field=malloc(dbfHead.head_length-32))==NULL){
		printf("malloc write head error");
		return -1;
	}

	if(flag==1 || flag==2){
		memcpy(write_head,&dbfHead,32);
	    //memcpy(write_field,&dbfHead+0x2,dbfHead.head_length-32);

	    if(flag==1)(* write_head).record_num=sort_area_rows;
		if(flag==2)(* write_head).record_num=rec_num;

	    fwrite(write_head,32,1,tmpF);
		fwrite(allf.pttr,dbfHead.head_length-32,1,tmpF);
	}
    if(flag!=2)
	fwrite(buf_sort,dbfHead.record_length,sort_area_rows,tmpF);

	free(write_head);
	free(write_field);

	return 0;
}

/*
功能：从指定文件里面读取一个小于等于SORT_AREA_ROWS记录数的段
      到内存段当中。返回实际读取字节数。
参数：infiles是文件结构数组指针
      k是结构数组中那个文件
	  seg是段句柄
返回0为已经到了文件末尾
  */
unsigned int read_data(SORT_TEMP_FILE * infiles,int k,SEG_MEM * seg){
	//判断是否已经到了文件结尾
	unsigned int redin=0;

	if(infiles[k].curr_read_row==infiles[k].dff.record_num) return 0;

	   redin=fread(seg[k].buf,dbfHead.record_length,SORT_AREA_ROWS,infiles[k].f);
    infiles[k].curr_read_row+=redin;
	seg[k].inrows=0;   //初始化读取记录数
	seg[k].allrows=redin;
	return redin;
}

/*
功能：将一组记录写入一个DBF格式的文件，组大小为<SORT_AREA_ROWS
  */
void writeInFile(FILE * infile,char * buf,int red_size,int num){
    fwrite(buf,red_size,num,infile);
}
/*
功能：说明使用方法。
  */
void useage(){
	printf("out_sort sourcefile targetfile sortcolumn [option]\n\n\
sourcefile : This is a source file unsorted dbf formated \n\
targetfile : A path of target file \n\
sortcolumn : Identified the column to sort \n\
[option] -Mxxx :xxx is limit the max memary \n\
		size by byte must in [1064960,536870912]\n");
#ifdef UNIX
printf("        -pxxx :xxx is number of parallel processes \n\
				       must less then 10\n");
#endif

}
/*
功能：输入失败树，从seg读取到ExNode
      如果seg已经到了最后，则重新读取到seg中。
参数：infiles是文件数组
      seg是段数值指针
      b是叶节点
	  k是到那一路输入
  */
void input_ex(pSORT_TEMP_FILE  infiles, pSEG_MEM seg ,pExNode b,int k){
	char tmp[MAX_CMP_CHAR];
	tmp[dbfField.length]='\0';

    if(seg[k].inrows==seg[k].allrows){
		if(read_data(infiles,k,seg)==0){   //文件已经到了末尾
		    strcpy(b[k].key,MAXKEY);
			return ;
		}
	}
	 memcpy(tmp,(seg[k].buf+dbfHead.record_length*seg[k].inrows+dbfField.addr),dbfField.length);
     strcpy(b[k].key,tmp);
	 seg[k].inrows++;
	
}

/*
功能：调整失败树其实只用到ls[1]...ls[k-1]，ls[0]不用。
  */
void Adjust(LoserTree *ls,pExNode b,int k,int s)        
{  
	int  tmp=0;
	int  t=0;
	char tmpc[MAX_CMP_CHAR]="";

	strcpy(tmpc,b[s].key);
    t=(int)((s+k)/2);                              
    while(t>0)
	{ 
        if(strcmp(b[s].key,b[ls[t].element].key)>=0){
			tmp=s;                   //值互换,s指示新的胜利者
			s=ls[t].element;
            ls[t].element=tmp;
		}
        t=(int)(t/2);
	}
    ls[0].element=s;   //最终s变成最小值数组索引，保存在ls[0]中
}

/*
功能：建立败者树已知b[0]到b[k-1]为完全二叉树
      ls的叶子结点存有k个关键码，沿从叶子到
	  根的k条路径。
*/
void CreateLoserTree(LoserTree *ls,pExNode b,int k){ 
	int i=0,j=0,h=0;
	char tmp[MAX_CMP_CHAR]="";
	ExNode c[MERGE_K_MAX];

	for(i=0;i<k;i++){
		strcpy(c[i].key,b[i].key);
	}

	for(i=0;i<k;i++){
		for(j=0;j<k;j++){
            if(strcmp(c[j].key,c[i].key)>0){
                strcpy(tmp,c[j].key);
				strcpy(c[j].key,c[i].key);
				strcpy(c[i].key,tmp);
			}
		}
	}

	for(i=0;i<k;i++){
		if(strcmp(c[0].key,b[i].key)==0) {
			h=i;
			break;
		}
	}

//    strcpy(b[k].key,c[0].key); /*设MINKEY为关键码可能的最小值*/

    for(i=0;i<k;i++) ls[i].element=h; /*设置ls中“败者”的初值*/
    for(i=k-1;i>=0;i--) Adjust(ls,b,k,i); /*依次从b[k-1],b[k-2],…,b[0]出发调整败者*/
}



/*
功能：输出失败树
	  满了则输出到指定文件，判断满不满要看External结构内记录数

      * infiles为输出文件