在 C++ 中使用多线程进行快速排序
Quick sort with multithreading in c++
我用 multi-thread
方法实现了一个 quicksort
程序,在 C++ 中有一个 Portfolio 任务。
The method of portfolio tasks is to maintain a queue of tasks. Each
free thread picks a task from the portfolio, executes it, if necessary
generating new subtasks and placing them in to the portfolio
但我不确定什么是对的!在我看来,在一个 thread
中,该算法比两个或四个 thread
运行得更快。我能以某种方式弄乱同步吗?
感谢任何人帮助我。
代码:
#include <thread>
#include <chrono>
#include <mutex>
#include <condition_variable>
#include <iostream>
#include <queue>
#include <vector>
#include <set>
#include <ctime>
#include <algorithm>
using namespace std;
//print array
template <typename T>
void print(const vector<T> &arr)
{
for (size_t i = 0; i < arr.size(); i++)
cout << arr[i] << " ";
cout << endl;
}
//queue tasks
queue< pair<int, int> > tasks;
//mutexs for set and queue task
mutex q_mutex, s_mutex;
//condition variable
condition_variable cv;
//set
set<int> ss;
//partition algorithm
template <typename T>
int partition(vector<T> &arr, int l, int r)
{
T tmp = arr[r]; //as pivot element
int i = l - 1;
for (int j = l; j <= r - 1; j++)
if (arr[j] < tmp)
{
i++;
swap(arr[i], arr[j]);
}
swap(arr[i + 1], arr[r]);
i++;
return i;
}
//quick sort
template <typename T>
void quick_sort(vector<T> &arr)
{
while (true)
{
unique_lock<mutex> u_lock(q_mutex); //lock mutex
//sort is fineshed
if ( ss.size() == arr.size() ) //u_lock.unlock()
return;
//if queue task is not empty
if ( tasks.size() > 0 )
{
//get task from queue
pair<int, int> cur_task = tasks.front();
tasks.pop();
int l = cur_task.first, r = cur_task.second;
if (l < r)
{
int q = partition(arr, l, r); //split array
//Add indexes in set
s_mutex.lock();
ss.insert(q);
ss.insert(l);
ss.insert(r);
s_mutex.unlock();
//push new tasks for left and right part
tasks.push( make_pair(l, q - 1) );
tasks.push( make_pair(q + 1, r) );
//wakeup some thread which waiting
cv.notify_one();
}
}
else
//if queue is empty
cv.wait(u_lock);
}
}
//Size array
const int ARR_SIZE = 100000;
//Count threads
const int THREAD_COUNT = 8;
thread thrs[THREAD_COUNT];
//generatin array
void generate_arr(vector<int> &arr)
{
srand(time( NULL ));
std::generate(arr.begin(), arr.end(), [](){return rand() % 10000; });
}
//check for sorting
bool is_sorted(const vector<int> &arr)
{
for (size_t i = 0; i < arr.size() - 1; i++)
if ( ! (arr[i] <= arr[i + 1]) )
return false;
return true;
}
int main()
{
//time
clock_t start, finish;
vector<int> arr(ARR_SIZE);
//generate array
generate_arr(arr);
cout << endl << "Generating finished!" << endl << endl;
cout << "Array before sorting" << endl << endl;
//Before sorting
print(arr);
cout << endl << endl;
cout << "Checking is_sorted finished! The result is " << (is_sorted(arr) == 0? "false": "true") << "." << endl << endl;
//add task
tasks.push( make_pair(0, arr.size() - 1) );
//==================================================
start = clock();
for (int i = 0; i < THREAD_COUNT; i++)
thrs[i] = thread( quick_sort<int>, ref(arr) );
finish = clock();
//==================================================
for (auto& th : thrs)
th.join();
cout << "Sorting finished!" << endl << endl;
cout << "Array after sorting" << endl << endl;
//After sorting
print(arr);
cout << endl << endl;
cout << "Checking is_sorted finished! The result is " << (is_sorted(arr) == 0? "false": "true") << "." << endl << endl;
cout << "Runtime: " << (double)(finish - start) / CLOCKS_PER_SEC << endl;
return 0;
}
影响性能的因素远不止你在这个问题上投入了多少线程。其中,
你需要有实际的并发性,而不仅仅是多线程。正如@Rakete1111 和@user1034749 都观察到的那样,你没有。
标准快速排序具有良好的参考局部性,尤其是当分区大小变小时,但您的技术丢弃了很多,因为给定数组元素的责任很可能被交换到不同的线程在每次分区时。
此外,互斥操作并不是特别便宜,当分区变小时,相对于实际排序的数量,您开始做相当多的操作。
使用比物理内核更多的线程没有意义。四个线程可能不会太多,但这取决于你的硬件。
以下是一些可以提高多线程性能的方法:
在方法 quick_sort()
中,不要像当前那样在实际排序期间保持互斥体 q_mutex
锁定(您正在使用的 unique_lock
构造函数会锁定mutex,并且在 unique_lock
).
的生命周期内您不会解锁它
对于小于某个阈值大小的分区,切换到普通递归技术。您必须进行测试才能找到一个好的特定阈值;也许它需要可调。
在每个分区中,每个线程 post 只有一个子分区到投资组合;让它递归地处理另一个——或者更好的是,迭代地处理另一个。事实上,将它设为 post 的 较小 子分区,因为这将更好地限制投资组合的大小。
您还可以考虑增加 运行 测试的元素数量。 100000 并没有那么多,对于更大的问题,您可能会看到不同的性能特征。 1000000 个元素对于在现代硬件上进行这样的测试一点也不合理。
在我看来,您应该将 组合任务 的行为捕获到 class.
template <typename TASK, unsigned CONCURRENCY>
class Portfolio {
std::array<std::thread, CONCURRENCY> workers_;
std::deque<TASK> tasks_;
std::mutex m_;
std::condition_variable cv_;
std::atomic<bool> quit_;
void work () {
while (!quit_) {
TASK t = get();
if (quit_) break;
t();
}
}
TASK get () {
std::unique_lock<std::mutex> lock(m_);
while (tasks_.empty()) {
cv_.wait(lock);
if (quit_) return TASK();
}
TASK t = tasks_.front();
tasks_.pop_front();
if (!tasks_.empty()) cv_.notify_one();
return t;
}
public:
void put (TASK t) {
std::unique_lock<std::mutex> lock(m_);
tasks_.push_back(t);
cv_.notify_one();
}
Portfolio ();
~Portfolio ();
};
构造函数将使用线程初始化工作线程,每个线程调用 work()
方法。析构函数将设置 quit_
,向所有线程发出信号,并加入它们。
那么,您的快速排序可以简化:
template <typename T, unsigned WORK_SIZE, typename PORTFOLIO>
QuickSortTask {
std::reference_wrapper<PORTFOLIO> portfolio_;
std::reference_wrapper<std::vector<T>> arr_;
int start_;
int end_;
QuickSortTask (PORTFOLIO &p, std::vector<T> &a, int s, int e)
: portfolio_(p), arr_(a), start_(s), end_(e)
{}
void operator () () {
if ((end_ - start_) > WORK_SIZE) {
int p = partition(arr_, start_, end_);
portfolio_.put(QuickSortTask(portfolio_, arr_, start_, p-1));
portfolio_.put(QuickSortTask(portfolio_, arr_, p+1, end_));
} else {
regular_quick_sort(arr_, start_, end_);
}
}
};
不幸的是,这种制定并行快速排序的方法不太可能产生较大的加速。您想要做的是并行化分区任务,这需要至少一个单线程计算通道(涉及数据比较和交换)才能开始并行化。
首先将数组分成 WORK_SIZE
个子数组,对每个子数组并行执行快速排序,然后合并结果以创建排序向量可能会更快。
我用 multi-thread
方法实现了一个 quicksort
程序,在 C++ 中有一个 Portfolio 任务。
The method of portfolio tasks is to maintain a queue of tasks. Each free thread picks a task from the portfolio, executes it, if necessary generating new subtasks and placing them in to the portfolio
但我不确定什么是对的!在我看来,在一个 thread
中,该算法比两个或四个 thread
运行得更快。我能以某种方式弄乱同步吗?
感谢任何人帮助我。
代码:
#include <thread>
#include <chrono>
#include <mutex>
#include <condition_variable>
#include <iostream>
#include <queue>
#include <vector>
#include <set>
#include <ctime>
#include <algorithm>
using namespace std;
//print array
template <typename T>
void print(const vector<T> &arr)
{
for (size_t i = 0; i < arr.size(); i++)
cout << arr[i] << " ";
cout << endl;
}
//queue tasks
queue< pair<int, int> > tasks;
//mutexs for set and queue task
mutex q_mutex, s_mutex;
//condition variable
condition_variable cv;
//set
set<int> ss;
//partition algorithm
template <typename T>
int partition(vector<T> &arr, int l, int r)
{
T tmp = arr[r]; //as pivot element
int i = l - 1;
for (int j = l; j <= r - 1; j++)
if (arr[j] < tmp)
{
i++;
swap(arr[i], arr[j]);
}
swap(arr[i + 1], arr[r]);
i++;
return i;
}
//quick sort
template <typename T>
void quick_sort(vector<T> &arr)
{
while (true)
{
unique_lock<mutex> u_lock(q_mutex); //lock mutex
//sort is fineshed
if ( ss.size() == arr.size() ) //u_lock.unlock()
return;
//if queue task is not empty
if ( tasks.size() > 0 )
{
//get task from queue
pair<int, int> cur_task = tasks.front();
tasks.pop();
int l = cur_task.first, r = cur_task.second;
if (l < r)
{
int q = partition(arr, l, r); //split array
//Add indexes in set
s_mutex.lock();
ss.insert(q);
ss.insert(l);
ss.insert(r);
s_mutex.unlock();
//push new tasks for left and right part
tasks.push( make_pair(l, q - 1) );
tasks.push( make_pair(q + 1, r) );
//wakeup some thread which waiting
cv.notify_one();
}
}
else
//if queue is empty
cv.wait(u_lock);
}
}
//Size array
const int ARR_SIZE = 100000;
//Count threads
const int THREAD_COUNT = 8;
thread thrs[THREAD_COUNT];
//generatin array
void generate_arr(vector<int> &arr)
{
srand(time( NULL ));
std::generate(arr.begin(), arr.end(), [](){return rand() % 10000; });
}
//check for sorting
bool is_sorted(const vector<int> &arr)
{
for (size_t i = 0; i < arr.size() - 1; i++)
if ( ! (arr[i] <= arr[i + 1]) )
return false;
return true;
}
int main()
{
//time
clock_t start, finish;
vector<int> arr(ARR_SIZE);
//generate array
generate_arr(arr);
cout << endl << "Generating finished!" << endl << endl;
cout << "Array before sorting" << endl << endl;
//Before sorting
print(arr);
cout << endl << endl;
cout << "Checking is_sorted finished! The result is " << (is_sorted(arr) == 0? "false": "true") << "." << endl << endl;
//add task
tasks.push( make_pair(0, arr.size() - 1) );
//==================================================
start = clock();
for (int i = 0; i < THREAD_COUNT; i++)
thrs[i] = thread( quick_sort<int>, ref(arr) );
finish = clock();
//==================================================
for (auto& th : thrs)
th.join();
cout << "Sorting finished!" << endl << endl;
cout << "Array after sorting" << endl << endl;
//After sorting
print(arr);
cout << endl << endl;
cout << "Checking is_sorted finished! The result is " << (is_sorted(arr) == 0? "false": "true") << "." << endl << endl;
cout << "Runtime: " << (double)(finish - start) / CLOCKS_PER_SEC << endl;
return 0;
}
影响性能的因素远不止你在这个问题上投入了多少线程。其中,
你需要有实际的并发性,而不仅仅是多线程。正如@Rakete1111 和@user1034749 都观察到的那样,你没有。
标准快速排序具有良好的参考局部性,尤其是当分区大小变小时,但您的技术丢弃了很多,因为给定数组元素的责任很可能被交换到不同的线程在每次分区时。
此外,互斥操作并不是特别便宜,当分区变小时,相对于实际排序的数量,您开始做相当多的操作。
使用比物理内核更多的线程没有意义。四个线程可能不会太多,但这取决于你的硬件。
以下是一些可以提高多线程性能的方法:
在方法
quick_sort()
中,不要像当前那样在实际排序期间保持互斥体q_mutex
锁定(您正在使用的unique_lock
构造函数会锁定mutex,并且在unique_lock
). 的生命周期内您不会解锁它
对于小于某个阈值大小的分区,切换到普通递归技术。您必须进行测试才能找到一个好的特定阈值;也许它需要可调。
在每个分区中,每个线程 post 只有一个子分区到投资组合;让它递归地处理另一个——或者更好的是,迭代地处理另一个。事实上,将它设为 post 的 较小 子分区,因为这将更好地限制投资组合的大小。
您还可以考虑增加 运行 测试的元素数量。 100000 并没有那么多,对于更大的问题,您可能会看到不同的性能特征。 1000000 个元素对于在现代硬件上进行这样的测试一点也不合理。
在我看来,您应该将 组合任务 的行为捕获到 class.
template <typename TASK, unsigned CONCURRENCY>
class Portfolio {
std::array<std::thread, CONCURRENCY> workers_;
std::deque<TASK> tasks_;
std::mutex m_;
std::condition_variable cv_;
std::atomic<bool> quit_;
void work () {
while (!quit_) {
TASK t = get();
if (quit_) break;
t();
}
}
TASK get () {
std::unique_lock<std::mutex> lock(m_);
while (tasks_.empty()) {
cv_.wait(lock);
if (quit_) return TASK();
}
TASK t = tasks_.front();
tasks_.pop_front();
if (!tasks_.empty()) cv_.notify_one();
return t;
}
public:
void put (TASK t) {
std::unique_lock<std::mutex> lock(m_);
tasks_.push_back(t);
cv_.notify_one();
}
Portfolio ();
~Portfolio ();
};
构造函数将使用线程初始化工作线程,每个线程调用 work()
方法。析构函数将设置 quit_
,向所有线程发出信号,并加入它们。
那么,您的快速排序可以简化:
template <typename T, unsigned WORK_SIZE, typename PORTFOLIO>
QuickSortTask {
std::reference_wrapper<PORTFOLIO> portfolio_;
std::reference_wrapper<std::vector<T>> arr_;
int start_;
int end_;
QuickSortTask (PORTFOLIO &p, std::vector<T> &a, int s, int e)
: portfolio_(p), arr_(a), start_(s), end_(e)
{}
void operator () () {
if ((end_ - start_) > WORK_SIZE) {
int p = partition(arr_, start_, end_);
portfolio_.put(QuickSortTask(portfolio_, arr_, start_, p-1));
portfolio_.put(QuickSortTask(portfolio_, arr_, p+1, end_));
} else {
regular_quick_sort(arr_, start_, end_);
}
}
};
不幸的是,这种制定并行快速排序的方法不太可能产生较大的加速。您想要做的是并行化分区任务,这需要至少一个单线程计算通道(涉及数据比较和交换)才能开始并行化。
首先将数组分成 WORK_SIZE
个子数组,对每个子数组并行执行快速排序,然后合并结果以创建排序向量可能会更快。