files

include/buffer.hpp

@@ -0,0 +1,69 @@
+#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

@@ -0,0 +1,70 @@
+#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 directory = "tmp";
+    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)
+    {
+        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/reader.hpp

@@ -0,0 +1,79 @@
+// #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

@@ -0,0 +1,53 @@
+#pragma once
+#include <array>
+#include <iostream>
+#include <random>
+
+class Record {
+public:
+    std::array<int, 5> a;
+    int x;
+
+    int 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 operator<(const Record& other) const
+    {
+        return evaluate() < other.evaluate();
+    }
+
+    friend 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;
+    }
+
+    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

@@ -0,0 +1,63 @@
+// #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;
+    }
+};

src/main.cpp

@@ -0,0 +1,67 @@
+#include "config.hpp"
+#include "record.hpp"
+#include "sort.cpp"
+#include "writer.hpp"
+#include <fstream>
+#include <iostream>
+#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();
+    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();
+        output.write(rec);
+    }
+}
+
+void read_and_evaluate(std::string filename, int n)
+{
+    Buffer main_buffer(n);
+    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);
+
+        if (records_read == 0) {
+            break;
+        }
+
+        for (size_t i = 0; i < records_read; ++i) {
+            const Record& rec = buff[i];
+            std::cout << "[Evaluate i = " << i << ": " << rec.evaluate() << "] " << rec << "\n";
+        }
+    }
+}
+
+int main(int argc, char** argv)
+{
+    try {
+        Configuration opts = Configuration::parse_args(argc, argv);
+
+        if (opts.generate_data) {
+            generate_file(opts.N, opts.output_file);
+            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;
+    }
+}

src/sort.cpp

@@ -0,0 +1,188 @@
+#pragma once
+#include "buffer.hpp"
+#include "config.hpp"
+#include "reader.hpp"
+#include "writer.hpp"
+#include <fstream>
+#include <queue>
+#include <string>
+
+// returns amount of records read
+size_t read_chunk(Reader& reader, SubBuffer& buff)
+{
+    for (size_t i = 0; i < buff.size(); i++) {
+        std::optional<Record> in = reader.read();
+        if (!in.has_value()) {
+            return i;
+        }
+        buff[i] = in.value();
+    }
+    return buff.size();
+}
+
+void write_to_file(const std::string& filename, SubBuffer& buff, size_t n)
+{
+    std::filesystem::path dirPath = std::filesystem::path(filename).parent_path();
+    std::filesystem::create_directories(dirPath);
+
+    std::ofstream out_stream(filename);
+    Writer output_writer(out_stream);
+    for (size_t j = 0; j < n; j++) {
+        output_writer.write(buff[j]);
+    }
+}
+
+size_t create_initial_runs(const std::string& input_filename, const std::string& directory, Buffer& main_buffer)
+{
+    auto buff = main_buffer.divide(1)[0];
+
+    std::ifstream in_stream(input_filename);
+    Reader input_reader(in_stream);
+    size_t run_index;
+
+    for (run_index = 0;; run_index++) {
+        size_t size = read_chunk(input_reader, buff);
+        if (size == 0) {
+            break;
+        }
+        std::sort(buff.begin(), buff.begin() + size);
+        write_to_file(directory + "/" + std::to_string(run_index) + ".run", buff, size);
+    }
+
+    return run_index;
+}
+
+void write_block(Writer& w, SubBuffer& buff, size_t n)
+{
+    for (size_t i = 0; i < n; i++) {
+        w.write(buff[i]);
+    }
+}
+
+void k_way_merge_and_write(std::vector<Reader>& input_readers, Writer& output_writer, std::vector<SubBuffer>& buffers)
+{
+    // for debug purposes
+    std::cout << "buffers size: " << buffers.size() << " input_readers: " << input_readers.size() << '\n';
+    for (size_t i = 0; i < input_readers.size(); i++) {
+        std::cout << "reader " << i << ": " << input_readers[i].read().value() << '\n';
+    }
+
+    struct HeapElement {
+        Record record;
+        size_t source_index;
+        // priority queue is a max heap by default, so comparison is backwards
+        bool operator<(const HeapElement& other) const
+        {
+            return !(record < other.record);
+        }
+    };
+    using MinHeap = std::priority_queue<HeapElement>;
+    MinHeap pq;
+
+    struct BufferState {
+        size_t items_total = 0;
+        size_t current_idx = 0;
+    };
+
+    std::vector<BufferState> buffer_states(input_readers.size());
+
+    // initial fill
+    for (size_t i = 0; i < input_readers.size(); i++) {
+        SubBuffer& input_buffer = buffers[i];
+        buffer_states[i].items_total = read_chunk(input_readers[i], input_buffer);
+        buffer_states[i].current_idx = 0;
+
+        if (buffer_states[i].items_total == 0) {
+            continue;
+        }
+
+        pq.push({ input_buffer[0], i });
+        buffer_states[i].current_idx++;
+    }
+
+    size_t output_index = 0;
+    SubBuffer& output_buffer = buffers[buffers.size() - 1];
+
+    while (!pq.empty()) {
+        HeapElement next = pq.top();
+        pq.pop();
+
+        output_buffer[output_index] = next.record;
+        output_index++;
+
+        if (output_index == output_buffer.size()) {
+            write_block(output_writer, output_buffer, output_index);
+            output_index = 0;
+        }
+
+        size_t source_index = next.source_index;
+
+        if (buffer_states[source_index].current_idx >= buffer_states[source_index].items_total) {
+            buffer_states[source_index].items_total = read_chunk(input_readers[source_index], buffers[source_index]);
+            buffer_states[source_index].current_idx = 0;
+            if (buffer_states[source_index].items_total == 0) {
+                continue;
+            }
+        }
+
+        pq.push({ buffers[source_index][buffer_states[source_index].current_idx], source_index });
+        buffer_states[source_index].current_idx++;
+    }
+}
+
+// I think more or less works correctly now
+// creates input readers, and output writer for a merge to be performed by another function
+// if not every file can be processed at once, splits them
+size_t run_merge_pass(std::vector<SubBuffer>& buffers, std::string current_dir, std::string next_dir, size_t file_count)
+{
+    const size_t K = buffers.size() - 1; // Merge factor (e.g., 3)
+
+    size_t new_runs = 0;
+    for (size_t i = 0; i < file_count; i += K) {
+        // std::cout << "pass 0 run " << i << '\n';
+        // create input readers, to turn into a function i would have to some complicated ownership semantics, and i don't want to bother with that
+        std::vector<std::ifstream> in_streams;
+        std::vector<Reader> input_readers;
+        in_streams.reserve(K);
+        input_readers.reserve(K);
+        for (size_t j = 0; j < K; j++) {
+            std::string filename = current_dir + "/" + std::to_string(i + j) + ".run";
+            in_streams.emplace_back(std::ifstream(filename));
+            input_readers.emplace_back(Reader(in_streams[j]));
+        }
+
+        // create the output writer -||-
+        std::string filename = next_dir + "/" + std::to_string(new_runs) + ".run";
+        std::filesystem::path dirPath = std::filesystem::path(filename).parent_path();
+        std::filesystem::create_directories(dirPath);
+        std::ofstream out_stream(filename);
+        Writer output_writer(out_stream);
+
+        k_way_merge_and_write(input_readers, output_writer, buffers);
+        new_runs++;
+    }
+
+    return new_runs;
+}
+
+int sort_file(const Configuration& opts)
+{
+    Buffer main_buffer(opts.n * opts.b);
+    size_t file_count;
+
+    file_count = create_initial_runs(opts.input_file, opts.directory + "/pass0", main_buffer);
+
+    auto buffers = main_buffer.divide(opts.n);
+    for (int i = 0;; i++) {
+        std::string current_dir = opts.directory + "/pass" + std::to_string(i);
+        std::string next_dir = opts.directory + "/pass" + std::to_string(i + 1);
+        file_count = run_merge_pass(buffers, current_dir, next_dir, file_count);
+        if (file_count == 1) {
+            // #todo move the resulting file to the output dir
+            break;
+        }
+    }
+
+    // #todo cleanup tmp files
+    return 0;
+}