在归并排序中,很重要的一步是将两个排序数组合并成一个数组,这个操作叫merge。merge操作可以用来解决某些Top K问题。
问题描述
在哼唱搜索中,用户通过哼唱一个音乐片段去搜索与其相似的音乐。后台的实现主要有两个步骤:特征提取和特征匹配。特征提取是从原始波形音乐文件中提取最能代表音乐的特征。特征匹配就是利用提取的特征与特征库进行匹配,找到最相似的音乐。在实际情况中,特征库往往很大,目前商用的特征库已达千万级别,这样的规模已经远远超过单机的处理能力,所以需要利用集群进行特征的匹配。
解决方案
在集群的每个节点上各存放特征库的一部分,所有不相交的特征库子集构成完整的特征库。集群的每个节点首先从主节点接收经过特征提取的用户哼唱特征,然后与自己的部分特征库进行匹配,返回Top K。最后利用MPI的规约函数将每个节点上的Top K规约成一个Top K返回给用户。
具体实现
主要的代码就是利用自己定义的规约操作完成merge操作。
1. 特征匹配的结果是一个表示两个音乐相似性的距离,越小说明两首歌越相似。首先定义规约的数据结构:
typedef struct
{
double dis; //表示计算的距离
char name[256]; //特征库文件表示的音乐
}distance;
2. 利用定义的数据结构声明一个MPI类型:
void new_type(MPI_Datatype* ctype)
{
int blockcounts[2];
MPI_Datatype oldtypes[2];
MPI_Aint offsets[2]; blockcounts[0]=1;
blockcounts[1]=256; offsets[0]=0;
offsets[1]=sizeof(double); oldtypes[0]=MPI_DOUBLE;
oldtypes[1]=MPI_CHAR; MPI_Type_struct(2,blockcounts,offsets,oldtypes,ctype);
MPI_Type_commit(ctype);
}
我们采用的方法是利用MPI的MPI_Type_struct声明一个结构数据类型。在distance结构体*有两个变量,所以MPI_Type_struct的中间三个参数都是长度为2的数组。这里需要注意的是,结构体的变量个数与结构体声明的个数无关,而与变量的类型数相关。例如结构体:
typedef struct
{
double x,y,z;
double velocity;
int n,type;
}Particle;
该结构体共有6个变量,但是在MPI结构类型中只有两个块{double,int},长度分别是{4,2}。
MPI_Type_struct的五个参数意义分别是:第一个参数指明结构体变量的块数,上面的两个例子都是2;第二个参数指明每个块的长度,上面的例子分别是{1,256}和{4,2};第三个参数指明每个块的偏移,简单的结构体可以利用sizeof获得,此外还可以利用MPI_Type_extent和MPI_Address获得;第四个参数指明每个块的变量类型;第五个参数就是根据我们声明的结构体返回的MPI变量类型。
3. 自定义归约操作实现merge:
void myProd(distance* in, distance* inout,int *len,MPI_Datatype* dptr)
{
int i,j,k; distance *result;
result=(distance*)malloc(sizeof(distance)*(*len)); for(i=0,j=0,k=0;i<*len;i++)
{
if(in[j].dis<inout[k].dis)
{
result[i].dis=in[j].dis;
strcpy(result[i].name,in[j].name);
j++;
}
else
{
result[i].dis=inout[k].dis;
strcpy(result[i].name,inout[k].name);
k++;
}
} for(int i=0;i<*len;i++)
{
inout[i].dis=result[i].dis;
strcpy(inout[i].name,result[i].name);
}
free(result);
}
用户自定义的归约操作是原型为:typedef void MPI_User_function(void *invec, void *inoutvec, int*len, MPI_Datatype *datatype);的函数。该函数有四个参数,第一个参数是数据输入,第二个是数据输入和输出,第三个参数是数据的长度,第四个是自定义归约操作的数据类型。
merge操作就是将两个排好序的数组合并成一个,这里有一点不同的是:合并的结果长度和输入数据长度相同,也即两个Top K结果合并成一个Top K结果。输入和输出的数据类型即是我们之前声明的类型distance。合并代码和常规的合并代码类似,但稍有不同。由于第二个变量既表示输入有代表输出,所以我们无法进行原地merge操作,在此我们引入一个临时变量result,将merge的结果先放入到result变量,最后再将result的结果拷贝到inout数组中。虽然这样显得浪费空间,但是这保证了正确性。
4. 主代码调用:
int main(int argc,char *argv[])
{
int n, myid, numprocs;
float *query, t1,t2;
int qline,scaned_file, sum=0;
int *accum_length;
int *seq_length,*all_seq, *small;
distance* dist, result[20]; MPI_Init(&argc,&argv);
MPI_Comm_size(MPI_COMM_WORLD,&numprocs);
MPI_Comm_rank(MPI_COMM_WORLD,&myid); /*
* step 1
* master get the humming sequence from the user, and broadcast to all the other nodes
*/
if(myid==MASTER)
{
query=get_sequence(argv[1],&qline);
if(query==NULL)
{
perror("get query error!\n");
return -1;
}
} /*
* step 2
* broadcast the sequence length to all the processes
*/
MPI_Bcast(&qline,1,MPI_INT,MASTER,MPI_COMM_WORLD); /*
* step 3
* the slave processes allocate the space for the humming sequence
*/
if(myid!=MASTER)
{
query=(float*)malloc(sizeof(float)*qline);
} /*
* step 4
* broadcast the humming sequence to all the processes
*/
MPI_Bcast(query,qline,MPI_FLOAT,MASTER,MPI_COMM_WORLD); /*
* step 5
* get all the sequence in the given directory which has the library
*/
get_dir_seq(argv[2],&scaned_file); /*
* step 6
* calculate the distance between the query and the library sequence
*/
small=match(query,qline,scaned_file,all_seq,sum,seq_length,accum_length,0.2); /*
* step 7
* sort the distance on every process
*/
dist=sort(small,scaned_file); /*
* step 8
* reduce the result to the MASTER process
*/
MPI_Op myop;
MPI_Datatype ctype; MPI_Op_create((MPI_User_function*)myProd,1,&myop);
new_type(&ctype);
MPI_Reduce(dist,result,20,ctype,myop,MASTER,MPI_COMM_WORLD); /*
* step 9
* free the allocated space
*/
free_space(query,accum_length,seq_length,all_seq,dist,scaned_file); MPI_Op_free(&myop);
MPI_Finalize();
return 0;
}
主代码流程比较简单,从命令行获取要匹配的序列,然后将该序列从MASTER广播到所有的进程。每个进程利用广播的序列与特征库进行匹配,然后将结果进行排序。最后利用自定义归约操作将排好序的文件归约到MASTER进程。