Add MT merger

task1/CMakeLists.txt

@@ -19,7 +19,8 @@ add_custom_target(task1_7_dataset
         ${CMAKE_CURRENT_BINARY_DIR}/dataset.dat)
 
 add_executable(task1_7 main.cpp
-        mergesort.h)
+        mergesort.h
+        mergesort_mt.h)
 
 target_link_libraries(task1_7
         fmt::fmt

task1/main.cpp

@@ -54,24 +54,25 @@ auto main(int argc, char *argv[]) -> int {
         auto dataset_seq = dataset;
 
         auto t1 = std::chrono::high_resolution_clock::now();
-        algo::MergeSort_v1::sort(dataset_seq.begin(), dataset_seq.end(), [](int32_t a, int32_t b) {
+        algo::MergeSort_mt::sort(dataset_seq, [](int32_t a, int32_t b) {
             return (a > b);
-        });
+        }, 0);
         auto t2 = std::chrono::high_resolution_clock::now();
 
         auto delay_ms = std::chrono::duration_cast<std::chrono::milliseconds>(t2 - t1);
         fmt::print("Sorted {} entries within {} ms in sequential\n", dataset_seq.size(), delay_ms.count());
 
 
-        const int nthreads = std::thread::hardware_concurrency();
+        //const int max_depth = std::thread::hardware_concurrency();
+        const int max_depth = 4;
         t1 = std::chrono::high_resolution_clock::now();
         algo::MergeSort_mt::sort(dataset_par, [](int32_t a, int32_t b) {
             return (a > b);
-        }, nthreads);
+        }, max_depth);
         t2 = std::chrono::high_resolution_clock::now();
 
         delay_ms = std::chrono::duration_cast<std::chrono::milliseconds>(t2 - t1);
-        fmt::print("Sorted {} entries within {} ms in parallel using {} threads\n", dataset_seq.size(), delay_ms.count(), nthreads);
+        fmt::print("Sorted {} entries within {} ms in parallel using {} threads\n", dataset_seq.size(), delay_ms.count(), max_depth);
 
         auto eq = (dataset_seq == dataset_par);
         fmt::print("Equality: {}\n", eq);

task1/mergesort.h

@@ -305,7 +305,7 @@ namespace algo {
 
             while (l < left.end() && r < right.end()) {
                 if (cmp(*l, *r)) {
-                    output.insert(o, *l);
+                    outpbufut.insert(o, *l);
                     l++;
                 } else {
                     output.insert(o, *r);
@@ -327,7 +327,7 @@ namespace algo {
         }
 
         template<typename T, typename Comparator>
-        static auto split(std::vector<T> data, Comparator cmp, int depth, int &num_threads,
+        static auto split(std::vector<T> data, Comparator cmp, int depth, int &max_depth,
                           std::mutex &mut) -> std::vector<T>{
 
             if (data.size() <= 1) {
@@ -350,15 +350,15 @@ namespace algo {
             std::vector<T> left(data.begin(), mid);
             std::vector<T> right(mid, data.end());
 
-            if (depth < num_threads) {
+            if (depth < max_depth) {
-                std::thread left_thread([&]() { left = split(left,  cmp, depth + 1, num_threads, mut); });
+                std::thread left_thread([&]() { left = split(left, cmp, depth + 1, max_depth, mut); });
-                std::thread right_thread([&]() { right = split(right,  cmp, depth + 1, num_threads, mut); });
+                std::thread right_thread([&]() { right = split(right, cmp, depth + 1, max_depth, mut); });
 
                 left_thread.join();
                 right_thread.join();
             } else {
-                left = split(left,  cmp, depth + 1, num_threads, mut);
+                left = split(left, cmp, depth + 1, max_depth, mut);
-                right = split(right,  cmp, depth + 1, num_threads, mut);
+                right = split(right, cmp, depth + 1, max_depth, mut);
             }
 
             return merge(left, right, cmp, mut);
@@ -367,12 +367,12 @@ namespace algo {
     public:
         template<typename T, typename Comparator>
         static auto
-        sort(std::vector<T> &data, Comparator cmp, int num_threads = std::thread::hardware_concurrency()) -> void {
+        sort(std::vector<T> &data, Comparator cmp, int max_depth = 0) -> void {
             std::mutex local_result_lock;
             std::vector<T> output;
             output.reserve(data.size());
 
-            output = split(data, cmp, 0, num_threads, local_result_lock);
+            output = split(data, cmp, 0, max_depth, local_result_lock);
             data.assign(output.begin(), output.end());
         }
     };

task1/mergesort_mt.h

@@ -0,0 +1,48 @@
+#include <vector>
+#include <span>
+#include <thread>
+#include <mutex>
+
+template<typename T, typename C>
+class MergeSorterMT {
+
+    C comp;
+    std::recursive_mutex mut;
+
+    auto merge(std::span<T> &output, std::span<T> left, std::span<T> right) -> void {
+        std::vector<T> buf;
+        buf.reserver(left.size() + right.size());
+
+        auto l = left.begin();
+        auto r = right.begin();
+        auto o = buf.begin();
+
+        hile (l < left.end() && r < right.end()) {
+            if (cmp(*l, *r)) {
+                buf.insert(o, *l);
+                l++;
+            } else {
+                buf.insert(o, *r);
+                r++;
+            }
+            o++;
+        }
+        while (l < left.end()) {
+            buf.insert(o, *l);
+            o++;
+            l++;
+        }
+        while (r < right.end()) {
+            buf.insert(o, *r);
+            o++;
+            r++;
+        }
+
+        {
+            std::lock_guard<std::recursive_mutex> lock(mut);
+            std::move(buf.begin(), buf.end(), output.begin());
+        }
+    }
+
+
+};