1
\$\begingroup\$

See the next iteration.

I have this C++ implementation of bottom-up mergesort (mergesort using iteration instead of recursion):

bottom_up_mergesort.h:

#ifndef BOTTOM_UP_MERGESORT_H
#define BOTTOM_UP_MERGESORT_H

#include <algorithm>
#include <iterator>

template <class RandomIt, class Compare>
class BottomUpMergesort {
    static const size_t INSERTIONSORT_THRESHOLD = 8;

    RandomIt array;
    RandomIt buffer;
    RandomIt source;
    RandomIt target;
    size_t range_length;
    Compare compare;

    public:

        BottomUpMergesort(RandomIt begin,
                          RandomIt end,
                          Compare& compare_) : array{begin}, compare(compare_) 
        {
            typedef typename std::iterator_traits<RandomIt>
                                ::value_type value_type;

            range_length = std::distance(begin, end);
            buffer = new value_type[range_length];
        }

        ~BottomUpMergesort() 
        {
            delete[] buffer;
        }

        void sort() {
            if (range_length < 2) 
            {
                return;
            }

            size_t runs = compute_run_amount();

            if (runs == 1) 
            {
                insertion_sort(array, 0, range_length, compare);
                return;
            }

            size_t merge_passes = compute_number_of_merge_passes(runs);

            if (merge_passes % 2 == 0) 
            {
                // We will need an even amount of merging passes over the input 
                // range in order to sort it. Let the input array be source so 
                // that the sorted range ends up in it.
                source = array;
                target = buffer;
            }
            else 
            {
                // We need an odd number of merging passes over the input range 
                // in order to sort it. Let the auxiliary buffer be the source 
                // so that the sorted range ends up in the input array.
                source = buffer;
                target = array;
            }

            // Presort small chunks using Insertionsort.
            presort_runs(runs);

            // Initial runs are ready to be merged. 'runLength <<= 1' multiplies
            // 'runLength' by 2.
            for (size_t run_length = INSERTIONSORT_THRESHOLD;
                        run_length < range_length;
                        run_length <<= 1) 
            {
                merge_pass(runs, run_length);
                // 'runs >>> 1' divides 'runs' by 2 ignoring the decimals.
                // '(runs & 1) != 0 ? 1 : 0' is zero if 'runs' is even, and one
                // otherwise. Basically, this computes the amount of remaining
                // runs.
                runs = (runs >> 1) + ((runs & 1) != 0 ? 1 : 0);
                // Now make the target array a source array, and vice versa.
                swap_array_roles();
            }
        }

    private:

        size_t compute_run_amount() 
        {
            return range_length / INSERTIONSORT_THRESHOLD +
                  (range_length % INSERTIONSORT_THRESHOLD != 0 ? 1 : 0);
        }

        size_t compute_number_of_merge_passes(size_t runs) 
        {
            return 8 * sizeof(size_t) - number_of_leading_zeros(runs - 1);
        }

        size_t number_of_leading_zeros(size_t num) 
        {
            size_t ret = 0;
            size_t one = 1;
            size_t mask = one << (8 * (sizeof(size_t)) - 1);

            while ((num & mask) == 0) 
            {
                ret++;
                mask >>= 1;
            }

            return ret;
        }

        void presort_runs(size_t runs) 
        {
            size_t local_from_index = 0;

            // Presort all but the last chunk in the source array.
            for (size_t i = 0; i < runs - 1; ++i) 
            {
                insertion_sort(source,
                               local_from_index,
                               local_from_index + INSERTIONSORT_THRESHOLD,
                               compare);

                local_from_index += INSERTIONSORT_THRESHOLD;
            }

            // Presort the last chunk that may be shorter than 
            // 'INSERTIONSORT_THRESHOLD'.
            insertion_sort(source,
                           local_from_index,
                           range_length,
                           compare);
        }

        inline void insertion_sort(RandomIt array,
                                   size_t begin,
                                   size_t end,
                                   Compare compare) 
        {
            typedef typename std::iterator_traits<RandomIt>
                                ::value_type value_type;

            for (size_t i = begin + 1; i < end; ++i) 
            {
                value_type element = array[i];
                size_t j = i;

                for (; j > begin && compare(element, array[j - 1]); --j) 
                {
                    array[j] = array[j - 1];
                }

                array[j] = element;
            }
        }

