多线程对800万个数据进行排序

浏览数：22 / 时间：2015年06月08日

分8个线程对800万个数据分别排序，等所有线程排完序之后，进行简单的多路归并。这和外排的多路归并是一样的，也可以使用优化的胜者树算法。

#include <iostream>
#include <stdlib.h>
#include <pthread.h>
#include <limits.h>
#include <sys/time.h>
using namespace std;
#define NTHR   8                /* number of threads */
#define NUMNUM 8000000L         /* number of numbers to sort */
#define TNUM   (NUMNUM/NTHR)    /* number to sort per thread */

int nums[NUMNUM];
int snums[NUMNUM];

pthread_barrier_t b;

void HeapAdjust(int *a, int i, int size)
{
    int lchild = 2*i;
    int rchild = 2*i + 1;
    int max = i;
    if(i <=size/2)
    {
        if(lchild <=size && a[lchild] > a[max] )
        {
            max = lchild;
        }
        if(rchild <=size && a[rchild] > a[max])
        {
            max = rchild;
        }
        if(max != i)
        {
            swap(a[i], a[max]);
            HeapAdjust(a, max, size);
        }
    }
}

void BuildHeap(int *a, int size)
{
    int i;
    for(i = size/2; i >= 1; i--)
    {
        HeapAdjust(a, i, size);
    }
}

void HeapSort(int *a, int size)
{
    int i;
    BuildHeap(a, size);
    for(i=size;i>=1;i--)
    {
        swap(a[1], a[i]);
        HeapAdjust(a, 1, i-1);
    }
}

/*
 * Worker thread to sort a portion of the set of numbers.
 */
void * thr_fn(void *arg)
{
    long idx = (long)arg;

    HeapSort(&nums[idx], TNUM);
    pthread_barrier_wait(&b);

    /*
     * Go off and perform more work ...
     */
    return((void *)0);
}

/*
 * Merge the results of the individual sorted ranges.
 */
void merge()
{
    long    idx[NTHR];
    long    i, minidx, sidx, num;

    for (i = 0; i < NTHR; i++)
        idx[i] = i * TNUM;
    for (sidx = 0; sidx < NUMNUM; sidx++) {
        num = LONG_MAX;
        for (i = 0; i < NTHR; i++) {
            if ((idx[i] < (i+1)*TNUM) && (nums[idx[i]] < num)) {
                num = nums[idx[i]];
                minidx = i;
            }
        }
        snums[sidx] = nums[idx[minidx]];
        idx[minidx]++;
    }
}

int
main()
{
    unsigned long   i;
    struct timeval  start, end;
    long long       startusec, endusec;
    double          elapsed;
    int             err;
    pthread_t       tid;

    /*
     * Create the initial set of numbers to sort.
     */
    srandom(1);
    for (i = 0; i < NUMNUM; i++)
        nums[i] = random();

    /*
     * Create 8 threads to sort the numbers.
     */
    gettimeofday(&start, NULL);
    pthread_barrier_init(&b, NULL, NTHR+1);
    for (i = 0; i < NTHR; i++) {
        err = pthread_create(&tid, NULL, thr_fn, (void *)(i * TNUM));
        if (err != 0)
            cout<<"can‘t create thread";
    }
    pthread_barrier_wait(&b);
    merge();
    gettimeofday(&end, NULL);

    /*
     * Print the sorted list.
     */
    startusec = start.tv_sec * 1000000 + start.tv_usec;
    endusec = end.tv_sec * 1000000 + end.tv_usec;
    elapsed = (double)(endusec - startusec) / 1000000.0;
    for (i = 0; i < NUMNUM; i++)
        cout<<snums[i]<<"\n";
    cout<<"sort took seconds"<<elapsed;
    exit(0);
}