Пример с передачей частей данных
This commit is contained in:
@@ -1,13 +1,74 @@
|
|||||||
#include <iostream>
|
|
||||||
#include <mpi.h>
|
#include <mpi.h>
|
||||||
|
#include <iostream>
|
||||||
|
#include <vector>
|
||||||
|
#include <map>
|
||||||
|
|
||||||
#include "csv_loader.hpp"
|
#include "csv_loader.hpp"
|
||||||
#include "utils.hpp"
|
#include "utils.hpp"
|
||||||
#include "mpi_utils.hpp"
|
#include "record.hpp"
|
||||||
|
|
||||||
|
// Функция: отобрать записи для конкретного ранга
|
||||||
|
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) {
|
int main(int argc, char** argv) {
|
||||||
MPI_Init(&argc, &argv);
|
MPI_Init(&argc, &argv);
|
||||||
|
|
||||||
mpi_test(); // просто тест, что всё работает
|
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();
|
MPI_Finalize();
|
||||||
return 0;
|
return 0;
|
||||||
|
|||||||
@@ -1,9 +1,9 @@
|
|||||||
|
#include "mpi_utils.hpp"
|
||||||
#include <mpi.h>
|
#include <mpi.h>
|
||||||
#include <iostream>
|
#include <iostream>
|
||||||
|
|
||||||
void mpi_test() {
|
void mpi_print_basic() {
|
||||||
int rank, size;
|
int rank, size;
|
||||||
|
|
||||||
MPI_Comm_rank(MPI_COMM_WORLD, &rank);
|
MPI_Comm_rank(MPI_COMM_WORLD, &rank);
|
||||||
MPI_Comm_size(MPI_COMM_WORLD, &size);
|
MPI_Comm_size(MPI_COMM_WORLD, &size);
|
||||||
|
|
||||||
|
|||||||
@@ -1,2 +1,2 @@
|
|||||||
#pragma once
|
#pragma once
|
||||||
void mpi_test();
|
void mpi_print_basic();
|
||||||
|
|||||||
Reference in New Issue
Block a user