Пример с передачей частей данных

bitcoin-project/.gitignore

@@ -1,2 +1,3 @@
 data
-build
+build
+out.txt

bitcoin-project/Makefile

@@ -1,5 +1,5 @@
-CXX = g++
-CXXFLAGS = -std=c++17 -O2 -Wall -Wextra
+CXX = mpic++
+CXXFLAGS = -std=c++17 -O2 -Wall -Wextra -Wno-cast-function-type
 
 SRC_DIR = src
 BUILD_DIR = build

bitcoin-project/README.md

@@ -0,0 +1,12 @@
+Проект обязательно должен быть расположен в общей директории для всех узлов,
+например, в `/mnt/shared/bitcoin-project`
+
+```sh
+make
+```
+
+```sh
+sbatch run.slurm
+```
+
+Обязательно должны быть запущены все 4 нода. Результат будет в out.txt.

bitcoin-project/run.slurm

@@ -0,0 +1,7 @@
+#!/bin/bash
+#SBATCH --job-name=btc
+#SBATCH --nodes=4
+#SBATCH --ntasks=4
+#SBATCH --output=out.txt
+
+mpirun -np 4 ./build/bitcoin_app

bitcoin-project/src/main.cpp

@@ -1,17 +1,75 @@
+#include <mpi.h>
 #include <iostream>
+#include <vector>
+#include <map>
+
 #include "csv_loader.hpp"
+#include "utils.hpp"
+#include "record.hpp"
 
-int main() {
-    auto records = load_csv("data/data.csv");
-
-    std::cout << "Loaded rows: " << records.size() << "\n";
-
-    for (int i = 0; i < 5 && i < records.size(); i++) {
-        std::cout << records[i].timestamp << " "
-                  << records[i].open << " "
-                  << records[i].high << " "
-                  << records[i].low << " "
-                  << records[i].close << " "
-                  << records[i].volume << "\n";
+// Функция: отобрать записи для конкретного ранга
+std::vector<Record> select_records_for_rank(
+    const std::map<long long, std::vector<Record>>& days,
+    const std::vector<long long>& 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;
+}
+
+int main(int argc, char** argv) {
+    MPI_Init(&argc, &argv);
+
+    int rank, size;
+    MPI_Comm_rank(MPI_COMM_WORLD, &rank);
+    MPI_Comm_size(MPI_COMM_WORLD, &size);
+
+    std::vector<Record> local_records;
+
+    if (rank == 0) {
+        std::cout << "Rank 0 loading CSV..." << std::endl;
+
+        auto records = load_csv("data/data.csv");
+
+        auto days = group_by_day(records);
+        auto parts = split_days(days, size);
+
+        // Рассылаем данные
+        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 = 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);
+        }
+    }
+    else {
+        // Принимает данные
+        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);
+    }
+
+    MPI_Barrier(MPI_COMM_WORLD);
+
+    std::cout << "Rank " << rank << " received "
+              << local_records.size() << " records" << std::endl;
+
+    MPI_Finalize();
+    return 0;
 }

bitcoin-project/src/mpi_utils.cpp

@@ -0,0 +1,11 @@
+#include "mpi_utils.hpp"
+#include <mpi.h>
+#include <iostream>
+
+void mpi_print_basic() {
+    int rank, size;
+    MPI_Comm_rank(MPI_COMM_WORLD, &rank);
+    MPI_Comm_size(MPI_COMM_WORLD, &size);
+
+    std::cout << "Hello from rank " << rank << " of " << size << std::endl;
+}

bitcoin-project/src/mpi_utils.hpp

@@ -0,0 +1,2 @@
+#pragma once
+void mpi_print_basic();

bitcoin-project/src/utils.cpp

@@ -0,0 +1,25 @@
+#include "utils.hpp"
+
+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;
+}
+

bitcoin-project/src/utils.hpp

@@ -0,0 +1,11 @@
+#pragma once
+
+#include "record.hpp"
+#include <map>
+#include <vector>
+
+using DayIndex = long long;
+
+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);
+