在 MPI 中发送和接收动态数组答案

【问题标题】：Send and receive dynamic arrays in MPI在 MPI 中发送和接收动态数组
【发布时间】：2014-09-17 00:10:51
【问题描述】：

我在使用 MPI 发送和（或）接收动态数组时遇到了一个大问题。下面只是我的代码的一部分，但应该足以看出我做错了什么。请帮助我，我经过一夜的搜索解决方案。我是这样的： * 检测到 glibc ./mv2.out: munmap_chunk(): 无效指针：0x0000000000da2a70 检测到 glibc ./mv2.out: malloc(): 内存损坏（快速）：0x0000000000da2a50 *

当我用静态替换动态数组时，一切正常。

#include "mpi.h"
#include <stdio.h>
#include <stdlib.h>

#define MASTER 0
#define FROM_MASTER 1
#define FROM_WORKER 2

double **alloc_2d_array(int rows, int cols) {
    int i;
    double *data = (double *)malloc(rows*cols*sizeof(double));
    double **array= (double **)malloc(rows*sizeof(double*));
    for (i=0; i<rows; i++)
        array[i] = &(data[cols*i]);

    return array;
}

int main (int argc, char *argv[])
{
    int degree,
    numtasks,
    taskid,
    numworkers,
    source,
    dest,
    mtype,
    rows,
    offset,
    averow,
    extra,
    i, j, k, rc;

    MPI_Status status;

    double **a, *b, *c;

    MPI_Init(&argc,&argv);
    MPI_Comm_rank(MPI_COMM_WORLD,&taskid);
    MPI_Comm_size(MPI_COMM_WORLD,&numtasks);
    if (numtasks < 2 )
    {
        printf("Aby rozpoczac obliczenia rownolegle potrzeba co najmniej 2 procesow.\n");
        MPI_Abort(MPI_COMM_WORLD, rc);
        exit(1);
    }
    numworkers = numtasks-1; 

    if (taskid == MASTER)
    {
        printf("Podaj stopien macierzy: \n");
        scanf ("%d", &degree);
        printf("Obecnie dostepnych jest %d procesow do dyspozycji mastera.\n", numtasks);

        FILE *file;

        file = fopen("matrix.txt", "r");
        if(file == NULL)
        {
            printf("Nie mozna otworzyc pliku!\n");
            MPI_Finalize();
            exit(0);
        }

        a = alloc_2d_array(degree, degree);
        b = (double*) malloc(sizeof(double) * degree);
        c = (double*) malloc(sizeof(double) * degree);

        printf("Tworzenie macierzy z pliku\n");
        for(i = 0; i < degree; i++)
            for(j = 0; j < degree; j++)
                fscanf(file, "%lf", &a[i][j]);

        for(i = 0; i < degree; i++)
        {
            for(j = 0; j < degree; j++)
            {
                printf("%f", a[i][j]);
            }
            printf("\n");
        }

        printf("Tworzenie wektora z pliku\n");
        for(i = 0; i < degree; i++)
            fscanf(file, "%lf", &b[i]);

        for(i = 0; i < degree; i++)
        {
            printf("%f\n", b[i]);
        }

        fclose(file);

        averow = degree / numworkers;
        extra = degree % numworkers;
        offset = 0;
        mtype = FROM_MASTER;
        for (dest = 1; dest <= numworkers; dest++)
        {
            rows = (dest <= extra) ? (averow + 1) : averow;
            printf("Wysylanie %d wierszy do procesu nr %d, z offset'em = %d\n", rows, dest, offset);
            MPI_Send(&degree, 1, MPI_INT, dest, mtype, MPI_COMM_WORLD);
            MPI_Send(&offset, 1, MPI_INT, dest, mtype, MPI_COMM_WORLD);
            MPI_Send(&rows, 1, MPI_INT, dest, mtype, MPI_COMM_WORLD);
            MPI_Send(&a[offset][0], rows * degree, MPI_DOUBLE, dest, mtype, MPI_COMM_WORLD);
            MPI_Send(&b, degree, MPI_DOUBLE, dest, mtype, MPI_COMM_WORLD);
            offset = offset + rows;
        }

        mtype = FROM_WORKER;
        for (i=1; i<=numworkers; i++)
        {
            source = i;
            MPI_Recv(&offset, 1, MPI_INT, source, mtype, MPI_COMM_WORLD, &status);
            MPI_Recv(&rows, 1, MPI_INT, source, mtype, MPI_COMM_WORLD, &status);
            MPI_Recv(&c[offset], rows, MPI_DOUBLE, source, mtype, MPI_COMM_WORLD, &status);
            printf("Otrzymalem wyniki od procesu nr %d\n", source);
        }

        printf("***\n");
        printf("Wektor wynikowy:\n");
        for (i = 0; i < degree; i++)
        {
            printf("\n%6.2f", c[i]);
        }
        printf("\n***\n");
        printf ("KONIEC\n");
    }


    if (taskid > MASTER)
    {
        b = (double*) malloc(sizeof(double) * degree);
        c = (double*) malloc(sizeof(double) * degree);

        mtype = FROM_MASTER;
        MPI_Recv(&degree, 1, MPI_INT, MASTER, mtype, MPI_COMM_WORLD, &status);
        MPI_Recv(&offset, 1, MPI_INT, MASTER, mtype, MPI_COMM_WORLD, &status);
        MPI_Recv(&rows, 1, MPI_INT, MASTER, mtype, MPI_COMM_WORLD, &status);
        MPI_Recv(&a, rows * degree, MPI_DOUBLE, MASTER, mtype, MPI_COMM_WORLD, &status); //HERE IS THE PROBLEM
        MPI_Recv(&b, degree, MPI_DOUBLE, MASTER, mtype, MPI_COMM_WORLD, &status);

        for (i = 0; i < rows; i++)
        {
            c[i] = 0.0;
            for (j = 0; j < degree; j++)
                c[i] += a[i][j] * b[j];
        }
        mtype = FROM_WORKER;
        MPI_Send(&offset, 1, MPI_INT, MASTER, mtype, MPI_COMM_WORLD);
        MPI_Send(&rows, 1, MPI_INT, MASTER, mtype, MPI_COMM_WORLD);
        MPI_Send(&c, rows, MPI_DOUBLE, MASTER, mtype, MPI_COMM_WORLD);
    }

    MPI_Finalize();
}

