Скрипт для усреднения результатов

benchmark.py

@@ -0,0 +1,115 @@
+"""
+Запускает make run <number_of_runs> раз и считает статистику по времени выполнения.
+Тупо парсит out.txt и берём значение из строки "Total execution time: <time> sec".
+
+python benchmark.py <number_of_runs>
+"""
+
+import os
+import re
+import sys
+import time
+import subprocess
+import statistics
+
+N = int(sys.argv[1]) if len(sys.argv) > 1 else 10
+OUT = "out.txt"
+
+TIME_RE = re.compile(r"Total execution time:\s*([0-9]*\.?[0-9]+)\s*sec")
+JOB_RE  = re.compile(r"Submitted batch job\s+(\d+)")
+
+APPEAR_TIMEOUT = 300.0     # ждать появления out.txt
+FINISH_TIMEOUT = 3600.0    # ждать появления Total execution time (сек)
+POLL = 0.2                 # частота проверки файла
+
+def wait_for_exists(path: str, timeout: float):
+    t0 = time.time()
+    while not os.path.exists(path):
+        if time.time() - t0 > timeout:
+            raise TimeoutError(f"{path} did not appear within {timeout} seconds")
+        time.sleep(POLL)
+
+def try_read(path: str) -> str:
+    try:
+        with open(path, "r", encoding="utf-8", errors="replace") as f:
+            return f.read()
+    except FileNotFoundError:
+        return ""
+    except OSError:
+        # бывает, что файл на NFS в момент записи недоступен на чтение
+        return ""
+
+def wait_for_time_line(path: str, timeout: float) -> float:
+    t0 = time.time()
+    last_report = 0.0
+    while True:
+        txt = try_read(path)
+        matches = TIME_RE.findall(txt)
+        if matches:
+            return float(matches[-1])  # последняя встреченная строка
+
+        now = time.time()
+        if now - t0 > timeout:
+            tail = txt[-800:] if txt else "<empty>"
+            raise TimeoutError("Timed out waiting for 'Total execution time' line.\n"
+                               f"Last 800 chars of out.txt:\n{tail}")
+
+        # иногда полезно печатать прогресс раз в ~5 сек
+        if now - last_report > 5.0:
+            last_report = now
+            if txt:
+                # показать последнюю непустую строку
+                lines = [l for l in txt.splitlines() if l.strip()]
+                if lines:
+                    print(f"  waiting... last line: {lines[-1][:120]}", flush=True)
+                else:
+                    print("  waiting... (out.txt empty)", flush=True)
+            else:
+                print("  waiting... (out.txt not readable yet)", flush=True)
+
+        time.sleep(POLL)
+
+times = []
+
+for i in range(N):
+    print(f"Run {i+1}/{N} ...", flush=True)
+
+    # удаляем out.txt перед запуском
+    try:
+        os.remove(OUT)
+    except FileNotFoundError:
+        pass
+
+    # запускаем make run и забираем stdout (там будет Submitted batch job XXX)
+    res = subprocess.run(["make", "run"], capture_output=True, text=True)
+    out = (res.stdout or "") + "\n" + (res.stderr or "")
+
+    job_id = None
+    m = JOB_RE.search(out)
+    if m:
+        job_id = m.group(1)
+        print(f"  submitted job {job_id}", flush=True)
+    else:
+        print("  (job id not detected; will only watch out.txt)", flush=True)
+
+    # ждём появления out.txt и появления строки с Total execution time
+    wait_for_exists(OUT, APPEAR_TIMEOUT)
+    t = wait_for_time_line(OUT, FINISH_TIMEOUT)
+
+    times.append(t)
+    print(f"  time = {t:.3f} sec", flush=True)
+
+    # опционально удалить out.txt после парсинга
+    try:
+        os.remove(OUT)
+    except FileNotFoundError:
+        pass
+
+print("\n=== RESULTS ===")
+print(f"Runs:   {len(times)}")
+print(f"Mean:   {statistics.mean(times):.3f} sec")
+print(f"Median: {statistics.median(times):.3f} sec")
+print(f"Min:    {min(times):.3f} sec")
+print(f"Max:    {max(times):.3f} sec")
+if len(times) > 1:
+    print(f"Stddev: {statistics.stdev(times):.3f} sec")

run.slurm

@@ -7,9 +7,10 @@
 
 # Путь к файлу данных (должен существовать на всех узлах)
 export DATA_PATH="/mnt/shared/supercomputers/data/data.csv"
