initial runs

.clangd

@@ -1,2 +1,5 @@
 CompileFlags:
-  Add: [-std=c++17]
+  Add:
+    - "-std=c++20"
+    - "-I./include"
+  CompilationDatabase: build

include/buffer.hpp

@@ -1,69 +0,0 @@
-#pragma once
-
-#include "record.hpp"
-#include <stdexcept>
-#include <vector>
-
-class Buffer;
-
-class SubBuffer {
-private:
-    Record* data_;
-    size_t size_;
-
-    friend class Buffer;
-
-    SubBuffer(Record* data, size_t size)
-        : data_(data)
-        , size_(size)
-    {
-    }
-
-public:
-    // so that i can create a  vecotr of set size, without the .reserve(), and other stuff
-    SubBuffer()
-    {
-    }
-    // for std::sort support
-    using iterator = Record*;
-    iterator begin() { return data_; }
-    iterator end() { return data_ + size_; }
-
-    Record& operator[](size_t index)
-    {
-        return data_[index];
-    }
-
-    size_t size() const { return size_; }
-};
-
-class Buffer {
-private:
-    std::vector<Record> storage_;
-
-public:
-    explicit Buffer(size_t total_size)
-        : storage_(total_size)
-    {
-    }
-
-    std::vector<SubBuffer> divide(size_t n)
-    {
-        if (n == 0) {
-            throw std::invalid_argument("Cannot divide buffer into 0 pieces");
-        }
-
-        size_t total_size = storage_.size();
-        if (total_size % n != 0) {
-            throw std::invalid_argument("Size not divisible by n");
-        }
-
-        std::vector<SubBuffer> result(n);
-        size_t piece_size = total_size / n;
-        for (size_t i = 0; i < n; ++i) {
-            result[i] = SubBuffer(storage_.data() + i * piece_size, piece_size);
-        }
-
-        return result;
-    }
-};

include/config.hpp

@@ -1,70 +1,17 @@
 #pragma once
 #include <getopt.h>
-#include <iostream>
-#include <stdexcept>
 #include <string>
 
 class Configuration {
 public:
     std::string input_file = "1.in";
-    std::string output_file = "1.in";
+    std::string output_file = "1.out";
-    std::string directory = "tmp";
+    std::string tmp_dir = "build/run";
     bool generate_data = false;
     bool evaluate_file = false;
     int N = 100000; // number of records in a file
     int n = 101; // number of buffers
     int b = 10; // blocking factor
 
-    static Configuration parse_args(int argc, char** argv)
+    Configuration(int argc, char** argv);
-    {
-        if (argc <= 1) {
-            std::cerr << "Please specify input arguments\n";
-            std::cerr << "Usage: " << argv[0] << " [-g] [-e] [-N count] [-i input] [-o output] [-d directory] [-n buffers] [-b block_size]\n";
-            throw std::invalid_argument("No input arguments");
-        }
-        Configuration config;
-        bool input_set = false;
-        int opt;
-
-        while ((opt = getopt(argc, argv, "i:o:d:geN:n:b:")) != -1) {
-            switch (opt) {
-            case 'i':
-                config.input_file = optarg;
-                input_set = true;
-                break;
-            case 'o':
-                config.output_file = optarg;
-                break;
-            case 'd':
-                config.directory = optarg;
-                break;
-            case 'g':
-                config.generate_data = true;
-                break;
-            case 'e':
-                config.evaluate_file = true;
-                break;
-            case 'N':
-                config.N = std::stoi(optarg);
-                break;
-            case 'n':
-                config.n = std::stoi(optarg);
-                break;
-            case 'b':
-                config.b = std::stoi(optarg);
-                break;
-            case '?':
-                throw std::invalid_argument("Invalid command line argument");
-            }
-        }
-
-        if (config.generate_data && input_set) {
-            throw std::invalid_argument("Cannot specify both -g and -i options");
-        }
-        if (config.generate_data && config.evaluate_file) {
-            throw std::invalid_argument("Cannot specify both -g and -e options");
-        }
-
-        return config;
-    }
 };

include/file_reader.hpp

@@ -0,0 +1,22 @@
+#pragma once
+#include "record.hpp"
+#include <fstream>
+#include <optional>
+#include <span>
+
+class FileReader {
+private:
+    std::ifstream stream_;
+    std::vector<Record> buffer_;
+    size_t current_pos_;
+    size_t valid_items_;
+    size_t total_reads_;
+
+    void refill_buffer();
+
+public:
+    FileReader(const std::string& filename, size_t buffer_size = 8);
+    std::optional<Record> read();
+    size_t read_chunk(std::span<Record>& chunk);
+    size_t total_reads() const;
+};

