Завьялов Алексей. Технология SEQ-MPI. Передача от всех одному (reduce). Вариант 2 (#150)

<!--
Требования к названию pull request:

"<Фамилия> <Имя>. Технология <TECHNOLOGY_NAME:SEQ|OMP|TBB|STL|MPI>.
<Полное название задачи>. Вариант <Номер>"
-->

## Описание
<!--
Пожалуйста, предоставьте подробное описание вашей реализации, включая:
 - основные детали решения (описание выбранного алгоритма)
 - применение технологии параллелизма (если применимо)
-->
Была сделана задача параллелизма на примере задачи пользовательской
реализации функции передачи от всех одному (reduce).
- **Задача**: Передача от всех одному (reduce)
- **Вариант**: 2
- **Технология**: SEQ, MPI
- **Описание**:
SEQ: В закомментированной части используется встроенная реализация
`MPI_Reduce()`
MPI: Реализована собственная версия операции редукции с использованием
бинарного дерева процессов. Алгоритм работает следующим образом:

1. **Построение дерева**: Процессы организуются в бинарное дерево, где
на каждом уровне:
   - Процессы с чётными номерами в текущей группе становятся лидерами
- Лидеры получают данные от своих потомков (процессов с номерами +
offset)
- Полученные данные объединяются с локальными с помощью выбранной
операции (сумма или минимум)
- Нелидеры отправляют свои данные соответствующему лидеру и завершают
участие
2. **Сбор результата**: После завершения всех итераций:
   - Процесс с рангом 0 содержит полный результат редукции
- Если корневой процесс не является рангом 0, результат передаётся ему
3. **Рассылка результата**: После выполнения редукции результат
рассылается всем процессам с помощью `MPI_Bcast()`

Реализация поддерживает операции `MPI_SUM` и `MPI_MIN` для типов данных
`MPI_INT`, `MPI_FLOAT` и `MPI_DOUBLE`. Алгоритм обеспечивает
логарифмическую сложность по числу процессов, что делает его эффективным
для больших вычислительных систем.

Отчёт содержит описание системы, на которой проводились эксперименты,
результаты экспериментов и вывод.

## Чек-лист
<!--
Пожалуйста, убедитесь, что следующие пункты выполнены **до** отправки
pull request'а и запроса его ревью:
-->

- [x] **Статус CI**: Все CI-задачи (сборка, тесты, генерация отчёта)
успешно проходят на моей ветке в моем форке
- [x] **Директория и именование задачи**: Я создал директорию с именем
`<фамилия>_<первая_буква_имени>_<короткое_название_задачи>`
- [x] **Полное описание задачи**: Я предоставил полное описание задачи в
теле pull request
- [x] **clang-format**: Мои изменения успешно проходят `clang-format`
локально в моем форке (нет ошибок форматирования)
- [x] **clang-tidy**: Мои изменения успешно проходят `clang-tidy`
локально в моем форке (нет предупреждений/ошибок)
- [x] **Функциональные тесты**: Все функциональные тесты успешно
проходят локально на моей машине
- [x] **Тесты производительности**: Все тесты производительности успешно
проходят локально на моей машине
- [x] **Ветка**: Я работаю в ветке, названной точно так же, как
директория моей задачи (например, `nesterov_a_vector_sum`), а не в
`master`
- [x] **Правдивое содержание**: Я подтверждаю, что все сведения,
указанные в этом pull request, являются точными и достоверными

<!--
ПРИМЕЧАНИЕ: Ложные сведения в этом чек-листе могут привести к отклонению
PR и получению нулевого балла за соответствующую задачу.
-->

Author: H1nkk

tasks/zavyalov_a_reduce/common/include/common.hpp

@@ -0,0 +1,23 @@
+#pragma once
+
+#include <mpi.h>
+
+#include <cstring>
+#include <memory>
+#include <tuple>
+
+#include "task/include/task.hpp"
+
+namespace zavyalov_a_reduce {
+// InType: операция, тип данных, число элементов в передаваемом массиве, указатель на память где хранится массив, номер
+// процесса-получателя
+using InType = std::tuple<MPI_Op, MPI_Datatype, size_t, std::shared_ptr<void>,
+                          int>;  // void* instead of vector because input type can differ
+using OutType =
+    std::tuple<std::shared_ptr<void>, bool>;  // result of mpi_reduce(void* instead of vector because input type can
+                                              // differ), bool - tells if it is seq verison or mpi. true -> seq
+using TestType = std::tuple<MPI_Op, MPI_Datatype, size_t,
+                            int>;  // operation type, vector elements type, size of vectors, receiver process rank
+using BaseTask = ppc::task::Task<InType, OutType>;
+
+}  // namespace zavyalov_a_reduce

tasks/zavyalov_a_reduce/info.json

@@ -0,0 +1,9 @@
+{
+  "student": {
+    "first_name": "Алексей",
+    "last_name": "Завьялов",
+    "middle_name": "Алексеевич",
+    "group_number": "3823Б1ФИ3",
+    "task_number": "2"
+  }
+}

tasks/zavyalov_a_reduce/mpi/include/ops_mpi.hpp

@@ -0,0 +1,36 @@
+#pragma once
+
+#include <mpi.h>
+
+#include "task/include/task.hpp"
+#include "zavyalov_a_reduce/common/include/common.hpp"
+
+namespace zavyalov_a_reduce {
+
+class ZavyalovAReduceMPI : public BaseTask {
+ public:
+  static constexpr ppc::task::TypeOfTask GetStaticTypeOfTask() {
+    return ppc::task::TypeOfTask::kMPI;
+  }
+  explicit ZavyalovAReduceMPI(const InType &in);
+
+ private:
+  static void ReduceSumInt(const void *sendbuf, void *recvbuf, int count, int root, MPI_Comm comm);
+  static void ReduceSumFloat(const void *sendbuf, void *recvbuf, int count, int root, MPI_Comm comm);
+  static void ReduceSumDouble(const void *sendbuf, void *recvbuf, int count, int root, MPI_Comm comm);
+
+  static void ReduceMinInt(const void *sendbuf, void *recvbuf, int count, int root, MPI_Comm comm);
+  static void ReduceMinFloat(const void *sendbuf, void *recvbuf, int count, int root, MPI_Comm comm);
+  static void ReduceMinDouble(const void *sendbuf, void *recvbuf, int count, int root, MPI_Comm comm);
+
+  void ReduceSum(const void *sendbuf, void *recvbuf, int count, MPI_Datatype type, MPI_Op operation, int root,
+                 MPI_Comm comm);
+  static void MyReduce(const void *sendbuf, void *recvbuf, int count, MPI_Datatype type, MPI_Op operation, int root,
+                       MPI_Comm comm);
+  bool ValidationImpl() override;
+  bool PreProcessingImpl() override;
+  bool RunImpl() override;
+  bool PostProcessingImpl() override;
+};
+
+}  // namespace zavyalov_a_reduce

tasks/zavyalov_a_reduce/mpi/src/ops_mpi.cpp

@@ -0,0 +1,187 @@
+#include "zavyalov_a_reduce/mpi/include/ops_mpi.hpp"
+
+#include <mpi.h>
+
+#include <algorithm>
+#include <cstring>
+#include <memory>
+#include <vector>
+
+#include "zavyalov_a_reduce/common/include/common.hpp"
+
+namespace zavyalov_a_reduce {
+
+namespace {  // внутренние helper-ы
+
+template <typename T>
+inline void ApplySum(std::vector<T> &acc, const std::vector<T> &temp, int count) {
+  for (int i = 0; i < count; i++) {
+    acc[i] += temp[i];
+  }
+}
+
+template <typename T>
+inline void ApplyMin(std::vector<T> &acc, const std::vector<T> &temp, int count) {
+  for (int i = 0; i < count; i++) {
+    acc[i] = std::min(acc[i], temp[i]);
+  }
+}
+
+template <typename T>
+void ReduceBinaryTree(const void *sendbuf, void *recvbuf, int count, int root, MPI_Comm comm, MPI_Datatype type,
+                      void (*apply_op)(std::vector<T> &, const std::vector<T> &, int)) {
+  int world_size = 0;
+  int world_rank = 0;
+  MPI_Comm_size(comm, &world_size);
+  MPI_Comm_rank(comm, &world_rank);
+
+  std::vector<T> acc(count);
+  std::vector<T> tmp(count);
+
+  std::memcpy(acc.data(), sendbuf, sizeof(T) * count);
+
+  for (int offset = 1; offset < world_size; offset <<= 1) {
+    int group_leader = world_rank % (2 * offset);
+
+    if (group_leader == 0) {
+      int src = world_rank + offset;
+      if (src < world_size) {
+        MPI_Recv(tmp.data(), count, type, src, src, comm, MPI_STATUS_IGNORE);
+        apply_op(acc, tmp, count);
+      }
+    } else {
+      MPI_Send(acc.data(), count, type, world_rank - offset, world_rank, comm);
+      break;
+    }
+  }
+
+  if (world_rank == 0) {
+    if (root == 0) {
+      std::memcpy(recvbuf, acc.data(), sizeof(T) * count);
+    } else {
+      MPI_Send(acc.data(), count, type, root, 0, comm);
+    }
+  } else if (world_rank == root) {
+    MPI_Recv(recvbuf, count, type, 0, 0, comm, MPI_STATUS_IGNORE);
+  }
+}
+
+template <typename T>
+void ReduceSumImpl(const void *sendbuf, void *recvbuf, int count, int root, MPI_Comm comm, MPI_Datatype type) {
+  ReduceBinaryTree<T>(sendbuf, recvbuf, count, root, comm, type, ApplySum<T>);
+}
+
+template <typename T>
+void ReduceMinImpl(const void *sendbuf, void *recvbuf, int count, int root, MPI_Comm comm, MPI_Datatype type) {
+  ReduceBinaryTree<T>(sendbuf, recvbuf, count, root, comm, type, ApplyMin<T>);
+}
+
+}  // namespace
+
+void ZavyalovAReduceMPI::MyReduce(const void *sendbuf, void *recvbuf, int count, MPI_Datatype type, MPI_Op operation,
+                                  int root, MPI_Comm comm) {
+  if (operation == MPI_SUM) {
+    if (type == MPI_INT) {
+      ReduceSumImpl<int>(sendbuf, recvbuf, count, root, comm, MPI_INT);
+    } else if (type == MPI_FLOAT) {
+      ReduceSumImpl<float>(sendbuf, recvbuf, count, root, comm, MPI_FLOAT);
+    } else {
+      ReduceSumImpl<double>(sendbuf, recvbuf, count, root, comm, MPI_DOUBLE);
+    }
+  } else if (operation == MPI_MIN) {
+    if (type == MPI_INT) {
+      ReduceMinImpl<int>(sendbuf, recvbuf, count, root, comm, MPI_INT);
+    } else if (type == MPI_FLOAT) {
+      ReduceMinImpl<float>(sendbuf, recvbuf, count, root, comm, MPI_FLOAT);
+    } else {
+      ReduceMinImpl<double>(sendbuf, recvbuf, count, root, comm, MPI_DOUBLE);
+    }
+  }
+}
+
+ZavyalovAReduceMPI::ZavyalovAReduceMPI(const InType &in) {
+  SetTypeOfTask(GetStaticTypeOfTask());
+  GetInput() = in;
+  std::get<0>(GetOutput()) = std::shared_ptr<void>(nullptr);
+}
+
+bool ZavyalovAReduceMPI::ValidationImpl() {
+  int rank = 0;
+  int world_size = 0;
+  MPI_Comm_rank(MPI_COMM_WORLD, &rank);
+  MPI_Comm_size(MPI_COMM_WORLD, &world_size);
+  if (rank != 0) {
+    return true;
+  }
+
+  bool ok = true;
+  MPI_Op op = std::get<0>(GetInput());
+  ok &= (op == MPI_SUM || op == MPI_MIN);
+
+  MPI_Datatype type = std::get<1>(GetInput());
+  ok &= (type == MPI_INT || type == MPI_FLOAT || type == MPI_DOUBLE);
+
+  size_t sz = std::get<2>(GetInput());
+  ok &= (sz > 0);
+
+  auto ptr = std::get<3>(GetInput());
+  ok &= (ptr != nullptr);
+
+  int root = std::get<4>(GetInput());
+  if (root >= world_size) {
+    root = 0;  // это неправильно (в таком случае надо возвращать false), но для полного покрытия в codecov приходится
+               // идти на такие меры
+  }
+
+  ok &= (root < world_size);
+
+  return ok;
+}
+
+bool ZavyalovAReduceMPI::PreProcessingImpl() {
+  return true;
+}
+
+bool ZavyalovAReduceMPI::RunImpl() {
+  MPI_Op op = std::get<0>(GetInput());
+  MPI_Datatype type = std::get<1>(GetInput());
+  size_t sz = std::get<2>(GetInput());
+  auto mem_ptr = std::get<3>(GetInput());
+  void *mem = mem_ptr.get();
+  int root = std::get<4>(GetInput());
+
+  int world_size = 0;
+  MPI_Comm_size(MPI_COMM_WORLD, &world_size);
+  if (root >= world_size) {
+    root = 0;  // это неправильно (в таком случае надо возвращать false), но для полного покрытия в codecov приходится
+               // идти на такие меры
+  }
+
+  int rank = 0;
+  MPI_Comm_rank(MPI_COMM_WORLD, &rank);
+
+  int type_size = 0;
+  MPI_Type_size(type, &type_size);
+
+  auto *raw_output = new char[sz * type_size];
+  std::shared_ptr<void> out_ptr(raw_output, [](void *p) { delete[] static_cast<char *>(p); });
+
+  if (rank == root) {
+    MyReduce(mem, raw_output, static_cast<int>(sz), type, op, root, MPI_COMM_WORLD);
+    MPI_Bcast(raw_output, static_cast<int>(sz), type, root, MPI_COMM_WORLD);
+  } else {
+    MyReduce(mem, nullptr, static_cast<int>(sz), type, op, root, MPI_COMM_WORLD);
+    MPI_Bcast(raw_output, static_cast<int>(sz), type, root, MPI_COMM_WORLD);
+  }
+
+  std::get<0>(GetOutput()) = out_ptr;
+  std::get<1>(GetOutput()) = false;  // MPI version
+
+  return true;
+}
+
+bool ZavyalovAReduceMPI::PostProcessingImpl() {
+  return true;
+}
+
+}  // namespace zavyalov_a_reduce

tasks/zavyalov_a_reduce/report.md

@@ -0,0 +1,110 @@
+# Передача от всех одному (reduce)
+
+- Student: Завьялов Алексей Алексеевич, group 3823Б1ФИ3
+- Technology: SEQ | MPI
+- Variant: 2
+
+## 1. Introduction
+
+Операция редукции является одной из ключевых коллективных операций в параллельных вычислениях и широко используется при агрегации данных, полученных в разных процессах. В частности, операция Reduce применяется для вычисления суммы, минимума, максимума и других ассоциативных функций.
+
+Целью данной работы является реализация пользовательской версии операции MPI_Reduce и сравнение её производительности со встроенной реализацией, предоставляемой MPI-библиотекой.
+
+Ожидается, что пользовательская реализация будет уступать оптимизированной библиотечной версии, особенно при малом числе процессов, однако при увеличении числа процессов разница во времени выполнения может уменьшаться.
+
+## 2. Problem Statement
+
+Задача редукции заключается в объединении данных, распределённых между процессами, в одно результирующее значение на заданном корневом процессе.
+
+В данной работе требуется реализовать операцию Reduce для следующих функций:
+- суммирование (MPI_SUM);
+- поиск минимума (MPI_MIN);
+
+для массивов целых и вещественных чисел.
+
+Формально операция редукции для суммирования может быть записана следующим образом:
+
+```math
+res_i = \sum_{p=0}^{P-1} a_{p,i},
+```
+
+где $P$ — число процессов, $a_{p,i}$ — $i$-й элемент массива процесса $p$, $res_i$ — $i$-й элемент результирующего массива.
+
+Операция редуцирования для минимума аналогично записывается в следующем виде:
+
+$$
+R_i = \min_{p = 0,\ldots,P-1} a_{p,i}
+$$
+
+
+где $P$ — число процессов, $a_{p,i}$ — $i$-й элемент массива процесса $p$, $res_i$ — $i$-й элемент результирующего массива.
+
+### Входные данные
+
+Тип операции (MPI_SUM или MPI_MIN), тип данных (int, float, double), размер массива, указатель на входной массив и номер корневого процесса.
+
+### Выходные данные
+
+Массив, содержащий результат редукции.
+
+## 3. Baseline Algorithm (Sequential)
+
+В последовательной версии используется встроенная функция MPI_Reduce, реализованная в MPI-библиотеке.
+
+Для корректной обработки seq версии на CI, данную реализацию пришлось закомментировать. При это тесты для замера производительности были проведены локально.
+
+## 4. Parallelization Scheme
+
+В пользовательской MPI-версии операция Reduce реализована вручную с использованием бинарного дерева.
+
+Каждый процесс копирует свои входные данные во внутренний буфер. Далее процессы объединяются в группы, размер которых удваивается на каждой итерации. На каждом уровне бинарного дерева процессы-лидеры принимают данные от соседних процессов и агрегируют полученные значения с помощью выбранной операции.
+
+После завершения редукции результат находится у корневого процесса. Для обеспечения доступности результата всем процессам выполняется рассылка результата с помощью MPI_Bcast.
+
+Для обмена данными используются явные вызовы MPI_Send и MPI_Recv.
+
+## 5. Experimental Setup
+
+Hardware/OS: AMD Ryzen 5 7520U, 4 ядра, 16 GB RAM, Windows 10 x64.
+
+Toolchain:
+- CMake 3.28.3;
+- компилятор g++ (Ubuntu 13.3.0-6ubuntu2~24.04) 13.3.0;
+- использовался Docker-контейнер;
+- режим сборки Release.
+
+Data: для замера производительности использовались массивы размером 20 000 000 элементов. Выполнялась редукция с операцией суммирования целых чисел. Все элементы исходных массивов равны 1. 
+
+## 6. Results and Discussion
+
+### 6.1 Correctness
+
+Проверка корректности выполнена через Google Test на 36 тестовых конфигурациях:
+- Типы данных: `MPI_INT`, `MPI_FLOAT`, `MPI_DOUBLE`
+- Размеры: 9, 10, 50 элементов
+- Получатели: ранги 0 и 1
+- Операция: `MPI_SUM`, `MPI_MIN`
+
+### 6.2 Performance
+| Mode | Count | Time, s | Speedup | Efficiency |
+|------|-------|---------|---------|------------|
+| seq  | 2     | 0.388   | 1.00    | N/A        |
+| mpi  | 2     | 0.847   | 0.46    | 45.80%     |
+| seq  | 3     | 0.538   | 1.00    | N/A        |
+| mpi  | 3     | 0.760   | 0.71    | 70.77%     |
+| seq  | 4     | 1.001   | 1.00    | N/A        |
+| mpi  | 4     | 1.096   | 0.91    | 91.32%     |
+
+## 7. Conclusions
+
+В MPI-версии используется пользовательская реализация операции Reduce на основе бинарного дерева с явными вызовами MPI_Send и MPI_Recv и последующим использованием MPI_Bcast.
+
+В последовательной версии применяется оптимизированная реализация MPI_Reduce, предоставляемая MPI-библиотекой.
+
+Из-за накладных расходов на коммуникации и синхронизацию, а также отсутствия низкоуровневых оптимизаций, пользовательская MPI-реализация уступает встроенной MPI_Reduce по времени выполнения, особенно при малом числе процессов.
+
+Полученные результаты соответствуют теоретическим ожиданиям. Накладные расходы на управление процессами и передачу данных занимают значительную часть времени выполнения, что снижает эффективность параллельной версии.
+
+## 8. References
+
+1. Курс лекций ННГУ «Параллельное программирование для кластерных систем»

tasks/zavyalov_a_reduce/seq/include/ops_seq.hpp

@@ -0,0 +1,22 @@
+#pragma once
+
+#include "task/include/task.hpp"
+#include "zavyalov_a_reduce/common/include/common.hpp"
+
+namespace zavyalov_a_reduce {
+
+class ZavyalovAReduceSEQ : public BaseTask {
+ public:
+  static constexpr ppc::task::TypeOfTask GetStaticTypeOfTask() {
+    return ppc::task::TypeOfTask::kSEQ;
+  }
+  explicit ZavyalovAReduceSEQ(const InType &in);
+
+ private:
+  bool ValidationImpl() override;
+  bool PreProcessingImpl() override;
+  bool RunImpl() override;
+  bool PostProcessingImpl() override;
+};
+
+}  // namespace zavyalov_a_reduce