6
\$\begingroup\$

Never written a merge sort for huge files before.

So after this mornings question I though I would give it a go.

#include <vector>
#include <fstream>
#include <iostream>
#include <type_traits>
#include <algorithm>

#include <unistd.h>

// Merger object used by sortFile()
template<typename T>
class Merger
{
    static_assert(std::is_pod<T>::value, "Currently only support POD types");

    // Each temporary file is represented by a part object.
    struct Part
    {
        std::string     fileName;
        std::ifstream   inputStream;
        T               currentValue;

        // Remember we need to remove the temporary files when done.
        ~Part()
        {
            if (!fileName.empty()) {
                unlink(fileName.c_str());
            }
        }

        // Create a Part object with a file name and unsorted
        // piece of data
        Part(std::string const& partName, std::vector<T>& data)
            : fileName(std::string("/tmp/par.") + partName)
        {
            std::sort(std::begin(data), std::end(data));
            {
                // Write the RAW data out to the temporary file.
                std::ofstream   tmpFile(fileName);
                tmpFile.write(reinterpret_cast<char const*>(data.data()), sizeof(T) * data.size());
            }

            // Open the temporary file for reading.
            // We now have a stream of sorted data we can read.
            inputStream.open(fileName);
        }
        Part(Part&& move) noexcept
            : fileName(std::move(move.fileName))
            , inputStream(std::move(move.inputStream))
            , currentValue(std::move(move.currentValue))
        {
            move.fileName.clear();
        }
        Part(Part const&)               = delete;
        Part& operator=(Part const&)    = delete;
        Part& operator=(Part&&)         = delete;


        // Read a new value from the stream into `currentValue`
        // return true if we read data or false otherwise.
        bool readValue()
        {
            return inputStream.read(reinterpret_cast<char*>(&currentValue), sizeof(currentValue)).good();
        }
    };


    // An array of all the temporary parts we have created.
    std::vector<Part>   parts;

    // We keep a sorted heap of all the parts we have created.
    // That way it is easy to find the next piece of data we want
    // to write to the output file. Note: The Part object contains
    // the stream and the already read next value we want consider.
    std::vector<Part*>  partHeap;

    // This is the number of items we read from the input file
    // before creating a partition.
    std::size_t const   partitionSize;

    // Comparison function to keep the `partHeap` correctly organized.
    static bool partHeapComp(Part const* lhs, Part const* rhs){return lhs->currentValue > rhs->currentValue;}

    public:
        Merger(std::size_t partitionSize)
            : partitionSize(partitionSize)
        {}

        Merger(Merger&& move) noexcept
            : parts(std::move(move.parts))
            , partHeap(std::move(move.partHeap))
            , partitionSize(move.partitionSize)
        {}

        Merger(Merger const&)            = delete;
        Merger& operator=(Merger const&) = delete;
        Merger& operator=(Merger&&)      = delete;

        bool hasValue() const {return !partHeap.empty();}

        // read the next partition from the input stream.
        // then add it to the `parts`
        void addPartition(std::istream& data)
        {
            T               value;
            std::size_t     count;
            std::vector<T>  dataValues;
            dataValues.reserve(partitionSize);
            for(count = 0; (count < partitionSize) && (data >> value); ++count) {
                dataValues.emplace_back(std::move(value));
            }

            parts.emplace_back(std::to_string(parts.size()), dataValues);
        }


        // When we have read all the partitions then create the heap.
        void buildHeap()
        {
            for(auto& part: parts) {
                if (part.readValue()) {

                    partHeap.push_back(&part);
                }
            }
            std::make_heap(std::begin(partHeap), std::end(partHeap), partHeapComp);
        }


        // write the next value to the output stream.
        // return true if there are more values
        bool writeNextValue(std::ostream& output)
        {
            // Pop the next value of the heap and write it to the output stream.
            // Note there are only a couple of values in the heap.
            // One for every partition so around 4 values.
            std::pop_heap(std::begin(partHeap), std::end(partHeap), partHeapComp);
            output << partHeap.back()->currentValue << "\n";

            // For that partition read the next value.
            // If there is a another value then insert it back into heap
            // otherwise we reduce the size of the heap as that partition is empty.
            if (partHeap.back()->readValue()) {
                std::push_heap(std::begin(partHeap), std::end(partHeap), partHeapComp);
            }
            else {
                partHeap.resize(partHeap.size() - 1);
            }

            // if we have more then return true.
            return hasValue();
        }
};

template<typename T>
Merger<T> partitionFile(std::string const& fileName)
{
    Merger<T>   result(25'000'000);
    std::ifstream   file(fileName);
    while(file) {
        result.addPartition(file);
    }
    result.buildHeap();

    return result;
}

template<typename T>
void mergeFiles(std::string const& fileName, Merger<T>& merger)
{
    std::ofstream   file(fileName);
    while(merger.writeNextValue(file)) {
        // do nothing
    }
}

template<typename T>
void sortFile(std::string const& fileName)
{
    Merger<T>      merger = partitionFile<T>(fileName);
    if (merger.hasValue()) {
        mergeFiles(fileName, merger);
    }
}

int main(int argc, char* argv[])
{
    sortFile<long>(argv[1]);
}

On a file of a billion randomly generated values.

> time ./a.out data1

real    1m46.914s
user    1m40.099s
sys 0m5.860s
> head data1
43
50
87
92
159
181
203
208
224
225
> tail data1
2147483380
2147483389
2147483436
2147483486
2147483506
2147483519
2147483531
2147483578
2147483579
2147483632
\$\endgroup\$
  • \$\begingroup\$ @TobySpeight: Sorry about that. Fixed. \$\endgroup\$ – Martin York Jul 3 '17 at 15:34
5
\$\begingroup\$
  • partitionFile functionality seems to belong to the constructor.

    Along the same line, I don't see the value in exposing addPartition and buildHeap.

  • writeNextValue is bound too strong to the IO (and looks very Javan). Consider implementing (a forward) Merger::iterator.

  • Part seems to not closing its inputStream.

  • "/tmp/par." + partName allows (unlikely but possible) conflict with some other process (for example, another instance of merge).

  • A portable temp file management is a huge pain. For example there is no guarantee that the /tmp directory does exists at all. Consider std::filesystem::temp_directory_path, even though it is only available in .

\$\endgroup\$
  • \$\begingroup\$ It's probably reasonable to not explicitly close the stream - that's why we have destructors, after all. \$\endgroup\$ – Toby Speight Jul 3 '17 at 16:04

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.