        void merge_pass(size_t runs, size_t run_length) 
        {
            size_t run_index = 0;

            // Perform pairwise merges.
            for (; run_index < runs - 1; run_index += 2) 
            {
                size_t left_end = (run_index + 1) * run_length;

                std::merge(source + run_index * run_length,
                           source + left_end,
                           source + left_end,
                           source + std::min(range_length, 
                                            (left_end  + run_length)),
                           target + run_index * run_length,
                           compare);
            }

            // Handle the orphan run, which occurs in the end of the range.
            if (run_index < runs) 
            {
                std::copy(source + run_index * run_length,
                          source + range_length,
                          target + run_index * run_length);
            }
        }

        void swap_array_roles() 
        {
            RandomIt tmp = source;
            source = target;
            target = tmp;
        }
};

template <class RandomIt, class Compare>
void bottom_up_mergesort(RandomIt first, RandomIt last, Compare comp)
{
    BottomUpMergesort<RandomIt, Compare> mergesort(first, last, comp);
    mergesort.sort();
}

#endif  // BOTTOM_UP_MERGESORT_H

main.cpp:

#include <algorithm>
#include <chrono>
#include <cstdlib>
#include <functional>
#include <iostream>
#include <random>
#include "bottom_up_mergesort.h"

using std::boolalpha;
using std::chrono::duration_cast;
using std::chrono::milliseconds;
using std::chrono::system_clock;
using std::copy;
using std::cout;
using std::default_random_engine;
using std::endl;
using std::equal;
using std::less;
using std::random_device;
using std::stable_sort;
using std::uniform_int_distribution;

static int** create_random_integer_pointer_array(
    size_t size, 
    default_random_engine& engine,
    uniform_int_distribution<int>& distribution) 
{
    int** array = new int*[size];

    for (size_t i = 0; i < size; ++i) 
    {
        int* p_int = new int(distribution(engine));
        array[i] = p_int;
    }

    return array;
}

class PointerCompare {
public:

    bool operator()(int* a, int* b) {
        return *a < *b;
    }
};

int main(int argc, char** argv) {
    random_device rd;
    default_random_engine random_engine(rd());
    uniform_int_distribution<int> distribution(0, 100000);

    const size_t m = 10000000;
    int** p_array1 = create_random_integer_pointer_array(m, 
                                                         random_engine, 
                                                         distribution);
    int** p_array2 = new int*[m];
    copy(p_array1, p_array1 + m, p_array2);

    PointerCompare pc;

    auto start = std::chrono::system_clock::now();
    stable_sort(p_array1, p_array1 + m, pc);
    auto end = std::chrono::system_clock::now();

    cout << "stable_sort in " 
         << duration_cast<milliseconds>(end - start).count()
         << " milliseconds."
         << endl;

    start = std::chrono::system_clock::now();
    bottom_up_mergesort(p_array2, p_array2 + m, pc);
    end = std::chrono::system_clock::now();

    cout << "bottom_up_mergesort in " 
         << duration_cast<milliseconds>(end - start).count()
         << " milliseconds."
         << endl;

    cout << "Same content: " 
         << std::boolalpha
         << equal(p_array1, p_array1 + m, p_array2) 
         << endl;

    return 0;
}

I don't have much experience in C++, so please tell me anything that comes to mind. Also, my demonstration shows that my implementation is a little bit faster than std::stable_sort, yet as soon as I pass the optimization option -O3 to the compiler (clang++), std::stable_sort becomes more efficient. Is there an explanation to that phenomenon?

\$\endgroup\$
4
\$\begingroup\$

Random

Avoid default_random_engine. Both libstdc++ and libc++, which are major standard library implementations, use minstd_rand0. Visual Studio uses mt19937 however and you can thank Stephan T. Lavavej for that.

Arrays

Avoid new. Avoid functions named create_random_integer_pointer_array. You could use vector or std::array instead combined with std::generate. Something like the following:

#include <iostream>
#include <algorithm>
#include <functional>
#include <random>
#include <vector>

using engine_type = std::mt19937;

engine_type& engine()
{
    thread_local static engine_type e;
    return e;
}

void seed(unsigned int s)
{
    engine().seed(s);
}

int randint(int A, int B)
{
    thread_local static std::uniform_int_distribution<> dist{A, B};

    return dist(engine());
}

int main()
{
    seed(std::random_device{}());

    const std::size_t m = 10000000;
    std::vector<int> v{m};
    std::generate(v.begin(), v.end(),
        std::bind(randint, 0, 100000));
}

Of course, if you still want to have a new-allocated array, go for it, but I believe this is an improvement.

Sorting algorithm

You may be interested in How to implement classic sorting algorithms in modern C++? by TemplateRex. Here's how one would implement Insertion Sort or Merge Sort in "modern C++":

