C++ Merge Sort Algorithm

Question

I have created this first-timer implementation of Merge Sort and I got confused by the C++ syntax once or twice.

So I have some questions about this code:

#include <iostream>
#include <thread>
#include <cstdlib>
#include <vector>

#define NUM 1048576*2

int sort(std::vector<int>* vec);
void run(std::vector<int>* vec, int left, int right);
void merge(std::vector<int>* vec, int left, int mid, int right);

int main() {
    srand(time(NULL));
    std::vector<int> tobesorted;
    for(int i = 0; i < NUM; i++){
        tobesorted.push_back(rand());
    }

    sort(&tobesorted);

    for(int i = 0; i < NUM; i++){
        std::cout << tobesorted[i] << ", ";
    }
    std::cout << std::endl;
    std::cout << "End of Program!";


}
int sort(std::vector<int>* pvec){
    int size = pvec->size();
    std::thread t1(run, pvec, 0, 0.25*size);
    std::thread t2(run, pvec, 0.25*size+1, 0.5*size);
    std::thread t3(run, pvec, 0.5*size+1, 0.75*size);
    std::thread t4(run, pvec, 0.75*size, size-1);

    t1.join();
    t2.join();
    t3.join();
    t4.join();

    std::thread t5(merge, pvec, 0, 0.25*size, 0.5*size);
    merge(pvec, 0.5*size+1, 0.75*size, size-1);
    t5.join();

    merge(pvec, 0, 0.5*size, size-1);

    return 1;
}

void run(std::vector<int>* pvec, int left, int right){
    if(right - left == 0){
        return;
    }
    int mid = (left+ right)/2;
    run(pvec, left,mid);
    run(pvec, mid+1, right);

    merge(pvec, left, mid, right);

}

void merge(std::vector<int>* pvec, int left, int mid, int right){
    int ileft = 0;
    int iright = 0;
    int index = 0;


    std::vector<int> leftvec;
    std::vector<int> rightvec;

    int j = 0;
    for(int i= left; i <= mid; i++){
        leftvec.push_back(pvec[0][i]);
    }
    j = 0;
    for(int i= mid+1; i <= right; i++){
        rightvec.push_back(pvec[0][i]);
    }

    while(ileft < leftvec.size() && iright < rightvec.size()){
        if(leftvec[ileft] > rightvec[iright]){
            pvec[0][left+index] = leftvec[ileft++];
            index++;
        } else {
            pvec[0][left+index] = rightvec[iright++];
            index++;
        }
    }

    while(ileft < leftvec.size()){
        pvec[0][left+index] = leftvec[ileft++];
        index++;
    }

    while(iright < rightvec.size()){
        pvec[0][left+index] = rightvec[iright++];
        index++;
    }

}

1. I think I should have better optimized for better cache use, do you see any opportunities there?

2. What could have been generally improved on that code?

Answer 1

Overview

You are passing a std::vector<> to be sorted. There are lots of container types that could be used in its place (std::array, C-Array, std::dequeue, std::string etc).

It is more traditional to allow sorting via iterators. You pass the beginning and end of the iterator sequence you want to sort.

template<typename I>
void sort(I begin, I end);

This way you can use the sort on any of the above types of container.

---

Threads are relatively expensive to start. You better have a lot of data too sort before you split that sort across multiple threads. You may get some benefit from using std::async rather than std::thread as the runtime should manage a pool of threads for handling async actions (thus the threads will be-reused and thus not cost as much to spin up).

But this will depend on the sophistication of the current runtime and will only be beneficial if you are using lots of async actions that can re-use threads from the pool (with only 5 threads not much re-use but with a larger program that does a lot of sorting that balance may change).

---

Code Review

The old #define is no longer used for constants. Macros should be reserved for Hardware/OS/Compiler variations. The language has better alternatives for most other uses of macros.

#define NUM 1048576*2

On this case:

constexpr int My_NUM = 1048576*2; // Has explicit type information associated.
                                  // Is not a text replacement which
                                  // could lead to unexpected errors.

What is the result of:

    int test1 = sizeof NUM;
    int test2 = sizeof My_NUM;

---

If the parameter is never nullptr then don't use pointers.

int sort(std::vector<int>* vec);
void run(std::vector<int>* vec, int left, int right);
void merge(std::vector<int>* vec, int left, int mid, int right);

Generally you should never be using pointers in C++. There is usually a better way to do it (when you get advanced then pointers come back but lets stay away from them until you master the rest of the language).

In this case passing by reference is a better choice:

int sort(std::vector<int>& vec);
void run(std::vector<int>& vec, int left, int right);
void merge(std::vector<int>& vec, int left, int mid, int right);

Though as mentioned above using iterator would even better (and more idiomatic).

template<typename I>
int sort(I begin, I end);
template<typename I>
void run(I begin, I end);
template<typename I>
void merge(I begin, I mid, I end);

---

Misspelled:

    srand(time(NULL));

    // Should be
    std::srand(std::time(nullptr));

---

