univer-7th-semester/supercomputers
Code Actions Releases

Compare commits

base: univer-7th-semester:main
Branches Tags
univer-7th-semester:main
..
compare: univer-7th-semester:f96b771fa96bdc97f9a910e010a9128a20505d7b
Branches Tags
univer-7th-semester:main

1 Commits
main ... f96b771fa9

Author SHA1 Message Date
Arity-T
f96b771fa9 Агрегация 2025-12-13 12:31:21 +00:00
16 changed files with 511 additions and 1222 deletions
Expand all files Collapse all files

30
README.md
View File

@@ -2,16 +2,6 @@
[Kaggle Bitcoin Historical Data](https://www.kaggle.com/datasets/mczielinski/bitcoin-historical-data)
Исходные данные хранят информацию по каждой минуте. Чтобы увеличить объём данных
для более наглядной демонстрации эффективности параллельных вычислений
и вычислений на GPU, с помощью линейной интерполяции данные были преобразованы
из данных о каждой минуте в данные о каждых 10 секундах, то есть объём данных увеличился
в 6 раз.
```
python3 upsample.py -i ./data/data.csv -o ./data/data_10s.csv -s 10
```
## Задание
Группируем данные по дням (Timestamp), за каждый день вычисляем среднюю цену
@@ -20,6 +10,26 @@ python3 upsample.py -i ./data/data.csv -o ./data/data_10s.csv -s 10
не менее чем на 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/
```
## Сборка
Проект обязательно должен быть расположен в общей директории для всех узлов,

115
benchmark.py
View File

@@ -1,115 +0,0 @@
"""
Запускает 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")

10
run.slurm
View File

@@ -6,19 +6,13 @@
#SBATCH --output=out.txt
# Путь к файлу данных (должен существовать на всех узлах)
export DATA_PATH="/mnt/shared/supercomputers/data/data_10s.csv"
export DATA_PATH="/mnt/shared/supercomputers/data/data.csv"
# Доли данных для каждого ранка (сумма определяет пропорции)
export DATA_READ_SHARES="10,11,13,14"
export DATA_READ_SHARES="10,12,13,13"
# Размер перекрытия в байтах для обработки границ строк
export READ_OVERLAP_BYTES=131072
# Интервал агрегации в секундах (60 = минуты, 600 = 10 минут, 86400 = дни)
export AGGREGATION_INTERVAL=60
# Использовать ли CUDA для агрегации (0 = нет, 1 = да)
export USE_CUDA=1
cd /mnt/shared/supercomputers/build
mpirun -np $SLURM_NTASKS ./bitcoin_app

71
src/aggregation.cpp
View File

@@ -1,50 +1,40 @@
#include "aggregation.hpp"
#include "utils.hpp"
#include <map>
#include <algorithm>
#include <cstdint>
#include <limits>
#include <vector>
std::vector<PeriodStats> aggregate_periods(const std::vector<Record>& records) {
const int64_t interval = get_aggregation_interval();
std::vector<PeriodStats> result;
if (records.empty()) return result;
struct PeriodAccumulator {
std::vector<DayStats> aggregate_days(const std::vector<Record>& 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;
void add(const Record& r) {
const double avg = (r.low + r.high) / 2.0;
avg_sum += avg;
open_min = std::min(open_min, r.open);
open_max = std::max(open_max, r.open);
close_min = std::min(close_min, r.close);
close_max = std::max(close_max, r.close);
++count;
}
};
PeriodIndex current_period =
static_cast<PeriodIndex>(records[0].timestamp) / interval;
std::map<DayIndex, DayAccumulator> days;
PeriodAccumulator acc;
acc.add(records[0]);
for (const auto& r : records) {
DayIndex day = static_cast<DayIndex>(r.timestamp) / 86400;
auto& acc = days[day];
for (size_t i = 1; i < records.size(); ++i) {
const Record& r = records[i];
const PeriodIndex period =
static_cast<PeriodIndex>(r.timestamp) / interval;
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++;
}
if (period != current_period) {
PeriodStats stats;
stats.period = current_period;
std::vector<DayStats> result;
result.reserve(days.size());
for (const auto& [day, acc] : days) {
DayStats stats;
stats.day = day;
stats.avg = acc.avg_sum / static_cast<double>(acc.count);
stats.open_min = acc.open_min;
stats.open_max = acc.open_max;
@@ -52,24 +42,7 @@ std::vector<PeriodStats> aggregate_periods(const std::vector<Record>& records) {
stats.close_max = acc.close_max;
stats.count = acc.count;
result.push_back(stats);
current_period = period;
acc = PeriodAccumulator{};
}
acc.add(r);
}
// последний период
PeriodStats stats;
stats.period = current_period;
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;
}

6
src/aggregation.hpp
View File

@@ -1,8 +1,8 @@
#pragma once
#include "record.hpp"
#include "period_stats.hpp"
#include "day_stats.hpp"
#include <vector>
// Агрегация записей по периодам на одном узле
std::vector<PeriodStats> aggregate_periods(const std::vector<Record>& records);
// Агрегация записей по дням на одном узле
std::vector<DayStats> aggregate_days(const std::vector<Record>& records);

15
src/day_stats.hpp Normal file
View File

@@ -0,0 +1,15 @@
#pragma once
#include <cstdint>
using DayIndex = int64_t;
// Агрегированные данные за один день
struct DayStats {
DayIndex day; // индекс дня (timestamp / 86400)
double avg; // среднее значение (Low + High) / 2 по всем записям
double open_min; // минимальный Open за день
double open_max; // максимальный Open за день
double close_min; // минимальный Close за день
double close_max; // максимальный Close за день
int64_t count; // количество записей, по которым агрегировали
};

196
src/gpu_loader.cpp
View File

@@ -1,133 +1,145 @@
#include "gpu_loader.hpp"
#include <dlfcn.h>
#include <map>
#include <algorithm>
#include <iostream>
#include <cstdint>
// Структура результата GPU (должна совпадать с gpu_plugin.cu)
struct GpuPeriodStats {
int64_t period;
double avg;
double open_min;
double open_max;
double close_min;
double close_max;
int64_t count;
};
// Типы функций из GPU плагина
using gpu_is_available_fn = int (*)();
using gpu_aggregate_periods_fn = int (*)(
const double* h_timestamps,
const double* h_open,
const double* h_high,
const double* h_low,
const double* h_close,
int num_ticks,
int64_t interval,
GpuPeriodStats** h_out_stats,
int* out_num_periods
);
using gpu_free_results_fn = void (*)(GpuPeriodStats*);
#include <iomanip>
#include <omp.h>
static void* get_gpu_lib_handle() {
static void* h = dlopen("./libgpu_compute.so", RTLD_NOW | RTLD_LOCAL);
return h;
}
bool gpu_is_available() {
gpu_is_available_fn load_gpu_is_available() {
void* h = get_gpu_lib_handle();
if (!h) return false;
if (!h) return nullptr;
auto fn = reinterpret_cast<gpu_is_available_fn>(dlsym(h, "gpu_is_available"));
if (!fn) return false;
return fn() != 0;
auto fn = (gpu_is_available_fn)dlsym(h, "gpu_is_available");
return fn;
}
bool aggregate_periods_gpu(
const std::vector<Record>& records,
int64_t aggregation_interval,
std::vector<PeriodStats>& out_stats)
{
if (records.empty()) {
out_stats.clear();
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) {
std::cerr << "GPU: Failed to load libgpu_compute.so" << std::endl;
if (!h) return nullptr;
auto fn = (gpu_aggregate_days_fn)dlsym(h, "gpu_aggregate_days");
return fn;
}
bool aggregate_days_gpu(
const std::vector<Record>& records,
std::vector<DayStats>& out_stats,
gpu_aggregate_days_fn gpu_fn)
{
if (!gpu_fn || records.empty()) {
return false;
}
auto aggregate_fn = reinterpret_cast<gpu_aggregate_periods_fn>(
dlsym(h, "gpu_aggregate_periods"));
auto free_fn = reinterpret_cast<gpu_free_results_fn>(
dlsym(h, "gpu_free_results"));
// Общий таймер всей функции
double t_total_start = omp_get_wtime();
if (!aggregate_fn || !free_fn) {
std::cerr << "GPU: Failed to load functions from plugin" << std::endl;
return false;
// Таймер CPU preprocessing
double t_preprocess_start = omp_get_wtime();
// Группируем записи по дням и подготавливаем данные для GPU
std::map<DayIndex, std::vector<size_t>> day_record_indices;
for (size_t i = 0; i < records.size(); i++) {
DayIndex day = static_cast<DayIndex>(records[i].timestamp) / 86400;
day_record_indices[day].push_back(i);
}
int num_ticks = static_cast<int>(records.size());
int num_days = static_cast<int>(day_record_indices.size());
// Конвертируем AoS в SoA
std::vector<double> timestamps(num_ticks);
std::vector<double> open(num_ticks);
std::vector<double> high(num_ticks);
std::vector<double> low(num_ticks);
std::vector<double> close(num_ticks);
// Подготавливаем массивы для GPU
std::vector<GpuRecord> gpu_records;
std::vector<int> day_offsets;
std::vector<int> day_counts;
std::vector<long long> day_indices;
for (int i = 0; i < num_ticks; i++) {
timestamps[i] = records[i].timestamp;
open[i] = records[i].open;
high[i] = records[i].high;
low[i] = records[i].low;
close[i] = records[i].close;
gpu_records.reserve(records.size());
day_offsets.reserve(num_days);
day_counts.reserve(num_days);
day_indices.reserve(num_days);
int current_offset = 0;
for (auto& [day, indices] : day_record_indices) {
day_indices.push_back(day);
day_offsets.push_back(current_offset);
day_counts.push_back(static_cast<int>(indices.size()));
// Добавляем записи этого дня
for (size_t idx : indices) {
const auto& r = records[idx];
GpuRecord gr;
gr.timestamp = r.timestamp;
gr.open = r.open;
gr.high = r.high;
gr.low = r.low;
gr.close = r.close;
gr.volume = r.volume;
gpu_records.push_back(gr);
}
// Вызываем GPU функцию
GpuPeriodStats* gpu_stats = nullptr;
int num_periods = 0;
current_offset += static_cast<int>(indices.size());
}
int result = aggregate_fn(
timestamps.data(),
open.data(),
high.data(),
low.data(),
close.data(),
num_ticks,
aggregation_interval,
&gpu_stats,
&num_periods
// Выделяем память для результата
std::vector<GpuDayStats> gpu_stats(num_days);
double t_preprocess_ms = (omp_get_wtime() - t_preprocess_start) * 1000.0;
std::cout << " GPU CPU preprocessing: " << std::fixed << std::setprecision(3)
<< std::setw(7) << t_preprocess_ms << " ms" << std::endl << std::flush;
// Вызываем GPU функцию (включает: malloc, memcpy H->D, kernel, memcpy D->H, free)
// Детальные тайминги выводятся внутри GPU функции
int result = gpu_fn(
gpu_records.data(),
static_cast<int>(gpu_records.size()),
day_offsets.data(),
day_counts.data(),
day_indices.data(),
num_days,
gpu_stats.data()
);
if (result != 0) {
std::cerr << "GPU: Aggregation failed with code " << result << std::endl;
std::cout << " GPU: Function returned error code " << result << std::endl;
return false;
}
// Конвертируем результат в PeriodStats
// Конвертируем результат в DayStats
out_stats.clear();
out_stats.reserve(num_periods);
out_stats.reserve(num_days);
for (int i = 0; i < num_periods; i++) {
PeriodStats ps;
ps.period = gpu_stats[i].period;
ps.avg = gpu_stats[i].avg;
ps.open_min = gpu_stats[i].open_min;
ps.open_max = gpu_stats[i].open_max;
ps.close_min = gpu_stats[i].close_min;
ps.close_max = gpu_stats[i].close_max;
ps.count = gpu_stats[i].count;
out_stats.push_back(ps);
for (const auto& gs : gpu_stats) {
DayStats ds;
ds.day = gs.day;
ds.avg = gs.avg;
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);
}
// Освобождаем память
free_fn(gpu_stats);
// Общее время всей GPU функции (включая preprocessing)
double t_total_ms = (omp_get_wtime() - t_total_start) * 1000.0;
std::cout << " GPU TOTAL (with prep): " << std::fixed << std::setprecision(3)
<< std::setw(7) << t_total_ms << " ms" << std::endl << std::flush;
return true;
}

51
src/gpu_loader.hpp
View File

@@ -1,15 +1,50 @@
#pragma once
#include "period_stats.hpp"
#include "day_stats.hpp"
#include "record.hpp"
#include <vector>
// Проверка доступности CUDA
bool gpu_is_available();
// Агрегация периодов на GPU
// Возвращает true если успешно, false если GPU недоступен или ошибка
bool aggregate_periods_gpu(
const std::vector<Record>& records,
int64_t aggregation_interval,
std::vector<PeriodStats>& out_stats
// Типы функций из GPU плагина
using gpu_is_available_fn = int (*)();
// Структуры для GPU (должны совпадать с gpu_plugin.cu)
struct GpuRecord {
double timestamp;
double open;
double high;
double low;
double close;
double volume;
};
struct GpuDayStats {
long long day;
double avg;
double open_min;
double open_max;
double close_min;
double close_max;
long long count;
};
using gpu_aggregate_days_fn = int (*)(
const GpuRecord* h_records,
int num_records,
const int* h_day_offsets,
const int* h_day_counts,
const long long* h_day_indices,
int num_days,
GpuDayStats* h_out_stats
);
// Загрузка функций из плагина
gpu_is_available_fn load_gpu_is_available();
gpu_aggregate_days_fn load_gpu_aggregate_days();
// Обёртка для агрегации на GPU (возвращает true если успешно)
bool aggregate_days_gpu(
const std::vector<Record>& records,
std::vector<DayStats>& out_stats,
gpu_aggregate_days_fn gpu_fn
);

558
src/gpu_plugin.cu
View File

@@ -1,430 +1,262 @@
#include <cuda_runtime.h>
#include <cub/cub.cuh>
#include <cstdint>
#include <cfloat>
#include <cstdio>
#include <ctime>
#include <string>
#include <sstream>
#include <iomanip>
// ============================================================================
// Структуры данных
// ============================================================================
// SoA (Structure of Arrays) для входных данных на GPU
struct GpuTicksSoA {
double* timestamp;
double* open;
double* high;
double* low;
double* close;
int n;
};
// Результат агрегации одного периода
struct GpuPeriodStats {
int64_t period;
double avg;
double open_min;
double open_max;
double close_min;
double close_max;
int64_t count;
};
// ============================================================================
// Вспомогательные функции
// ============================================================================
// CPU таймер в миллисекундах
static double get_time_ms() {
struct timespec ts;
clock_gettime(CLOCK_MONOTONIC, &ts);
return ts.tv_sec * 1000.0 + ts.tv_nsec / 1000000.0;
}
#define CUDA_CHECK(call) do { \
cudaError_t err = call; \
if (err != cudaSuccess) { \
printf("CUDA error at %s:%d: %s\n", __FILE__, __LINE__, cudaGetErrorString(err)); \
return -1; \
} \
} while(0)
// Структуры данных (должны совпадать с C++ кодом)
struct GpuRecord {
double timestamp;
double open;
double high;
double low;
double close;
double volume;
};
// ============================================================================
// Kernel: вычисление period_id для каждого тика
// ============================================================================
__global__ void compute_period_ids_kernel(
const double* __restrict__ timestamps,
int64_t* __restrict__ period_ids,
int n,
int64_t interval)
{
int idx = blockIdx.x * blockDim.x + threadIdx.x;
if (idx < n) {
period_ids[idx] = static_cast<int64_t>(timestamps[idx]) / interval;
}
}
// ============================================================================
// Kernel: агрегация одного периода (один блок на период)
// ============================================================================
__global__ void aggregate_periods_kernel(
const double* __restrict__ open,
const double* __restrict__ high,
const double* __restrict__ low,
const double* __restrict__ close,
const int64_t* __restrict__ unique_periods,
const int* __restrict__ offsets,
const int* __restrict__ counts,
int num_periods,
GpuPeriodStats* __restrict__ out_stats)
{
int period_idx = blockIdx.x;
if (period_idx >= num_periods) return;
int offset = offsets[period_idx];
int count = counts[period_idx];
// Используем shared memory для редукции внутри блока
__shared__ double s_avg_sum;
__shared__ double s_open_min;
__shared__ double s_open_max;
__shared__ double s_close_min;
__shared__ double s_close_max;
// Инициализация shared memory первым потоком
if (threadIdx.x == 0) {
s_avg_sum = 0.0;
s_open_min = DBL_MAX;
s_open_max = -DBL_MAX;
s_close_min = DBL_MAX;
s_close_max = -DBL_MAX;
}
__syncthreads();
// Локальные аккумуляторы для каждого потока
double local_avg_sum = 0.0;
double local_open_min = DBL_MAX;
double local_open_max = -DBL_MAX;
double local_close_min = DBL_MAX;
double local_close_max = -DBL_MAX;
// Каждый поток обрабатывает свою часть тиков
for (int i = threadIdx.x; i < count; i += blockDim.x) {
int tick_idx = offset + i;
double avg = (low[tick_idx] + high[tick_idx]) / 2.0;
local_avg_sum += avg;
local_open_min = min(local_open_min, open[tick_idx]);
local_open_max = max(local_open_max, open[tick_idx]);
local_close_min = min(local_close_min, close[tick_idx]);
local_close_max = max(local_close_max, close[tick_idx]);
}
// Редукция с использованием атомарных операций
atomicAdd(&s_avg_sum, local_avg_sum);
atomicMin(reinterpret_cast<unsigned long long*>(&s_open_min),
__double_as_longlong(local_open_min));
atomicMax(reinterpret_cast<unsigned long long*>(&s_open_max),
__double_as_longlong(local_open_max));
atomicMin(reinterpret_cast<unsigned long long*>(&s_close_min),
__double_as_longlong(local_close_min));
atomicMax(reinterpret_cast<unsigned long long*>(&s_close_max),
__double_as_longlong(local_close_max));
__syncthreads();
// Первый поток записывает результат
if (threadIdx.x == 0) {
GpuPeriodStats stats;
stats.period = unique_periods[period_idx];
stats.avg = s_avg_sum / static_cast<double>(count);
stats.open_min = s_open_min;
stats.open_max = s_open_max;
stats.close_min = s_close_min;
stats.close_max = s_close_max;
stats.count = count;
out_stats[period_idx] = stats;
}
}
// ============================================================================
// Простой kernel для агрегации (один поток на период)
// Используется когда периодов много и тиков в каждом мало
// ============================================================================
__global__ void aggregate_periods_simple_kernel(
const double* __restrict__ open,
const double* __restrict__ high,
const double* __restrict__ low,
const double* __restrict__ close,
const int64_t* __restrict__ unique_periods,
const int* __restrict__ offsets,
const int* __restrict__ counts,
int num_periods,
GpuPeriodStats* __restrict__ out_stats)
{
int period_idx = blockIdx.x * blockDim.x + threadIdx.x;
if (period_idx >= num_periods) return;
int offset = offsets[period_idx];
int count = counts[period_idx];
double avg_sum = 0.0;
double open_min = DBL_MAX;
double open_max = -DBL_MAX;
double close_min = DBL_MAX;
double close_max = -DBL_MAX;
for (int i = 0; i < count; i++) {
int tick_idx = offset + i;
double avg = (low[tick_idx] + high[tick_idx]) / 2.0;
avg_sum += avg;
open_min = min(open_min, open[tick_idx]);
open_max = max(open_max, open[tick_idx]);
close_min = min(close_min, close[tick_idx]);
close_max = max(close_max, close[tick_idx]);
}
GpuPeriodStats stats;
stats.period = unique_periods[period_idx];
stats.avg = avg_sum / static_cast<double>(count);
stats.open_min = open_min;
stats.open_max = open_max;
stats.close_min = close_min;
stats.close_max = close_max;
stats.count = count;
out_stats[period_idx] = stats;
}
// ============================================================================
// Проверка доступности GPU
// ============================================================================
struct GpuDayStats {
long long day;
double avg;
double open_min;
double open_max;
double close_min;
double close_max;
long long count;
};
extern "C" int gpu_is_available() {
int n = 0;
cudaError_t err = cudaGetDeviceCount(&n);
if (err != cudaSuccess) return 0;
if (n > 0) {
cudaFree(0); // Форсируем инициализацию контекста
// Инициализируем CUDA контекст заранее (cudaFree(0) форсирует инициализацию)
cudaFree(0);
}
return (n > 0) ? 1 : 0;
}
// ============================================================================
// Главная функция агрегации на GPU
// ============================================================================
extern "C" int gpu_aggregate_periods(
const double* h_timestamps,
const double* h_open,
const double* h_high,
const double* h_low,
const double* h_close,
int num_ticks,
int64_t interval,
GpuPeriodStats** h_out_stats,
int* out_num_periods)
// Kernel для агрегации (каждый поток обрабатывает один день)
__global__ void aggregate_kernel(
const GpuRecord* records,
int num_records,
const int* day_offsets, // начало каждого дня в массиве records
const int* day_counts, // количество записей в каждом дне
const long long* day_indices, // индексы дней
int num_days,
GpuDayStats* out_stats)
{
if (num_ticks == 0) {
*h_out_stats = nullptr;
*out_num_periods = 0;
return 0;
// Глобальный индекс потока = индекс дня
int d = blockIdx.x * blockDim.x + threadIdx.x;
if (d >= num_days) return;
int offset = day_offsets[d];
int count = day_counts[d];
GpuDayStats stats;
stats.day = day_indices[d];
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];
// Accumulate avg = (low + high) / 2
avg_sum += (r.low + r.high) / 2.0;
// 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;
}
std::ostringstream output;
double total_start = get_time_ms();
stats.avg = avg_sum / static_cast<double>(count);
out_stats[d] = stats;
}
// ========================================================================
// Шаг 1: Выделение памяти и копирование данных на GPU
// ========================================================================
double step1_start = get_time_ms();
// Функция агрегации, вызываемая из C++
extern "C" int gpu_aggregate_days(
const GpuRecord* h_records,
int num_records,
const int* h_day_offsets,
const int* h_day_counts,
const long long* h_day_indices,
int num_days,
GpuDayStats* h_out_stats)
{
double cpu_total_start = get_time_ms();
double* d_timestamps = nullptr;
double* d_open = nullptr;
double* d_high = nullptr;
double* d_low = nullptr;
double* d_close = nullptr;
int64_t* d_period_ids = nullptr;
// === Создаём CUDA события для измерения времени ===
double cpu_event_create_start = get_time_ms();
size_t ticks_bytes = num_ticks * sizeof(double);
cudaEvent_t start_malloc, stop_malloc;
cudaEvent_t start_transfer, stop_transfer;
cudaEvent_t start_kernel, stop_kernel;
cudaEvent_t start_copy_back, stop_copy_back;
cudaEvent_t start_free, stop_free;
CUDA_CHECK(cudaMalloc(&d_timestamps, ticks_bytes));
CUDA_CHECK(cudaMalloc(&d_open, ticks_bytes));
CUDA_CHECK(cudaMalloc(&d_high, ticks_bytes));
CUDA_CHECK(cudaMalloc(&d_low, ticks_bytes));
CUDA_CHECK(cudaMalloc(&d_close, ticks_bytes));
CUDA_CHECK(cudaMalloc(&d_period_ids, num_ticks * sizeof(int64_t)));
cudaEventCreate(&start_malloc);
cudaEventCreate(&stop_malloc);
cudaEventCreate(&start_transfer);
cudaEventCreate(&stop_transfer);
cudaEventCreate(&start_kernel);
cudaEventCreate(&stop_kernel);
cudaEventCreate(&start_copy_back);
cudaEventCreate(&stop_copy_back);
cudaEventCreate(&start_free);
cudaEventCreate(&stop_free);
CUDA_CHECK(cudaMemcpy(d_timestamps, h_timestamps, ticks_bytes, cudaMemcpyHostToDevice));
CUDA_CHECK(cudaMemcpy(d_open, h_open, ticks_bytes, cudaMemcpyHostToDevice));
CUDA_CHECK(cudaMemcpy(d_high, h_high, ticks_bytes, cudaMemcpyHostToDevice));
CUDA_CHECK(cudaMemcpy(d_low, h_low, ticks_bytes, cudaMemcpyHostToDevice));
CUDA_CHECK(cudaMemcpy(d_close, h_close, ticks_bytes, cudaMemcpyHostToDevice));
double cpu_event_create_ms = get_time_ms() - cpu_event_create_start;
double step1_ms = get_time_ms() - step1_start;
// === ИЗМЕРЕНИЕ cudaMalloc ===
cudaEventRecord(start_malloc);
// ========================================================================
// Шаг 2: Вычисление period_id для каждого тика
// ========================================================================
double step2_start = get_time_ms();
GpuRecord* d_records = nullptr;
int* d_day_offsets = nullptr;
int* d_day_counts = nullptr;
long long* d_day_indices = nullptr;
GpuDayStats* d_out_stats = nullptr;
const int BLOCK_SIZE = 256;
int num_blocks = (num_ticks + BLOCK_SIZE - 1) / BLOCK_SIZE;
cudaError_t err;
compute_period_ids_kernel<<<num_blocks, BLOCK_SIZE>>>(
d_timestamps, d_period_ids, num_ticks, interval);
CUDA_CHECK(cudaGetLastError());
CUDA_CHECK(cudaDeviceSynchronize());
err = cudaMalloc(&d_records, num_records * sizeof(GpuRecord));
if (err != cudaSuccess) return -1;
double step2_ms = get_time_ms() - step2_start;
err = cudaMalloc(&d_day_offsets, num_days * sizeof(int));
if (err != cudaSuccess) { cudaFree(d_records); return -2; }
// ========================================================================
// Шаг 3: RLE (Run-Length Encode) для нахождения уникальных периодов
// ========================================================================
double step3_start = get_time_ms();
err = cudaMalloc(&d_day_counts, num_days * sizeof(int));
if (err != cudaSuccess) { cudaFree(d_records); cudaFree(d_day_offsets); return -3; }
int64_t* d_unique_periods = nullptr;
int* d_counts = nullptr;
int* d_num_runs = nullptr;
err = cudaMalloc(&d_day_indices, num_days * sizeof(long long));
if (err != cudaSuccess) { cudaFree(d_records); cudaFree(d_day_offsets); cudaFree(d_day_counts); return -4; }
CUDA_CHECK(cudaMalloc(&d_unique_periods, num_ticks * sizeof(int64_t)));
CUDA_CHECK(cudaMalloc(&d_counts, num_ticks * sizeof(int)));
CUDA_CHECK(cudaMalloc(&d_num_runs, sizeof(int)));
err = cudaMalloc(&d_out_stats, num_days * sizeof(GpuDayStats));
if (err != cudaSuccess) { cudaFree(d_records); cudaFree(d_day_offsets); cudaFree(d_day_counts); cudaFree(d_day_indices); return -5; }
// Определяем размер временного буфера для CUB
void* d_temp_storage = nullptr;
size_t temp_storage_bytes = 0;
cudaEventRecord(stop_malloc);
cudaEventSynchronize(stop_malloc);
cub::DeviceRunLengthEncode::Encode(
d_temp_storage, temp_storage_bytes,
d_period_ids, d_unique_periods, d_counts, d_num_runs,
num_ticks);
float time_malloc_ms = 0;
cudaEventElapsedTime(&time_malloc_ms, start_malloc, stop_malloc);
CUDA_CHECK(cudaMalloc(&d_temp_storage, temp_storage_bytes));
// === ИЗМЕРЕНИЕ memcpy H->D ===
cudaEventRecord(start_transfer);
cub::DeviceRunLengthEncode::Encode(
d_temp_storage, temp_storage_bytes,
d_period_ids, d_unique_periods, d_counts, d_num_runs,
num_ticks);
CUDA_CHECK(cudaGetLastError());
err = cudaMemcpy(d_records, h_records, num_records * sizeof(GpuRecord), cudaMemcpyHostToDevice);
if (err != cudaSuccess) return -10;
// Копируем количество уникальных периодов
int num_periods = 0;
CUDA_CHECK(cudaMemcpy(&num_periods, d_num_runs, sizeof(int), cudaMemcpyDeviceToHost));
err = cudaMemcpy(d_day_offsets, h_day_offsets, num_days * sizeof(int), cudaMemcpyHostToDevice);
if (err != cudaSuccess) return -11;
cudaFree(d_temp_storage);
d_temp_storage = nullptr;
err = cudaMemcpy(d_day_counts, h_day_counts, num_days * sizeof(int), cudaMemcpyHostToDevice);
if (err != cudaSuccess) return -12;
double step3_ms = get_time_ms() - step3_start;
err = cudaMemcpy(d_day_indices, h_day_indices, num_days * sizeof(long long), cudaMemcpyHostToDevice);
if (err != cudaSuccess) return -13;
// ========================================================================
// Шаг 4: Exclusive Scan для вычисления offsets
// ========================================================================
double step4_start = get_time_ms();
cudaEventRecord(stop_transfer);
cudaEventSynchronize(stop_transfer);
int* d_offsets = nullptr;
CUDA_CHECK(cudaMalloc(&d_offsets, num_periods * sizeof(int)));
float time_transfer_ms = 0;
cudaEventElapsedTime(&time_transfer_ms, start_transfer, stop_transfer);
temp_storage_bytes = 0;
cub::DeviceScan::ExclusiveSum(
d_temp_storage, temp_storage_bytes,
d_counts, d_offsets, num_periods);
// === ИЗМЕРЕНИЕ kernel ===
const int THREADS_PER_BLOCK = 256;
int num_blocks = (num_days + THREADS_PER_BLOCK - 1) / THREADS_PER_BLOCK;
CUDA_CHECK(cudaMalloc(&d_temp_storage, temp_storage_bytes));
cudaEventRecord(start_kernel);
cub::DeviceScan::ExclusiveSum(
d_temp_storage, temp_storage_bytes,
d_counts, d_offsets, num_periods);
CUDA_CHECK(cudaGetLastError());
aggregate_kernel<<<num_blocks, THREADS_PER_BLOCK>>>(
d_records, num_records,
d_day_offsets, d_day_counts, d_day_indices,
num_days, d_out_stats
);
cudaFree(d_temp_storage);
err = cudaGetLastError();
if (err != cudaSuccess) {
cudaFree(d_records);
cudaFree(d_day_offsets);
cudaFree(d_day_counts);
cudaFree(d_day_indices);
cudaFree(d_out_stats);
return -7;
}
double step4_ms = get_time_ms() - step4_start;
cudaEventRecord(stop_kernel);
cudaEventSynchronize(stop_kernel);
// ========================================================================
// Шаг 5: Агрегация периодов
// ========================================================================
double step5_start = get_time_ms();
float time_kernel_ms = 0;
cudaEventElapsedTime(&time_kernel_ms, start_kernel, stop_kernel);
GpuPeriodStats* d_out_stats = nullptr;
CUDA_CHECK(cudaMalloc(&d_out_stats, num_periods * sizeof(GpuPeriodStats)));
// === ИЗМЕРЕНИЕ memcpy D->H ===
cudaEventRecord(start_copy_back);
cudaMemcpy(h_out_stats, d_out_stats, num_days * sizeof(GpuDayStats), cudaMemcpyDeviceToHost);
cudaEventRecord(stop_copy_back);
cudaEventSynchronize(stop_copy_back);
// Используем простой kernel (один поток на период)
// т.к. обычно тиков в периоде немного
int agg_blocks = (num_periods + BLOCK_SIZE - 1) / BLOCK_SIZE;
float time_copy_back_ms = 0;
cudaEventElapsedTime(&time_copy_back_ms, start_copy_back, stop_copy_back);
aggregate_periods_simple_kernel<<<agg_blocks, BLOCK_SIZE>>>(
d_open, d_high, d_low, d_close,
d_unique_periods, d_offsets, d_counts,
num_periods, d_out_stats);
CUDA_CHECK(cudaGetLastError());
CUDA_CHECK(cudaDeviceSynchronize());
// === ИЗМЕРЕНИЕ cudaFree ===
cudaEventRecord(start_free);
double step5_ms = get_time_ms() - step5_start;
// ========================================================================
// Шаг 6: Копирование результатов на CPU
// ========================================================================
double step6_start = get_time_ms();
GpuPeriodStats* h_stats = new GpuPeriodStats[num_periods];
CUDA_CHECK(cudaMemcpy(h_stats, d_out_stats, num_periods * sizeof(GpuPeriodStats),
cudaMemcpyDeviceToHost));
double step6_ms = get_time_ms() - step6_start;
// ========================================================================
// Шаг 7: Освобождение GPU памяти
// ========================================================================
double step7_start = get_time_ms();
cudaFree(d_timestamps);
cudaFree(d_open);
cudaFree(d_high);
cudaFree(d_low);
cudaFree(d_close);
cudaFree(d_period_ids);
cudaFree(d_unique_periods);
cudaFree(d_counts);
cudaFree(d_offsets);
cudaFree(d_num_runs);
cudaFree(d_records);
cudaFree(d_day_offsets);
cudaFree(d_day_counts);
cudaFree(d_day_indices);
cudaFree(d_out_stats);
double step7_ms = get_time_ms() - step7_start;
cudaEventRecord(stop_free);
cudaEventSynchronize(stop_free);
// ========================================================================
// Итого
// ========================================================================
double total_ms = get_time_ms() - total_start;
float time_free_ms = 0;
cudaEventElapsedTime(&time_free_ms, start_free, stop_free);
// Формируем весь вывод одной строкой
output << " GPU aggregation (" << num_ticks << " ticks, interval=" << interval << " sec):\n";
output << " 1. Malloc + H->D copy: " << std::fixed << std::setprecision(3) << std::setw(7) << step1_ms << " ms\n";
output << " 2. Compute period_ids: " << std::setw(7) << step2_ms << " ms\n";
output << " 3. RLE (CUB): " << std::setw(7) << step3_ms << " ms (" << num_periods << " periods)\n";
output << " 4. Exclusive scan: " << std::setw(7) << step4_ms << " ms\n";
output << " 5. Aggregation kernel: " << std::setw(7) << step5_ms << " ms\n";
output << " 6. D->H copy: " << std::setw(7) << step6_ms << " ms\n";
output << " 7. Free GPU memory: " << std::setw(7) << step7_ms << " ms\n";
output << " GPU TOTAL: " << std::setw(7) << total_ms << " ms\n";
// Общее время GPU
float time_total_ms = time_malloc_ms + time_transfer_ms + time_kernel_ms + time_copy_back_ms + time_free_ms;
// Выводим всё одним принтом
printf("%s", output.str().c_str());
// === Освобождаем события ===
double cpu_event_destroy_start = get_time_ms();
cudaEventDestroy(start_malloc);
cudaEventDestroy(stop_malloc);
cudaEventDestroy(start_transfer);
cudaEventDestroy(stop_transfer);
cudaEventDestroy(start_kernel);
cudaEventDestroy(stop_kernel);
cudaEventDestroy(start_copy_back);
cudaEventDestroy(stop_copy_back);
cudaEventDestroy(start_free);
cudaEventDestroy(stop_free);
double cpu_event_destroy_ms = get_time_ms() - cpu_event_destroy_start;
double cpu_total_ms = get_time_ms() - cpu_total_start;
// Выводим детальную статистику
printf(" GPU Timings (%d records, %d days):\n", num_records, num_days);
printf(" cudaMalloc: %7.3f ms\n", time_malloc_ms);
printf(" memcpy H->D: %7.3f ms\n", time_transfer_ms);
printf(" kernel execution: %7.3f ms\n", time_kernel_ms);
printf(" memcpy D->H: %7.3f ms\n", time_copy_back_ms);
printf(" cudaFree: %7.3f ms\n", time_free_ms);
printf(" GPU TOTAL: %7.3f ms\n", cpu_total_ms);
fflush(stdout);
*h_out_stats = h_stats;
*out_num_periods = num_periods;
return 0;
}
// ============================================================================
// Освобождение памяти результатов
// ============================================================================
extern "C" void gpu_free_results(GpuPeriodStats* stats) {
delete[] stats;
}

310
src/intervals.cpp
View File

@@ -1,301 +1,69 @@
#include "intervals.hpp"
#include "utils.hpp"
#include <mpi.h>
#include <algorithm>
#include <cmath>
#include <fstream>
#include <iomanip>
#include <sstream>
#include <ctime>
#include <limits>
// Вспомогательная структура для накопления min/max в интервале
struct IntervalAccumulator {
PeriodIndex start_period;
double start_avg;
double open_min;
double open_max;
double close_min;
double close_max;
void init(const PeriodStats& p) {
start_period = p.period;
start_avg = p.avg;
open_min = p.open_min;
open_max = p.open_max;
close_min = p.close_min;
close_max = p.close_max;
std::vector<Interval> find_intervals(const std::vector<DayStats>& days, double threshold) {
if (days.empty()) {
return {};
}
void update(const PeriodStats& p) {
open_min = std::min(open_min, p.open_min);
open_max = std::max(open_max, p.open_max);
close_min = std::min(close_min, p.close_min);
close_max = std::max(close_max, p.close_max);
}
std::vector<Interval> intervals;
Interval finalize(const PeriodStats& end_period, double change) const {
Interval iv;
iv.start_period = start_period;
iv.end_period = end_period.period;
iv.start_avg = start_avg;
iv.end_avg = end_period.avg;
iv.change = change;
iv.open_min = std::min(open_min, end_period.open_min);
iv.open_max = std::max(open_max, end_period.open_max);
iv.close_min = std::min(close_min, end_period.close_min);
iv.close_max = std::max(close_max, end_period.close_max);
return iv;
}
};
// Упакованная структура PeriodStats для MPI передачи (8 doubles)
struct PackedPeriodStats {
double period; // PeriodIndex 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 PeriodStats& ps) {
period = static_cast<double>(ps.period);
avg = ps.avg;
open_min = ps.open_min;
open_max = ps.open_max;
close_min = ps.close_min;
close_max = ps.close_max;
count = static_cast<double>(ps.count);
valid = 1.0;
}
PeriodStats unpack() const {
PeriodStats ps;
ps.period = static_cast<PeriodIndex>(period);
ps.avg = avg;
ps.open_min = open_min;
ps.open_max = open_max;
ps.close_min = close_min;
ps.close_max = close_max;
ps.count = static_cast<int64_t>(count);
return ps;
}
bool is_valid() const { return valid > 0.5; }
void set_invalid() { valid = 0.0; }
};
IntervalResult find_intervals_parallel(
const std::vector<PeriodStats>& periods,
int rank, int size,
double threshold)
{
IntervalResult result;
result.compute_time = 0.0;
result.wait_time = 0.0;
if (periods.empty()) {
if (rank < size - 1) {
PackedPeriodStats 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) ? periods.size() : periods.size() - 1;
IntervalAccumulator acc;
size_t start_idx = 0;
bool have_pending_interval = false;
double price_base = days[start_idx].avg;
if (rank > 0) {
double wait_start = MPI_Wtime();
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;
PackedPeriodStats received;
MPI_Recv(&received, 8, MPI_DOUBLE, rank - 1, 0, MPI_COMM_WORLD, MPI_STATUS_IGNORE);
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;
result.wait_time = MPI_Wtime() - wait_start;
compute_start = MPI_Wtime();
// Находим min/max Open и Close в интервале
interval.open_min = days[start_idx].open_min;
interval.open_max = days[start_idx].open_max;
interval.close_min = days[start_idx].close_min;
interval.close_max = days[start_idx].close_max;
if (received.is_valid()) {
PeriodStats prev_period = received.unpack();
for (size_t j = start_idx + 1; j <= i; j++) {
interval.open_min = std::min(interval.open_min, days[j].open_min);
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);
}
for (start_idx = 0; start_idx < periods.size(); start_idx++) {
if (periods[start_idx].period > prev_period.period) {
intervals.push_back(interval);
// Начинаем новый интервал
start_idx = i + 1;
if (start_idx >= days.size()) {
break;
}
}
if (start_idx < process_until) {
acc.init(prev_period);
have_pending_interval = true;
for (size_t i = start_idx; i < process_until; i++) {
acc.update(periods[i]);
double change = std::abs(periods[i].avg - acc.start_avg) / acc.start_avg;
if (change >= threshold) {
result.intervals.push_back(acc.finalize(periods[i], change));
have_pending_interval = false;
start_idx = i + 1;
if (start_idx < process_until) {
acc.init(periods[start_idx]);
have_pending_interval = true;
}
}
}
}
} else {
if (process_until > 0) {
acc.init(periods[0]);
have_pending_interval = true;
start_idx = 0;
}
}
} else {
if (process_until > 0) {
acc.init(periods[0]);
have_pending_interval = true;
start_idx = 0;
price_base = days[start_idx].avg;
}
}
if (rank == 0 && have_pending_interval) {
for (size_t i = 1; i < process_until; i++) {
acc.update(periods[i]);
double change = std::abs(periods[i].avg - acc.start_avg) / acc.start_avg;
if (change >= threshold) {
result.intervals.push_back(acc.finalize(periods[i], change));
have_pending_interval = false;
start_idx = i + 1;
if (start_idx < process_until) {
acc.init(periods[start_idx]);
have_pending_interval = true;
}
}
}
return intervals;
}
if (rank == size - 1 && have_pending_interval && !periods.empty()) {
const auto& last_period = periods.back();
double change = std::abs(last_period.avg - acc.start_avg) / acc.start_avg;
result.intervals.push_back(acc.finalize(last_period, change));
}
result.compute_time = MPI_Wtime() - compute_start;
if (rank < size - 1) {
PackedPeriodStats to_send;
if (have_pending_interval) {
PeriodStats start_period;
start_period.period = acc.start_period;
start_period.avg = acc.start_avg;
start_period.open_min = acc.open_min;
start_period.open_max = acc.open_max;
start_period.close_min = acc.close_min;
start_period.close_max = acc.close_max;
start_period.count = 0;
to_send.pack(start_period);
} else if (periods.size() >= 2) {
to_send.pack(periods[periods.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;
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_period = static_cast<PeriodIndex>(buffer[i * 9 + 0]);
iv.end_period = static_cast<PeriodIndex>(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;
}
std::sort(local_intervals.begin(), local_intervals.end(),
[](const Interval& a, const Interval& b) {
return a.start_period < b.start_period;
});
} else {
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_period);
buffer[i * 9 + 1] = static_cast<double>(iv.end_period);
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 period_index_to_datetime(PeriodIndex period) {
int64_t interval = get_aggregation_interval();
time_t ts = static_cast<time_t>(period) * interval;
std::string day_index_to_date(DayIndex day) {
time_t ts = static_cast<time_t>(day) * 86400;
struct tm* tm_info = gmtime(&ts);
std::ostringstream oss;
oss << std::setfill('0')
<< (tm_info->tm_year + 1900) << "-"
<< std::setw(2) << (tm_info->tm_mon + 1) << "-"
<< std::setw(2) << tm_info->tm_mday << " "
<< std::setw(2) << tm_info->tm_hour << ":"
<< std::setw(2) << tm_info->tm_min << ":"
<< std::setw(2) << tm_info->tm_sec;
<< std::setw(2) << tm_info->tm_mday;
return oss.str();
}
@@ -304,11 +72,11 @@ void write_intervals(const std::string& filename, const std::vector<Interval>& i
std::ofstream out(filename);
out << std::fixed << std::setprecision(2);
out << "start_datetime,end_datetime,open_min,open_max,close_min,close_max,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 << period_index_to_datetime(iv.start_period) << ","
<< period_index_to_datetime(iv.end_period) << ","
out << day_index_to_date(iv.start_day) << ","
<< day_index_to_date(iv.end_day) << ","
<< iv.open_min << ","
<< iv.open_max << ","
<< iv.close_min << ","

39
src/intervals.hpp
View File

@@ -1,44 +1,27 @@
#pragma once
#include "period_stats.hpp"
#include "day_stats.hpp"
#include <vector>
#include <string>
// Интервал с изменением >= threshold
struct Interval {
PeriodIndex start_period;
PeriodIndex end_period;
double open_min;
double open_max;
double close_min;
double close_max;
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;
};
// Результат параллельного построения интервалов
struct IntervalResult {
std::vector<Interval> intervals;
double compute_time; // время вычислений
double wait_time; // время ожидания данных от предыдущего ранка
};
// Параллельное построение интервалов с использованием MPI
IntervalResult find_intervals_parallel(
const std::vector<PeriodStats>& periods,
int rank, int size,
double threshold = 0.10
);
// Сбор интервалов со всех ранков на ранк 0
double collect_intervals(
std::vector<Interval>& local_intervals,
int rank, int size
);
// Вычисление интервалов с изменением >= threshold (по умолчанию 10%)
std::vector<Interval> find_intervals(const std::vector<DayStats>& days, double threshold = 0.10);
// Вывод интервалов в файл
void write_intervals(const std::string& filename, const std::vector<Interval>& intervals);
// Преобразование PeriodIndex в строку даты/времени
std::string period_index_to_datetime(PeriodIndex period);
// Преобразование DayIndex в строку даты (YYYY-MM-DD)
std::string day_index_to_date(DayIndex day);

83
src/main.cpp
View File

@@ -5,31 +5,16 @@
#include "csv_loader.hpp"
#include "record.hpp"
#include "period_stats.hpp"
#include "day_stats.hpp"
#include "aggregation.hpp"
#include "intervals.hpp"
#include "utils.hpp"
#include "gpu_loader.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);
// Проверяем доступность GPU
bool use_cuda = get_use_cuda();
bool have_gpu = gpu_is_available();
bool use_gpu = use_cuda && have_gpu;
std::cout << "Rank " << rank
<< ": USE_CUDA=" << use_cuda
<< ", GPU available=" << have_gpu
<< ", using " << (use_gpu ? "GPU" : "CPU")
<< std::endl;
// Параллельное чтение данных
double read_start = MPI_Wtime();
std::vector<Record> records = load_csv_parallel(rank, size);
@@ -40,76 +25,16 @@ int main(int argc, char** argv) {
<< " in " << std::fixed << std::setprecision(3) << read_time << " sec"
<< std::endl;
// Агрегация по периодам
// Агрегация по дням
double agg_start = MPI_Wtime();
std::vector<PeriodStats> periods;
if (use_gpu) {
int64_t interval = get_aggregation_interval();
if (!aggregate_periods_gpu(records, interval, periods)) {
std::cerr << "Rank " << rank << ": GPU aggregation failed, falling back to CPU" << std::endl;
periods = aggregate_periods(records);
}
} else {
periods = aggregate_periods(records);
}
std::vector<DayStats> days = aggregate_days(records);
double agg_time = MPI_Wtime() - agg_start;
std::cout << "Rank " << rank
<< ": aggregated " << periods.size() << " periods"
<< " [" << (periods.empty() ? 0 : periods.front().period)
<< ".." << (periods.empty() ? 0 : periods.back().period) << "]"
<< ": aggregated " << days.size() << " days"
<< " in " << std::fixed << std::setprecision(3) << agg_time << " sec"
<< std::endl;
// Удаляем крайние периоды (могут быть неполными из-за параллельного чтения)
trim_edge_periods(periods, rank, size);
std::cout << "Rank " << rank
<< ": after trim " << periods.size() << " periods"
<< " [" << (periods.empty() ? 0 : periods.front().period)
<< ".." << (periods.empty() ? 0 : periods.back().period) << "]"
<< std::endl;
// Параллельное построение интервалов
IntervalResult iv_result = find_intervals_parallel(periods, 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;
}

15
src/period_stats.hpp
View File

@@ -1,15 +0,0 @@
#pragma once
#include <cstdint>
using PeriodIndex = int64_t;
// Агрегированные данные за один период
struct PeriodStats {
PeriodIndex period; // индекс периода (timestamp / AGGREGATION_INTERVAL)
double avg; // среднее значение (Low + High) / 2 по всем записям
double open_min; // минимальный Open за период
double open_max; // максимальный Open за период
double close_min; // минимальный Close за период
double close_max; // максимальный Close за период
int64_t count; // количество записей, по которым агрегировали
};

51
src/utils.cpp
View File

@@ -4,6 +4,29 @@
#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;
for (const auto& r : recs) {
DayIndex day = static_cast<DayIndex>(r.timestamp) / 86400;
days[day].push_back(r);
}
return days;
}
std::vector<std::vector<DayIndex>> split_days(const std::map<DayIndex, std::vector<Record>>& days, int parts) {
std::vector<std::vector<DayIndex>> out(parts);
int i = 0;
for (auto& kv : days) {
out[i % parts].push_back(kv.first);
i++;
}
return out;
}
int get_num_cpu_threads() {
const char* env_threads = std::getenv("NUM_CPU_THREADS");
int num_cpu_threads = 1;
@@ -40,14 +63,6 @@ int64_t get_read_overlap_bytes() {
return std::stoll(get_env("READ_OVERLAP_BYTES"));
}
int64_t get_aggregation_interval() {
return std::stoll(get_env("AGGREGATION_INTERVAL"));
}
bool get_use_cuda() {
return std::stoi(get_env("USE_CUDA")) != 0;
}
int64_t get_file_size(const std::string& path) {
std::ifstream file(path, std::ios::binary | std::ios::ate);
if (!file.is_open()) {
@@ -58,6 +73,7 @@ int64_t get_file_size(const std::string& path) {
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;
@@ -66,6 +82,8 @@ ByteRange calculate_byte_range(int rank, int size, int64_t file_size,
}
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;
@@ -75,31 +93,22 @@ ByteRange calculate_byte_range(int rank, int size, int64_t file_size,
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;
}
void trim_edge_periods(std::vector<PeriodStats>& periods, int rank, int size) {
if (periods.empty()) return;
if (rank == 0) {
periods.pop_back();
} else if (rank == size - 1) {
periods.erase(periods.begin());
} else {
periods.pop_back();
periods.erase(periods.begin());
}
}

11
src/utils.hpp
View File

@@ -1,20 +1,22 @@
#pragma once
#include "record.hpp"
#include "period_stats.hpp"
#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();
int64_t get_aggregation_interval();
bool get_use_cuda();
// Структура для хранения диапазона байт для чтения
struct ByteRange {
@@ -28,6 +30,3 @@ ByteRange calculate_byte_range(int rank, int size, int64_t file_size,
// Получение размера файла
int64_t get_file_size(const std::string& path);
// Удаляет крайние периоды, которые могут быть неполными из-за параллельного чтения
void trim_edge_periods(std::vector<PeriodStats>& periods, int rank, int size);

136
upsample.py
View File

@@ -1,136 +0,0 @@
#!/usr/bin/env python3
import argparse
import sys
import time
def parse_row(line: str):
# Timestamp,Open,High,Low,Close,Volume
ts, o, h, l, c, v = line.split(',')
return int(float(ts)), float(o), float(h), float(l), float(c), float(v)
def fmt_row(ts, o, h, l, c, v):
return f"{ts},{o:.2f},{h:.2f},{l:.2f},{c:.2f},{v:.8f}\n"
def count_lines_fast(path: str) -> int:
with open(path, "rb") as f:
return sum(1 for _ in f) - 1 # минус header
def main(inp, out, step, flush_every):
# считаем количество строк для прогресса
total_lines = count_lines_fast(inp)
print(f"Total input rows: {total_lines:,}", file=sys.stderr)
start_time = time.time()
processed = 0
last_report = start_time
with open(inp, "r", buffering=8 * 1024 * 1024) as fin, \
open(out, "w", buffering=8 * 1024 * 1024) as fout:
fin.readline() # пропускаем header
fout.write("Timestamp,Open,High,Low,Close,Volume\n")
first = fin.readline()
if not first:
return
prev = parse_row(first.strip())
out_buf = []
out_rows = 0
for line in fin:
line = line.strip()
if not line:
continue
cur = parse_row(line)
t1, o1, h1, l1, c1, v1 = prev
t2, o2, h2, l2, c2, v2 = cur
dt = t2 - t1
steps = dt // step
if steps > 0:
do = o2 - o1
dh = h2 - h1
dl = l2 - l1
dc = c2 - c1
dv = v2 - v1
inv = 1.0 / steps
for i in range(steps):
a = i * inv
out_buf.append(fmt_row(
t1 + i * step,
o1 + do * a,
h1 + dh * a,
l1 + dl * a,
c1 + dc * a,
v1 + dv * a
))
out_rows += steps
prev = cur
processed += 1
# прогресс
if processed % 100_000 == 0:
now = time.time()
if now - last_report >= 0.5:
pct = processed * 100.0 / total_lines
elapsed = now - start_time
speed = processed / elapsed if elapsed > 0 else 0
eta = (total_lines - processed) / speed if speed > 0 else 0
print(
f"\rprocessed: {processed:,} / {total_lines:,} "
f"({pct:5.1f}%) | "
f"out ~ {out_rows:,} | "
f"{speed:,.0f} rows/s | "
f"ETA {eta/60:5.1f} min",
end="",
file=sys.stderr,
flush=True,
)
last_report = now
# сброс буфера
if out_rows >= flush_every:
fout.write("".join(out_buf))
out_buf.clear()
out_rows = 0
# остатки
if out_buf:
fout.write("".join(out_buf))
# последнюю строку пишем как есть
t, o, h, l, c, v = prev
fout.write(fmt_row(t, o, h, l, c, v))
total_time = time.time() - start_time
print(
f"\nDone in {total_time/60:.1f} min",
file=sys.stderr
)
if __name__ == "__main__":
ap = argparse.ArgumentParser()
ap.add_argument("-i", "--input", required=True)
ap.add_argument("-o", "--output", required=True)
ap.add_argument("-s", "--step", type=int, default=10)
ap.add_argument("--flush-every", type=int, default=200_000)
args = ap.parse_args()
if args.step <= 0:
raise SystemExit("step must be > 0")
main(args.input, args.output, args.step, args.flush_every)