【C++11】多线程并行化快速排序的实现

基础

并行化可以优化许多算法，前提是并行化的任务之间不会产生冲突。快速排序的核心逻辑为分治，通过一次遍历，使得区间分成了两个子区间，其中一个子区间的元素恒小于等于另一个区间的元素，再用同样的方法分别处理这两个子区间。可以发现，这两个子区间不重叠，并且后续的操作都是在一个区间内继续完成，不会跨区间，因此满足了并行化的条件。

通常的一种用多线程来解决问题的方式是使用线程池，将需要完成的操作细化成任务，传入线程池，空闲的线程会去完成这些任务。对于并行化快排来说，对较大的区间[p, q]进行partition操作分为两个子区间或对较小的区间直接进行快速排序就是它的任务。不过与普通的线程池操作不同，该任务的完成可能会创造新的任务，即这个任务可能有“返回值”，因此需要对通用线程池进行扩展，满足并行化快速排序的特定需求。具体实现细节见后文。

其他的一些优化

由于线程的切换存在时间开销，因此并行化任务不会无限对区间进行细分，当区间长度小于阈值时，使用串行化快排。

由于插入排序对小范围区间的排序处理效果更好，因此在区间范围小于16的情况下，不再使用快速排序，而是插入排序（STL里的sort也是这么做的）。

实现

Task包括一个需要排序的数组，和具体需要排序的区间 [p, q]

template 
class Task {
    public:
        Task(T &vec, int p, int q) : vec_(vec), p_(p), q_(q) {}
        T &vec_;
        int p_, q_;
};

任务队列主要的数据为一个包含Task的Queue，为了保证Queue操作的原子性，使用了mutex，condition_variable用于使得Get操作在Queue为空时，能够阻塞等待，直到新插入一个Task

template 
class TaskQueue {
    public:
        Task Get();
        void Push(Task &&task);
        bool Empty();

    private:
        std::queue > que_;
        std::mutex mutex_;
        std::condition_variable cond_var_;
};

template 
Task TaskQueue::Get() {
    std::unique_lock lk(mutex_);
    while(que_.empty()) {
        cond_var_.wait(lk);
    }
    Task ans = que_.front();
    que_.pop();
    return ans;
}

template 
void TaskQueue::Push(Task &&task) {
    std::lock_guard lk(mutex_);
    que_.push(std::move(task));
    cond_var_.notify_one();
}

template 
bool TaskQueue::Empty() {
    std::lock_guard lk(mutex_);
    return que_.empty();
}

CQSort 为主要的并行排序类，整体逻辑类似一个线程池

void SequentialSort_(T &vec, int p, int q);

void InsertSort_(T &vec, int p, int q);

int Partition_(T &vec, int p, int q);

上述三个函数是基础的排序方法，分别为串行化快排、插入排序和快排的Partition操作

void ThreadFunc_() 为线程池中线程执行的入口，不断从Queue中取出Task执行直到排序完成

void ConcurrentSort_(T &vec) 为并行化快排开始的入口，创建一个Task并插入到Queue中

void doConcurrentSort_(T &vec, int p, int q) 具体执行快排的代码，在Partiton后继续执行其中一半，另一半存入任务队列。若待排序区间长度小于512，则使用串行化快排。

mThr_ 为线程的数量

idleThr_ 为空闲线程的数量

std::vector < std::future < void>> asyncVec 存储执行的线程

bool isEnd 为线程池结束的标识符，当所有线程都空闲，并且任务队列为空，则认为排序已完成

template 
class CQSort {
public:
    CQSort(int mthr);
    ~CQSort();
    void operator()(T &vec);

private:
    void SequentialSort_(T &vec, int p, int q);
    void InsertSort_(T &vec, int p, int q);
    int Partition_(T &vec, int p, int q);
    void ConcurrentSort_(T &vec);
    void doConcurrentSort_(T &vec, int p, int q);
    void ThreadFunc_();

private:
    const int mThr_;
    std::mutex idleMutex_;
    int idleThr_;
    std::vector> asyncVec;
    TaskQueue taskQue;
    std::mutex endMutex_;
    std::condition_variable endCond_;
    bool isEnd;
};

template
CQSort::CQSort(int mthr) : mThr_(mthr), idleThr_(mthr), isEnd(false) { //mthr 线程数量
    for(int i = 0; i < mThr_; ++i) {
        asyncVec.push_back(std::async(std::launch::async, bind(&CQSort::ThreadFunc_, this)));
    }
}

