MPI代码中Rank自发送/接收的竞态条件问题排查

架构老林

2026-6-12

MPI主从模式多线程实现中的任务重复分配问题

问题现象

运行代码时主rank会执行重复任务，同一任务被多次分配，问题随机出现。示例如下：

预期输出：

1-(1,1),(0,1)
2-(2,2),(0,2)
3-(3,3),(0,3)
3-(0,3),(3,4)
4-(4,4),(0,4)
5-(5,5),(1,2)
6-(6,6),(1,3)
7-(7,7),(1,4)
0-(0,0),(2,3)
1-(0,1),(2,4)

实际输出：

0-(0,0),(0,0)
1-(1,1),(0,1)
2-(2,2),(0,2)
3-(3,3),(0,3)
3-(0,3),(3,4)
4-(4,4),(0,4)
5-(5,5),(1,2)
6-(6,6),(1,3)
7-(7,7),(1,4)
0-(0,0),(2,3)
1-(0,1),(2,4)
0-(2,3),(2,3)

实现代码

#include <array>
#include <atomic>
#include <cstdlib>
#include <ctime>
#include <iostream>
#include <mutex>
#include <thread>
#include <vector>

#include <mpi.h>

const int TAG_REQUEST = 0;
const int TAG_TASK = 1;
const int TAG_STOP = 2;

const int master_rank = 0;

std::mutex task_mutex;
std::atomic<bool> done(false);

using Task = std::array<int, 2>;
using TaskList = std::vector<Task>;

TaskList generate_tasks(int N) {
  TaskList tasks;
  for (int i = 0; i < N; i++) {
    for (int j = i + 1; j < N; j++) {
      tasks.emplace_back(Task{i, j});
    }
  }
  return tasks;
}

int score2(int i) {
  return i;
}

int score1(const Task &task) {
  return task[0] * task[1];
}

int do_work(const Task &task, int rank) {
  int result;
  if (task[0] == task[1]) {
    result = score2(task[0]);
  } else {
    result = score1(task) + rank;
  }

  return result;
}

void master_comm_thread(int N) {
  int tasks_finished = 0, ranks_finished = 0;
  int size;
  MPI_Comm_size(MPI_COMM_WORLD, &size);
  std::srand(static_cast<unsigned>(std::time(nullptr)));
  TaskList tasks = generate_tasks(N);
  const int tasks_size = tasks.size();
  Task task = {-1, -1};
  while (true) {
    MPI_Status status;
    MPI_Probe(MPI_ANY_SOURCE, TAG_REQUEST, MPI_COMM_WORLD, &status);

    MPI_Recv(task.data(), 2, MPI_INT, status.MPI_SOURCE, TAG_REQUEST,
             MPI_COMM_WORLD, MPI_STATUS_IGNORE);
    bool have_task = !tasks.empty();
    {
      std::lock_guard<std::mutex> lock(task_mutex);
      if (!tasks.empty()) {
        // int idx = std::rand() % tasks.size();
        int idx = 0;
        task = tasks[idx];
        tasks.erase(tasks.begin() + idx);
        have_task = true;
        tasks_finished++;
      }
    }

    if (have_task) {
      MPI_Send(task.data(), 2, MPI_INT, status.MPI_SOURCE, TAG_TASK,
               MPI_COMM_WORLD);
    } else {
      Task finish{-1, -1};
      MPI_Send(&finish, 2, MPI_INT, status.MPI_SOURCE, TAG_STOP,
               MPI_COMM_WORLD);
      ranks_finished++;
      if (tasks_finished >= tasks_size) {
        done = true;
        break;
      }
    }
  }
  while (ranks_finished < size) {
    MPI_Status status;
    MPI_Probe(MPI_ANY_SOURCE, TAG_REQUEST, MPI_COMM_WORLD, &status);
    MPI_Recv(task.data(), 2, MPI_INT, status.MPI_SOURCE, TAG_REQUEST,
             MPI_COMM_WORLD, MPI_STATUS_IGNORE);

    MPI_Send(task.data(), 2, MPI_INT, status.MPI_SOURCE, TAG_STOP,
             MPI_COMM_WORLD);
    ranks_finished++;
  }
}

void data_loop(int rank, bool &active, Task &result) {
  Task task = {-1, -1};
  MPI_Status status;

  // Post non-blocking send and receive for the next task
  MPI_Send(result.data(), 2, MPI_INT, master_rank, TAG_REQUEST, MPI_COMM_WORLD);
  MPI_Recv(task.data(), 2, MPI_INT, master_rank, MPI_ANY_TAG, MPI_COMM_WORLD, &status);
  if (status.MPI_TAG == TAG_STOP) {
    result = {-1, -1};
    active = false;
    return;
  }

  std::cout<<rank<<"-"<<"("<<result[0]<<","<<result[1]<<"), ("<<task[0]<<","<<task[1]<<")"<<std::endl;

  result = task;
  active = true;
  return;
}