include/file_writer.hpp

@@ -0,0 +1,23 @@
+#pragma once
+#include "record.hpp"
+#include <fstream>
+#include <span>
+
+class FileWriter {
+private:
+    std::ofstream stream_;
+    std::vector<Record> buffer_;
+    size_t current_pos_;
+    size_t total_writes_;
+
+    void write_buffer();
+
+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);
+
+    size_t total_writes() const;
+};

include/merge_sorter.hpp

@@ -0,0 +1,29 @@
+#pragma once
+#include "file_reader.hpp"
+#include "file_writer.hpp"
+#include "record.hpp"
+#include <vector>
+
+class MergeSorter {
+private:
+    std::vector<Record> buffer_;
+    const std::string& tmp_dir_;
+    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);
+    void perform_k_way_merge(std::vector<std::unique_ptr<FileReader>>& readers,
+        FileWriter& writer);
+
+public:
+    MergeSorter(size_t buffer_rows, size_t buffer_cols, const std::string& tmp_dir);
+    bool sort_file(const std::string& input_file, const std::string& output_file);
+
+    size_t disk_reads() const;
+    size_t disk_writes() const;
+    size_t phases() const;
+};

include/reader.hpp

@@ -1,79 +0,0 @@
-// #todo maybe overload >> operators insted of read()
-#pragma once
-#include "record.hpp"
-#include <istream>
-#include <optional>
-#include <vector>
-
-class Reader {
-private:
-    std::vector<Record> read_buffer;
-    std::istream& input_stream;
-    size_t buffer_size;
-    size_t total_reads_count;
-    int last_element_idx; // last valid element in buffer
-    int current_idx; // current reading position
-
-    // returns true if at least one record was returned
-    bool refill_buffer()
-    {
-        last_element_idx = -1;
-        current_idx = 0;
-
-        for (size_t i = 0; i < buffer_size; i++) {
-            if (!(input_stream >> read_buffer[i])) {
-                break;
-            }
-            last_element_idx = i;
-        }
-
-        return last_element_idx >= 0;
-    }
-
-public:
-    explicit Reader(std::istream& input_stream, size_t buffer_size = 8)
-        : read_buffer(buffer_size)
-        , input_stream(input_stream)
-        , buffer_size(buffer_size)
-        , total_reads_count(0)
-        , last_element_idx(-1)
-        , current_idx(0)
-    {
-    }
-
-    std::optional<Record> read()
-    {
-        if (current_idx > last_element_idx) {
-            bool read_new_items = refill_buffer();
-            if (!read_new_items) {
-                return std::nullopt;
-            }
-        }
-
-        Record record = read_buffer[current_idx];
-        current_idx++;
-        total_reads_count++;
-
-        return record;
-    }
-
-    size_t total_reads() const
-    {
-        return total_reads_count;
-    }
-
-    // #todo remove if not used in the end
-    // bool has_more() const
-    // {
-    //     return current_idx <= last_element_idx || !input_stream.eof();
-    // }
-
-    // void reset()
-    // {
-    //     input_stream.clear();
-    //     input_stream.seekg(0);
-    //     last_element_idx = -1;
-    //     current_idx = 0;
-    //     total_reads_count = 0;
-    // }
-};

include/record.hpp

@@ -1,53 +1,17 @@
 #pragma once
 #include <array>
 #include <iostream>
-#include <random>
 
 class Record {
 public:
     std::array<int, 5> a;
     int x;
 
-    int evaluate() const
+    int evaluate() const;
-    {
+    bool operator<(const Record& other) const;
-        int out = 0;
-        int x_n = 1;
-        for (int i = 0; i < 5; i++) {
-            out += a[i] * x_n;
-            x_n *= x;
-        }
-        return out;
-    }
 
-    bool operator<(const Record& other) const
+    friend std::ostream& operator<<(std::ostream& os, const Record& r);
-    {
+    friend std::istream& operator>>(std::istream& is, Record& r);
-        return evaluate() < other.evaluate();
-    }
 
-    friend std::ostream& operator<<(std::ostream& os, const Record& r)
+    static Record random();
-    {
+};
-        os << r.a[0] << ' ' << r.a[1] << ' ' << r.a[2] << ' '
-           << r.a[3] << ' ' << r.a[4] << ' ' << r.x;
-        return os;
-    }
-
-    friend std::istream& operator>>(std::istream& is, Record& r)
-    {
-        is >> r.a[0] >> r.a[1] >> r.a[2] >> r.a[3] >> r.a[4] >> r.x;
-        return is;
-    }
-
-    static Record random()
-    {
-        static std::random_device rd;
-        static std::mt19937 gen(rd());
-        static std::uniform_int_distribution<int> dist;
-
-        Record r;
-        for (int& val : r.a) {
-            val = dist(gen);
-        }
-        r.x = dist(gen);
-        return r;
-    }
-};

