mpi_scatter_init.h
Go to the documentation of this file.
#ifndef _RHEO_MPI_SCATTER_INIT_H
#define _RHEO_MPI_SCATTER_INIT_H
#include "rheolef/compiler.h"
#include "rheolef/distributed.h"
#include "rheolef/scatter_message.h"
#include "rheolef/msg_sort_with_permutation.h"
#include "rheolef/msg_to_context.h"
#include "rheolef/msg_from_context_pattern.h"
#include "rheolef/msg_from_context_indices.h"
#include "rheolef/msg_local_context.h"
#include "rheolef/msg_local_optimize.h"
#include "rheolef/msg_util.h"
#include <boost/functional.hpp>
#include <boost/iterator/transform_iterator.hpp>
/*F:
NAME: mpi_scatter_init -- gather/scatter initialize (@PACKAGE@ @VERSION@)
DESCRIPTION:
Initialize communication
for distributed to sequential scatter context.
COMPLEXITY:
Time and memory complexity is O(nidx+nproc).
For finite-element problems in d dimenion
| nidx ~ N^((d-1)/d)
where N is the number of degrees of freedom.
IMPLEMENTATION
Inspirated from petsc-2.0/vpscat.c: VecScatterCreate_PtoS()
AUTHORS:
LMC-IMAG, 38041 Grenoble cedex 9, France
| Pierre.Saramito@imag.fr
DATE: 23 march 1999
END:
*/
namespace rheolef {
//<mpi_scatter_init:
template <class Message, class Size, class SizeRandomIterator1,
class SizeRandomIterator2, class SizeRandomIterator3, class Tag>
void
mpi_scatter_init (
// input:
Size nidx,
SizeRandomIterator1 idx,
Size nidy,
SizeRandomIterator2 idy,
Size idy_maxval,
SizeRandomIterator3 ownership,
Tag tag,
const distributor::communicator_type& comm,
// output:
Message& from,
Message& to)
{
typedef Size size_type;
size_type my_proc = comm.rank();
size_type nproc = comm.size();
// -------------------------------------------------------
// 1) first count number of contributors to each processor
// -------------------------------------------------------
std::vector<size_type> msg_size(nproc, 0);
std::vector<size_type> msg_mark(nproc, 0);
std::vector<size_type> owner (nidx);
size_type send_nproc = 0;
{
size_type iproc = 0;
for (size_type i = 0; i < nidx; i++) {
for (; iproc < nproc; iproc++) {
if (idx[i] >= ownership[iproc] && idx[i] < ownership[iproc+1]) {
owner[i] = iproc;
msg_size [iproc]++;
if (!msg_mark[iproc]) {
msg_mark[iproc] = 1;
send_nproc++;
}
break;
}
}
check_macro (iproc != nproc, "bad stash data: idx["<<i<<"]="<<idx[i]<<" out of range [0:"<<ownership[nproc]<<"[");
}
} // end block
// -------------------------------------------------------
// 2) avoid to send message to my-proc in counting
// -------------------------------------------------------
size_type n_local = msg_size[my_proc];
if (n_local != 0) {
msg_size [my_proc] = 0;
msg_mark [my_proc] = 0;
send_nproc--;
}
// ----------------------------------------------------------------
// 3) compute number of messages to be send to my_proc
// ----------------------------------------------------------------
std::vector<size_type> work(nproc);
mpi::all_reduce (
comm,
msg_mark.begin().operator->(),
nproc,
work.begin().operator->(),
std::plus<size_type>());
size_type receive_nproc = work [my_proc];
// ----------------------------------------------------------------
// 4) compute messages max size to be send to my_proc
// ----------------------------------------------------------------
mpi::all_reduce (
comm,
msg_size.begin().operator->(),
nproc,
work.begin().operator->(),
mpi::maximum<size_type>());
size_type receive_max_size = work [my_proc];
// ----------------------------------------------------------------
// 5) post receive: exchange the buffer adresses between processes
// ----------------------------------------------------------------
std::list<std::pair<size_type,mpi::request> > receive_waits;
std::vector<size_type> receive_data (receive_nproc*receive_max_size);
for (size_type i_receive = 0; i_receive < receive_nproc; i_receive++) {
mpi::request i_req = comm.irecv (
mpi::any_source,
tag,
receive_data.begin().operator->() + i_receive*receive_max_size,
receive_max_size);
receive_waits.push_back (std::make_pair(i_receive, i_req));
}
// ---------------------------------------------------------------------------
// 6) compute the send indexes
// ---------------------------------------------------------------------------
// comme idx est trie, on peut faire une copie de idx dans send_data
// et du coup owner et send_data_ownership sont inutiles
std::vector<size_type> send_data (nidx);
std::copy (idx, idx+nidx, send_data.begin());
// ---------------------------------------------------------------------------
// 7) do send
// ---------------------------------------------------------------------------
std::list<std::pair<size_type,mpi::request> > send_waits;
{
size_type i_send = 0;
size_type i_start = 0;
for (size_type iproc = 0; iproc < nproc; iproc++) {
size_type i_msg_size = msg_size[iproc];
if (i_msg_size == 0) continue;
mpi::request i_req = comm.isend (
iproc,
tag,
send_data.begin().operator->() + i_start,
i_msg_size);
send_waits.push_back(std::make_pair(i_send,i_req));
i_send++;
i_start += i_msg_size;
}
} // end block
// ---------------------------------------------------------------------------
// 8) wait on receives
// ---------------------------------------------------------------------------
// note: for wait_all, build an iterator adapter that scan the pair.second in [index,request]
// and then get an iterator in the pair using iter.base(): retrive the corresponding index
// for computing the position in the receive.data buffer
typedef boost::transform_iterator<select2nd<size_t,mpi::request>, std::list<std::pair<size_t,mpi::request> >::iterator>
request_iterator;
std::vector<size_type> receive_size (receive_nproc);
std::vector<size_type> receive_proc (receive_nproc);
size_type receive_total_size = 0;
while (receive_waits.size() != 0) {
typedef size_type data_type; // exchanged data is of "size_type"
request_iterator iter_r_waits (receive_waits.begin(), select2nd<size_t,mpi::request>()),
last_r_waits (receive_waits.end(), select2nd<size_t,mpi::request>());
// waits on any receive...
std::pair<mpi::status,request_iterator> pair_status = mpi::wait_any (iter_r_waits, last_r_waits);
// check status
boost::optional<int> i_msg_size_opt = pair_status.first.count<data_type>();
check_macro (i_msg_size_opt, "receive wait failed");
int iproc = pair_status.first.source();
check_macro (iproc >= 0, "receive: source iproc = "<<iproc<<" < 0 !");
// get size of receive and number in data
size_type i_msg_size = (size_t)i_msg_size_opt.get();
std::list<std::pair<size_t,mpi::request> >::iterator i_pair_ptr = pair_status.second.base();
size_type i_receive = (*i_pair_ptr).first;
receive_proc [i_receive] = iproc;
receive_size [i_receive] = i_msg_size;
receive_total_size += i_msg_size;
receive_waits.erase (i_pair_ptr);
}
// ---------------------------------------------------------------------------
// 9) allocate the entire send(to) scatter context
// ---------------------------------------------------------------------------
to.resize (receive_total_size, receive_nproc);
// ---------------------------------------------------------------------------
// 10) compute the permutation of values that gives the sorted source[] sequence
// ---------------------------------------------------------------------------
// init: perm[i] = i
std::vector<size_type> perm(receive_nproc);
copy(index_iterator<size_type>(), index_iterator<size_type>(receive_nproc), perm.begin());
sort_with_permutation (
receive_proc.begin().operator->(),
perm.begin().operator->(),
receive_nproc);
// ---------------------------------------------------------------------------
// 11) Computes the receive compresed message pattern for send(to)
// ---------------------------------------------------------------------------
size_type istart = ownership[my_proc]; // = ownership.first_index()
msg_to_context (
perm.begin(),
perm.end(),
receive_proc.begin(),
receive_size.begin(),
receive_data.begin(),
receive_max_size,
istart,
to.procs().begin(),
to.starts().begin(),
to.indices().begin());
// ---------------------------------------------------------------------------
// 12) allocate the entire receive(from) scatter context
// ---------------------------------------------------------------------------
from.resize(nidy, send_nproc);
// ---------------------------------------------------------------------------
// 13) Computes the receive compresed message pattern for receive(from)
// ---------------------------------------------------------------------------
std::vector<size_type> proc2from_proc(nproc);
msg_from_context_pattern (
msg_size.begin(),
msg_size.end(),
from.procs().begin(),
from.starts().begin(),
proc2from_proc.begin());
// ---------------------------------------------------------------------------
// 14) Computes the receive compresed message indices for receive(from)
// ---------------------------------------------------------------------------
// assume that indices are sorted by increasing order
std::vector<size_type> start(send_nproc+1);
copy (from.starts().begin(), from.starts().end(), start.begin());
msg_from_context_indices (
owner.begin(),
owner.end(),
idy,
proc2from_proc.begin(),
my_proc,
idy_maxval,
start.begin(),
from.indices().begin());
// ---------------------------------------------------------------------------
// 15) wait on sends
// ---------------------------------------------------------------------------
request_iterator iter_s_waits (send_waits.begin(), select2nd<size_type,mpi::request>()),
last_s_waits (send_waits.end(), select2nd<size_type,mpi::request>());
mpi::wait_all (iter_s_waits, last_s_waits);
// ---------------------------------------------------------------------------
// 16) Computes the receive compresed message local pattern,
// i.e. the only part that does not requires communication.
// ---------------------------------------------------------------------------
from.local_slots.resize(n_local);
to.local_slots.resize(n_local);
size_type ilast = ownership[my_proc+1]; // = ownership.last_index()
msg_local_context (
idx,
idx+nidx,
idy,
idy_maxval,
istart,
ilast,
to.local_slots.begin(),
to.local_slots.end(),
from.local_slots.begin());
// ---------------------------------------------------------------------------
// 17) Optimize local exchanges during gatter/scatter
// ---------------------------------------------------------------------------
bool has_opt = msg_local_optimize (
to.local_slots.begin(),
to.local_slots.end(),
from.local_slots.begin());
if (has_opt && n_local != 0) {
to.local_is_copy = true;
to.local_copy_start = to.local_slots[0];
to.local_copy_length = n_local;
from.local_is_copy = true;
from.local_copy_start = from.local_slots[0];
from.local_copy_length = n_local;
}
}
//>mpi_scatter_init:
} // namespace rheolef
#endif // _RHEO_MPI_SCATTER_INIT_H
check_macro
check_macro(expr1.have_homogeneous_space(Xh1), "dual(expr1,expr2); expr1 should have homogeneous space. HINT: use dual(interpolate(Xh, expr1),expr2)")
rheolef::msg_from_context_indices
void msg_from_context_indices(InputIterator1 owner, InputIterator1 last_owner, InputIterator2 idy, InputRandomIterator proc2from_proc, Proc my_proc, Size idy_maxval, MutableRandomIterator ptr, OutputIterator from_idx)
Definition: msg_from_context_indices.h:76
rheolef::msg_local_context
void msg_local_context(InputIterator1 idx, InputIterator1 last_idx, InputIterator2 idy, Size idy_maxval, Size istart, Size ilast, OutputIterator1 to_loc_idx, OutputIterator1 last_to_loc_idx, OutputIterator2 from_loc_idy)
Definition: msg_local_context.h:73
rheolef::index_iterator
Definition: msg_util.h:79
rheolef::distributor::communicator_type
communicator communicator_type
Definition: distributor.h:72
rheolef::size_type
size_t size_type
Definition: basis_get.cc:76
rheolef::msg_from_context_pattern
void msg_from_context_pattern(InputIterator1 msg_size, InputIterator1 last_msg_size, OutputIterator1 from_proc, OutputIterator2 from_ptr, OutputIterator3 proc2from_proc)
Definition: msg_from_context_pattern.h:72
rheolef::sort_with_permutation
void sort_with_permutation(RandomIterator v, SizeRandomIterator p, Size n)
Definition: msg_sort_with_permutation.h:91
rheolef::msg_local_optimize
bool msg_local_optimize(InputIterator1 to_loc_idx, InputIterator1 last_to_loc_idx, InputIterator2 from_loc_idy)
Definition: msg_local_optimize.h:75
rheolef::select2nd
Definition: msg_util.h:45
rheolef::mpi_scatter_init
void mpi_scatter_init(Size nidx, SizeRandomIterator1 idx, Size nidy, SizeRandomIterator2 idy, Size idy_maxval, SizeRandomIterator3 ownership, Tag tag, const distributor::communicator_type &comm, Message &from, Message &to)
Definition: mpi_scatter_init.h:67
rheolef::msg_to_context
void msg_to_context(InputIterator1 perm, InputIterator1 last_perm, InputRandomIterator2 r_iproc, InputRandomIterator3 r_size, InputRandomIterator4 r_idx, Size receive_max_size, Size istart, OutputIterator1 to_proc, OutputIterator2 to_ptr, OutputIterator3 to_idx)
Definition: msg_to_context.h:78
rheolef
This file is part of Rheolef.
Definition: compiler_eigen.h:37
size_type
field::size_type size_type
Definition: branch.cc:425