+# export DATA_PATH="/data/data.csv"
 
 # Доли данных для каждого ранка (сумма определяет пропорции)
-export DATA_READ_SHARES="10,12,13,13"
+export DATA_READ_SHARES="10,14,18,22"
 
 # Размер перекрытия в байтах для обработки границ строк
 export READ_OVERLAP_BYTES=131072

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) {
-        // Сортируем по timestamp для определения first/last
-        std::sort(recs.begin(), recs.end(), 
-            [](const Record* a, const Record* b) {
-                return a->timestamp < b->timestamp;
-            });
-        
+    for (const auto& [day, acc] : days) {
         DayStats stats;
         stats.day = day;
-        stats.low = std::numeric_limits<double>::max();
-        stats.high = std::numeric_limits<double>::lowest();
-        stats.open = recs.front()->open;
-        stats.close = recs.back()->close;
-        stats.first_ts = recs.front()->timestamp;
-        stats.last_ts = recs.back()->timestamp;
-        
-        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;
-        
+        stats.avg = acc.avg_sum / static_cast<double>(acc.count);
+        stats.open_min = acc.open_min;
+        stats.open_max = acc.open_max;
+        stats.close_min = acc.close_min;
+        stats.close_max = acc.close_max;
+        stats.count = acc.count;
         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/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 high;        // максимальный High за день
-    double open;        // первый Open за день
-    double close;       // последний Close за день
-    double avg;         // среднее = (low + high) / 2
-    double first_ts;    // timestamp первой записи (для определения порядка open)
-    double last_ts;     // timestamp последней записи (для определения close)
+    double avg;         // среднее значение (Low + High) / 2 по всем записям
+    double open_min;    // минимальный Open за день
+    double open_max;    // максимальный Open за день
+    double close_min;   // минимальный Close за день
+    double close_max;   // максимальный Close за день
+    int64_t count;      // количество записей, по которым агрегировали
 };
-
-// Интервал с изменением >= 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

@@ -127,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.last_ts = gs.last_ts;
+        ds.open_min = gs.open_min;
+        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

@@ -20,13 +20,12 @@ struct GpuRecord {
 
 struct GpuDayStats {
     long long day;
-    double low;
-    double high;
-    double open;
-    double close;
     double avg;
-    double first_ts;
-    double last_ts;
+    double open_min;
+    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 last_ts;
+    double open_min;
+    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.high = -DBL_MAX;
-    stats.first_ts = DBL_MAX;
-    stats.last_ts = -DBL_MAX;
-    stats.open = 0;
-    stats.close = 0;
+    stats.open_min = DBL_MAX;
+    stats.open_max = -DBL_MAX;
+    stats.close_min = DBL_MAX;
+    stats.close_max = -DBL_MAX;
+    stats.count = count;
+    
+    double avg_sum = 0.0;
     
     for (int i = 0; i < count; i++) {
         const GpuRecord& r = records[offset + i];
         
-        // min/max
-        if (r.low < stats.low) stats.low = r.low;
-        if (r.high > stats.high) stats.high = r.high;
+        // Accumulate avg = (low + high) / 2
+        avg_sum += (r.low + r.high) / 2.0;
         
-        // first/last по timestamp
-        if (r.timestamp < stats.first_ts) {
-            stats.first_ts = r.timestamp;
-            stats.open = r.open;
-        }
-        if (r.timestamp > stats.last_ts) {
-            stats.last_ts = r.timestamp;
-            stats.close = r.close;
-        }
+        // min/max Open
+        if (r.open < stats.open_min) stats.open_min = r.open;
+        if (r.open > stats.open_max) stats.open_max = r.open;
+        
+        // min/max Close
+        if (r.close < stats.close_min) stats.close_min = r.close;
+        if (r.close > stats.close_max) stats.close_max = 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

@@ -1,54 +1,309 @@
 #include "intervals.hpp"
+#include <mpi.h>
 #include <algorithm>
 #include <cmath>
 #include <fstream>
 #include <iomanip>
 #include <sstream>
 #include <ctime>
+#include <limits>
 
-std::vector<Interval> find_intervals(const std::vector<DayStats>& days, double threshold) {
-    if (days.empty()) {
-        return {};
+// Вспомогательная структура для накопления min/max в интервале
+struct IntervalAccumulator {
+    DayIndex start_day;
+    double start_avg;
+    double open_min;
+    double open_max;
+    double close_min;
+    double close_max;
+    
+    void init(const DayStats& day) {
+        start_day = day.day;
+        start_avg = day.avg;
+        open_min = day.open_min;
+        open_max = day.open_max;
+        close_min = day.close_min;
+        close_max = day.close_max;
     }
     
-    std::vector<Interval> intervals;
+    void update(const DayStats& day) {
+        open_min = std::min(open_min, day.open_min);
+        open_max = std::max(open_max, day.open_max);
+        close_min = std::min(close_min, day.close_min);
+        close_max = std::max(close_max, day.close_max);
+    }
     
+    Interval finalize(const DayStats& end_day, double change) const {
+        Interval iv;
+        iv.start_day = start_day;
+        iv.end_day = end_day.day;
+        iv.start_avg = start_avg;
+        iv.end_avg = end_day.avg;
+        iv.change = change;
+        iv.open_min = std::min(open_min, end_day.open_min);
+        iv.open_max = std::max(open_max, end_day.open_max);
+        iv.close_min = std::min(close_min, end_day.close_min);
+        iv.close_max = std::max(close_max, end_day.close_max);
+        return iv;
+    }
+};
+
+// Упакованная структура DayStats для MPI передачи (8 doubles)
+struct PackedDayStats {
+    double day;      // DayIndex as double
+    double avg;
+    double open_min;
+    double open_max;
+    double close_min;
+    double close_max;
+    double count;    // int64_t as double
+    double valid;    // флаг валидности (1.0 = valid, 0.0 = invalid)
+    
+    void pack(const DayStats& ds) {
+        day = static_cast<double>(ds.day);
+        avg = ds.avg;
+        open_min = ds.open_min;
+        open_max = ds.open_max;
+        close_min = ds.close_min;
+        close_max = ds.close_max;
+        count = static_cast<double>(ds.count);
+        valid = 1.0;
+    }
+    
+    DayStats unpack() const {
+        DayStats ds;
+        ds.day = static_cast<DayIndex>(day);
+        ds.avg = avg;
+        ds.open_min = open_min;
+        ds.open_max = open_max;
+        ds.close_min = close_min;
+        ds.close_max = close_max;
+        ds.count = static_cast<int64_t>(count);
+        return ds;
+    }
+    
+    bool is_valid() const { return valid > 0.5; }
+    void set_invalid() { valid = 0.0; }
+};
+
+IntervalResult find_intervals_parallel(
+    const std::vector<DayStats>& days,
+    int rank, int size,
+    double threshold)
+{
+    IntervalResult result;
+    result.compute_time = 0.0;
+    result.wait_time = 0.0;
+    
+    if (days.empty()) {
+        // Передаём невалидный DayStats следующему ранку
+        if (rank < size - 1) {
+            PackedDayStats invalid;
+            invalid.set_invalid();
+            MPI_Send(&invalid, 8, MPI_DOUBLE, rank + 1, 0, MPI_COMM_WORLD);
+        }
+        return result;
+    }
+    
+    double compute_start = MPI_Wtime();
+    
+    // Определяем, до какого индекса обрабатывать
+    // Для последнего ранка - до конца, для остальных - до предпоследнего дня
+    size_t process_until = (rank == size - 1) ? days.size() : days.size() - 1;
+    
+    IntervalAccumulator acc;
     size_t start_idx = 0;
-    double price_base = days[start_idx].avg;
+    bool have_pending_interval = false;
     
-    for (size_t i = 1; i < days.size(); i++) {
-        double price_now = days[i].avg;
-        double change = std::abs(price_now - price_base) / price_base;
+    // Если не первый ранк - ждём данные от предыдущего
+    if (rank > 0) {
+        double wait_start = MPI_Wtime();
         
-        if (change >= threshold) {
-            Interval interval;
-            interval.start_day = days[start_idx].day;
-            interval.end_day = days[i].day;
-            interval.start_avg = price_base;
-            interval.end_avg = price_now;
-            interval.change = change;
+        PackedDayStats received;
+        MPI_Recv(&received, 8, MPI_DOUBLE, rank - 1, 0, MPI_COMM_WORLD, MPI_STATUS_IGNORE);
+        
+        result.wait_time = MPI_Wtime() - wait_start;
+        compute_start = MPI_Wtime();
+        
+        if (received.is_valid()) {
+            DayStats prev_day = received.unpack();
             
-            // Находим min(Open) и max(Close) в интервале
-            interval.min_open = days[start_idx].open;
-            interval.max_close = days[start_idx].close;
-            
-            for (size_t j = start_idx; j <= i; j++) {
-                interval.min_open = std::min(interval.min_open, days[j].open);
-                interval.max_close = std::max(interval.max_close, days[j].close);
+            // Ищем первый день с индексом > prev_day.day
+            for (start_idx = 0; start_idx < days.size(); start_idx++) {
+                if (days[start_idx].day > prev_day.day) {
+                    break;
+                }
             }
             
-            intervals.push_back(interval);
-            
-            // Начинаем новый интервал
-            start_idx = i + 1;
-            if (start_idx >= days.size()) {
-                break;
+            if (start_idx < process_until) {
+                // Инициализируем аккумулятор данными от предыдущего ранка
+                acc.init(prev_day);
+                have_pending_interval = true;
+                
+                // Продолжаем строить интервал
+                for (size_t i = start_idx; i < process_until; i++) {
+                    acc.update(days[i]);
+                    
+                    double change = std::abs(days[i].avg - acc.start_avg) / acc.start_avg;
+                    
+                    if (change >= threshold) {
+                        result.intervals.push_back(acc.finalize(days[i], change));
+                        have_pending_interval = false;
+                        
+                        // Начинаем новый интервал
+                        start_idx = i + 1;
+                        if (start_idx < process_until) {
+                            acc.init(days[start_idx]);
+                            have_pending_interval = true;
+                        }
+                    }
+                }
             }
-            price_base = days[start_idx].avg;
+        } else {
+            // Предыдущий ранк не передал валидные данные, начинаем с начала
+            if (process_until > 0) {
+                acc.init(days[0]);
+                have_pending_interval = true;
+                start_idx = 0;
+            }
+        }
+    } else {
+        // Первый ранк - начинаем с первого дня
+        if (process_until > 0) {
+            acc.init(days[0]);
+            have_pending_interval = true;
+            start_idx = 0;
         }
     }
     
-    return intervals;
+    // Обрабатываем дни (если ещё не обработали выше)
+    if (rank == 0 && have_pending_interval) {
+        for (size_t i = 1; i < process_until; i++) {
+            acc.update(days[i]);
+            
+            double change = std::abs(days[i].avg - acc.start_avg) / acc.start_avg;
+            
+            if (change >= threshold) {
+                result.intervals.push_back(acc.finalize(days[i], change));
+                have_pending_interval = false;
+                
+                // Начинаем новый интервал
+                start_idx = i + 1;
+                if (start_idx < process_until) {
+                    acc.init(days[start_idx]);
+                    have_pending_interval = true;
+                }
+            }
+        }
+    }
+    
+    // Для последнего ранка: завершаем последний интервал на последнем дне
+    if (rank == size - 1 && have_pending_interval && !days.empty()) {
+        const auto& last_day = days.back();
+        double change = std::abs(last_day.avg - acc.start_avg) / acc.start_avg;
+        result.intervals.push_back(acc.finalize(last_day, change));
+    }
+    
+    result.compute_time = MPI_Wtime() - compute_start;
+    
+    // Передаём данные следующему ранку
+    if (rank < size - 1) {
+        PackedDayStats to_send;
+        
+        if (have_pending_interval) {
+            // Передаём день, с которого начался незавершённый интервал
+            DayStats start_day;
+            start_day.day = acc.start_day;
+            start_day.avg = acc.start_avg;
+            start_day.open_min = acc.open_min;
+            start_day.open_max = acc.open_max;
+            start_day.close_min = acc.close_min;
+            start_day.close_max = acc.close_max;
+            start_day.count = 0;
+            to_send.pack(start_day);
+        } else if (!days.empty()) {
+            // Интервал завершился, передаём предпоследний день
+            to_send.pack(days[days.size() - 2]);
+        } else {
+            to_send.set_invalid();
+        }
+        
+        MPI_Send(&to_send, 8, MPI_DOUBLE, rank + 1, 0, MPI_COMM_WORLD);
+    }
+    
+    return result;
+}
+
+double collect_intervals(
+    std::vector<Interval>& local_intervals,
+    int rank, int size)
+{
+    double wait_time = 0.0;
+    
+    // Упакованный Interval для MPI (9 doubles)
+    // start_day, end_day, open_min, open_max, close_min, close_max, start_avg, end_avg, change
+    
+    if (rank == 0) {
+        // Собираем интервалы со всех остальных ранков
+        for (int r = 1; r < size; r++) {
+            double wait_start = MPI_Wtime();
+            
+            // Сначала получаем количество интервалов
+            int count;
+            MPI_Recv(&count, 1, MPI_INT, r, 1, MPI_COMM_WORLD, MPI_STATUS_IGNORE);
+            
+            if (count > 0) {
+                std::vector<double> buffer(count * 9);
+                MPI_Recv(buffer.data(), count * 9, MPI_DOUBLE, r, 2, MPI_COMM_WORLD, MPI_STATUS_IGNORE);
+                
+                // Распаковываем
+                for (int i = 0; i < count; i++) {
+                    Interval iv;
+                    iv.start_day = static_cast<DayIndex>(buffer[i * 9 + 0]);
+                    iv.end_day = static_cast<DayIndex>(buffer[i * 9 + 1]);
+                    iv.open_min = buffer[i * 9 + 2];
+                    iv.open_max = buffer[i * 9 + 3];
+                    iv.close_min = buffer[i * 9 + 4];
+                    iv.close_max = buffer[i * 9 + 5];
+                    iv.start_avg = buffer[i * 9 + 6];
+                    iv.end_avg = buffer[i * 9 + 7];
+                    iv.change = buffer[i * 9 + 8];
+                    local_intervals.push_back(iv);
+                }
+            }
+            
+            wait_time += MPI_Wtime() - wait_start;
+        }
+        
+        // Сортируем по start_day
+        std::sort(local_intervals.begin(), local_intervals.end(),
+            [](const Interval& a, const Interval& b) {
+                return a.start_day < b.start_day;
+            });
+    } else {
+        // Отправляем свои интервалы на ранк 0
+        int count = static_cast<int>(local_intervals.size());
+        MPI_Send(&count, 1, MPI_INT, 0, 1, MPI_COMM_WORLD);
+        
+        if (count > 0) {
+            std::vector<double> buffer(count * 9);
+            for (int i = 0; i < count; i++) {
+                const auto& iv = local_intervals[i];
+                buffer[i * 9 + 0] = static_cast<double>(iv.start_day);
+                buffer[i * 9 + 1] = static_cast<double>(iv.end_day);
+                buffer[i * 9 + 2] = iv.open_min;
+                buffer[i * 9 + 3] = iv.open_max;
+                buffer[i * 9 + 4] = iv.close_min;
+                buffer[i * 9 + 5] = iv.close_max;
+                buffer[i * 9 + 6] = iv.start_avg;
+                buffer[i * 9 + 7] = iv.end_avg;
+                buffer[i * 9 + 8] = iv.change;
+            }
+            MPI_Send(buffer.data(), count * 9, MPI_DOUBLE, 0, 2, MPI_COMM_WORLD);
+        }
+    }
+    
+    return wait_time;
 }
 
 std::string day_index_to_date(DayIndex day) {
@@ -68,16 +323,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.max_close << ","
+            << iv.open_min << ","
+            << 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,12 +4,43 @@
 #include <vector>
 #include <string>
 
-// Вычисление интервалов с изменением >= threshold (по умолчанию 10%)
-std::vector<Interval> find_intervals(const std::vector<DayStats>& days, double threshold = 0.10);
+// Интервал с изменением >= threshold
+struct Interval {
+    DayIndex start_day;
+    DayIndex end_day;
+    double open_min;
+    double open_max;
+    double close_min;
+    double close_max;
+    double start_avg;
+    double end_avg;
+    double change;
+};
+
+// Результат параллельного построения интервалов
+struct IntervalResult {
+    std::vector<Interval> intervals;
+    double compute_time;  // время вычислений
+    double wait_time;     // время ожидания данных от предыдущего ранка
+};
+
+// Параллельное построение интервалов с использованием MPI
+// Каждый ранк обрабатывает свою часть дней и передаёт незавершённый интервал следующему
+IntervalResult find_intervals_parallel(
+    const std::vector<DayStats>& days,
+    int rank, int size,
+    double threshold = 0.10
+);
+
+// Сбор интервалов со всех ранков на ранк 0
+// Возвращает время ожидания данных
+double collect_intervals(
+    std::vector<Interval>& local_intervals,
+    int rank, int size
+);
 
 // Вывод интервалов в файл
 void write_intervals(const std::string& filename, const std::vector<Interval>& intervals);
 
 // Преобразование DayIndex в строку даты (YYYY-MM-DD)
 std::string day_index_to_date(DayIndex day);
-

src/main.cpp

@@ -4,29 +4,89 @@
 #include <iomanip>
 
 #include "csv_loader.hpp"
-#include "utils.hpp"
 #include "record.hpp"
+#include "day_stats.hpp"
+#include "aggregation.hpp"
+#include "intervals.hpp"
+#include "utils.hpp"
 
 int main(int argc, char** argv) {
     MPI_Init(&argc, &argv);
+    double total_start = MPI_Wtime();
 
     int rank, size;
     MPI_Comm_rank(MPI_COMM_WORLD, &rank);
     MPI_Comm_size(MPI_COMM_WORLD, &size);
 
     // Параллельное чтение данных
-    double start_time = MPI_Wtime();
-    
+    double read_start = MPI_Wtime();
     std::vector<Record> records = load_csv_parallel(rank, size);
-    
-    double end_time = MPI_Wtime();
-    double read_time = end_time - start_time;
+    double read_time = MPI_Wtime() - read_start;
 
     std::cout << "Rank " << rank 
               << ": read " << records.size() << " records"
               << " in " << std::fixed << std::setprecision(3) << read_time << " sec"
               << std::endl;
 
+    // Агрегация по дням
+    double agg_start = MPI_Wtime();
+    std::vector<DayStats> days = aggregate_days(records);
+    double agg_time = MPI_Wtime() - agg_start;
+
+    std::cout << "Rank " << rank 
+              << ": aggregated " << days.size() << " days"
+              << " [" << (days.empty() ? 0 : days.front().day) 
+              << ".." << (days.empty() ? 0 : days.back().day) << "]"
+              << " in " << std::fixed << std::setprecision(3) << agg_time << " sec"
+              << std::endl;
+
+    // Удаляем крайние дни (могут быть неполными из-за параллельного чтения)
+    trim_edge_days(days, rank, size);
+
+    std::cout << "Rank " << rank 
+              << ": after trim " << days.size() << " days"
+              << " [" << (days.empty() ? 0 : days.front().day) 
+              << ".." << (days.empty() ? 0 : days.back().day) << "]"
+              << std::endl;
+
+    // Параллельное построение интервалов
+    IntervalResult iv_result = find_intervals_parallel(days, rank, size);
+
+    std::cout << "Rank " << rank 
+              << ": found " << iv_result.intervals.size() << " intervals"
+              << ", compute " << std::fixed << std::setprecision(6) << iv_result.compute_time << " sec"
+              << ", wait " << iv_result.wait_time << " sec"
+              << std::endl;
+
+    // Сбор интервалов на ранке 0
+    double collect_wait = collect_intervals(iv_result.intervals, rank, size);
+
+    if (rank == 0) {
+        std::cout << "Rank 0: collected " << iv_result.intervals.size() << " total intervals"
+                  << ", wait " << std::fixed << std::setprecision(3) << collect_wait << " sec"
+                  << std::endl;
+    }
+
+    // Запись результатов в файл (только ранк 0)
+    if (rank == 0) {
+        double write_start = MPI_Wtime();
+        write_intervals("result.csv", iv_result.intervals);
+        double write_time = MPI_Wtime() - write_start;
+
+        std::cout << "Rank 0: wrote result.csv"
+                  << " in " << std::fixed << std::setprecision(3) << write_time << " sec"
+                  << std::endl;
+    }
+
+    // Вывод общего времени выполнения
+    MPI_Barrier(MPI_COMM_WORLD);
+    double total_time = MPI_Wtime() - total_start;
+    if (rank == 0) {
+        std::cout << "Total execution time: "
+                  << std::fixed << std::setprecision(3)
+                  << total_time << " sec" << std::endl;
+    }
+
     MPI_Finalize();
     return 0;
 }

src/utils.cpp

@@ -112,3 +112,16 @@ ByteRange calculate_byte_range(int rank, int size, int64_t file_size,
     
     return range;
 }
+
+void trim_edge_days(std::vector<DayStats>& days, int rank, int size) {
+    if (days.empty()) return;
+    
+    if (rank == 0) {
+        days.pop_back();
+    } else if (rank == size - 1) {
+        days.erase(days.begin());
+    } else {
+        days.pop_back();
+        days.erase(days.begin());
+    }
+}

src/utils.hpp

@@ -30,3 +30,9 @@ ByteRange calculate_byte_range(int rank, int size, int64_t file_size,
 
 // Получение размера файла
 int64_t get_file_size(const std::string& path);
+
+// Удаляет крайние дни, которые могут быть неполными из-за параллельного чтения
+// rank 0: удаляет последний день
+// последний rank: удаляет первый день  
+// промежуточные: удаляют первый и последний дни
+void trim_edge_days(std::vector<DayStats>& days, int rank, int size);