重现错误的最短版本：

#include "mpi.h"
#include <stdio.h>
#include <stdlib.h>

#define MASTER 0
#define FROM_MASTER 1
#define FROM_WORKER 2

double **alloc_2d_array(int rows, int cols) {
    int i;
    double *data = (double *)malloc(rows*cols*sizeof(double));
    double **array= (double **)malloc(rows*sizeof(double*));
    for (i=0; i<rows; i++)
        array[i] = &(data[cols*i]);

    return array;
}

int main (int argc, char *argv[])
{
    int degree,
    taskid,
    source,
    dest,
    mtype,
    rows,
    offset,
    i, j;

    MPI_Status status;

    double **a, *b, *c;

    MPI_Init(&argc,&argv);
    MPI_Comm_rank(MPI_COMM_WORLD,&taskid);
    MPI_Comm_size(MPI_COMM_WORLD,&numtasks);

    if (taskid == MASTER)
    {
        FILE *file;

        file = fopen("matrix.txt", "r");

        a = alloc_2d_array(degree, degree);
        b = (double*) malloc(sizeof(double) * degree);
        c = (double*) malloc(sizeof(double) * degree);

        for(i = 0; i < degree; i++)
            for(j = 0; j < degree; j++)
                fscanf(file, "%lf", &a[i][j]);      

        for(i = 0; i < degree; i++)
            fscanf(file, "%lf", &b[i]);

        fclose(file);

        offset = 0;
        mtype = FROM_MASTER;
        MPI_Send(&degree, 1, MPI_INT, dest, mtype, MPI_COMM_WORLD);
        MPI_Send(&offset, 1, MPI_INT, dest, mtype, MPI_COMM_WORLD);
        MPI_Send(&rows, 1, MPI_INT, dest, mtype, MPI_COMM_WORLD);
        MPI_Send(&a[offset][0], rows * degree, MPI_DOUBLE, dest, mtype, MPI_COMM_WORLD);
        MPI_Send(&b, degree, MPI_DOUBLE, dest, mtype, MPI_COMM_WORLD);
        }
    }


    if (taskid > MASTER)
    {
        a = alloc_2d_array(degree, degree);
        b = (double*) malloc(sizeof(double) * degree);
        c = (double*) malloc(sizeof(double) * degree);

        mtype = FROM_MASTER;
        MPI_Recv(&degree, 1, MPI_INT, MASTER, mtype, MPI_COMM_WORLD, &status);
        MPI_Recv(&offset, 1, MPI_INT, MASTER, mtype, MPI_COMM_WORLD, &status);
        MPI_Recv(&rows, 1, MPI_INT, MASTER, mtype, MPI_COMM_WORLD, &status);
        // v HERE IS THE PROBLEM v
        MPI_Recv(&a, rows * degree, MPI_DOUBLE, MASTER, mtype, MPI_COMM_WORLD, &status);
        MPI_Recv(&b, degree, MPI_DOUBLE, MASTER, mtype, MPI_COMM_WORLD, &status);
    }

    MPI_Finalize();
}

编辑：没有从工作人员发送到主机的版本可以正常工作，但是当我添加响应应用程序时会挂起。在下面的链接中，您可以找到静态数组的示例。当我用动态替换它们时，应用程序无法正常工作。 https://computing.llnl.gov/tutorials/mpi/samples/C/mpi_mm.c

