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*>(¤tValue), 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
