通过 MPI 进程同步数组，如果每个线程都改变了它的一部分？

Question

我有一个程序要使用 MPI 进行并行化。我以前没有使用过 MPI。

程序随着时间的推移计算大量对象的行为。的数据 这些对象存储在数组中，例如x 坐标为double precision :: body_x(10000)。

要计算一个对象的行为，需要所有其他对象的信息， 所以每个线程都需要保存所有数据，但只会更新其中的一部分。但在之前 新的时间步长，每个线程都需要从所有其他线程获取信息。

据我了解，MPI_Allgather 可以用于此目的，但它需要一个发送缓冲区和一个 接收缓冲区。如果每个线程都更新了，我如何在不同的线程上同步一个数组 数组的不同部分？我是否必须将整个数组从每个线程发送到 master 在接收缓冲区中，更新 masters 数组的特定部分，毕竟 线程已经从主服务器发送了他们的数据重新广播？

这是一个非常基本的问题，但我对 MPI 很陌生，我发现的所有示例都是 很简单，不涉及这个。感谢您的帮助。

伪示例（假设第一个索引为 1 的 Fortran 样式向量）： （是的，发送/接收最好是非阻塞的，这是为了简单起见）

if (master) then
   readInputFile
end if

MPI_Bcast(numberOfObject)

allocate body_arrays(numberOfObjects)

if (master) then
   fill body_arrays ! with the data from the input file
end if

MPI_Bcast(body_arrays)

objectsPerThread = numberOfObjects / threadCount

myStart = threadID * objectsPerThread + 1
myEnd   = (threadID + 1) * objectsPerThread

do while (t < t_end)

   do i = myStart, myEnd
      do stuff for body_arrays(i)
   end do

   ! here is the question
   if (.not. master)
      MPI_Send(body_arrays, toMaster)
   else
      do i = 1, threadCount - 1
         MPI_Recive(body_arrays_recive, senderID)
         body_arrays(senderID*objectsPerThread+1, (senderId+1)*objectsPerThread) = body_arrays_recive(senderID*objectsPerThread+1, (senderId+1)*objectsPerThread)
   end if

   MPI_Bcast(body_arrays)
   ! ----

   t = t + dt
end do

Answer 1

听起来你想要 MPI_Allgather。为避免需要单独的发送缓冲区，您可以使用 MPI_IN_PLACE 值。这告诉 MPI 使用相同的缓冲区进行发送和接收。

见http://mpi-forum.org/docs/mpi-2.2/mpi22-report/node99.htm#Node99

Answer 2

可以通过调用MPI_Allgatherv 来组合来自所有进程的数组块。以下是 Fortran 中的完整示例。它定义了一个大小为 50 的数组。然后每个进程将该数组的一部分设置为某个复数。最后，对MPI_allgatherv 的调用将所有块拉到一起。块大小的计算，以及一些需要传递给MPI_allgatherv的参数都封装在mpi_split例程中。

program test
  use mpi

  implicit none

  integer, parameter :: idp = 8
  integer, parameter :: n_tasks = 11
  real(idp), parameter :: zero = 0.0d0
  complex(idp), parameter :: czero = cmplx(zero, zero, kind=idp)

  integer :: mpi_n_procs, mpi_proc_id, error
  integer :: i, i_from, i_to
  complex(idp) :: c(-5:5)
  real(idp) :: split_size

  integer, allocatable :: recvcount(:), displs(:)

  call MPI_Init(error)
  call MPI_Comm_size(MPI_COMM_WORLD, mpi_n_procs, error)
  call MPI_Comm_rank(MPI_COMM_WORLD, mpi_proc_id, error)

  allocate(recvcount(mpi_n_procs))
  allocate(displs(mpi_n_procs))

  i_from = -5
  i_to = 5

  ! each process covers only part of the array
  call mpi_split(i_from, i_to, counts=recvcount, displs=displs)
  write(*,*) "ID", mpi_proc_id,":", i_from, "..", i_to
  if (mpi_proc_id == 0) then
    write(*,*) "Counts: ", recvcount
    write(*,*) "Displs: ", displs
  end if

  c(:) = czero
  do i = i_from, i_to
    c(i) = cmplx(real(i, idp), real(i+1, idp), kind=idp)
  end do

  call MPI_Allgatherv(c(i_from), i_to-i_from+1, MPI_DOUBLE_COMPLEX, c,         &
  &                   recvcount, displs, MPI_DOUBLE_COMPLEX, MPI_COMM_WORLD,   &
  &                   error)

  if (mpi_proc_id == 0) then
    do i = -5, 5
      write(*,*) i, ":", c(i)
    end do
  end if

  deallocate(recvcount, displs)
  call MPI_Finalize(error)

contains

  !! @description: split the range (a,b) into equal chunks, where each chunk is
  !! handled by a different MPI process
  !! @param: a        On input, the lower bound of an array to be processed. On
  !!                  output, the lower index of the chunk that the MPI process
  !!                  `proc_id` should process
  !! @param: b        On input, the upper bound of an array. On, output the
  !!                  upper index of the chunk that process `proc_id` should
  !!                  process.
  !! @param: n_procs  The total number of available processes. If not given,
  !!                  this is determined automatically from the MPI environment.
  !! @param: proc_id  The (zero-based) process ID (`0 <= proc_id < n_procs`). If
  !!                  not given, the ID of the current MPI process
  !! @param: counts   If given, must be of size `n_procs`. On output, the chunk
  !!                  size for each MPI process
  !! @param: displs   If given, must be of size `n_procs`. On output, the offset
  !!                  if the first index processed by each MPI process, relative
  !!                  to the input value of `a`
  subroutine mpi_split(a, b, n_procs, proc_id, counts, displs)

    integer,           intent(inout) :: a
    integer,           intent(inout) :: b
    integer, optional, intent(in)    :: n_procs
    integer, optional, intent(in)    :: proc_id
    integer, optional, intent(inout) :: counts(:)
    integer, optional, intent(inout) :: displs(:)

    integer :: mpi_n_procs, n_tasks, mpi_proc_id, error
    integer :: aa, bb
    real(idp) :: split_size
    logical :: mpi_is_initialized

    mpi_n_procs = 1
    if (present(n_procs)) mpi_n_procs = n_procs
    mpi_proc_id = 0
    if (present(proc_id)) mpi_proc_id = proc_id
    if (.not. present(n_procs)) then
      call MPI_Comm_size(MPI_COMM_WORLD, mpi_n_procs, error)
    end if
    if (.not. present(proc_id)) then
      call MPI_Comm_rank(MPI_COMM_WORLD, mpi_proc_id, error)
    end if

    aa = a
    bb = b

    n_tasks = bb - aa + 1
    split_size = real(n_tasks, idp) / real(max(mpi_n_procs, 1), idp)
    a = nint(mpi_proc_id * split_size) + aa
    b = min(aa + nint((mpi_proc_id+1) * split_size) - 1, bb)

    if (present(counts)) then
      do mpi_proc_id = 0, mpi_n_procs-1
        counts(mpi_proc_id+1) = max(nint((mpi_proc_id+1) * split_size)         &
        &                           - nint((mpi_proc_id) * split_size), 0)
      end do
    end if

    if (present(displs)) then
      do mpi_proc_id = 0, mpi_n_procs-1
        displs(mpi_proc_id+1) = min(nint(mpi_proc_id * split_size), bb-aa)
      end do
    end if

  end subroutine mpi_split

end program