// 正式开始排序
template     
void CQSort::operator()(T &vec) {
    if(mThr_ <= 1) {
        SequentialSort_(vec, 0, vec.size() - 1);
    }
    else {
        ConcurrentSort_(vec);
    }
}

template 
CQSort::~CQSort() {
    if(mThr_ <= 1) {
        return;
    }
    T tmp;
    int n = mThr_;
    while((n--) != 0) {
        taskQue.Push(Task{tmp, 0, 0});
    }
    while(!asyncVec.empty()) {
        asyncVec.pop_back();
    }
}

template 
void CQSort::SequentialSort_(T &vec, int p, int q) {
    if(p < q) {
        if(q - p < 16) {
            InsertSort_(vec, p, q);
        } else {
            int mid = Partition_(vec, p, q);
            SequentialSort_(vec, p, mid - 1);
            SequentialSort_(vec, mid + 1, q);
        }
    }
}

template 
void CQSort::InsertSort_(T &vec, int p, int q) {
    for(int i = p + 1; i <= q; ++i) {
        typename T::value_type tmp = vec[i];
        int j = i - 1;
        while(j >= p && vec[j] > tmp) {
            vec[j + 1] = vec[j];
            --j;
        }
        vec[j + 1] = tmp;
    }
}

// 并行化快排开始的入口，创建一个Task并插入到Queue中，并等待排序的完成
template 
void CQSort::ConcurrentSort_(T &vec) {
    if(vec.size() < ConcurrentThreshold) {
        return SequentialSort_(vec, 0, vec.size() - 1);
    }
    isEnd = false;
    taskQue.Push(Task{vec, 0, (int)vec.size() - 1});
    {
        std::unique_lock lk(endMutex_);
        while(!isEnd) endCond_.wait(lk);
    }
}

// 具体执行并行化快排的代码，在Partiton后继续执行其中一半，另一半存入任务队列。若待排序区间长度小于512，则使用串行化快排。
template 
void CQSort::doConcurrentSort_(T &vec, int p, int q) {
    if(q - p < ConcurrentThreshold) {
        return SequentialSort_(vec, p , q);
    }
    int mid = Partition_(vec, p, q);
    if(p < mid - 1 && q > mid + 1) {
        taskQue.Push(Task{vec, p, mid - 1});
        doConcurrentSort_(vec, mid + 1, q);
    } else if(p < mid - 1) {
        doConcurrentSort_(vec, p, mid - 1);
    }
    else if(q > mid + 1) {
        doConcurrentSort_(vec, mid + 1, q);
    }
}

// 线程池中线程执行的入口，不断从Queue中取出Task执行直到排序完成
template 
void CQSort::ThreadFunc_() {
    while(true) {
        Task task = taskQue.Get();
        if(task.p_ >= task.q_) {
            return;
        }
        {
            std::lock_guard lk(idleMutex_);
            --idleThr_;
        }
        doConcurrentSort_(task.vec_, task.p_, task.q_);
        {
            std::lock_guard lk1(idleMutex_);
            // check finsh
            if(++idleThr_ == mThr_ && taskQue.Empty()) {
                std::lock_guard lk2(endMutex_);
                isEnd = true;
                endCond_.notify_one();
            }
        }
    }
}

template 
int CQSort::Partition_(T &vec, int p, int q) {
    // three nums pick mid
    int mid = (p + q) >> 1;
    int pivot = p;
    const typename T::value_type &pv = vec[p], &midv = vec[mid], &qv = vec[q];
    if(pv > qv) {
        if(pv > midv) {
            pivot = midv < qv ? q : mid;
        } else { 
            pivot = p;
        }
    } else {
        if(qv > midv) {
            pivot = midv < pv ? p : mid;
        } else {
            pivot = q;
        }
    }
    // partation
    typename T::value_type tmp = vec[pivot];
    vec[pivot] = vec[p];
    while(p < q) {
        while(p < q && vec[q] >= tmp) {
            --q;
        }
        vec[p] = vec[q];
        while(p < q && vec[p] <= tmp) {
            ++p;
        }
        vec[q] = vec[p];
    }
    vec[p] = tmp;
    return p;
}

完整代码：

// cqsort.hpp
#ifndef CQSORT_HPP
#define CQSORT_HPP
#include 
#include 
#include 
#include 
#include 
#include 
#include 
#include 
#include 
#include 

const int ConcurrentThreshold = 512;

