实验5 蒙特卡罗法求PI及排序算法

#浙江大学城市学院实验报告课程名称多核与并行程序设计实验项目名称实验五蒙特卡罗法求PI及排序算法学生姓名专业班级学号实验成绩指导老师（签名）日期【实验环境】硬件平台：联想4核,4GZ内存编译器：MicrosoftVisualStudioC++6.0操作系统：Windows2003serversp2测试数据集合：由随机数函数产生的数据集合【实验1】一、问题描述蒙特卡洛算法可理解为通过大量实验，模拟实际行为，来收集统计数据。本例中，算法随机产生一系列点，模拟这些点落在如下图所示的正方形区域内的情况。其几何解释如下图1如图1所示，正方形边长为1,左下顶点与原点重合，两边分别与x,y轴重合。曲线为1/4圆弧，圆心位于原点，与正方形左下定点重合，半径为1。正方形面积S1=1，圆弧内11—兀r2=—兀面积s2=44。算法模拟大量点随机落在此正方形区域内，落在圆弧内的点的数量（n2）与点的总数（叫）的比例与面积成正比关系。即nS41—1—(1)nS兀22由此可得4n兀=2_n(2)1因此，只要计算出落在圆弧内的点的数量在点总数中所占的比例，就能求出兀的值。由图1可知，所有点均落在正方形范围内，因此点的x坐标满足0<X<1。又，当点落在圆弧范围内，则点的二维坐标关系满足x2+y2<1。检验每一个点是否满足此关系即可判定改点是否落在圆弧内。二、串行算法描述本项目中使用了标准c语言库中的产生随机数函数。该函数原型为：intrand(void);此函数产生随机数列，每次调用时均返回0到RAND_MAX之间的一个整数。voidsrand(unsignedintseed);此函数为rand()函数所生成的伪随机数序列设置起始点，使之产生不同的伪随机数。算法：产生2n个随机数据，范围［0,1］,对每个数据点计算其坐标是否满足x2+y2<1,count兀=统计满足此关系的点的数量count,则~4n—三、并行算法3.1并行算法描述算法步骤：1、确定需要产生的点的个数n参与运行的处理器数m；2、对每一个处理器，生成两个随机数x，y,范围［0,1］；3、判断两个随机数x,y是否满足x2+y2<1；4、若满足，则变量COUNTi++；5、重复步骤2-4,直至每个处理器均生成n/m个随机点；6、收集COUNT匚的值，并累加至变量COUNT中，此即为随机点落在圆弧内的数量；7、通过(2)式计算的值。3.2并行算法在Windows下的一个例子#include<stdio.h>#include<windows.h>#include<time.h>//#include<process.h>#include<iostream>#include<fstream>#include<stdlib.h>usingnamespacestd;HANDLEevFinish;longcs=0;//总循环次数longcount=0;//主线程有效次数longcount_thread=0;//thread线程有效次数time_tstart,finish;//定义开始结束时间//thread线程计算量为总数的一半DWORDWINAPIthread(LPVOIDparam)//这个函数在PDF里WaitForSingleObject(evFinish,INFINITE);//两线程同步count+=count_thread;finish=time(NULL);//记录结束时间printf("并行情况：\n\n");printf(”用时=%f秒\n",difftime(finish,start));//计算时间差printf(”总共的循环次数=%d次\n",cs);printf("线程有效次数=%d次\n",count);printf("pi=%f\n",4*(double)count/(double)cs);printf("串行行情况：\n");count=0;start=time(NULL);//记录开始时间for(i=0;i<cs;i++){x=(longdouble)rand()/(longdouble)RAND_MAX;y=(longdouble)rand()/(longdouble)RAND_MAX;if((x*x+y*y)<=1)count++;//printf("±%d",i);//printf("count%d",count);}finish=time(NULL);//记录结束时间printf（"用时=%f秒\n",difftime（finish,start））;printf（"总共的循环次数=%d次\n",cs）;printf（"线程有效次数=%d次\n",count）;printf（"pi=%f\n",4*（double）count/（double）cs）;return（0）;}四、实验结果1、实验运行结果是多少2、本例中加速比（串行时间/并行时间）：S=【实验2】一、问题描述在单核计算环境中，排序算法关注的核心问题是怎样减少要排序数据之间的比较次数或算法所需要的内存空间。在多核计算环境中，每个核以线程为执行单元，排序程序可以通过生成相互协作的线程来完成排序。与单核计算环境不同的是，在多核计算环境中更关注数据集的合理划分，更致力于识别可并行执行的任务。一旦完成这些工作，程序设计上就可以生成对应的线程去执行任务。理论上，基于相同的串行算法和相同的cache命中率，多核计算速度可以无限接近单核计算速度的P倍，其中P为核的数目。多核上的并行排序算法所面临的问题在于：未排序的数据集合理划分到每个线程后，最后怎么汇合，形成完整的排好序的数据集呢？怎么保证可并行执行的线程的数目和核的数目相等或稍微多于核的数目，同时也保证这些线程之间的工作量也尽可能的相同呢？在这个实验中，串行的算法采用标准C语言库中的快速排序函数。并行算法中，先将要处理的数据集均等的分到每个线程中，并使用C语言库中的快速排序函数各自排序。然后所有线程开始根据相同的数对自己的数据集进行划分，这个数是依据一定的方法选出来的（详见并行算法描述）。每个线程的数据集都会被分成K份，（其中P<=K<P2，P为核的数目），每份将被称为一桶。很显然这个过程选出了K个数，这些数将被成为bound_value，记为X1,X2,X3XK。最后每个线程中小于或等于X1的数会被一个独立的线程去归并排序，同样小于或等于x2的数也会被另外一个独立的线程去归并排序，依次类推，直到排好序。需要指出的是：这个并行版本最消耗时间的部分是一开始每个线程各自的排序，时间为：O（nlogn）；不过其数据划分和线程生成也相对简单。最后的归并排序所需时间是线性增长的，即：O（n），因此即使在最后归并部分线程执行的任务已经是不均衡的，也不会对整个程序的性能产生很大的影响。、串行算法描述本项目中使用了标准C语言库中的快速排序函数。该函数原型为：voidvoid*base,size_tnum,size_twidth,int(cdecl*compare)(constvoid*elem1,constvoid*elem2))其中base:待排序数据的数组基址；num:待排序数据的数目；width:待排序数据元组的大小；compare:排序比较函数，当elemlvelem2时返回值小于0，当eleml=elem2时返回值等于0，当elem1>elem2时返回值大于0；三、并行算法3.1并行算法描述算法:1、将原始待排序的数据分成P等份，每个处理器上对NO个数据进行排序，称每个被排序后的子集为B0,...,Bp-12、Remain_data=N,设定第0组归并起始位置全部为0,i=0，设置第0组在目标数组中的起始位置为03、循环直至remian_data<L(L=N0/P)3.1选取所有子集中起始位置后续L个元素的最小值bound_value，并获得bound_value的桶号bucket3.2在所有子集中从起始位置到后续L个元素中选取边界位置，使得边界位置的最后一个元素小于或等于bound_value，而边界位置后的第一元素大于bound_value。3.3记录所有的边界位置，并设置所有第i+1组的起始位置为第i组的起始位置加上边界位置3.4累积所有边界值，得到该归并组的大小3.5根据归并组大小和本组起始位置计算第i+1组在目标数组中的起始位置。4、设置最后一个归并组的边界为N05、对所有归并组进行并行P路归并排序。说明：P和多核处理器的数目相同。比如是双核的，那么P=2；Remain_data是每个线程处理的数据集中还没有被X”X2,X3划分过的数据集的总数目。比如，根据X］每个线程划分出来的数据集为x,y，z，那么Remain_data=n-x-y-z3.2并行算法在Windows下的一个例子#include<stdlib.h>#include<stdio.h>#include<time.h>#include<search.h>#include<windows.h>#include<process.h>#ifndef_BASIC_SORT_H#define_BASIC_SORT_H#ifndef_SORT_P#define_SORT_Pvoid*sort(void*parameter);voidgenerate_data(int*a,intn);voidsort_s(int*a,intn);voidview_data(int*a,intn);intcheck_data_sort(int*a,intn);intcompare(constvoid*arg1,constvoid*arg2);#defineMILLION1000000L#defineP2#defineN04332539//#defineN01000000#defineNN0*P#defineLN0/Pvoidsort_p(int**d,int*b);time_tstart,finish;//定义开始结束时间structmerge{//归并结构intbegin[P];//数组beginintcount[P];//数组countintmerge_size;//归并大小intglobal_pos;//全局位置intmerged;//归并是否完成};structarg_merge_data{//归并数据结构int**d;//数组的指针structmerge*mp;//归并结构int*b;//整数bintk;////////////////////////////////////////////////};structarg_merge_data*tmp2;structforsort{int*d;intk;};structforsortforsorta[P];intfind_bound_value(int**d,structmerge*mp,int*bucket);intfind_min(int**d,structmerge*mp,int*bucket);voidfind_bound(int**d,structmerge*mp,intbucket,intbound_value);voiddo_last_merge_block(structmerge*mp);voidmerge_data(void*arg);voidview_merge(int**d,structmerge*mp,intlast_merge_block);intstart_time();intdiff_time();#endif#endifintk;//////////////HANDLESwait[P];HANDLEPwait[P];HANDLEpthread[P];HANDLEqthread[P];///////externint*d[P];///////////////////////////////////////////////////////////////////voidgenerate_data(int*a,intn){//产生数据inti;for(i=0;i<n;i++){a[i]=rand()%10000;〃产生数组a有n个元素}for(i=0;i<P;i++){d[i]=&(a[i*NO]);〃产生数组d对应a[i]中每个n0块的第i个元素}}////////////////////////////////////////////////////////////////////voidsort_s(int*a,intn){//快排a[i]qsort((void*)a,n,sizeof(int),compare);}////////////////////////////////////////////////////////////////////void*sort(void*parameter){〃快排参数(数组)的前NO个元素structforsort*forsortb=(structforsort*)parameter;//printf("\nSetEvent(Swait[%d])",forsortb->k);/////////////////=====================intkk=forsortb->k;qsort(/*(void*)*/forsortb->d,NO,sizeof(int),compare);SetEvent(Swait[kk]);//printf("\n%d",kk);return(void*)O;}///////////////////////////////////////////////////////////////////voidview_data(int*a,intn){inti=n,j;intindex=O;while(i>N0){for(j=0;j<N0;j++){printf(”％d\t",a[index++]);〃输出a[i]中前NO个数}printf("\n");i-=NO;}for(;i>0;i--){〃输出a[i]中NO后面的个数printf("%d\t",a[index++]);}printf("\n");}//////////////////////////////////////////////////////////////////////intcheck_data_sort(int*a,intn){//找出不服和大小顺序的位置inti;for(i=O;i<n-1;i++){if(a[i]>a[i+1])break;}returni;}//////////////////////////////////////////////////////////////////////intcompare(constvoid*arg1,constvoid*arg2){//返回arg1与arg2的差inta=*(int*)arg1,b=*(int*)arg2;returna-b;}////////////////////////////////////////////////////////////////////inta[N];//dataforparallelsortintb[N];//theresu