domain_indirect_mpi.cc

Go to the documentation of this file.

 
#include "rheolef/config.h"
 
#ifdef _RHEOLEF_HAVE_MPI
#include "rheolef/domain_indirect.h"
#include "rheolef/geo.h"
 
namespace rheolef {
 
// ----------------------------------------------------------------------------
// @brief init domain ios from the whole mesh ios data
// ----------------------------------------------------------------------------
template <class T>
void
domain_indirect_rep<distributed>::init_ios (
    const geo_abstract_rep<T,distributed>& omega)
{
  // compute i/o _ioige2ini_dis_ioige & _ioige2ini_dis_ioige numbering tables
  // --------------------------------------------
  // 1) mark used elements in omega ios numbering
  // --------------------------------------------
  const size_type unset = std::numeric_limits<size_type>::max();
  distributor ios_ownership = omega.ios_sizes().ownership_by_dimension [base::_map_dim];
  disarray<size_type,distributed> ios_mark (ios_ownership, unset);
  distributor dom_ownership = base::ownership();
  size_type first_dom_dis_ie = dom_ownership.first_index();
  size_type dom_ie = 0;
  for (const_iterator_ioige iter = base::ioige_begin(), last = base::ioige_end();
        iter != last; iter++, dom_ie++) {
    size_type ie = (*iter).index();
    size_type ios_dis_ie = omega.ige2ios_dis_ige (base::_map_dim, ie);
    size_type dom_dis_ie = first_dom_dis_ie + dom_ie;
    ios_mark.dis_entry (ios_dis_ie) = dom_dis_ie;
  }
  ios_mark.dis_entry_assembly();
  // --------------------------------------------
  // 2) count used elements in omega ios numbering
  // --------------------------------------------
  size_type dom_ini_size = 0;
  for (size_type ios_ie = 0, ios_ne = ios_mark.size(); ios_ie < ios_ne; ios_ie++) {
    if (ios_mark [ios_ie] != unset) dom_ini_size++;
  }
  // --------------------------------------------
  // 3) set _ini_ioige2dis_ioige
  // --------------------------------------------
  distributor dom_ini_ownership (distributor::decide, base::comm(), dom_ini_size);
  _ini_ioige2dis_ioige.resize (dom_ini_ownership, unset);
  size_type dom_ini_ie = 0;
  for (size_type ios_ie = 0, ios_ne = ios_mark.size(); ios_ie < ios_ne; ios_ie++) {
    if (ios_mark [ios_ie] == unset) continue;
    _ini_ioige2dis_ioige [dom_ini_ie] = ios_mark [ios_ie];
    dom_ini_ie++;
  }
  // --------------------------------------------
  // 4) set _ioige2ini_dis_ioige
  // --------------------------------------------
  _ioige2ini_dis_ioige.resize (dom_ownership, unset);
  _ini_ioige2dis_ioige.reverse_permutation (_ioige2ini_dis_ioige);
}
// ----------------------------------------------------------------------------
// ----------------------------------------------------------------------------
template<class U>
idiststream&
domain_indirect_rep<distributed>::get (
    idiststream&                   ips,
    const geo_rep<U,distributed>& omega)
{
  const size_type unset = std::numeric_limits<size_type>::max();
  // -----------------------
  // 1) get initial disarray
  // -----------------------
  // 1.1) get header
  communicator comm = omega.comm();
  if ( ! dis_scatch (ips, comm, "\ndomain")) {
    ips.is().setstate (std::ios::badbit);
    return ips;
  }
  size_type version, dis_noige;
  ips >> base::_name
      >> version
      >> base::_map_dim
      >> dis_noige;
  check_macro (version == 2, "unsupported version="<<version<<" domain format");
 
  // 1.2) get data
  distributor ini_ioige_ownership (dis_noige, comm);
  disarray<geo_element_indirect,distributed> ini_oige (ini_ioige_ownership);
  ini_oige.get_values (ips);
  // ---------------------------
  // 2) first renumbering (ios)
  // ---------------------------
  // 2.1) compute ios_owner for each oriented-side
  distributor ios_ige_ownership = omega.geo_element_ios_ownership (base::_map_dim);
  disarray<size_type>  ios_owner (ini_ioige_ownership, unset);
  for (size_type ini_ioige = 0, ini_noige = ini_oige.size();
        ini_ioige < ini_noige; ini_ioige++) {
    size_type ios_ige = ini_oige [ini_ioige].index();
    ios_owner [ini_ioige] = ios_ige_ownership.find_owner (ios_ige);
  }
  // 2.2) ios repartition
  disarray<geo_element_indirect>  ios_oige;
  disarray<size_type>  ini_ioige2ios_dis_ioige;
  disarray<size_type>  ios_ioige2ini_dis_ioige;
  ini_oige.repartition (
        ios_owner,
        ios_oige,
        ios_ioige2ini_dis_ioige,
        ini_ioige2ios_dis_ioige);
 
  // ---------------------
  // 3) first renumbering
  // ---------------------
  // 3.1) geo_element renumbering
  distributor ios_ioige_ownership = ios_oige.ownership();
  disarray<geo_element_indirect>  tmp_oige (ios_ioige_ownership);
  for (size_type ios_ioige = 0, ios_noige = ios_ioige_ownership.size();
        ios_ioige < ios_noige; ios_ioige++) {
    orientation_type orient      = ios_oige [ios_ioige].orientation();
    size_type        ios_dis_ige = ios_oige [ios_ioige].index();
    size_type        ios_ige     = ios_dis_ige - ios_ige_ownership.first_index();
    size_type        dis_ige     = omega.ios_ige2dis_ige (base::_map_dim, ios_ige);
    tmp_oige [ios_ioige].set (orient, dis_ige);
  }
  // 3.2) compute ownership for each oriented-side
  distributor ige_ownership = omega.geo_element_ownership (base::_map_dim);
  disarray<size_type>  partition (ios_ioige_ownership, unset);
  for (size_type ios_ioige = 0, ios_noige = ios_ioige_ownership.size();
        ios_ioige < ios_noige; ios_ioige++) {
    size_type ige = tmp_oige [ios_ioige].index();
    partition [ios_ioige] = ige_ownership.find_owner (ige);
  }
  // 3.3) repartition
  disarray<size_type>  ios_ioige2dis_ioige;
  disarray<size_type>  ioige2ios_dis_ioige;
  tmp_oige.repartition (
        partition,
        *this,
        ioige2ios_dis_ioige,
        ios_ioige2dis_ioige);
  
  // 3.4) shift from "dis_ioige" to a "ioige" numbering local to each process:
  distributor ioige_ownership = disarray<geo_element_indirect>::ownership();
  for (size_type ioige = 0, noige = ioige_ownership.size(); ioige < noige; ioige++) {
    size_type        dis_ige = operator[] (ioige).index();
    size_type        ige = dis_ige - ige_ownership.first_index();
    operator[] (ioige).set_index (ige);
  }
  // ----------------------------------------------
  // 4) Back to "ini" renumbering: set permutations
  // ----------------------------------------------
  _ini_ioige2dis_ioige.resize (ini_ioige_ownership, unset);
  _ioige2ini_dis_ioige.resize (    ioige_ownership, unset);
  for (size_type ios_ioige = 0, ios_noige = ios_ioige_ownership.size();
        ios_ioige < ios_noige; ios_ioige++) {
    size_type ini_dis_ioige = ios_ioige2ini_dis_ioige [ios_ioige];
    size_type dis_ioige     = ios_ioige2dis_ioige     [ios_ioige];
    _ini_ioige2dis_ioige.dis_entry (ini_dis_ioige) = dis_ioige;
    _ioige2ini_dis_ioige.dis_entry (dis_ioige)     = ini_dis_ioige;
 }
 _ioige2ini_dis_ioige.dis_entry_assembly();
 _ini_ioige2dis_ioige.dis_entry_assembly();
 
  return ips;
}
template<class U>
odiststream&
domain_indirect_rep<distributed>::put (
    odiststream&                   ops,
    const geo_rep<U,distributed>& omega) const
{
  using namespace std;
  ops << "domain" << endl
      << base::_name << endl
      << "2 " << base::_map_dim << " " << dis_size() << endl;
  distributor ioige_ownership = disarray<geo_element_indirect>::ownership();
  distributor ini_ioige_ownership = _ini_ioige2dis_ioige.ownership();
  distributor ige_ownership = omega.geo_element_ownership (base::_map_dim);
  disarray<geo_element_indirect> ini_oige (ini_ioige_ownership);
  for (size_type ioige = 0, noige = ioige_ownership.size(); ioige < noige; ioige++) {
    size_type ini_dis_ioige = _ioige2ini_dis_ioige [ioige];
    orientation_type orient = operator[] (ioige).orientation();
    size_type        ige    = operator[] (ioige).index();
    size_type   ios_dis_ige = omega.ige2ios_dis_ige (base::_map_dim, ige);
    ini_oige.dis_entry (ini_dis_ioige) = geo_element_indirect (orient, ios_dis_ige);
  }
  ini_oige.dis_entry_assembly();
  ini_oige.put_values (ops);
  return ops;
}
// ----------------------------------------------------------------------------
// instanciation in library
// ----------------------------------------------------------------------------
template
idiststream&
domain_indirect_rep<distributed>::get (
    idiststream&                       ips,
    const geo_rep<Float,distributed>& omega);
 
template
odiststream&
domain_indirect_rep<distributed>::put (
    odiststream&                       ops,
    const geo_rep<Float,distributed>& omega) const;
 
#define _RHEOLEF_instanciation(T,M)                                     \
template                                                                \
void                                                                    \
domain_indirect_rep<M>::init_ios (const geo_abstract_rep<T,M>&);                \
 
_RHEOLEF_instanciation(Float,distributed)
 
} // namespace rheolef
#endif // _RHEOLEF_HAVE_MPI