【问题讨论】：

有什么问题？你能把你的代码减少到最低限度来重新创建它吗？
我添加了完整的源代码。当我用静态替换动态数组时，一切都很完美。
如果您可以编写一个最小的示例来重现您的问题，这将是一个更清晰的问题。例如，只需分配一个数组并发送/接收它。
为什么你认为这是问题所在？你得到什么输出，你期望什么输出？
然后使用-g 编译您的代码并进行一些调试。目前，任何人都可以帮助您的唯一方法是目视检查您的源代码，这几乎是最糟糕的调试方式。它似乎依赖一个名为matrix.txt 的文件才能运行，而我们没有。就像我之前说的，如果您将代码减少到可以重现问题的最少行数，对您和我们都会有所帮助。

标签： c arrays mpi dynamic-arrays

【解决方案1】：

这里必须有点通灵，因为你还没有发布一个可以运行和测试的完整程序（请在以后这样做，例如这里你没有包含所有必需的#includes 和#defines，并且你也不提供输入文件）。但是，据我所知，您在 recv 中的问题有两个方面。首先，您没有为“工作”进程分配任何内存。其次，您为 a 的 recv 指定的起始地址不正确。这是您的代码的精简版本，它可以满足您的需求（但我再次猜测）。另请注意，C 是我的第二语言，所以请检查我做了什么。无论如何，它是：

ian@ian-pc:~/test$ cat mpi.c

#include <stdio.h>
#include <stdlib.h>

#include "mpi.h"

#define FROM_MASTER 1
#define FROM_WORKER 2

#define MASTER 0

double **alloc_2d_array(int rows, int cols) {
  int i;
  double *data = (double *)malloc(rows*cols*sizeof(double));
  double **array= (double **)malloc(rows*sizeof(double*));
  for (i=0; i<rows; i++)
    array[i] = &(data[cols*i]);

  return array;
}

