Linux 下多线程排序的实现

对于计算密集型的任务，如果能采用合理的多线程处理，能够大大的提升计算效率。这篇博文实现了多线程排序，同时讲解了一些需要注意的问题。

首先，说一下总体的思路：将元素分成n段，使用快速排序多个线程并行处理。最后需要等待这些线程都将分段排好序之后，进行类似归并排序的过程。

这样时间复杂度算下来是(假设我是４核的机器)　O(n+n/4log(n/4))，比O(nlogn)大概快了一倍的样子。（请带入数值具体计算）

先来介绍一下pthread_barrier系列函数。

函数原型：
#include <pthread.h>
int pthread_barrier_init(pthread_barrier_t *restrict barrier, const pthread_barrierattr_t *restrict attr, unsigned count);
int pthread_barrier_wait(pthread_barrier_t *barrier);
int pthread_barrier_destroy(pthread_barrier_t *barrier);

参数解释：

pthread_barrier_t，是一个计数锁，对该锁的操作都包含在三个函数内部，我们不用关心也无法直接操作。只需要实例化一个对象丢给它就好。

pthread_barrierattr_t，锁的属性设置，设为NULL让函数使用默认属性即可。

count，你要指定的等待个数。

通俗解释:
pthread_barrier_*其实只做且只能做一件事，就是充当栏杆（barrier意为栏杆)。形象的说就是把先后到达的多个线程挡在同一栏杆前，直到所有线程到齐，然后撤下栏杆同时放行。1）init函数负责指定要等待的线程个数；2） wait()函数由每个线程主动调用，它告诉栏杆“我到起跑线前了”。wait(）执行末尾栏杆会检查是否所有人都到栏杆前了，如果是，栏杆就消失所有线程继续执行下一句代码；如果不是，则所有已到wait()的线程停在该函数不动，剩下没执行到wait()的线程继续执行；3）destroy函数释放init申请的资源。

单线程排序：

#include <unistd.h>
#include <sys/time.h>
#include <pthread.h>
#include <string.h>
#include <string>
#include <cstdio>
#include <iostream>
#include <errno.h>
#include <stdlib.h>
#include <algorithm>
using namespace std;

//错误检查函数
inline void ERR_EXIT(const string &msg,int retnum)
{
if(retnum!=0)
{
cerr<<msg<<": "<<strerror(retnum)<<endl;
exit(EXIT_FAILURE);
}
}
#define NUMMAX 8000000L
long int nums[NUMMAX];

int main()
{
srandom(time(NULL));
for(unsigned long i=0;i<NUMMAX;i++)
nums[i]=random();

//计时开始
gettimeofday(&start,NULL);
sort(nums,nums+NUMMAX);
gettimeofday(&end,NULL);

//计算用时
long long startusec=start.tv_sec*1000000+start.tv_usec;
long long endusec=end.tv_sec*1000000+end.tv_usec;
double elapsed=(double)(endusec-startusec)/1000000.0;
printf("sort took %.4f seconds\n",elapsed);

FILE *fp=fopen("save.txt","w+");
for(unsigned long i=0;i<NUMMAX;i++)
fprintf(fp,"%ld ",nums[i]);
return 0;
}

排序时间花费如下：



多线程排序：

#include <unistd.h>
#include <sys/time.h>
#include <pthread.h>
#include <string.h>
#include <string>
#include <cstdio>
#include <iostream>
#include <errno.h>
#include <climits>
#include <stdlib.h>
#include <algorithm>
using namespace std;

//错误检查函数
inline void ERR_EXIT(const string &msg,int retnum)
{
if(retnum!=0)
{
cerr<<msg<<": "<<strerror(retnum)<<endl;
exit(EXIT_FAILURE);
}
}
#define NUMMAX 8000000L
#define NTHR 4
#define TNUM (NUMMAX/NTHR)
long int nums[NUMMAX];
long int snums[NUMMAX];
pthread_barrier_t b;

void * workThread(void *args)
{
long index=(long)args;
sort(nums+index,nums+index+TNUM);
pthread_barrier_wait(&b);
pthread_exit(NULL);
}

void merge()
{
long index[NTHR];
for(long i=0;i<NTHR;i++)
index[i]=i*TNUM;

for(long i=0;i<NUMMAX;i++)
{
long min_index;
long min_num=LONG_MAX;
for(long j=0;j<NTHR;j++)
{
if((index[j]<(j+1)*TNUM)&&(nums[index[j]]<min_num))
{
min_num=nums[index[j]];
min_index=j;
}
}
snums[i]=nums[index[min_index]];
index[min_index]++;
}

}
int main()
{
srandom(time(NULL));
for(unsigned long i=0;i<NUMMAX;i++)
nums[i]=random();

struct timeval start,end;
pthread_t tid;
//计时开始
gettimeofday(&start,NULL);
pthread_barrier_init(&b,NULL,NTHR+1);
for(unsigned long i=0;i<NTHR;i++)
pthread_create(&tid,NULL,workThread,(void*)(i*TNUM));

pthread_barrier_wait(&b);
merge();
gettimeofday(&end,NULL);

//计算用时
long long startusec=start.tv_sec*1000000+start.tv_usec;
long long endusec=end.tv_sec*1000000+end.tv_usec;
double elapsed=(double)(endusec-startusec)/1000000.0;
printf("sort took %.4f seconds\n",elapsed);

FILE *fp=fopen("save.txt","w+");
for(unsigned long i=0;i<NUMMAX;i++)
fprintf(fp,"%ld ",snums[i]);
return 0;
}

运行结果如下：

线程数为２时:



线程数为４时:



线程数为８时:



由于我的电脑是4核的CPU，所以能够看到当线程数是4时的运 算时间最短，恰好达到并行的结果。当线程数再多时，就会有线程之间额外切换的开销。