void worker_loop_async() {
  int rank;
  MPI_Comm_rank(MPI_COMM_WORLD, &rank);

  bool active = true;
  // First task is pair-wise for my local data
  Task task_current = {-1, -1};
  Task task_next = {rank, rank};
  while (active) {
    task_current = task_next;
    std::thread data_thread(data_loop, rank, std::ref(active),
                            std::ref(task_next));
    int result = do_work(task_current, rank);
    // Do something with result
    data_thread.join();
  }
}

int main(int argc, char *argv[]) {
  int provided;
  MPI_Init_thread(&argc, &argv, MPI_THREAD_MULTIPLE, &provided);
  if (provided < MPI_THREAD_MULTIPLE) {
    std::cerr << "MPI threading not sufficient: " << provided << std::endl;
    MPI_Abort(MPI_COMM_WORLD, 1);
  }

  int rank;
  MPI_Comm_rank(MPI_COMM_WORLD, &rank);

  int N = 5;
  std::thread comm_thread;
  if (rank == master_rank) {
    comm_thread = std::thread(master_comm_thread, N);
  }

  worker_loop_async();

  if (rank == master_rank) {
    comm_thread.join();
  }

  MPI_Finalize();
  return 0;
}

问题根源分析

请求接收与任务变量复用冲突：
在master_comm_thread中，接收请求时直接复用了task变量，导致该变量被worker发送的请求数据覆盖。当任务队列空了之后，终止阶段的代码错误地将这个被覆盖的旧数据发送给worker，而非标准的终止标记{-1,-1}，这就造成worker（包括主rank自己）收到重复的旧任务。
终止阶段逻辑错误：
第一个循环退出后，处理剩余请求的循环中，代码直接把接收的请求数据作为终止信号发送，完全违背了终止逻辑，应该始终发送{-1,-1}作为终止标记。
任务计数逻辑混淆：
tasks_finished实际统计的是已分配任务数，而非已完成任务数。当任务全部分配后主rank就退出第一个循环，但此时worker可能仍在处理任务，后续请求会被错误返回旧数据。

修复方案

隔离请求与任务变量：使用单独的变量接收请求数据，避免覆盖要发送的任务或终止标记。
修正终止信号发送逻辑：无论何时发送终止信号，都使用{-1,-1}，而非复用请求数据。
调整任务计数逻辑：可考虑新增任务完成反馈机制，确保所有任务处理完毕后再终止主rank的通信线程。

示例修复代码片段：

// 修改master_comm_thread的第一个循环，隔离请求与任务变量
void master_comm_thread(int N) {
  int tasks_finished = 0, ranks_finished = 0;
  int size;
  MPI_Comm_size(MPI_COMM_WORLD, &size);
  std::srand(static_cast<unsigned>(std::time(nullptr)));
  TaskList tasks = generate_tasks(N);
  const int tasks_size = tasks.size();
  Task req_task = {-1, -1};
  Task send_task = {-1, -1};
  
  while (true) {
    MPI_Status status;
    MPI_Probe(MPI_ANY_SOURCE, TAG_REQUEST, MPI_COMM_WORLD, &status);

    MPI_Recv(req_task.data(), 2, MPI_INT, status.MPI_SOURCE, TAG_REQUEST,
             MPI_COMM_WORLD, MPI_STATUS_IGNORE);
    bool have_task = !tasks.empty();
    {
      std::lock_guard<std::mutex> lock(task_mutex);
      if (!tasks.empty()) {
        int idx = 0;
        send_task = tasks[idx];
        tasks.erase(tasks.begin() + idx);
        have_task = true;
        tasks_finished++;
      }
    }

    if (have_task) {
      MPI_Send(send_task.data(), 2, MPI_INT, status.MPI_SOURCE, TAG_TASK,
               MPI_COMM_WORLD);
    } else {
      Task finish{-1, -1};
      MPI_Send(finish.data(), 2, MPI_INT, status.MPI_SOURCE, TAG_STOP,
               MPI_COMM_WORLD);
      ranks_finished++;
      if (tasks_finished >= tasks_size) {
        done = true;
        break;
      }
    }
  }
  
  // 修改终止阶段逻辑，固定发送终止标记
  while (ranks_finished < size) {
    MPI_Status status;
    MPI_Probe(MPI_ANY_SOURCE, TAG_REQUEST, MPI_COMM_WORLD, &status);
    MPI_Recv(req_task.data(), 2, MPI_INT, status.MPI_SOURCE, TAG_REQUEST,
             MPI_COMM_WORLD, MPI_STATUS_IGNORE);

    Task finish{-1, -1};
    MPI_Send(finish.data(), 2, MPI_INT, status.MPI_SOURCE, TAG_STOP,
             MPI_COMM_WORLD);
    ranks_finished++;
  }
}

内容的提问来源于stack exchange，提问作者Niteya Shah

MPI代码中Rank自发送/接收的竞态条件问题排查

问题现象 #

实现代码 #

问题根源分析 #

修复方案 #

问题现象

实现代码

问题根源分析

修复方案