bronku/file_sorting
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

file sctructure

Browse source
This commit is contained in:
bronku 2025-11-15 14:09:33 +01:00
parent 5e8ebb6270
commit e34dc036b0
8 changed files with 13 additions and 601 deletions
Download patch file Download diff file Expand all files Collapse all files

69
buffer.hpp
View file

@ -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;
}
};

70
config.hpp
View file

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

67
main.cpp
View file

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

25
makefile
View file

@ -1,10 +1,14 @@
CXX = clang++
CFLAGS = -g -lc++ -std=c++17
build/main: main.cpp sort.cpp config.hpp buffer.hpp reader.hpp writer.hpp build
clang $(CFLAGS) main.cpp -o build/main
SRC := $(wildcard src/*.cpp)
TARGET = build/main
$(TARGET): $(SRC) | build
$(CXX) $(CXXFLAGS) $(SRC) -o $(TARGET)
build:
mkdir -p build
mkdir -p build build/run
clean:
rm -rf build
@ -13,15 +17,12 @@ generate: build/run/1.in
# actually should depend on build/main and build/run, but that will trigger it every recompile, and i don't need that
build/run/1.in:
build/main -g -o build/run/1.in
$(TARGET) -g -o build/run/1.in
run: build/main build/run/1.in
build/main -i build/run/1.in -d build/run/tmp
run: $(TARGET) build/run/1.in
$(TARGET) -i build/run/1.in -d build/run/tmp
run_alt: build/main build/run/1.in
build/main -i build/run/1.in -d build/run/tmp -n 11 -b 100
run_alt: $(TARGET) build/run/1.in
$(TARGET) -i build/run/1.in -d build/run/tmp -n 11 -b 100
debug: build/main tests/1.in
lldb -- build/main
.PHONY: clean run debug test generate run_alt
.PHONY: clean run generate run_alt

79
reader.hpp
View file

@ -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;
// }
};

53
record.hpp
View file

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

188
sort.cpp
View file

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

63
writer.hpp
View file

@ -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;
}
};