// Task包括一个需要排序的数组，和具体需要排序的区间 [p, q]
template 
class Task {
    public:
        Task(T &vec, int p, int q) : vec_(vec), p_(p), q_(q) {}
        T &vec_;
        int p_, q_;
};
// 任务队列主要的数据为一个包含Task的Queue，为了保证Queue操作的原子性，使用了mutex，
// condition_variable用于使得Get操作在Queue为空时，能够阻塞等待，直到新插入一个Task
template 
class TaskQueue {
    public:
        Task Get();
        void Push(Task &&task);
        bool Empty();

    private:
        std::queue > que_;
        std::mutex mutex_;
        std::condition_variable cond_var_;
};

template 
Task TaskQueue::Get() {
    std::unique_lock lk(mutex_);
    while(que_.empty()) {
        cond_var_.wait(lk);
    }
    Task ans = que_.front();
    que_.pop();
    return ans;
}

template 
void TaskQueue::Push(Task &&task) {
    std::lock_guard lk(mutex_);
    que_.push(std::move(task));
    cond_var_.notify_one();
}

template 
bool TaskQueue::Empty() {
    std::lock_guard lk(mutex_);
    return que_.empty();
}
 
template 
class CQSort {
public:
    CQSort(int mthr);
    ~CQSort();
    void operator()(T &vec);

private:
    void SequentialSort_(T &vec, int p, int q);
    void InsertSort_(T &vec, int p, int q);
    int Partition_(T &vec, int p, int q);
    void ConcurrentSort_(T &vec);
    void doConcurrentSort_(T &vec, int p, int q);
    void ThreadFunc_();

private:
    const int mThr_;
    std::mutex idleMutex_;
    int idleThr_;
    std::vector> asyncVec;
    TaskQueue taskQue;
    std::mutex endMutex_;
    std::condition_variable endCond_;
    bool isEnd;
};

template
CQSort::CQSort(int mthr) : mThr_(mthr), idleThr_(mthr), isEnd(false) { //mthr 线程数量
    for(int i = 0; i < mThr_; ++i) {
        asyncVec.push_back(std::async(std::launch::async, bind(&CQSort::ThreadFunc_, this)));
    }
}

// 正式开始排序
template     
void CQSort::operator()(T &vec) {
    if(mThr_ <= 1) {
        SequentialSort_(vec, 0, vec.size() - 1);
    }
    else {
        ConcurrentSort_(vec);
    }
}

template 
CQSort::~CQSort() {
    if(mThr_ <= 1) {
        return;
    }
    T tmp;
    int n = mThr_;
    while((n--) != 0) {
        taskQue.Push(Task{tmp, 0, 0});
    }
    while(!asyncVec.empty()) {
        asyncVec.pop_back();
    }
}

template 
void CQSort::SequentialSort_(T &vec, int p, int q) {
    if(p < q) {
        if(q - p < 16) {
            InsertSort_(vec, p, q);
        } else {
            int mid = Partition_(vec, p, q);
            SequentialSort_(vec, p, mid - 1);
            SequentialSort_(vec, mid + 1, q);
        }
    }
}

template 
void CQSort::InsertSort_(T &vec, int p, int q) {
    for(int i = p + 1; i <= q; ++i) {
        typename T::value_type tmp = vec[i];
        int j = i - 1;
        while(j >= p && vec[j] > tmp) {
            vec[j + 1] = vec[j];
            --j;
        }
        vec[j + 1] = tmp;
    }
}

template 
void CQSort::ConcurrentSort_(T &vec) {
    if(vec.size() < ConcurrentThreshold) {
        return SequentialSort_(vec, 0, vec.size() - 1);
    }
    isEnd = false;
    taskQue.Push(Task{vec, 0, (int)vec.size() - 1});
    {
        std::unique_lock lk(endMutex_);
        while(!isEnd) endCond_.wait(lk);
    }
}

template 
void CQSort::doConcurrentSort_(T &vec, int p, int q) {
    if(q - p < ConcurrentThreshold) {
        return SequentialSort_(vec, p , q);
    }
    int mid = Partition_(vec, p, q);
    if(p < mid - 1 && q > mid + 1) {
        taskQue.Push(Task{vec, p, mid - 1});
        doConcurrentSort_(vec, mid + 1, q);
    } else if(p < mid - 1) {
        doConcurrentSort_(vec, p, mid - 1);
    }
    else if(q > mid + 1) {
        doConcurrentSort_(vec, mid + 1, q);
    }
}

template 
void CQSort::ThreadFunc_() {
    while(true) {
        Task task = taskQue.Get();
        if(task.p_ >= task.q_) {
            return;
        }
        {
            std::lock_guard lk(idleMutex_);
            --idleThr_;
        }
        doConcurrentSort_(task.vec_, task.p_, task.q_);
        {
            std::lock_guard lk1(idleMutex_);
            // check finsh
            if(++idleThr_ == mThr_ && taskQue.Empty()) {
                std::lock_guard lk2(endMutex_);
                isEnd = true;
                endCond_.notify_one();
            }
        }
    }
}