int main (int argc, char *argv[])
{

  int degree, numtasks, taskid, numworkers, dest, mtype, rows, offset, averow, extra, i, j;

  MPI_Status status;

  double **a;

  MPI_Init(&argc,&argv);
  MPI_Comm_rank(MPI_COMM_WORLD,&taskid);
  MPI_Comm_size(MPI_COMM_WORLD,&numtasks);
  if (numtasks < 2 )
    {
      printf("Aby rozpoczac obliczenia rownolegle potrzeba co najmniej 2 procesow.\n");
      MPI_Abort(MPI_COMM_WORLD, EXIT_FAILURE);
      exit(1);
    }
  numworkers = numtasks-1; 

  if (taskid == MASTER)
    {
      printf("Podaj stopien macierzy: \n");
      scanf ("%d", &degree);
      printf("Obecnie dostepnych jest %d procesow do dyspozycji mastera.\n", numtasks);

      a = alloc_2d_array(degree, degree);

      printf("Tworzenie macierzy z pliku\n");
      for(i = 0; i < degree; i++)
    for(j = 0; j < degree; j++)
      a[ i ][ j ] = i + 10 * j;

      printf( "Initial\n" );
      for(i = 0; i < degree; i++) {
    for(j = 0; j < degree; j++) 
      printf("%f ", a[i][j]);
    printf("\n");
      }

      averow = degree / numworkers;
      extra = degree % numworkers;
      offset = 0;
      mtype = FROM_MASTER;
      for (dest = 1; dest <= numworkers; dest++)
        {
      rows = (dest <= extra) ? (averow + 1) : averow;
      MPI_Send(&degree, 1, MPI_INT, dest, mtype, MPI_COMM_WORLD);
      MPI_Send(&rows, 1, MPI_INT, dest, mtype, MPI_COMM_WORLD);
      MPI_Send(&(a[offset][0]), rows * degree, MPI_DOUBLE, dest, mtype, MPI_COMM_WORLD);
      offset = offset + rows;
        }

    }


  /* Attempt to order output, not gauranteed to work */
  MPI_Barrier( MPI_COMM_WORLD );

  if (taskid > MASTER)
    {

      mtype = FROM_MASTER;
      MPI_Recv(&degree, 1, MPI_INT, MASTER, mtype, MPI_COMM_WORLD, &status);
      MPI_Recv(&rows, 1, MPI_INT, MASTER, mtype, MPI_COMM_WORLD, &status);
      a = alloc_2d_array( rows, degree);
      /* MPI_Recv(&a, rows * degree, MPI_DOUBLE, MASTER, mtype, MPI_COMM_WORLD, &status);  */
      MPI_Recv(&(a[0][0]), rows * degree, MPI_DOUBLE, MASTER, mtype, MPI_COMM_WORLD, &status); 

      printf( "Final from %d\n", taskid );
      for(i = 0; i < rows; i++) {
    printf( "%d ", taskid );
    for(j = 0; j < degree; j++)
      printf("%f ", a[i][j]);
    printf("\n");
      }
    }

  MPI_Finalize();

  return EXIT_SUCCESS;

}
ian@ian-pc:~/test$ mpicc -std=c89 -Wall -Wextra -pedantic -O mpi.c
In file included from mpi.c:5:0:
/usr/lib/openmpi/include/mpi.h:220:9: warning: ISO C90 does not support ‘long long’ [-Wlong-long]
mpi.c: In function ‘main’:
mpi.c:45:13: warning: ignoring return value of ‘scanf’, declared with attribute warn_unused_result [-Wunused-result]
ian@ian-pc:~/test$ mpirun -np 3 ./a.out
Podaj stopien macierzy: 
2
Obecnie dostepnych jest 3 procesow do dyspozycji mastera.
Tworzenie macierzy z pliku
Initial
0.000000 10.000000 
1.000000 11.000000 
Final from 1
1 0.000000 10.000000 
Final from 2
2 1.000000 11.000000 
ian@ian-pc:~/test$ mpirun -np 3 ./a.out
Podaj stopien macierzy: 
4
Final from 1
1 0.000000 10.000000 20.000000 30.000000 
1 1.000000 11.000000 21.000000 31.000000 
Final from 2
2 2.000000 12.000000 22.000000 32.000000 
2 3.000000 13.000000 23.000000 33.000000 
Obecnie dostepnych jest 3 procesow do dyspozycji mastera.
Tworzenie macierzy z pliku
Initial
0.000000 10.000000 20.000000 30.000000 
1.000000 11.000000 21.000000 31.000000 
2.000000 12.000000 22.000000 32.000000 
3.000000 13.000000 23.000000 33.000000 
ian@ian-pc:~/test$ mpirun -np 3 ./a.out
Podaj stopien macierzy: 
5
Final from 2
2 3.000000 13.000000 23.000000 33.000000 43.000000 
2 4.000000 14.000000 24.000000 34.000000 44.000000 
Obecnie dostepnych jest 3 procesow do dyspozycji mastera.
Tworzenie macierzy z pliku
Initial
0.000000 10.000000 20.000000 30.000000 40.000000 
1.000000 11.000000 21.000000 31.000000 41.000000 
2.000000 12.000000 22.000000 32.000000 42.000000 
3.000000 13.000000 23.000000 33.000000 43.000000 
4.000000 14.000000 24.000000 34.000000 44.000000 
Final from 1
1 0.000000 10.000000 20.000000 30.000000 40.000000 
1 1.000000 11.000000 21.000000 31.000000 41.000000 
1 2.000000 12.000000 22.000000 32.000000 42.000000 
ian@ian-pc:~/test$

不过也要了解 MPI_Bcast，在这里会很有用...

【讨论】：

非常感谢！你救我！我真的很感激你对我所做的一切。
真的很抱歉，但这并没有解决我的问题。当我从 master 发送并在 worker 中接收时没有问题，但是当我尝试从 worker 发送到 master 时，它被阻止了。我该怎么办？
首先显示不起作用的代码。尽量避免输入文件，我们需要的是一个我们可以自己测试的示例。

【解决方案2】：

动态分配数组的方式看起来很奇怪。通常你会这样做：

double **alloc_2d_array(int rows, int cols) {
    int i;
    double **array= (double **)malloc(rows*sizeof(double*));
    for (i=0; i<rows; i++)
        array[i] = (double *)malloc(rows*cols*sizeof(double));
    return array;
}

然后释放也将分两步进行：

void free_2d_array(double*** arrayPtr){
  double **array = *arrayPtr;
  for (i=0; i<rows; i++) free(array[i]);
  free(array);
}

更新：假设按照最初建议的方式执行效率更高，分配应该是这样的：

double **alloc_2d_array(int rows, int cols) {
    int i;
    double *data = (double *)malloc(rows*cols*sizeof(double));
    double **array= (double **)malloc(rows*sizeof(double*));
    for (i=0; i<rows; i++)
        array[i] = data + cols*i;

    return array;
}

在这种情况下，释放也很简单：为两个数组调用free。

【讨论】：

最好一步分配，然后将行指针设置为连续数组。在 MPI 中使用它要容易得多。
是的，我没有时间更正它，在另一个问题中，Jonathan Dursi 的回答中有一个很好的解决方案，但我找不到它。
是的，差不多就是这样，+1 来自我。只是你可能想要cols*i。
data + cols*i 不只是 OP 的指针算术版本吗？另外，请不要将您的mallocs 投射到 C 中。
也许是这样，但它不会改变行为，因此它不能成为问题的解决方案。我不明白这是一个答案。