Агрегация

README.md

@@ -10,10 +10,30 @@
 не менее чем на 10% от даты начала интервала, вместе с минимальными и максимальными 
 значениями Open и Close за все дни внутри интервала.
 
+## Параллельное чтение данных
+
+Нет смысла параллельно читать данные из NFS, так как в реальности файлы с данными
+будут лежать только на NFS сервере. То есть другие узлы лишь отправляют сетевые запросы
+на NFS сервер, который уже читает реальные данные с диска и лишь затем отправляет
+их другим узлам.
+
+Чтобы этого избежать, нужно на всех машинах скопировать файлы с данными в их реальные
+файловые системы. Например в папку `/data`.
+
+```sh
+# На каждом узле создаем директорию /data
+sudo mkdir /data
+sudo chown $USER /data
+
+# Копируем данные
+cd /mnt/shared/supercomputers/data
+cp data.csv /data/
+```
+
 ## Сборка
 
 Проект обязательно должен быть расположен в общей директории для всех узлов,
-например, в `/mnt/shared/supercomputers/bitcoin-project/build`. 
+например, в `/mnt/shared/supercomputers/build`. 
 Перед запуском указать актуальный путь в `run.slurm`.
 
 ```sh

run.slurm

@@ -5,8 +5,14 @@
 #SBATCH --cpus-per-task=2
 #SBATCH --output=out.txt
 
-# Количество CPU потоков на узел (должно соответствовать cpus-per-task)
+# Путь к файлу данных (должен существовать на всех узлах)
-export NUM_CPU_THREADS=2
+export DATA_PATH="/mnt/shared/supercomputers/data/data.csv"
+
+# Доли данных для каждого ранка (сумма определяет пропорции)
+export DATA_READ_SHARES="10,12,13,13"
+
+# Размер перекрытия в байтах для обработки границ строк
+export READ_OVERLAP_BYTES=131072
 
 cd /mnt/shared/supercomputers/build
 mpirun -np $SLURM_NTASKS ./bitcoin_app

src/aggregation.cpp

@@ -1,87 +1,48 @@
 #include "aggregation.hpp"
+#include <map>
 #include <algorithm>
 #include <limits>
-#include <cmath>
 
 std::vector<DayStats> aggregate_days(const std::vector<Record>& records) {
-    // Группируем записи по дням
+    // Накопители для каждого дня
-    std::map<DayIndex, std::vector<const Record*>> day_records;
+    struct DayAccumulator {
+        double avg_sum = 0.0;
+        double open_min = std::numeric_limits<double>::max();
+        double open_max = std::numeric_limits<double>::lowest();
+        double close_min = std::numeric_limits<double>::max();
+        double close_max = std::numeric_limits<double>::lowest();
+        int64_t count = 0;
+    };
+    
+    std::map<DayIndex, DayAccumulator> days;
     
     for (const auto& r : records) {
         DayIndex day = static_cast<DayIndex>(r.timestamp) / 86400;
-        day_records[day].push_back(&r);
+        auto& acc = days[day];
+        
+        double avg = (r.low + r.high) / 2.0;
+        acc.avg_sum += avg;
+        acc.open_min = std::min(acc.open_min, r.open);
+        acc.open_max = std::max(acc.open_max, r.open);
+        acc.close_min = std::min(acc.close_min, r.close);
+        acc.close_max = std::max(acc.close_max, r.close);
+        acc.count++;
     }
     
     std::vector<DayStats> result;
-    result.reserve(day_records.size());
+    result.reserve(days.size());
     
-    for (auto& [day, recs] : day_records) {
+    for (const auto& [day, acc] : days) {
-        // Сортируем по timestamp для определения first/last
-        std::sort(recs.begin(), recs.end(), 
-            [](const Record* a, const Record* b) {
-                return a->timestamp < b->timestamp;
-            });
-        
         DayStats stats;
         stats.day = day;
-        stats.low = std::numeric_limits<double>::max();
+        stats.avg = acc.avg_sum / static_cast<double>(acc.count);
-        stats.high = std::numeric_limits<double>::lowest();
+        stats.open_min = acc.open_min;
-        stats.open = recs.front()->open;
+        stats.open_max = acc.open_max;
-        stats.close = recs.back()->close;
+        stats.close_min = acc.close_min;
-        stats.first_ts = recs.front()->timestamp;
+        stats.close_max = acc.close_max;
-        stats.last_ts = recs.back()->timestamp;
+        stats.count = acc.count;
-        
-        for (const auto* r : recs) {
-            stats.low = std::min(stats.low, r->low);
-            stats.high = std::max(stats.high, r->high);
-        }
-        
-        stats.avg = (stats.low + stats.high) / 2.0;
-        
         result.push_back(stats);
     }
     
     return result;
 }
-
-std::vector<DayStats> merge_day_stats(const std::vector<DayStats>& all_stats) {
-    // Объединяем статистику по одинаковым дням (если такие есть)
-    std::map<DayIndex, DayStats> merged;
-    
-    for (const auto& s : all_stats) {
-        auto it = merged.find(s.day);
-        if (it == merged.end()) {
-            merged[s.day] = s;
-        } else {
-            // Объединяем данные за один день
-            auto& m = it->second;
-            m.low = std::min(m.low, s.low);
-            m.high = std::max(m.high, s.high);
-            
-            // open берём от записи с меньшим timestamp
-            if (s.first_ts < m.first_ts) {
-                m.open = s.open;
-                m.first_ts = s.first_ts;
-            }
-            
-            // close берём от записи с большим timestamp
-            if (s.last_ts > m.last_ts) {
-                m.close = s.close;
-                m.last_ts = s.last_ts;
-            }
-            
-            m.avg = (m.low + m.high) / 2.0;
-        }
-    }
-    
-    // Преобразуем в отсортированный вектор
-    std::vector<DayStats> result;
-    result.reserve(merged.size());
-    
-    for (auto& [day, stats] : merged) {
-        result.push_back(stats);
-    }
-    
-    return result;
-}
-

src/aggregation.hpp

@@ -3,12 +3,6 @@
 #include "record.hpp"
 #include "day_stats.hpp"
 #include <vector>
-#include <map>
 
 // Агрегация записей по дням на одном узле
 std::vector<DayStats> aggregate_days(const std::vector<Record>& records);
-
-// Объединение агрегированных данных с разных узлов
-// (на случай если один день попал на разные узлы - но в нашей схеме это не должно случиться)
-std::vector<DayStats> merge_day_stats(const std::vector<DayStats>& all_stats);
-

src/csv_loader.cpp

@@ -2,46 +2,134 @@
 #include <fstream>
 #include <sstream>
 #include <iostream>
+#include <stdexcept>
 
-std::vector<Record> load_csv(const std::string& filename) {
+bool parse_csv_line(const std::string& line, Record& record) {
+    if (line.empty()) {
+        return false;
+    }
+    
+    std::stringstream ss(line);
+    std::string item;
+    
+    try {
+        // timestamp
+        if (!std::getline(ss, item, ',') || item.empty()) return false;
+        record.timestamp = std::stod(item);
+        
+        // open
+        if (!std::getline(ss, item, ',') || item.empty()) return false;
+        record.open = std::stod(item);
+        
+        // high
+        if (!std::getline(ss, item, ',') || item.empty()) return false;
+        record.high = std::stod(item);
+        
+        // low
+        if (!std::getline(ss, item, ',') || item.empty()) return false;
+        record.low = std::stod(item);
+        
+        // close
+        if (!std::getline(ss, item, ',') || item.empty()) return false;
+        record.close = std::stod(item);
+        
+        // volume
+        if (!std::getline(ss, item, ',')) return false;
+        // Volume может быть пустым или содержать данные
+        if (item.empty()) {
+            record.volume = 0.0;
+        } else {
+            record.volume = std::stod(item);
+        }
+        
+        return true;
+    } catch (const std::exception&) {
+        return false;
+    }
+}
+
+std::vector<Record> load_csv_parallel(int rank, int size) {
     std::vector<Record> data;
-    std::ifstream file(filename);
+    
-
+    // Читаем настройки из переменных окружения
+    std::string data_path = get_data_path();
+    std::vector<int> shares = get_data_read_shares();
+    int64_t overlap_bytes = get_read_overlap_bytes();
+    
+    // Получаем размер файла
+    int64_t file_size = get_file_size(data_path);
+    
+    // Вычисляем диапазон байт для этого ранка
+    ByteRange range = calculate_byte_range(rank, size, file_size, shares, overlap_bytes);
+    
+    // Открываем файл и читаем нужный диапазон
+    std::ifstream file(data_path, std::ios::binary);
     if (!file.is_open()) {
-        throw std::runtime_error("Cannot open file: " + filename);
+        throw std::runtime_error("Cannot open file: " + data_path);
     }
-
+    
-    std::string line;
+    // Переходим к началу диапазона
-
+    file.seekg(range.start);
-    // читаем первую строку (заголовок)
+    
-    std::getline(file, line);
+    // Читаем данные в буфер
-
+    int64_t bytes_to_read = range.end - range.start;
-    while (std::getline(file, line)) {
+    std::vector<char> buffer(bytes_to_read);
-        std::stringstream ss(line);
+    file.read(buffer.data(), bytes_to_read);
-        std::string item;
+    int64_t bytes_read = file.gcount();
-
+    
-        Record row;
+    file.close();
-
+    
-        std::getline(ss, item, ',');
+    // Преобразуем в строку для удобства парсинга
-        row.timestamp = std::stod(item);
+    std::string content(buffer.data(), bytes_read);
-
+    
-        std::getline(ss, item, ',');
+    // Находим позицию начала первой полной строки
-        row.open = std::stod(item);
+    size_t parse_start = 0;
-
+    if (rank == 0) {
-        std::getline(ss, item, ',');
+        // Первый ранк: пропускаем заголовок (первую строку)
-        row.high = std::stod(item);
+        size_t header_end = content.find('\n');
-
+        if (header_end != std::string::npos) {
-        std::getline(ss, item, ',');
+            parse_start = header_end + 1;
-        row.low = std::stod(item);
+        }
-
+    } else {
-        std::getline(ss, item, ',');
+        // Остальные ранки: начинаем с первого \n (пропускаем неполную строку)
-        row.close = std::stod(item);
+        size_t first_newline = content.find('\n');
-
+        if (first_newline != std::string::npos) {
-        std::getline(ss, item, ',');
+            parse_start = first_newline + 1;
-        row.volume = std::stod(item);
+        }
-
-        data.push_back(row);
     }
-
+    
+    // Находим позицию конца последней полной строки
+    size_t parse_end = content.size();
+    if (rank != size - 1) {
+        // Не последний ранк: ищем последний \n
+        size_t last_newline = content.rfind('\n');
+        if (last_newline != std::string::npos && last_newline > parse_start) {
+            parse_end = last_newline;
+        }
+    }
+    
+    // Парсим строки
+    size_t pos = parse_start;
+    while (pos < parse_end) {
+        size_t line_end = content.find('\n', pos);
+        if (line_end == std::string::npos || line_end > parse_end) {
+            line_end = parse_end;
+        }
+        
+        std::string line = content.substr(pos, line_end - pos);
+        
+        // Убираем \r если есть (Windows line endings)
+        if (!line.empty() && line.back() == '\r') {
+            line.pop_back();
+        }
+        
+        Record record;
+        if (parse_csv_line(line, record)) {
+            data.push_back(record);
+        }
+        
+        pos = line_end + 1;
+    }
+    
     return data;
 }

src/csv_loader.hpp

@@ -2,5 +2,14 @@
 #include <string>
 #include <vector>
 #include "record.hpp"
+#include "utils.hpp"
 
-std::vector<Record> load_csv(const std::string& filename);
+// Параллельное чтение CSV файла для MPI
+// rank - номер текущего ранка
+// size - общее количество ранков
+// Возвращает вектор записей, прочитанных этим ранком
+std::vector<Record> load_csv_parallel(int rank, int size);
+
+// Парсинг одной строки CSV в Record
+// Возвращает true если парсинг успешен
+bool parse_csv_line(const std::string& line, Record& record);

src/day_stats.hpp

@@ -1,28 +1,15 @@
 #pragma once
 #include <cstdint>
 
-using DayIndex = long long;
+using DayIndex = int64_t;
 
 // Агрегированные данные за один день
 struct DayStats {
     DayIndex day;       // индекс дня (timestamp / 86400)
-    double low;         // минимальный Low за день
+    double avg;         // среднее значение (Low + High) / 2 по всем записям
-    double high;        // максимальный High за день
+    double open_min;    // минимальный Open за день
-    double open;        // первый Open за день
+    double open_max;    // максимальный Open за день
-    double close;       // последний Close за день
+    double close_min;   // минимальный Close за день
-    double avg;         // среднее = (low + high) / 2
+    double close_max;   // максимальный Close за день
-    double first_ts;    // timestamp первой записи (для определения порядка open)
+    int64_t count;      // количество записей, по которым агрегировали
-    double last_ts;     // timestamp последней записи (для определения close)
 };
-
-// Интервал с изменением >= 10%
-struct Interval {
-    DayIndex start_day;
-    DayIndex end_day;
-    double min_open;
-    double max_close;
-    double start_avg;
-    double end_avg;
-    double change;
-};
-

src/gpu_loader.cpp

@@ -19,6 +19,16 @@ gpu_is_available_fn load_gpu_is_available() {
     return fn;
 }
 
+bool gpu_is_available() {    
+    auto gpu_is_available_fn = load_gpu_is_available();
+    
+    if (gpu_is_available_fn && gpu_is_available_fn()) {
+        return true;
+    }
+    
+    return false;
+}
+
 gpu_aggregate_days_fn load_gpu_aggregate_days() {
     void* h = get_gpu_lib_handle();
     if (!h) return nullptr;
@@ -117,13 +127,12 @@ bool aggregate_days_gpu(
     for (const auto& gs : gpu_stats) {
         DayStats ds;
         ds.day = gs.day;
-        ds.low = gs.low;
-        ds.high = gs.high;
-        ds.open = gs.open;
-        ds.close = gs.close;
         ds.avg = gs.avg;
-        ds.first_ts = gs.first_ts;
+        ds.open_min = gs.open_min;
-        ds.last_ts = gs.last_ts;
+        ds.open_max = gs.open_max;
+        ds.close_min = gs.close_min;
+        ds.close_max = gs.close_max;
+        ds.count = gs.count;
         out_stats.push_back(ds);
     }
     

src/gpu_loader.hpp

@@ -3,6 +3,8 @@
 #include "record.hpp"
 #include <vector>
 
+bool gpu_is_available();
+
 // Типы функций из GPU плагина
 using gpu_is_available_fn = int (*)();
 
@@ -18,13 +20,12 @@ struct GpuRecord {
 
 struct GpuDayStats {
     long long day;
-    double low;
-    double high;
-    double open;
-    double close;
     double avg;
-    double first_ts;
+    double open_min;
-    double last_ts;
+    double open_max;
+    double close_min;
+    double close_max;
+    long long count;
 };
 
 using gpu_aggregate_days_fn = int (*)(

src/gpu_plugin.cu

@@ -23,13 +23,12 @@ struct GpuRecord {
 
 struct GpuDayStats {
     long long day;
-    double low;
-    double high;
-    double open;
-    double close;
     double avg;
-    double first_ts;
+    double open_min;
-    double last_ts;
+    double open_max;
+    double close_min;
+    double close_max;
+    long long count;
 };
 
 extern "C" int gpu_is_available() {
@@ -63,32 +62,30 @@ __global__ void aggregate_kernel(
     
     GpuDayStats stats;
     stats.day = day_indices[d];
-    stats.low = DBL_MAX;
+    stats.open_min = DBL_MAX;
-    stats.high = -DBL_MAX;
+    stats.open_max = -DBL_MAX;
-    stats.first_ts = DBL_MAX;
+    stats.close_min = DBL_MAX;
-    stats.last_ts = -DBL_MAX;
+    stats.close_max = -DBL_MAX;
-    stats.open = 0;
+    stats.count = count;
-    stats.close = 0;
+    
+    double avg_sum = 0.0;
     
     for (int i = 0; i < count; i++) {
         const GpuRecord& r = records[offset + i];
         
-        // min/max
+        // Accumulate avg = (low + high) / 2
-        if (r.low < stats.low) stats.low = r.low;
+        avg_sum += (r.low + r.high) / 2.0;
-        if (r.high > stats.high) stats.high = r.high;
         
-        // first/last по timestamp
+        // min/max Open
-        if (r.timestamp < stats.first_ts) {
+        if (r.open < stats.open_min) stats.open_min = r.open;
-            stats.first_ts = r.timestamp;
+        if (r.open > stats.open_max) stats.open_max = r.open;
-            stats.open = r.open;
+        
-        }
+        // min/max Close
-        if (r.timestamp > stats.last_ts) {
+        if (r.close < stats.close_min) stats.close_min = r.close;
-            stats.last_ts = r.timestamp;
+        if (r.close > stats.close_max) stats.close_max = r.close;
-            stats.close = r.close;
-        }
     }
     
-    stats.avg = (stats.low + stats.high) / 2.0;
+    stats.avg = avg_sum / static_cast<double>(count);
     out_stats[d] = stats;
 }
 

src/intervals.cpp

@@ -28,13 +28,17 @@ std::vector<Interval> find_intervals(const std::vector<DayStats>& days, double t
             interval.end_avg = price_now;
             interval.change = change;
             
-            // Находим min(Open) и max(Close) в интервале
+            // Находим min/max Open и Close в интервале
-            interval.min_open = days[start_idx].open;
+            interval.open_min = days[start_idx].open_min;
-            interval.max_close = days[start_idx].close;
+            interval.open_max = days[start_idx].open_max;
+            interval.close_min = days[start_idx].close_min;
+            interval.close_max = days[start_idx].close_max;
             
-            for (size_t j = start_idx; j <= i; j++) {
+            for (size_t j = start_idx + 1; j <= i; j++) {
-                interval.min_open = std::min(interval.min_open, days[j].open);
+                interval.open_min = std::min(interval.open_min, days[j].open_min);
-                interval.max_close = std::max(interval.max_close, days[j].close);
+                interval.open_max = std::max(interval.open_max, days[j].open_max);
+                interval.close_min = std::min(interval.close_min, days[j].close_min);
+                interval.close_max = std::max(interval.close_max, days[j].close_max);
             }
             
             intervals.push_back(interval);
@@ -68,16 +72,17 @@ void write_intervals(const std::string& filename, const std::vector<Interval>& i
     std::ofstream out(filename);
     
     out << std::fixed << std::setprecision(2);
-    out << "start_date,end_date,min_open,max_close,start_avg,end_avg,change\n";
+    out << "start_date,end_date,open_min,open_max,close_min,close_max,start_avg,end_avg,change\n";
     
     for (const auto& iv : intervals) {
         out << day_index_to_date(iv.start_day) << ","
             << day_index_to_date(iv.end_day) << ","
-            << iv.min_open << ","
+            << iv.open_min << ","
-            << iv.max_close << ","
+            << iv.open_max << ","
+            << iv.close_min << ","
+            << iv.close_max << ","
             << iv.start_avg << ","
             << iv.end_avg << ","
             << std::setprecision(6) << iv.change << "\n";
     }
 }
-

src/intervals.hpp

@@ -4,6 +4,19 @@
 #include <vector>
 #include <string>
 
+// Интервал с изменением >= threshold
+struct Interval {
+    DayIndex start_day;
+    DayIndex end_day;
+    double open_min;    // минимальный Open в интервале
+    double open_max;    // максимальный Open в интервале
+    double close_min;   // минимальный Close в интервале
+    double close_max;   // максимальный Close в интервале
+    double start_avg;
+    double end_avg;
+    double change;
+};
+
 // Вычисление интервалов с изменением >= threshold (по умолчанию 10%)
 std::vector<Interval> find_intervals(const std::vector<DayStats>& days, double threshold = 0.10);
 
@@ -12,4 +25,3 @@ void write_intervals(const std::string& filename, const std::vector<Interval>& i
 
 // Преобразование DayIndex в строку даты (YYYY-MM-DD)
 std::string day_index_to_date(DayIndex day);
-

src/main.cpp

@@ -1,54 +1,12 @@
 #include <mpi.h>
-#include <omp.h>
 #include <iostream>
 #include <vector>
-#include <map>
 #include <iomanip>
-#include <cstdlib>
 
 #include "csv_loader.hpp"
-#include "utils.hpp"
 #include "record.hpp"
 #include "day_stats.hpp"
 #include "aggregation.hpp"
-#include "intervals.hpp"
-#include "gpu_loader.hpp"
-
-// Функция: отобрать записи для конкретного ранга
-std::vector<Record> select_records_for_rank(
-    const std::map<DayIndex, std::vector<Record>>& days,
-    const std::vector<DayIndex>& day_list)
-{
-    std::vector<Record> out;
-    for (auto d : day_list) {
-        auto it = days.find(d);
-        if (it != days.end()) {
-            const auto& vec = it->second;
-            out.insert(out.end(), vec.begin(), vec.end());
-        }
-    }
-    return out;
-}
-
-// Разделить записи на N частей (по дням)
-std::vector<std::vector<Record>> split_records(const std::vector<Record>& records, int n_parts) {
-    // Группируем по дням
-    std::map<DayIndex, std::vector<Record>> by_day;
-    for (const auto& r : records) {
-        DayIndex day = static_cast<DayIndex>(r.timestamp) / 86400;
-        by_day[day].push_back(r);
-    }
-    
-    // Распределяем дни по частям
-    std::vector<std::vector<Record>> parts(n_parts);
-    int i = 0;
-    for (auto& [day, recs] : by_day) {
-        parts[i % n_parts].insert(parts[i % n_parts].end(), recs.begin(), recs.end());
-        i++;
-    }
-    
-    return parts;
-}
 
 int main(int argc, char** argv) {
     MPI_Init(&argc, &argv);
@@ -57,225 +15,25 @@ int main(int argc, char** argv) {
     MPI_Comm_rank(MPI_COMM_WORLD, &rank);
     MPI_Comm_size(MPI_COMM_WORLD, &size);
 
-    // Читаем количество CPU потоков из переменной окружения
+    // Параллельное чтение данных
-    int num_cpu_threads = 2;
+    double read_start = MPI_Wtime();
-    const char* env_threads = std::getenv("NUM_CPU_THREADS");
+    std::vector<Record> records = load_csv_parallel(rank, size);
-    if (env_threads) {
+    double read_time = MPI_Wtime() - read_start;
-        num_cpu_threads = std::atoi(env_threads);
-        if (num_cpu_threads < 1) num_cpu_threads = 1;
-    }
-    omp_set_num_threads(num_cpu_threads + 1); // +1 для GPU потока если есть
 
-    // ====== ЗАГРУЗКА GPU ФУНКЦИЙ ======
+    std::cout << "Rank " << rank 
-    auto gpu_is_available = load_gpu_is_available();
+              << ": read " << records.size() << " records"
-    auto gpu_aggregate = load_gpu_aggregate_days();
+              << " in " << std::fixed << std::setprecision(3) << read_time << " sec"
-    
+              << std::endl;
-    bool have_gpu = false;
-    if (gpu_is_available && gpu_is_available()) {
-        have_gpu = true;
-        std::cout << "Rank " << rank << ": GPU available + " << num_cpu_threads << " CPU threads" << std::endl;
-    } else {
-        std::cout << "Rank " << rank << ": " << num_cpu_threads << " CPU threads only" << std::endl;
-    }
 
-    std::vector<Record> local_records;
+    // Агрегация по дням
+    double agg_start = MPI_Wtime();
+    std::vector<DayStats> days = aggregate_days(records);
+    double agg_time = MPI_Wtime() - agg_start;
 
-    // ====== ТАЙМЕРЫ ======
+    std::cout << "Rank " << rank 
-    double time_load_data = 0.0;
+              << ": aggregated " << days.size() << " days"
-    double time_distribute = 0.0;
+              << " in " << std::fixed << std::setprecision(3) << agg_time << " sec"
-
+              << std::endl;
-    if (rank == 0) {
-        std::cout << "Rank 0 loading CSV..." << std::endl;
-
-        // Таймер загрузки данных
-        double t_load_start = MPI_Wtime();
-        
-        // Запускаем из build
-        auto records = load_csv("../data/data.csv");
-
-        auto days = group_by_day(records);
-        auto parts = split_days(days, size);
-        
-        time_load_data = MPI_Wtime() - t_load_start;
-        std::cout << "Rank 0: Data loading time: " << std::fixed << std::setprecision(3) 
-                  << time_load_data << "s" << std::endl;
-
-        // Таймер рассылки данных
-        double t_distribute_start = MPI_Wtime();
-        
-        // Рассылаем данные
-        for (int r = 0; r < size; r++) {
-            auto vec = select_records_for_rank(days, parts[r]);
-
-            if (r == 0) {  
-                // себе не отправляем — сразу сохраняем
-                local_records = vec;
-                continue;
-            }
-
-            int count = static_cast<int>(vec.size());
-            MPI_Send(&count, 1, MPI_INT, r, 0, MPI_COMM_WORLD);
-            MPI_Send(vec.data(), count * sizeof(Record), MPI_BYTE, r, 1, MPI_COMM_WORLD);
-        }
-        
-        time_distribute = MPI_Wtime() - t_distribute_start;
-    }
-    else {
-        // Таймер получения данных
-        double t_receive_start = MPI_Wtime();
-        
-        // Принимает данные
-        int count = 0;
-        MPI_Recv(&count, 1, MPI_INT, 0, 0, MPI_COMM_WORLD, MPI_STATUS_IGNORE);
-
-        local_records.resize(count);
-        MPI_Recv(local_records.data(), count * sizeof(Record),
-                 MPI_BYTE, 0, 1, MPI_COMM_WORLD, MPI_STATUS_IGNORE);
-        
-        time_distribute = MPI_Wtime() - t_receive_start;
-    }
-
-    MPI_Barrier(MPI_COMM_WORLD);
-    
-    // Вывод времени рассылки/получения данных
-    std::cout << "Rank " << rank << ": Data distribution time: " << std::fixed 
-              << std::setprecision(3) << time_distribute << "s" << std::endl;
-
-    std::cout << "Rank " << rank << " received "
-              << local_records.size() << " records" << std::endl;
-
-    // ====== АГРЕГАЦИЯ НА КАЖДОМ УЗЛЕ ======
-    std::vector<DayStats> local_stats;
-    double time_start = omp_get_wtime();
-    
-    // Время работы: [0] = GPU (если есть), [1..n] = CPU потоки
-    std::vector<double> worker_times(num_cpu_threads + 1, 0.0);
-    
-    if (have_gpu && gpu_aggregate) {
-        // GPU узел: делим на (1 + num_cpu_threads) частей
-        int n_workers = 1 + num_cpu_threads;
-        auto parts = split_records(local_records, n_workers);
-        
-        std::vector<std::vector<DayStats>> results(n_workers);
-        std::vector<bool> success(n_workers, true);
-        
-        #pragma omp parallel
-        {
-            int tid = omp_get_thread_num();
-            if (tid < n_workers) {
-                double t0 = omp_get_wtime();
-                if (tid == 0) {
-                    // GPU поток
-                    success[0] = aggregate_days_gpu(parts[0], results[0], gpu_aggregate);
-                } else {
-                    // CPU потоки
-                    results[tid] = aggregate_days(parts[tid]);
-                }
-                worker_times[tid] = omp_get_wtime() - t0;
-            }
-        }
-        
-        // Объединяем результаты
-        for (int i = 0; i < n_workers; i++) {
-            if (i == 0 && !success[0]) {
-                // GPU failed - обработаем на CPU
-                std::cout << "Rank " << rank << ": GPU failed, processing on CPU" << std::endl;
-                double t0 = omp_get_wtime();
-                results[0] = aggregate_days(parts[0]);
-                worker_times[0] = omp_get_wtime() - t0;
-            }
-            local_stats.insert(local_stats.end(), results[i].begin(), results[i].end());
-        }
-        
-    } else {
-        // CPU-only узел
-        auto parts = split_records(local_records, num_cpu_threads);
-        std::vector<std::vector<DayStats>> results(num_cpu_threads);
-        
-        #pragma omp parallel
-        {
-            int tid = omp_get_thread_num();
-            if (tid < num_cpu_threads) {
-                double t0 = omp_get_wtime();
-                results[tid] = aggregate_days(parts[tid]);
-                worker_times[tid + 1] = omp_get_wtime() - t0; // +1 т.к. [0] для GPU
-            }
-        }
-        
-        for (int i = 0; i < num_cpu_threads; i++) {
-            local_stats.insert(local_stats.end(), results[i].begin(), results[i].end());
-        }
-    }
-    
-    double time_total = omp_get_wtime() - time_start;
-
-    // Вывод времени
-    std::cout << std::fixed << std::setprecision(3);
-    std::cout << "Rank " << rank << " aggregated " << local_stats.size() << " days in " 
-              << time_total << "s (";
-    if (have_gpu) {
-        std::cout << "GPU: " << worker_times[0] << "s, ";
-    }
-    for (int i = 0; i < num_cpu_threads; i++) {
-        int idx = have_gpu ? (i + 1) : (i + 1);
-        std::cout << "CPU" << i << ": " << worker_times[idx] << "s";
-        if (i < num_cpu_threads - 1) std::cout << ", ";
-    }
-    std::cout << ")" << std::endl;
-
-    // ====== СБОР АГРЕГИРОВАННЫХ ДАННЫХ НА RANK 0 ======
-    std::vector<DayStats> all_stats;
-
-    if (rank == 0) {
-        // Добавляем свои данные
-        all_stats.insert(all_stats.end(), local_stats.begin(), local_stats.end());
-
-        // Получаем данные от других узлов
-        for (int r = 1; r < size; r++) {
-            int count = 0;
-            MPI_Recv(&count, 1, MPI_INT, r, 2, MPI_COMM_WORLD, MPI_STATUS_IGNORE);
-
-            std::vector<DayStats> remote_stats(count);
-            MPI_Recv(remote_stats.data(), count * sizeof(DayStats),
-                     MPI_BYTE, r, 3, MPI_COMM_WORLD, MPI_STATUS_IGNORE);
-
-            all_stats.insert(all_stats.end(), remote_stats.begin(), remote_stats.end());
-        }
-    } else {
-        // Отправляем свои агрегированные данные на rank 0
-        int count = static_cast<int>(local_stats.size());
-        MPI_Send(&count, 1, MPI_INT, 0, 2, MPI_COMM_WORLD);
-        MPI_Send(local_stats.data(), count * sizeof(DayStats), MPI_BYTE, 0, 3, MPI_COMM_WORLD);
-    }
-
-    // ====== ВЫЧИСЛЕНИЕ ИНТЕРВАЛОВ НА RANK 0 ======
-    if (rank == 0) {
-        std::cout << "Rank 0: merging " << all_stats.size() << " day stats..." << std::endl;
-
-        // Объединяем и сортируем
-        auto merged_stats = merge_day_stats(all_stats);
-        std::cout << "Rank 0: total " << merged_stats.size() << " unique days" << std::endl;
-
-        // Вычисляем интервалы
-        auto intervals = find_intervals(merged_stats, 0.10);
-        std::cout << "Found " << intervals.size() << " intervals with >=10% change" << std::endl;
-
-        // Записываем результат
-        write_intervals("../result.csv", intervals);
-        std::cout << "Results written to result.csv" << std::endl;
-
-        // Выводим первые несколько интервалов
-        std::cout << "\nFirst 5 intervals:\n";
-        std::cout << "start_date,end_date,min_open,max_close,change\n";
-        for (size_t i = 0; i < std::min(intervals.size(), size_t(5)); i++) {
-            const auto& iv = intervals[i];
-            std::cout << day_index_to_date(iv.start_day) << ","
-                      << day_index_to_date(iv.end_day) << ","
-                      << iv.min_open << ","
-                      << iv.max_close << ","
-                      << iv.change << "\n";
-        }
-    }
 
     MPI_Finalize();
     return 0;

src/utils.cpp

@@ -1,4 +1,8 @@
 #include "utils.hpp"
+#include <fstream>
+#include <sstream>
+#include <stdexcept>
+#include <numeric>
 
 std::map<DayIndex, std::vector<Record>> group_by_day(const std::vector<Record>& recs) {
     std::map<DayIndex, std::vector<Record>> days;
@@ -23,3 +27,88 @@ std::vector<std::vector<DayIndex>> split_days(const std::map<DayIndex, std::vect
     return out;
 }
 
+int get_num_cpu_threads() {
+    const char* env_threads = std::getenv("NUM_CPU_THREADS");
+    int num_cpu_threads = 1;
+    if (env_threads) {
+        num_cpu_threads = std::atoi(env_threads);
+        if (num_cpu_threads < 1) num_cpu_threads = 1;
+    }
+    return num_cpu_threads;
+}
+
+std::string get_env(const char* name) {
+    const char* env = std::getenv(name);
+    if (!env) {
+        throw std::runtime_error(std::string("Environment variable not set: ") + name);
+    }
+    return std::string(env);
+}
+
+std::string get_data_path() {
+    return get_env("DATA_PATH");
+}
+
+std::vector<int> get_data_read_shares() {
+    std::vector<int> shares;
+    std::stringstream ss(get_env("DATA_READ_SHARES"));
+    std::string item;
+    while (std::getline(ss, item, ',')) {
+        shares.push_back(std::stoi(item));
+    }
+    return shares;
+}
+
+int64_t get_read_overlap_bytes() {
+    return std::stoll(get_env("READ_OVERLAP_BYTES"));
+}
+
+int64_t get_file_size(const std::string& path) {
+    std::ifstream file(path, std::ios::binary | std::ios::ate);
+    if (!file.is_open()) {
+        throw std::runtime_error("Cannot open file: " + path);
+    }
+    return static_cast<int64_t>(file.tellg());
+}
+
+ByteRange calculate_byte_range(int rank, int size, int64_t file_size,
+                               const std::vector<int>& shares, int64_t overlap_bytes) {
+    // Если shares пустой или не соответствует size, используем равные доли
+    std::vector<int> effective_shares;
+    if (shares.size() == static_cast<size_t>(size)) {
+        effective_shares = shares;
+    } else {
+        effective_shares.assign(size, 1);
+    }
+    
+    int total_shares = std::accumulate(effective_shares.begin(), effective_shares.end(), 0);
+    
+    // Вычисляем базовые границы для каждого ранка
+    int64_t bytes_per_share = file_size / total_shares;
+    
+    int64_t base_start = 0;
+    for (int i = 0; i < rank; i++) {
+        base_start += bytes_per_share * effective_shares[i];
+    }
+    
+    int64_t base_end = base_start + bytes_per_share * effective_shares[rank];
+    
+    // Применяем overlap
+    ByteRange range;
+    
+    if (rank == 0) {
+        // Первый ранк: начинаем с 0, добавляем overlap в конце
+        range.start = 0;
+        range.end = std::min(base_end + overlap_bytes, file_size);
+    } else if (rank == size - 1) {
+        // Последний ранк: вычитаем overlap в начале, читаем до конца файла
+        range.start = std::max(base_start - overlap_bytes, static_cast<int64_t>(0));
+        range.end = file_size;
+    } else {
+        // Промежуточные ранки: overlap с обеих сторон
+        range.start = std::max(base_start - overlap_bytes, static_cast<int64_t>(0));
+        range.end = std::min(base_end + overlap_bytes, file_size);
+    }
+    
+    return range;
+}

src/utils.hpp

@@ -4,6 +4,29 @@
 #include "day_stats.hpp"
 #include <map>
 #include <vector>
+#include <string>
+#include <cstdlib>
+#include <cstdint>
 
+// Группировка записей по дням
 std::map<DayIndex, std::vector<Record>> group_by_day(const std::vector<Record>& recs);
 std::vector<std::vector<DayIndex>> split_days(const std::map<DayIndex, std::vector<Record>>& days, int parts);
+
+// Чтение переменных окружения
+int get_num_cpu_threads();
+std::string get_data_path();
+std::vector<int> get_data_read_shares();
+int64_t get_read_overlap_bytes();
+
+// Структура для хранения диапазона байт для чтения
+struct ByteRange {
+    int64_t start;
+    int64_t end;  // exclusive
+};
+
+// Вычисляет диапазон байт для конкретного ранка
+ByteRange calculate_byte_range(int rank, int size, int64_t file_size,
+                               const std::vector<int>& shares, int64_t overlap_bytes);
+
+// Получение размера файла
+int64_t get_file_size(const std::string& path);