Решил реализовать быструю сортировку(Quick Sort) с использованием многопоточности C++(thread) и портфеля задач.
Метод портфеля задач заключается в поддержании очереди из задач. Каждый свободный поток берёт задачу из портфеля, выполняет её, при необходимости генерируя новые подзадачи и помещая их в портфель.
Алгоритм вроде бы работает, но у меня складывается такое впечатление, что на одном потоке работает быстрее, чем, например, на 2-х или 4-х. Пожалуйста, подскажите что не так в моём коде? Быть может я что-то намудрил с синхронизацией?
Вот код:
#include <thread>
#include <chrono>
#include <mutex>
#include <condition_variable>
#include <iostream>
#include <queue>
#include <vector>
#include <set>
#include <ctime>
#include <algorithm>
//очередь задач(сюда помещаются индексы после разбиения, т.е l и r
queue< pair<int, int> > tasks;
//мьютекс для очереди и множества
mutex q_mutex, s_mutex;
//условная переменная
condition_variable cv;
//множество
set<int> ss;
//алгоритм разбиения
template <typename T>
int partition(vector<T> &arr, int l, int r)
{
T tmp = arr[r]; //опорный элемент
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;
}
//быстрая сортировка
template <typename T>
void quick_sort(vector<T> &arr)
{
while (true)
{
unique_lock<mutex> u_lock(q_mutex); //блокируем мьютекс
//массив отсортирован
if ( ss.size() == arr.size() ) //u_lock.unlock()
return;
//Если очередь не пуста, берем из неё задачу
if ( tasks.size() > 0 )
{
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); //сделали разбиение
//Добавили в множество
s_mutex.lock();
ss.insert(q);
ss.insert(l);
ss.insert(r);
s_mutex.unlock();
//добавили в очередь задачи для левого и правого конца
tasks.push( make_pair(l, q - 1) );
tasks.push( make_pair(q + 1, r) );
//разбудили ожидающий поток
cv.notify_one();
}
}
else
//Если очередь пуста, то ожидаем
cv.wait(u_lock);
}
}
//Размер массива
const int ARR_SIZE = 100000;
//Количество потоков
const int THREAD_COUNT = 8;
//массив потоков
thread thrs[THREAD_COUNT];
//генерируем массив чисел
void generate_arr(vector<int> &arr)
{
srand(time( NULL ));
std::generate(arr.begin(), arr.end(), [](){return rand() % 10000; });
}
//проверка на сортировку
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_arr(arr);
cout << endl << "Generating finished!" << endl << endl;
cout << "Array before sorting" << endl << endl;
print(arr);
cout << endl << endl;
cout << "Checking is_sorted finished! The result is " << (is_sorted(arr) == 0? "false": "true") << "." << endl << endl;
//Добавили задачу в портфель(первую)
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;
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;
}