主辅助进程动态负载平衡中的列组无效

我有一个master worker动态负载平衡,它在我的计算机上运行得很好,但是当我试图在超级计算机上运行我的程序时,我在这个块中从master发送作业时遇到了无效的秩错误:

  while (mesg[0] < nindices) {
    MPI_Probe(MPI_ANY_SOURCE, 1, comm, &status);
    int isource = status.MPI_SOURCE;
    MPI_Recv(recv_mesg, 1, MPI_INT, isource, 1, comm, &status);
    MPI_Send(mesg, 1, MPI_INT, isource, 1, comm);
    mesg[0] += 1;
  }

如果工作的数量是2个或3个,我有一个工人和一个师傅,它的工作绝对好!(即使在超级计算机中)但当我将作业数增加到4或更高时,我将得到无效的秩错误

示例代码是:

#include "mpi.h"
#include <stdio.h>
#include <stdlib.h>
#include <iostream>
#include <array>
#include <cassert>
#include <chrono>
#include <cmath>
using namespace std;

#define MAX_SIZE 1

void function_test(MPI_Comm comm) {
  int mpi_world_size;
MPI_Comm_size(comm, &mpi_world_size);
int mpi_rank;
MPI_Comm_rank(comm, &mpi_rank);

  std::cout<<"hi"<<std::endl;
  int nindices =3;

  if (nindices > 0) {
    if(mpi_rank == 0) {
      MPI_Status status;

      int mesg[1]={0};
      int initial_id = 0;
      int recv_mesg[1]={0};

      //-- send out initial ids to workers --//
      while (initial_id < mpi_world_size - 1) {
        if (initial_id < nindices) {
          MPI_Send(mesg, 1, MPI_INT, initial_id + 1, 1, comm);
          mesg[0] += 1;
          ++initial_id;
        }
      }

      //--  id to workers dynamically --//
      while (mesg[0] < nindices) {
        MPI_Probe(MPI_ANY_SOURCE, 1, comm, &status);
        int isource = status.MPI_SOURCE;
        MPI_Recv(recv_mesg, 1, MPI_INT, isource, 1, comm, &status);
        MPI_Send(mesg, 1, MPI_INT, isource, 1, comm);
        mesg[0] += 1;
      }

      //-- ending signals--//
      for (int rank = 1; rank < mpi_world_size; ++rank) {
        mesg[0] = -1;
        MPI_Send(mesg, 1, MPI_INT, rank, 0, comm);
      }
    } else { // worker
      MPI_Status status;
      int id[1]={0};
      while(true) {
        
        MPI_Probe(MPI_ANY_SOURCE, MPI_ANY_TAG, comm, &status);
        int itag = status.MPI_TAG;
        MPI_Recv(id, 1, MPI_INT, 0, itag, comm, &status);

        int jindex = id[0];
        if (jindex < 0) break;
        MPI_Send(id, 1, MPI_INT, 0, 1, comm);
      }
    }
  }

}

int main(int argc, char *argv[])
{
  // MPI Start
  MPI_Init(NULL, NULL);
  int world_rank, world_size;
  MPI_Comm world_comm = MPI_COMM_WORLD;

  int color_rank_count = 3;
  // ****************************************************************************** Geting the size and rank of the world_comm
  MPI_Comm_rank(world_comm, &world_rank);
  MPI_Comm_size(world_comm, &world_size);

  // ****************************************************************************** Creat two commiunicators
  MPI_Comm c_comm;
  MPI_Comm c_comm_0;

  int color_integral =
      (world_rank == 0)
          ? 0
          : 1 + static_cast<int>(floor((world_rank - 1) / color_rank_count));
  int key_integral =
      (world_rank == 0) ? 0 : (world_rank - 1) % color_rank_count;
  int color_zeros =
      (world_rank == 0) ? 0 : (world_rank - 1) % color_rank_count;
  int key_zeros =
      (world_rank == 0)
          ? 0
          : 1 + static_cast<int>(floor((world_rank - 1) / color_rank_count));


  MPI_Comm_split(world_comm, color_integral, key_integral,
                 &c_comm);
  MPI_Comm_split(world_comm, color_zeros, key_zeros, &c_comm_0);

  int c_size, c_0_size;
  MPI_Comm_size(c_comm, &c_size);
  MPI_Comm_size(c_comm_0, &c_0_size);

  int c_rank, c_0_rank;
  MPI_Comm_rank(c_comm, &c_rank);
  MPI_Comm_rank(c_comm_0, &c_0_rank);


  if (world_rank == 0) { // controller rank
        int mesg[1] = {0};
            int nindices = 3;
           int initial_task = 0;
           int nProc = c_0_size;
           bool recv_mesg[1]={true};
           //-- send out initial tasks to slaves --//
           while (initial_task < c_0_size - 1) {
             if (initial_task < nindices) {
               MPI_Send(mesg, 1, MPI_INT, initial_task + 1, 1, c_comm_0);
               mesg[0] += 1;
               ++initial_task;
             }
           }
             int proc = initial_task;
             //-- hand out quartets to slaves dynamically --//
             while (mesg[0] != nindices) {
               if(proc >= nProc) {
                 proc = 1;
               }
               MPI_Send(mesg, 1, MPI_INT, proc, 1, c_comm_0);
               ++proc;
               mesg[0] += 1;

             }
        //-- hand out ending signals once done --//
        for (int rank = 1; rank < c_0_size; ++rank) {
          mesg[0] = -1;
          MPI_Send(mesg, 1, MPI_INT, rank, 0, c_comm_0);
        }

      } else { // worker ranks
      MPI_Status status;
      int task[1]={0};

      while (true) {
        if (color_zeros == 0) {
          MPI_Probe(MPI_ANY_SOURCE, MPI_ANY_TAG, c_comm_0, &status);
          int isource = status.MPI_SOURCE;
          int itag = status.MPI_TAG;
          MPI_Recv(task, 1, MPI_INT, isource, itag, c_comm_0, &status);
        }
        MPI_Barrier(c_comm);
        MPI_Bcast(task, 1, MPI_INT, 0, c_comm);

        int ifrag = task[0];

        if (ifrag < 0)
          break;
          function_test(c_comm);

         }
       }

  /* All done */
  MPI_Finalize();
  return 0;

}

另外,在代码中,为了达到我的目的,我不得不将等级分开,这是我必须使用的MPI的简单框架。