Minor optimizations:

    std::thread t1(run, pvec, 0, 0.25*size);
    std::thread t2(run, pvec, 0.25*size+1, 0.5*size);
    std::thread t3(run, pvec, 0.5*size+1, 0.75*size);
    std::thread t4(run, pvec, 0.75*size, size-1);
                                  ///  ^ missing +1 is this a bug?
    t1.join();
    t2.join();

    // Why are you waiting on t3 and t4 before starting the merge.
    // If t1 and t2 are finished then start the merge operation.
    // You can then wait on t3 and t4 once this has started.

    t3.join();
    t4.join();

    std::thread t5(merge, pvec, 0, 0.25*size, 0.5*size);
    merge(pvec, 0.5*size+1, 0.75*size, size-1);
    t5.join();

    merge(pvec, 0, 0.5*size, size-1);

    // If a function can only return one value
    // Why have it?    
    return 1;
}

---

You create temporary vectors every recursive call to merge:

void merge(std::vector<int>* pvec, int left, int mid, int right){

    std::vector<int> leftvec;
    std::vector<int> rightvec;

This can get very expensive. Why not create the temporary vectors once (up in sort() then re-use parts at every level. This will make the code more efferent overall.

---

Copying is fine for simple objects (like int). But what if you want to sort a vector of large objects? Then copying (the result of a push) is very expensive. So you should prefer to try and move objects.

        leftvec.push_back(pvec[0][i]);
        rightvec.push_back(pvec[0][i]);

        leftvec.push_back(std::move(pvec[0][i]));
        rightvec.push_back(std::move(pvec[0][i]));

If you already have the containers pre-built (as suggested by the last option for optimizing) then you can simply move the objects between containers.

        std::move(pvec[0][i], destination);

---

You could use a standard algorithm for this:

    while(ileft < leftvec.size()){
        pvec[0][left+index] = leftvec[ileft++];
        index++;
    }

    while(iright < rightvec.size()){
        pvec[0][left+index] = rightvec[iright++];
        index++;
    }

    /// Why not

    std::move( std::begin(leftvec) + iLeftm std::end(leftVec), std::begin(pvec) +  left+index);
    std::move( std::begin(rightvec) + iLeftm std::end(rightvec), std::begin(pvec) +  left+index);

Answer 2

This code doesn't work, as can be shown by including <algorithm> and changing the output:

std::cout << std::is_sorted(tobesorted.begin(), tobesorted.end()) << std::endl;

---

Other things that are surprising:

• We pass pointers but always assume they are not null. That suggests we should be using references instead.
• Misspelt std::srand and std::time - though I'd recommend using <random> instead.
• std::endl used where there's no need to flush output - just write \n instead.
• Why are we using int rather than std::size_t for our left and right indices?
• Use of floating-point multiplication for integer arithmetic.
• Unused variable j.

---

There's a risk of overflow here:

int mid = (left+ right)/2;

Better would be

auto mid = left + (right - left) / 2;

---

This code can be re-written:

std::vector<int> leftvec;
std::vector<int> rightvec;

for(int i= left; i <= mid; i++){
    leftvec.push_back(pvec[0][i]);
}
for(int i= mid+1; i <= right; i++){
    rightvec.push_back(pvec[0][i]);
}

Those vectors will likely re-allocate several times as they are filled, because we didn't reserve() capacity. Instead of those loops, we could use the appropriate constructor:

std::vector<int> leftvec(pvec->begin()+left, pvec->begin()+mid);
std::vector<int> rightvec(pvec->begin()+mid, pvec->>begin()+right+1);

It would be easier if we passed iterators rather than indices here.

Answer 3

Some other observations, in addition to Toby Speight's answer:

• Sort your includes.

• #define NUM 1048576*2 should be constexpr std::size_t num = 1048576*2 (or 2097152) instead.

• This loop:

  std::vector<int> tobesorted;
  for(int i = 0; i < NUM; i++){
      tobesorted.push_back(rand());
  }
  

  can be simplified: (requires <algorithm>)

  std::vector<int> numbers(num);
  std::generate_n(numbers.begin(), num, []{ return std::rand(); });
  

  or, better, with <random> and a separate function: (requires <algorithm> and <random>)

  std::vector<int> generate()
  {
      static std::mt19937_64 engine{std::time(nullptr)};
      std::uniform_int_distribution<int> dist{}; // [0, INT_MAX] by default
  
      std::vector<int> result(num);
      std::generate(result.begin(), result.end(), [&]{ return dist(engine); });
      return result;
  }
  
  // in main()
  auto numbers = generate(); // guaranteed copy elision
  

• This loop:

  for(int i = 0; i < NUM; i++){
      std::cout << tobesorted[i] << ", ";
  }
  

  can be replaced with (requires <algorithm> and <iterator>)

  std::copy(numbers.begin(), numbers.end(),
            std::ostream_iterator<int>{std::cout, ", "});
  

• Avoid std::endl unless the flushing semantics is necessary. See std::endl vs \n for more information.

• Use [begin, end) (including begin, excluding end) style instead of [begin, end] (including both begin and end) in order to eliminate the +1 / -1. Rewrite 0.25*size as size / 4, 0.5*size as size / 2, and 0.75*size as size / 4 * 3.

• Instead of using std::thread directly, consider the more high-level std::async and std::future:

  constexpr std::size_t thread_num = 4;
  auto chunk_size = vec.size() / thread_num;
  
  std::vector<std::future<void>> tasks;
  tasks.reserve(thread_num);
  for (std::size_t i = 0; i < thread_num; ++i) {
      tasks.push_back(std::async(std::launch::async,
          [](auto first, auto last) { std::sort(first, last); },
          vec.begin() + chunk_size * i,
          vec.begin() + chunk_size * (i + 1)
      ));
  }
  for (auto& task : tasks) {
      task.get();
  }
  // similarly for merging