如何使用MPI发送GMP或MPFR类型数据

Question

我尝试使用 MPI_Scatter 发送 mpfr_t 类型的变量。例如：

mpfr_t *v1 = new mpfr_t[10];  
mpfr_t *v2 = new mpfr_t[10];   
MPI_Scatter(v1, 5, MPI_BYTE, v2, 5, MPI_BYTE, 0, MPI_COMM_WORLD ); 
for (int i = 0; i < 5; i++) 
    mpfr_printf("value rank %d -  %RNf \n", ProcRank, v2[i]);

它会打印出:

value rank 0 - nan
value rank 0 - nan
value rank 0 - nan
.....
value rank 1 - nan
value rank 0 - nan

但这是MPI_Bcast的工作。我做错了什么？C/C++代码，使用的是OpenMPI-1.6库。

- Sidny Sho

---

你似乎忘记了将sendcount乘以mpfr_t的大小。如果你只有5个进程，为什么要分配10个数组？ - Dima Chubarov

是的，数组的大小是5。如果我将数量乘以sizeof(mpfr_t)，它会正常工作。谢谢。我该如何重写缩减函数？例如MPI_MINLOC和MPI_MAXLOC对于mpfr类型的变量？ - Sidny Sho

你应该定义自己的归约运算符，然后使用 MPI_Op_create() 进行注册。请参见这里。 - Hristo Iliev

这与约简运算符有什么关系？MPI_Scatter根本不使用它们。 - timos

Answer 1

您指定了sendcount为5，datatype为MPI_BYTE，这似乎很奇怪。如果您希望使用MPI_BYTE并且想要发送5个mpfr_t值，则应指定sendcount为5*sizeof(mpfr_t)。另一种选择是创建自己的MPI派生数据类型（如果您想要摆脱sizeof()）。

- timos

---

MPFR的所有数据是否都封装在该分配中？我从来没有确定过该结构的一部分是否是指向堆内存的指针。 - Jeff Hammond

@JeffHammond：MPFR数字是一个具有4个元素的“struct”的单元素数组。其中之一，_mpfr_d不幸地是指向在堆上分配的“limbs”数组的指针。请参见https://www.mpfr.org/mpfr-current/mpfr.html#Internals。 - András Aszódi

Answer 2

如@LaryxDecidua所指出的那样，MPFR数字使用堆上的动态内存，因此不能直接用于MPI操作。但是，如果您可以使用MPFR版本4.0.0或更高版本，则可以使用MPFR函数mpfr_fpif_export将数字序列化为线性内存缓冲区，并使用MPI发送该缓冲区。接收端可以通过调用mpfr_fpif_import来恢复原始数字。这两个函数都在FILE *句柄上操作，因此您还需要使用open_memstream和/或fmemopen将文件句柄映射到内存缓冲区。不幸的是，序列化的数字的长度不仅取决于精度，还取决于值（例如值0占用的字节数较少），因此需要每个排名固定缓冲区大小的组通信（例如MPI_Scatter或MPI_Gather）将无法使用。请改用MPI_Send和MPI_recv对。

这是一个完整的C++17示例，其中包含mpreal，为C++代码库提供了与MPFR数字交互的良好接口：

#include <cstdio>
#include <iostream>
#include <mpi.h>
#include <mpreal.h>
#include <numeric>
#include <optional>


int main(int argc, char **argv) {
    MPI_Init(&argc, &argv);
    int world_rank;
    MPI_Comm_rank(MPI_COMM_WORLD, &world_rank);
    int world_size;
    MPI_Comm_size(MPI_COMM_WORLD, &world_size);


    std::vector<mpfr::mpreal> real_vec{world_rank, M_PI};

    // Serialize the mpreal vector to memory (send_buf)
    char *send_buf;
    size_t send_buf_size;
    FILE *real_stream = open_memstream(&send_buf, &send_buf_size);
    for (auto &real : real_vec) {
        mpfr_fpif_export(real_stream, real.mpfr_ptr());
    }
    fclose(real_stream);

    // Gather the buffer length of all processes
    std::optional<std::vector<size_t>> send_buf_size_vec;
    if (world_rank == 0) {
        send_buf_size_vec = std::vector<size_t>(world_size);
    }
    MPI_Gather(&send_buf_size, 1, MPI_UNSIGNED_LONG, (send_buf_size_vec ? send_buf_size_vec->data() : nullptr), 1,
               MPI_UNSIGNED_LONG, 0, MPI_COMM_WORLD);

    if (world_rank != 0) {
        // Send the serialized mpreal vector to rank 0
        MPI_Send(send_buf, send_buf_size, MPI_BYTE, 0, 0, MPI_COMM_WORLD);
    } else {
        // Create a recv buffer which can hold the data from all processes
        size_t recv_buf_size = std::accumulate(send_buf_size_vec->begin(), send_buf_size_vec->end(), 0UL);
        std::vector<char> recv_buf(recv_buf_size);
        auto all_buf_it = recv_buf.begin();

        // Directly copy the send buffer of process 0
        std::memcpy(&*all_buf_it, send_buf, send_buf_size);
        all_buf_it += (*send_buf_size_vec)[0];

        // Receive serialized numbers from all other ranks
        MPI_Status status;
        for (int i = 1; i < world_size; ++i) {
            MPI_Recv(&*all_buf_it, (*send_buf_size_vec)[i], MPI_BYTE, i, MPI_ANY_TAG, MPI_COMM_WORLD, &status);
            all_buf_it += (*send_buf_size_vec)[i];
        }

        // Import all mpreals from the receive buffer
        real_stream = fmemopen(recv_buf.data(), recv_buf.size(), "rb");
        std::vector<mpfr::mpreal> all_real_vec(world_size * real_vec.size());
        for (auto &real : all_real_vec) {
            mpfr_fpif_import(real.mpfr_ptr(), real_stream);
        }
        fclose(real_stream);

        // Print all received values
        std::cout << "Read values:" << std::endl;
        for (auto &real : all_real_vec) {
            std::cout << real << std::endl;
        }
    }

    MPI_Finalize();

    return 0;
}

- IngoMeyer