template 
int CQSort::Partition_(T &vec, int p, int q) {
    // three nums pick mid
    int mid = (p + q) >> 1;
    int pivot = p;
    const typename T::value_type &pv = vec[p], &midv = vec[mid], &qv = vec[q];
    if(pv > qv) {
        if(pv > midv) {
            pivot = midv < qv ? q : mid;
        } else { 
            pivot = p;
        }
    } else {
        if(qv > midv) {
            pivot = midv < pv ? p : mid;
        } else {
            pivot = q;
        }
    }
    // partation
    typename T::value_type tmp = vec[pivot];
    vec[pivot] = vec[p];
    while(p < q) {
        while(p < q && vec[q] >= tmp) {
            --q;
        }
        vec[p] = vec[q];
        while(p < q && vec[p] <= tmp) {
            ++p;
        }
        vec[q] = vec[p];
    }
    vec[p] = tmp;
    return p;
}

测试代码：

// cqbench.cpp
#include 
#include 
#include 
#include 
#include 
#include 
#include 
#include "cqsort.hpp"

using namespace std;
using chrono::steady_clock;
using chrono::duration;
using chrono::duration_cast;

vector vec1, vec2, vec3;

int main()
{
    default_random_engine rand;
    rand.seed((unsigned int)time(NULL));
    
    const double INF = 1e10;
    CQSort > cq1(0), cq2(7);
    int listSize = 1;
    fstream fout("benchResult", ios::out);
    fout << "List Size       Sequential      Sort      Time (s)            Concurrent      Sort      Time (s)" << endl;
    fout << "              min            max            average         min            max            average" << endl;
    fout << "-----------   ------------   ------------   ------------    ------------   ------------   ------------" << endl;
    for(int i = 0; i < 8; ++i, listSize *= 10) {
        double seqMin = INF, seqMax = -INF, seqAvg = 0.0;
        double conMin = INF, conMax = -INF, conAvg = 0.0;
        for(int j = 0; j < 10; ++j) {
            vec1.resize(listSize);
            for(auto &val : vec1) {
                val = rand();
            }
            vec2 = vec1;
            vec3 = vec1;
            steady_clock::time_point t1 = steady_clock::now();
            cq1(vec1);
            steady_clock::time_point t2 = steady_clock::now();
            cq2(vec2);
            steady_clock::time_point t3 = steady_clock::now();

            duration ts1 = duration_cast >(t2 - t1);
            duration ts2 = duration_cast >(t3 - t2);
            seqMin = min(seqMin, ts1.count());
            seqMax = max(seqMax, ts1.count());
            seqAvg += ts1.count();
            conMin = min(conMin, ts2.count());
            conMax = max(conMax, ts2.count());
            conAvg += ts2.count();
        }
        seqAvg /= 10.0;
        conAvg /= 10.0;
        fout << setiosflags(ios::left);
        fout << setw(9) << listSize << "     ";
        fout << setw(12) << seqMin << "   " << setw(12) << seqMax << "   " << setw(12) << seqAvg << "    ";
        fout << setw(12) << conMin << "   " << setw(12) << conMax << "   " << setw(12) << conAvg << endl;
    }
    return EXIT_SUCCESS;
}

实验结果：

List Size       Sequential      Sort      Time (s)            Concurrent      Sort      Time (s)
              min            max            average         min            max            average
-----------   ------------   ------------   ------------    ------------   ------------   ------------
1             0              4e-07          7e-08           0              1e-07          7e-08       
10            2e-07          4e-07          3e-07           2e-07          3e-07          2.8e-07     
100           5.4e-06        6.1e-06        5.68e-06        4.2e-06        4.8e-06        4.45e-06    
1000          8.14e-05       9.97e-05       8.436e-05       9.28e-05       0.0002154      0.0001312   
10000         0.0010873      0.0014095      0.00115039      0.0004317      0.0010223      0.00060445  
100000        0.0137637      0.0145         0.0139018       0.0037528      0.0051069      0.00443984  
1000000       0.165105       0.203724       0.172742        0.0347065      0.0477473      0.0386616

如果大家对文章内容有疑问或建议，欢迎评论区留言，我们可以一起讨论，共同提高
转载请标明原作者:https://blog.csdn.net/qq_36622751

【C++11】多线程并行化快速排序的实现

C/C++/C#相关栏目本月热门文章