template<class FwdIt, class Compare = std::less<>>
void insertion_sort(FwdIt first, FwdIt last, Compare cmp = Compare{})
{
    for (auto it = first; it != last; ++it) {
        auto const insertion = std::upper_bound(first, it, *it, cmp);
        std::rotate(insertion, it, std::next(it)); 
        assert(std::is_sorted(first, std::next(it), cmp));
    }
}

template<class BiDirIt, class Compare = std::less<>>
void merge_sort(BiDirIt first, BiDirIt last, Compare cmp = Compare{})
{
    auto const N = std::distance(first, last);
    if (N <= 1) return;                   
    auto const middle = std::next(first, N / 2);
    merge_sort(first, middle, cmp); // assert(std::is_sorted(first, middle, cmp));
    merge_sort(middle, last, cmp);  // assert(std::is_sorted(middle, last, cmp));
    std::inplace_merge(first, middle, last, cmp); // assert(std::is_sorted(first, last, cmp));
}

Please visit the linked page for omitted details. You might've noticed a few details:

  • Stronger iterator "guarantees", rather than just using Random Access Iterator.

  • Delegation to standard library functions which are almost always going to be faster than hand-rolled solutions.

  • C++14's transparent comparators: in this case, std::less.

Minor things

swap_array_roles could probably be replaced with std::swap.

typedef could probably be replaced with type aliases.

\$\endgroup\$
  • \$\begingroup\$ A good excerpt for merge_sort yet it relies on inplace_merge which worsens its time complexity to \$\Theta(n \log^2 n)\$ (as far as I recall). Could you elaborate your answer on how to allocate the auxiliary buffer such that I can accept vectors, lists, etc.? \$\endgroup\$ – coderodde Sep 29 '15 at 15:57
  • \$\begingroup\$ thread_local static? Don't see any threading. Seems overkill, especially since a lot of systems still don't support thread_local yet. Also thread_local static seems redundant. Are't all thread_local already static to the thread. Seems like it should be just thread_local or just static. \$\endgroup\$ – Martin York Sep 29 '15 at 17:16
0
\$\begingroup\$

What's with all the pointers?

I could sum up my review with just that comment. So many pointers. Why? You're adding all this extra work for yourself in having to do memory managementwhen you don't have to. You want a container of ints? Use std::vector<int>.

Here's how I would write code to generate a bunch of random integers:

template <typename D>
std::vector<int> random_ints(size_t count, D distribution)
{
    static std::random_device rd;
    static std::mt19937 engine(rd());

    std::vector<int> v;
    v.reserve(count);
    for (size_t i = 0; i < count; ++i) {
        v.push_back(distribution(engine));
    }
    return v;
}

auto values = random_ints(10000000, std::uniform_int_distribution<int>{0, 100000});

Now you don't even need a custom comparator, or have to worry about delete (which you don't have in your code btw), or worry about how to copy this into another container.

using std::X;

I'd be surprised if you saved any characters with all your using declarations as opposed to simply typing std::X everywhere. Using namespaces really isn't bad. Especially when you're talking about five characters... stick to the using declarations when you actually solve a problem. Don't use them to avoid laziness. Also lots of the time, you're writing std:: anyway so I'm just confused at the decision.

Your algorithm

I, truly, have no idea what you're doing. It doesn't help that you have 6 members in your merge sort class, 4 of which are iterators:

RandomIt array;
RandomIt buffer;
RandomIt source;
RandomIt target;

where none of those names sound like iterators. array seems like it should be, you know, an array. buffer should be a buffer. What do these things even refer to?

And then:

buffer = new value_type[range_length];

Ah, so RandomIt is a raw pointer. So I can't even sort a std::vector? That makes the code incredibly confusing since really it's like you're sorting T*s, but you're using type names as if you could sort arbitrary containers.

Advice

Start by writing a recursive merge-sort that actually works with arbitrary iterators. Test that you can sort an int[10], a std::vector<int>, and even a std::list<int> (will obviously be less efficient, but no reason for it not to work!) Once you develop the tools to do that, then go back and write an purely iterative version. And only then, add in insertion sort as an optimization point.

\$\endgroup\$
  • \$\begingroup\$ FYI, std::stable_sort (even with -std=c++1y) does not accept iterators of std::list. \$\endgroup\$ – coderodde Oct 2 '15 at 16:23
  • \$\begingroup\$ Your list point was misleading. \$\endgroup\$ – coderodde Oct 2 '15 at 16:40

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.