working

include/file_writer.hpp

@@ -16,8 +16,8 @@ public:
     FileWriter(const std::string& filename, size_t buffer_size = 8);
     ~FileWriter();
 
-    void write(const Record& record);
-    void write_chunk(std::span<const Record>& chunk);
+    void write(Record record);
+    void write_chunk(std::span<Record> chunk);
 
     size_t total_writes() const;
 };

include/merge_sorter.hpp

@@ -8,14 +8,14 @@ class MergeSorter {
 private:
     std::vector<Record> buffer_;
     const std::string& tmp_dir_;
-    size_t buffer_rows = 0;
+    size_t buffer_rows_ = 0;
     size_t disk_reads_ = 0;
     size_t disk_writes_ = 0;
     size_t phases_ = 0;
 
-    size_t create_initial_runs(const std::string& input_file);
-    void merge_runs(const std::vector<std::string>& input_files,
-        const std::string& output_file);
+    std::span<Record> get_chunk_span(size_t chunk_index);
+    size_t create_initial_runs(const std::string& input_file, const std::string& next_dir);
+    size_t merge_pass(const std::string& current_dir, const std::string& next_dir, size_t run_count);
     void perform_k_way_merge(std::vector<std::unique_ptr<FileReader>>& readers,
         FileWriter& writer);
 

src/file_writer.cpp

@@ -24,7 +24,7 @@ FileWriter::~FileWriter()
     write_buffer();
 }
 
