Параллельное чтение данных

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/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/main.cpp

@@ -1,54 +1,11 @@
 #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,222 +14,18 @@ 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 start_time = MPI_Wtime();
-    const char* env_threads = std::getenv("NUM_CPU_THREADS");
-    if (env_threads) {
-        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
-    bool have_gpu = gpu_is_available();
     
-    if (have_gpu) {
+    std::vector<Record> records = load_csv_parallel(rank, size);
-        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 time_load_data = 0.0;
-    double time_distribute = 0.0;
-
-    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);
     
-    // Вывод времени рассылки/получения данных
+    double end_time = MPI_Wtime();
-    std::cout << "Rank " << rank << ": Data distribution time: " << std::fixed 
+    double read_time = end_time - start_time;
-              << std::setprecision(3) << time_distribute << "s" << std::endl;
 
-    std::cout << "Rank " << rank << " received "
+    std::cout << "Rank " << rank 
-              << local_records.size() << " records" << std::endl;
+              << ": read " << records.size() << " records"
-
+              << " in " << std::fixed << std::setprecision(3) << read_time << " sec"
-    // ====== АГРЕГАЦИЯ НА КАЖДОМ УЗЛЕ ======
+              << 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 узел: делим на (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_simple(parts[0], results[0]);
-                } 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);