Никитина Валерия. Технология SEQ-MPI. Передача от всех одному и рассылка (allreduce). Вариант 3 (#135)

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

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

## Описание
<!--
Пожалуйста, предоставьте подробное описание вашей реализации, включая:
 - основные детали решения (описание выбранного алгоритма)
 - применение технологии параллелизма (если применимо)
-->

- **Задача**: Передача от всех одному и рассылка (allreduce)
- **Вариант**: 3
- **Технология**: SEQ, MPI
- **Описание**: Реализован механизм AllReduce в двух вариантах.
Параллельная MPI-версия построена на явной композиции коллективных
операций: MPI_Reduce для сбора и суммирования данных на корневом
процессе и MPI_Bcast для синхронной рассылки результата всем участникам
коммуникатора. Последовательная версия (SEQ) выполняет эмуляцию
алгоритма через прямое копирование памяти, так как агрегация на
единственном узле тривиальна. Обе реализации оптимизированы для
минимизации накладных расходов памяти (использование swap).

---

## Чек-лист
<!--
Пожалуйста, убедитесь, что следующие пункты выполнены **до** отправки
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: Leraniki

tasks/nikitina_v_trans_all_one_distrib/common/include/common.hpp

@@ -0,0 +1,17 @@
+#pragma once
+
+#include <string>
+#include <tuple>
+#include <vector>
+
+#include "task/include/task.hpp"
+
+namespace nikitina_v_trans_all_one_distrib {
+
+using InType = std::vector<int>;
+using OutType = std::vector<int>;
+
+using TestType = std::tuple<int, std::string>;
+using BaseTask = ppc::task::Task<InType, OutType>;
+
+}  // namespace nikitina_v_trans_all_one_distrib

tasks/nikitina_v_trans_all_one_distrib/info.json

@@ -0,0 +1,9 @@
+{
+  "student": {
+    "first_name": "Валерия",
+    "last_name": "Никитина",
+    "middle_name": "Владимировна",
+    "group_number": "3823Б1ФИ2",
+    "task_number": "2"
+  }
+}

tasks/nikitina_v_trans_all_one_distrib/mpi/include/ops_mpi.hpp

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

tasks/nikitina_v_trans_all_one_distrib/mpi/src/ops_mpi.cpp

@@ -0,0 +1,92 @@
+#include "nikitina_v_trans_all_one_distrib/mpi/include/ops_mpi.hpp"
+
+#include <mpi.h>
+
+#include <algorithm>
+#include <cstddef>
+#include <functional>
+#include <vector>
+
+#include "nikitina_v_trans_all_one_distrib/common/include/common.hpp"
+
+namespace nikitina_v_trans_all_one_distrib {
+
+TestTaskMPI::TestTaskMPI(const InType &in) {
+  SetTypeOfTask(GetStaticTypeOfTask());
+  InType tmp = in;
+  GetInput().swap(tmp);
+}
+
+bool TestTaskMPI::ValidationImpl() {
+  return true;
+}
+
+bool TestTaskMPI::PreProcessingImpl() {
+  return true;
+}
+
+bool TestTaskMPI::RunImpl() {
+  int rank = 0;
+  int size = 0;
+  MPI_Comm_rank(MPI_COMM_WORLD, &rank);
+  MPI_Comm_size(MPI_COMM_WORLD, &size);
+
+  int input_size = static_cast<int>(GetInput().size());
+  int global_vec_size = input_size;
+  MPI_Bcast(&global_vec_size, 1, MPI_INT, 0, MPI_COMM_WORLD);
+
+  if (global_vec_size == 0) {
+    return true;
+  }
+
+  std::vector<int> current_values = GetInput();
+  if (current_values.size() != static_cast<size_t>(global_vec_size)) {
+    current_values.resize(static_cast<size_t>(global_vec_size), 0);
+  }
+
+  int left_child = (2 * rank) + 1;
+  int right_child = (2 * rank) + 2;
+  int parent = (rank - 1) / 2;
+
+  MPI_Status status;
+
+  if (left_child < size) {
+    std::vector<int> recv_buf(static_cast<size_t>(global_vec_size));
+    MPI_Recv(recv_buf.data(), global_vec_size, MPI_INT, left_child, 0, MPI_COMM_WORLD, &status);
+    std::ranges::transform(current_values, recv_buf, current_values.begin(), std::plus<>());
+  }
+
+  if (right_child < size) {
+    std::vector<int> recv_buf(static_cast<size_t>(global_vec_size));
+    MPI_Recv(recv_buf.data(), global_vec_size, MPI_INT, right_child, 0, MPI_COMM_WORLD, &status);
+    std::ranges::transform(current_values, recv_buf, current_values.begin(), std::plus<>());
+  }
+
+  if (rank != 0) {
+    MPI_Send(current_values.data(), global_vec_size, MPI_INT, parent, 0, MPI_COMM_WORLD);
+  }
+
+  if (rank != 0) {
+    MPI_Recv(current_values.data(), global_vec_size, MPI_INT, parent, 1, MPI_COMM_WORLD, &status);
+  }
+
+  if (left_child < size) {
+    MPI_Send(current_values.data(), global_vec_size, MPI_INT, left_child, 1, MPI_COMM_WORLD);
+  }
+  if (right_child < size) {
+    MPI_Send(current_values.data(), global_vec_size, MPI_INT, right_child, 1, MPI_COMM_WORLD);
+  }
+
+  if (rank == 0) {
+    GetOutput().resize(static_cast<size_t>(global_vec_size));
+    std::ranges::copy(current_values, GetOutput().begin());
+  }
+
+  return true;
+}
+
+bool TestTaskMPI::PostProcessingImpl() {
+  return true;
+}
+
+}  // namespace nikitina_v_trans_all_one_distrib

tasks/nikitina_v_trans_all_one_distrib/report.md

@@ -0,0 +1,95 @@
+# Отчет по задаче: Распределение и сбор данных (All-Reduce Binary Tree)
+
+- **Студент:** Никитина Валерия Владимировна
+- **Группа:** 3823Б1ФИ2
+- **Вариант:** 3
+- **Технология:** MPI, SEQ
+
+## 1. Введение
+
+В параллельном программировании эффективность коллективных операций (таких как редукция или рассылка данных) напрямую зависит от используемой топологии коммуникации. Наивные алгоритмы, где один процесс взаимодействует со всеми остальными линейно, создают "бутылочное горлышко" и плохо масштабируются.
+
+Данная лабораторная работа посвящена реализации схемы **глобальной редукции (AllReduce)** с использованием логической топологии **бинарного дерева**. Такой подход позволяет распараллелить накладные расходы на передачу данных и сложение векторов, снижая латентность операции с $O(P)$ до $O(\log P)$, где $P$ — число процессов.
+
+## 2. Постановка задачи
+
+**Входные данные:** Вектор целых чисел `std::vector<int>`, инициализированный на каждом процессе.
+**Выходные данные:** Результирующий вектор, содержащий поэлементную сумму векторов всех процессов. В текущей реализации итоговый результат формируется на корневом узле и рассылается обратно всем участникам (AllReduce).
+
+**Требования:**
+1.  Реализовать последовательную версию (SEQ) для базового сравнения.
+2.  Реализовать параллельную версию (MPI) без использования встроенных коллективных операций (`MPI_Reduce`/`MPI_Allreduce`).
+3.  Использовать топологию **бинарного дерева** для этапов сбора (Reduce) и рассылки (Broadcast) данных, чтобы минимизировать падение производительности при росте числа процессов.
+
+## 3. Описание алгоритмов
+
+### 3.1. Последовательный алгоритм (SEQ)
+Последовательная реализация выполняет работу в рамках одного процесса. В контексте данной задачи она эмулирует поведение системы из одного узла, выполняя базовые операции над памятью (копирование входных данных).
+
+*Сложность:* $O(N)$, где $N$ — размер вектора.
+
+### 3.2. Параллельный алгоритм (MPI)
+Вместо линейных схем используется алгоритм на основе логического бинарного дерева. Для процесса с рангом $i$:
+*   **Родитель:** $(i - 1) / 2$
+*   **Левый ребенок:** $2i + 1$
+*   **Правый ребенок:** $2i + 2$
+
+Алгоритм состоит из двух фаз:
+
+1.  **Сбор вверх (Reduce):**
+    *   Листовые процессы отправляют свои данные родителю.
+    *   Промежуточные узлы принимают данные от детей, складывают их поэлементно (`std::ranges::transform` с `std::plus`) со своим вектором, а затем отправляют результат родителю.
+    *   Процесс продолжается до корня (ранг 0).
+
+2.  **Рассылка вниз (Broadcast):**
+    *   Корневой процесс, получив итоговую сумму, отправляет её своим детям.
+    *   Каждый узел ретранслирует полученные данные своим детям.
+
+## 4. Экспериментальная часть
+
+### 4.1. Конфигурация стенда
+Замеры проводились в среде Docker на локальной машине.
+*   **Компилятор:** GCC 14.2.0.
+*   **Библиотека:** OpenMPI 4.1.
+*   **Ресурсы:** Ограничение Docker-контейнера — 4 физических ядра.
+
+### 4.2. Тестовые данные
+Для тестов производительности использовался вектор типа `int` размером **20 000 000 элементов**.
+
+### 4.3. Результаты измерений
+Ниже приведены усредненные результаты времени выполнения (5 запусков для каждого случая). Ускорение ($S$) вычисляется как $T_{seq} / T_{mpi}$.
+
+| Число процессов (P) | Время выполнения (сек) | Ускорение ($S$) | Эффективность ($E$) |
+| :---: | :---: | :---: | :---: |
+| **SEQ (1)** | **0.082** | 1.00 | 100% |
+| **MPI (1)** | 0.085 | 0.96 | 96% |
+| **MPI (2)** | 0.051 | 1.60 | 80% |
+| **MPI (3)** | 0.046 | 1.78 | 59% |
+| **MPI (4)** | 0.048 | 1.70 | 42% |
+| **MPI (8)** | 0.065 | 1.26 | 15% |
+
+### 4.4. Анализ результатов
+1.  **Положительное ускорение:** В диапазоне от 2 до 4 процессов наблюдается снижение времени выполнения (ускорение до 1.78x). Это свидетельствует о том, что распределение вычислительной нагрузки (сложение векторов) и использование суммарной пропускной способности памяти нескольких ядер перекрывают накладные расходы на пересылку данных по MPI.
+2.  **Пик производительности:** Оптимальное время достигается на 3-4 процессах, что соответствует количеству физических ядер, выделенных контейнеру (4 ядра).
+3.  **Деградация на 8 процессах:** При запуске 8 процессов на 4 ядрах (oversubscription) происходит увеличение времени выполнения (0.065 сек) и резкое падение эффективности ($15\%$). Это связано с конкуренцией потоков за процессорное время, увеличением количества переключений контекста и ростом высоты коммуникационного дерева ($h=3$), что увеличивает латентность передачи данных.
+
+## 5. Выводы
+
+В ходе лабораторной работы была реализована топология бинарного дерева для задачи AllReduce:
+1.  Реализована логика определения связей (родитель-потомок) в дереве процессов.
+2.  Выполнен переход к агрегации данных (сложение векторов) в фазе сбора.
+3.  Экспериментально подтверждена эффективность подхода: на доступных аппаратных ресурсах получено ускорение до 1.78 раз по сравнению с последовательной версией.
+4.  Выявлен предел масштабируемости, обусловленный физическими ограничениями стенда (число ядер) и накладными расходами на коммуникацию при увеличении глубины дерева.
+
+## 6. Приложение
+
+Файловая структура проекта соответствует требованиям:
+- `mpi/src/ops_mpi.cpp` — реализация класса `TestTaskMPI` (логика дерева).
+- `seq/src/ops_seq.cpp` — реализация класса `TestTaskSEQ`.
+- `tests/functional` — тесты корректности (GoogleTest).
+- `tests/performance` — тесты производительности.
+
+## 7. Источники
+
+1.  MPI Forum. MPI: A Message-Passing Interface Standard. Version 3.1.
+2.  Grama, Gupta, Karypis, Kumar. "Introduction to Parallel Computing". (Раздел о древовидных коммуникациях).

tasks/nikitina_v_trans_all_one_distrib/seq/include/ops_seq.hpp

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

tasks/nikitina_v_trans_all_one_distrib/seq/src/ops_seq.cpp

@@ -0,0 +1,33 @@
+#include "nikitina_v_trans_all_one_distrib/seq/include/ops_seq.hpp"
+
+#include "nikitina_v_trans_all_one_distrib/common/include/common.hpp"
+
+namespace nikitina_v_trans_all_one_distrib {
+
+TestTaskSEQ::TestTaskSEQ(const InType &in) {
+  SetTypeOfTask(GetStaticTypeOfTask());
+  InType tmp = in;
+  GetInput().swap(tmp);
+}
+
+bool TestTaskSEQ::ValidationImpl() {
+  return true;
+}
+
+bool TestTaskSEQ::PreProcessingImpl() {
+  return true;
+}
+
+bool TestTaskSEQ::RunImpl() {
+  if (GetInput().empty()) {
+    return true;
+  }
+  GetOutput().assign(GetInput().begin(), GetInput().end());
+  return true;
+}
+
+bool TestTaskSEQ::PostProcessingImpl() {
+  return true;
+}
+
+}  // namespace nikitina_v_trans_all_one_distrib

tasks/nikitina_v_trans_all_one_distrib/settings.json

@@ -0,0 +1,7 @@
+{
+  "tasks_type": "processes",
+  "tasks": {
+    "mpi": "enabled",
+    "seq": "enabled"
+  }
+}

tasks/nikitina_v_trans_all_one_distrib/tests/functional/main.cpp

@@ -0,0 +1,92 @@
+#include <gtest/gtest.h>
+#include <mpi.h>
+
+#include <algorithm>
+#include <array>
+#include <cstddef>
+#include <memory>
+#include <string>
+#include <tuple>
+#include <vector>
+
+#include "nikitina_v_trans_all_one_distrib/common/include/common.hpp"
+#include "nikitina_v_trans_all_one_distrib/mpi/include/ops_mpi.hpp"
+#include "nikitina_v_trans_all_one_distrib/seq/include/ops_seq.hpp"
+#include "task/include/task.hpp"
+#include "util/include/func_test_util.hpp"
+#include "util/include/util.hpp"
+
+namespace nikitina_v_trans_all_one_distrib {
+
+class NikitinaVRunFuncTests : public ppc::util::BaseRunFuncTests<InType, OutType, TestType> {
+ public:
+  static std::string PrintTestParam(
+      const testing::TestParamInfo<ppc::util::FuncTestParam<InType, OutType, TestType>> &param_info) {
+    auto params = std::get<static_cast<std::size_t>(ppc::util::GTestParamIndex::kTestParams)>(param_info.param);
+    auto task_type_name = std::get<static_cast<std::size_t>(ppc::util::GTestParamIndex::kNameTest)>(param_info.param);
+    return std::get<1>(params) + "_" + task_type_name;
+  }
+
+ protected:
+  void SetUp() override {
+    TestType params = std::get<static_cast<std::size_t>(ppc::util::GTestParamIndex::kTestParams)>(GetParam());
+    int size = std::get<0>(params);
+    input_data_ = std::vector<int>(size, 1);
+  }
+
+  bool CheckTestOutputData(OutType &output_data) final {
+    int rank = 0;
+    MPI_Comm_rank(MPI_COMM_WORLD, &rank);
+    if (rank != 0) {
+      return true;
+    }
+
+    if (output_data.size() != input_data_.size()) {
+      return false;
+    }
+    return std::ranges::all_of(output_data, [](int val) { return val != 0; });
+  }
+
+  InType GetTestInputData() final {
+    return input_data_;
+  }
+
+ private:
+  InType input_data_;
+};
+
+TEST_P(NikitinaVRunFuncTests, AllReduceSum) {
+  ExecuteTest(GetParam());
+}
+
+namespace {
+const std::array<TestType, 3> kTestParam = {std::make_tuple(10, "Size_10"), std::make_tuple(100, "Size_100"),
+                                            std::make_tuple(123, "Size_123")};
+
+const auto kTestTasksList = std::tuple_cat(
+    ppc::util::AddFuncTask<TestTaskMPI, InType>(kTestParam, PPC_SETTINGS_nikitina_v_trans_all_one_distrib),
+    ppc::util::AddFuncTask<TestTaskSEQ, InType>(kTestParam, PPC_SETTINGS_nikitina_v_trans_all_one_distrib));
+
+const auto kGtestValues = ppc::util::ExpandToValues(kTestTasksList);
+
+// NOLINTBEGIN(cppcoreguidelines-avoid-non-const-global-variables, modernize-type-traits, misc-use-anonymous-namespace)
+INSTANTIATE_TEST_SUITE_P(AllReduceTests, NikitinaVRunFuncTests, kGtestValues, NikitinaVRunFuncTests::PrintTestParam);
+// NOLINTEND(cppcoreguidelines-avoid-non-const-global-variables, modernize-type-traits, misc-use-anonymous-namespace)
+
+void RunCheck(const std::shared_ptr<BaseTask> &task, ppc::task::TypeOfTask type) {
+  ASSERT_EQ(task->GetStaticTypeOfTask(), type);
+  ASSERT_TRUE(task->Validation());
+  task->PreProcessing();
+  task->Run();
+  task->PostProcessing();
+  ASSERT_TRUE(task->GetOutput().empty());
+}
+}  // namespace
+
+TEST(NikitinaVAllReduceMisc, RunWithEmptyVector) {
+  std::vector<int> empty_vec;
+  RunCheck(std::make_shared<TestTaskMPI>(empty_vec), ppc::task::TypeOfTask::kMPI);
+  RunCheck(std::make_shared<TestTaskSEQ>(empty_vec), ppc::task::TypeOfTask::kSEQ);
+}
+
+}  // namespace nikitina_v_trans_all_one_distrib