-void FileWriter::write(const Record& record)
+void FileWriter::write(Record record)
 {
     if (current_pos_ >= buffer_.size()) {
         write_buffer();
@@ -34,7 +34,7 @@ void FileWriter::write(const Record& record)
     total_writes_++;
 }
 
-void FileWriter::write_chunk(std::span<const Record>& chunk)
+void FileWriter::write_chunk(std::span<Record> chunk)
 {
     for (const auto& record : chunk) {
         write(record);

src/merge_sorter.cpp

@@ -1,12 +1,19 @@
 #include "../include/merge_sorter.hpp"
 
-size_t MergeSorter::create_initial_runs(const std::string& input_file)
+std::span<Record> MergeSorter::get_chunk_span(size_t chunk_index)
+{
+    size_t chunk_size = buffer_.size() / buffer_rows_;
+    size_t start = chunk_index * chunk_size;
+    return std::span<Record>(buffer_).subspan(start, chunk_size);
+}
+
+size_t MergeSorter::create_initial_runs(const std::string& input_file, const std::string& next_dir)
 {
     FileReader reader(input_file);
     std::span<Record> chunk(buffer_);
     size_t run_count = 0;
 
-    for (run_count = 0;; run_count++) {
+    for (;; run_count++) {
         size_t records_read = reader.read_chunk(chunk);
         if (records_read == 0) {
             break;
@@ -14,9 +21,9 @@ size_t MergeSorter::create_initial_runs(const std::string& input_file)
 
         std::sort(chunk.begin(), chunk.begin() + records_read);
 
-        std::string run_file = tmp_dir_ + "pass0/run_" + std::to_string(run_count) + ".dat";
+        std::string run_file = next_dir + std::to_string(run_count) + ".dat";
         FileWriter writer(run_file);
-        auto output = std::span<const Record>(chunk.data(), records_read);
+        auto output = std::span<Record>(chunk.data(), records_read);
         writer.write_chunk(output);
 
         disk_writes_ += writer.total_writes();
@@ -25,10 +32,103 @@ size_t MergeSorter::create_initial_runs(const std::string& input_file)
 
     return run_count;
 }
+
+void MergeSorter::perform_k_way_merge(std::vector<std::unique_ptr<FileReader>>& readers,
+    FileWriter& writer)
+{
+    using HeapElement = std::pair<Record, size_t>;
+    auto heap_compare = [](const HeapElement& a, const HeapElement& b) {
+        return !(a.first < b.first);
+    };
+    std::priority_queue<HeapElement, std::vector<HeapElement>, decltype(heap_compare)>
+        min_heap(heap_compare);
+
+    std::vector<std::span<Record>> buffers;
+    std::vector<size_t> positions;
+    std::vector<size_t> chunk_sizes;
+
+    buffers.reserve(readers.size());
+    positions.reserve(readers.size());
+    chunk_sizes.reserve(readers.size());
+
+    for (size_t i = 0; i < readers.size(); ++i) {
+        buffers.push_back(get_chunk_span(i));
+        chunk_sizes.push_back(readers[i]->read_chunk(buffers.back()));
+        positions.push_back(0);
+
+        if (chunk_sizes[i] > 0) {
+            min_heap.emplace(buffers[i][0], i);
+            positions[i] = 1;
+        }
+    }
+
+    auto output_buffer = get_chunk_span(buffer_rows_ - 1);
+    size_t output_pos = 0;
+
+    while (!min_heap.empty()) {
+        auto [record, source_idx] = min_heap.top();
+        min_heap.pop();
+
+        output_buffer[output_pos++] = record;
+
+        if (output_pos >= output_buffer.size()) {
+            writer.write_chunk(output_buffer);
+            output_pos = 0;
+        }
+
+        if (positions[source_idx] >= chunk_sizes[source_idx]) {
+            chunk_sizes[source_idx] = readers[source_idx]->read_chunk(buffers[source_idx]);
+            positions[source_idx] = 0;
+            if (chunk_sizes[source_idx] == 0)
+                continue;
+        }
+
+        if (positions[source_idx] < chunk_sizes[source_idx]) {
+            min_heap.emplace(buffers[source_idx][positions[source_idx]], source_idx);
+            positions[source_idx]++;
+        }
+    }
+
+    if (output_pos > 0) {
+        writer.write_chunk(std::span<Record>(output_buffer.data(), output_pos));
+    }
+}
+
+size_t MergeSorter::merge_pass(const std::string& current_dir, const std::string& next_dir, size_t run_count)
+{
+    const size_t merge_factor = buffer_rows_ - 1;
+    const size_t chunk_size = buffer_.size() / buffer_rows_;
+    size_t new_run_count = 0;
+
+    for (size_t i = 0; i < run_count; i += merge_factor) {
+        size_t end = std::min(i + merge_factor, run_count);
+
+        std::vector<std::unique_ptr<FileReader>> readers;
+        for (size_t j = i; j < end; ++j) {
+            std::string input_filename = current_dir + std::to_string(j) + ".dat";
+            readers.push_back(std::make_unique<FileReader>(input_filename, chunk_size));
+        }
+
+        std::string output_filename = next_dir + std::to_string(new_run_count) + ".dat";
+        FileWriter writer(output_filename, chunk_size);
+
+        perform_k_way_merge(readers, writer);
+
+        for (const auto& reader : readers) {
+            disk_reads_ += reader->total_reads();
+        }
+        disk_writes_ += writer.total_writes();
+
+        new_run_count++;
+    }
+
+    return new_run_count;
+}
+
 MergeSorter::MergeSorter(size_t buffer_rows, size_t buffer_cols, const std::string& tmp_dir)
     : buffer_(buffer_rows * buffer_cols)
     , tmp_dir_(tmp_dir)
-    , buffer_rows(buffer_rows)
+    , buffer_rows_(buffer_rows)
     , disk_reads_(0)
     , disk_writes_(0)
     , phases_(0)
@@ -40,8 +140,21 @@ MergeSorter::MergeSorter(size_t buffer_rows, size_t buffer_cols, const std::stri
 
 bool MergeSorter::sort_file(const std::string& input_file, const std::string& output_file)
 {
-    create_initial_runs(input_file);
+    std::string current_dir = tmp_dir_ + "/pass0/run_";
+    std::string next_dir = tmp_dir_ + "/pass1/run_";
+    size_t run_count = 0;
+
+    run_count = create_initial_runs(input_file, current_dir);
+    phases_++;
     std::cout << "created inital runs, disk reads: " << disk_reads_ << ", disk writes: " << disk_writes_ << '\n';
+
+    while (run_count > 1) {
+        run_count = merge_pass(current_dir, next_dir, run_count);
+        phases_++;
+        current_dir = next_dir;
+        next_dir = tmp_dir_ + "/pass" + std::to_string(phases_) + "/run_";
+    }
+
     std::cout << output_file << '\n';
     return true;
 }