include/writer.hpp

@@ -1,63 +0,0 @@
-// #todo maybe overload << operators insted of write()
-#pragma once
-#include "record.hpp"
-#include <ostream>
-#include <vector>
-
-class Writer {
-private:
-    std::vector<Record> write_buffer;
-    std::ostream& output_stream;
-    size_t buffer_size;
-    size_t total_writes_count;
-    size_t current_index;
-
-    // returns true on success
-    // doesn't really flush for performace reasons
-    // could be changed to really flush to monitor the impact of amount of writes
-    bool flush_buffer()
-    {
-        for (size_t i = 0; i < current_index; ++i) {
-            if (!(output_stream << write_buffer[i] << " ")) {
-                return false;
-            }
-        }
-
-        current_index = 0;
-
-        return true;
-    }
-
-public:
-    explicit Writer(std::ostream& output_stream, size_t buffer_size = 8)
-        : write_buffer(buffer_size)
-        , output_stream(output_stream)
-        , buffer_size(buffer_size)
-        , total_writes_count(0)
-        , current_index(0)
-    {
-    }
-
-    ~Writer()
-    {
-        flush_buffer();
-    }
-
-    bool write(const Record& record)
-    {
-        write_buffer[current_index] = record;
-        current_index++;
-        total_writes_count++;
-
-        if (current_index >= buffer_size) {
-            return flush_buffer();
-        }
-
-        return true;
-    }
-
-    size_t total_writes() const
-    {
-        return total_writes_count;
-    }
-};

makefile

@@ -1,11 +1,18 @@
 CXX = clang++
-CFLAGS = -g -lc++ -std=c++17
+CXXFLAGS = -g -std=c++20 -Iinclude -Wall -Wextra -Werror
+LDFLAGS =
 
 SRC := $(wildcard src/*.cpp)
+OBJ := $(SRC:src/%.cpp=build/%.o)
 TARGET = build/main
 
-$(TARGET): $(SRC) | build
+all: $(TARGET)
-	$(CXX) $(CXXFLAGS) $(SRC) -o $(TARGET)
+
+$(TARGET): $(OBJ)
+	$(CXX) $(CXXFLAGS) $^ -o $@ $(LDFLAGS)
+
+build/%.o: src/%.cpp | build
+	$(CXX) $(CXXFLAGS) -c $< -o $@
 
 build:
 	mkdir -p build build/run
@@ -25,4 +32,4 @@ run: $(TARGET) build/run/1.in
 run_alt: $(TARGET) build/run/1.in
 	$(TARGET) -i build/run/1.in -d build/run/tmp -n 11 -b 100
 
-.PHONY: clean run generate run_alt
+.PHONY: clean run generate run_alt all

src/config.cpp

@@ -0,0 +1,55 @@
+#include "../include/config.hpp"
+#include <iostream>
+#include <stdexcept>
+
+Configuration::Configuration(int argc, char** argv)
+{
+    if (argc <= 1) {
+        std::cerr << "Please specify input arguments\n";
+        std::cerr << "Usage: " << argv[0] << " [-g] [-e] [-N count] [-i input] [-o output] [-d dir] [-n buffers] [-b block_size]\n";
+        throw std::invalid_argument("No input arguments");
+    }
+    bool input_set = false;
+    int opt;
+
+    while ((opt = getopt(argc, argv, "i:d:o:geN:n:b:")) != -1) {
+        switch (opt) {
+        case 'i':
+            input_file = optarg;
+            input_set = true;
+            break;
+        case 'o':
+            output_file = optarg;
+            break;
+        case 'd':
+            tmp_dir = optarg;
+            break;
+        case 'g':
+            generate_data = true;
+            break;
+        case 'e':
+            evaluate_file = true;
+            break;
+        case 'N':
+            N = std::stoi(optarg);
+            break;
+        case 'n':
+            n = std::stoi(optarg);
+            break;
+        case 'b':
+            b = std::stoi(optarg);
+            break;
+        case '?':
+            throw std::invalid_argument("Invalid command line argument");
+        default:
+            std::cerr << "Please specify input arguments\n";
+        }
+    }
+
+    if (generate_data && input_set) {
+        throw std::invalid_argument("Cannot specify both -g and -i options");
+    }
+    if (generate_data && evaluate_file) {
+        throw std::invalid_argument("Cannot specify both -g and -e options");
+    }
+}

src/file_reader.cpp

@@ -0,0 +1,62 @@
+#include "../include/file_reader.hpp"
+
+void FileReader::refill_buffer()
+{
+    current_pos_ = 0;
+    valid_items_ = 0;
+
+    for (size_t i = 0; i < buffer_.size(); ++i) {
+        Record record;
+        if (!(stream_ >> record)) {
+            break;
+        }
+        buffer_[i] = record;
+        valid_items_ = i + 1;
+    }
+}
+
+FileReader::FileReader(const std::string& filename, size_t buffer_size)
+    : stream_(filename)
+    , buffer_(buffer_size)
+    , current_pos_(0)
+    , valid_items_(0)
+    , total_reads_(0)
+{
+    refill_buffer();
+}
+
+std::optional<Record> FileReader::read()
+{
+    if (current_pos_ >= valid_items_) {
+        refill_buffer();
+    }
+
+    if (valid_items_ == 0) {
+        return std::nullopt;
+    }
+
+    Record record = buffer_[current_pos_];
+    current_pos_++;
+    total_reads_++;
+
+    return record;
+}
+
+size_t FileReader::read_chunk(std::span<Record>& chunk)
+{
+    size_t count = 0;
+    for (size_t i = 0; i < chunk.size(); i++) {
+        auto record = read();
+        if (!record.has_value()) {
+            break;
+        }
+        chunk[count] = record.value();
+        count++;
+    }
+    return count;
+}
+
+size_t FileReader::total_reads() const
+{
+    return total_reads_;
+}

src/file_writer.cpp

@@ -0,0 +1,47 @@
+#include "../include/file_writer.hpp"
+
+void FileWriter::write_buffer()
+{
+    for (size_t i = 0; i < current_pos_; i++) {
+        stream_ << buffer_[i] << "\n";
+    }
+    current_pos_ = 0;
+}
+
+FileWriter::FileWriter(const std::string& filename, size_t buffer_size)
+    : buffer_(buffer_size)
+    , current_pos_(0)
+    , total_writes_(0)
+{
+    std::filesystem::path file_path(filename);
+    std::filesystem::path dir_path = file_path.parent_path();
+    std::filesystem::create_directories(dir_path);
+    stream_.open(filename);
+}
+
+FileWriter::~FileWriter()
+{
+    write_buffer();
+}
+
+void FileWriter::write(const Record& record)
+{
+    if (current_pos_ >= buffer_.size()) {
+        write_buffer();
+    }
+    buffer_[current_pos_] = record;
+    current_pos_++;
+    total_writes_++;
+}
+
+void FileWriter::write_chunk(std::span<const Record>& chunk)
+{
+    for (const auto& record : chunk) {
+        write(record);
+    }
+}
+
+size_t FileWriter::total_writes() const
+{
+    return total_writes_;
+}

src/main.cpp

@@ -1,19 +1,16 @@
-#include "config.hpp"
+#include "../include/config.hpp"
-#include "record.hpp"
+#include "../include/file_reader.hpp"
-#include "sort.cpp"
+#include "../include/file_writer.hpp"
-#include "writer.hpp"
+#include "../include/merge_sorter.hpp"
-#include <fstream>
+#include "../include/record.hpp"
 #include <iostream>
+#include <string>
 #include <sys/stat.h>
 #include <sys/types.h>
 
 void generate_file(int N, const std::string& filename)
 {
-    std::filesystem::path dirPath = std::filesystem::path(filename).parent_path();
+    FileWriter output(filename);
-    std::filesystem::create_directories(dirPath);
-
-    std::ofstream out_stream(filename);
-    Writer output(out_stream);
 
     for (int i = 0; i < N; i++) {
         Record rec = Record::random();
@@ -21,47 +18,33 @@ void generate_file(int N, const std::string& filename)
     }
 }
 
-void read_and_evaluate(std::string filename, int n)
+void read_and_evaluate(std::string filename)
 {
-    Buffer main_buffer(n);
+    FileReader input(filename);
-    auto buff = main_buffer.divide(1)[0];
-
-    std::ifstream in_stream(filename);
-    Reader input_reader(in_stream);
-
     while (true) {
-        size_t records_read = read_chunk(input_reader, buff);
+        auto rec = input.read();
-
+        if (!rec.has_value()) {
-        if (records_read == 0) {
+            return;
-            break;
         }
 
-        for (size_t i = 0; i < records_read; ++i) {
+        std::cout << "[Evaluate: " << rec.value().evaluate() << "] " << rec.value() << "\n";
-            const Record& rec = buff[i];
-            std::cout << "[Evaluate i = " << i << ": " << rec.evaluate() << "] " << rec << "\n";
-        }
     }
 }
 
 int main(int argc, char** argv)
 {
-    try {
+    Configuration opts(argc, argv);
-        Configuration opts = Configuration::parse_args(argc, argv);
 
-        if (opts.generate_data) {
+    if (opts.generate_data) {
-            generate_file(opts.N, opts.output_file);
+        generate_file(opts.N, opts.output_file);
-            return 0;
+        return 0;
-        }
-
-        if (opts.evaluate_file) {
-            read_and_evaluate(opts.input_file, opts.N);
-            return 0;
-        }
-
-        sort_file(opts);
-
-    } catch (const std::exception& e) {
-        std::cerr << "Error: " << e.what() << '\n';
-        return 1;
     }
+
+    if (opts.evaluate_file) {
+        read_and_evaluate(opts.input_file);
+        return 0;
+    }
+
+    MergeSorter sorter(opts.b, opts.n, opts.tmp_dir);
+    sorter.sort_file(opts.input_file, opts.output_file);
 }

src/merge_sorter.cpp

@@ -0,0 +1,46 @@
+#include "../include/merge_sorter.hpp"
+
+size_t MergeSorter::create_initial_runs(const std::string& input_file)
+{
+    FileReader reader(input_file);
+    std::span<Record> chunk(buffer_);
+    size_t run_count = 0;
+
+    for (run_count = 0;; run_count++) {
+        size_t records_read = reader.read_chunk(chunk);
+        if (records_read == 0) {
+            break;
+        }
+
+        std::sort(chunk.begin(), chunk.begin() + records_read);
+
+        std::string run_file = tmp_dir_ + "pass0/run_" + std::to_string(run_count) + ".dat";
+        FileWriter writer(run_file);
+        auto output = std::span<const Record>(chunk.data(), records_read);
+        writer.write_chunk(output);
+
+        disk_reads_ += reader.total_reads() - disk_reads_;
+        disk_writes_ += writer.total_writes() - disk_writes_;
+    }
+
+    return 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)
+    , disk_reads_(0)
+    , disk_writes_(0)
+    , phases_(0)
+{
+    if (buffer_rows < 3) {
+        throw std::invalid_argument("Buffer must have at least 3 rows for merging");
+    }
+}
+
+bool MergeSorter::sort_file(const std::string& input_file, const std::string& output_file)
+{
+    create_initial_runs(input_file);
+    std::cout << output_file << '\n';
+    return true;
+}

src/record.cpp

@@ -0,0 +1,46 @@
+#include "../include/record.hpp"
+#include <random>
+
+int Record::evaluate() const
+{
+    int out = 0;
+    int x_n = 1;
+    for (int i = 0; i < 5; i++) {
+        out += a[i] * x_n;
+        x_n *= x;
+    }
+    return out;
+}
+
+bool Record::operator<(const Record& other) const
+{
+    return evaluate() < other.evaluate();
+}
+
+Record Record::random()
+{
+    static std::random_device rd;
+    static std::mt19937 gen(rd());
+    static std::uniform_int_distribution<int> dist;
+
+    Record r {};
+    for (int& val : r.a) {
+        val = dist(gen);
+    }
+    r.x = dist(gen);
+    return r;
+}
+
+std::ostream& operator<<(std::ostream& os, const Record& r)
+{
+
+    os << r.a[0] << ' ' << r.a[1] << ' ' << r.a[2] << ' '
+       << r.a[3] << ' ' << r.a[4] << ' ' << r.x;
+    return os;
+}
+
+std::istream& operator>>(std::istream& is, Record& r)
+{
+    is >> r.a[0] >> r.a[1] >> r.a[2] >> r.a[3] >> r.a[4] >> r.x;
+    return is;
+}