communication_scheme.h

Go to the documentation of this file.

/*
 * Copyright (c) 2011-2015:  G-CSC, Goethe University Frankfurt
 * Author: Martin Rupp
 * 
 * This file is part of UG4.
 * 
 * UG4 is free software: you can redistribute it and/or modify it under the
 * terms of the GNU Lesser General Public License version 3 (as published by the
 * Free Software Foundation) with the following additional attribution
 * requirements (according to LGPL/GPL v3 §7):
 * 
 * (1) The following notice must be displayed in the Appropriate Legal Notices
 * of covered and combined works: "Based on UG4 (www.ug4.org/license)".
 * 
 * (2) The following notice must be displayed at a prominent place in the
 * terminal output of covered works: "Based on UG4 (www.ug4.org/license)".
 * 
 * (3) The following bibliography is recommended for citation and must be
 * preserved in all covered files:
 * "Reiter, S., Vogel, A., Heppner, I., Rupp, M., and Wittum, G. A massively
 *   parallel geometric multigrid solver on hierarchically distributed grids.
 *   Computing and visualization in science 16, 4 (2013), 151-164"
 * "Vogel, A., Reiter, S., Rupp, M., Nägel, A., and Wittum, G. UG4 -- a novel
 *   flexible software system for simulating pde based models on high performance
 *   computers. Computing and visualization in science 16, 4 (2013), 165-179"
 * 
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
 * GNU Lesser General Public License for more details.
 */
 
#ifndef COMMUNICATION_SCHEME_H_
#define COMMUNICATION_SCHEME_H_
 
#include "pcl/pcl.h"
 
#ifdef UG_PARALLEL
namespace ug
{
 
// CommunicationScheme
//----------------------
template<typename TDerived, typename TValue>
class CommunicationScheme : public pcl::ICommunicationPolicy<IndexLayout>
{
public:
  TDerived& derived() {
         return static_cast<TDerived&>(*this);
  }
 
  virtual bool
  collect(ug::BinaryBuffer& buff, const Interface& interface)
 
  {
    int pid = interface.get_target_proc();
    for(IndexLayout::Interface::const_iterator iter = interface.begin(); iter != interface.end(); ++iter)
      Serialize(buff, derived().send(pid, interface.get_element(iter)));
    return true;
  }
 
 
  virtual bool
  extract(ug::BinaryBuffer& buff, const Interface& interface)
  {
    int pid = interface.get_target_proc();
    TValue val;
    for(IndexLayout::Interface::const_iterator iter = interface.begin(); iter != interface.end(); ++iter)
    {
      Deserialize(buff, val);
      derived().receive(pid, interface.get_element(iter), val);
    }
    return true;
  }
 
  virtual int
  get_required_buffer_size(const Interface& interface)
  {
    int s = derived().get_element_size();
    if(s == -1)
      return -1;
    else
      return s * interface.size();
  }
};
 
 
template<typename TDerived>
class CommunicationScheme<TDerived, bool> : public pcl::ICommunicationPolicy<IndexLayout>
{
public:
  TDerived& derived() { return static_cast<TDerived&>(*this); }
 
  virtual bool
  collect(ug::BinaryBuffer& buff, const Interface& interface)
  {
    int pid = interface.get_target_proc();
    int j=0;
    char a=0;
    for(IndexLayout::Interface::const_iterator iter = interface.begin(); iter != interface.end(); ++iter)
    {
      bool b = derived().send(pid, interface.get_element(iter));
      if(b) a |= (1 << j);
      j++;
      if(j == 8)
      {
        Serialize(buff, a);
        a = 0;
        j = 0;
      }
    }
    if(j) Serialize(buff, a);
    return true;
  }
 
  virtual bool
  extract(ug::BinaryBuffer& buff, const Interface& interface)
  {
    int pid = interface.get_target_proc();
    int j=8;
    char a=0;
    for(IndexLayout::Interface::const_iterator iter = interface.begin(); iter != interface.end(); ++iter)
    {
      size_t index = interface.get_element(iter);
      if(j==8)
      {
        Deserialize(buff, a);
        j=0;
      }
      bool b = a & (1 << j);
      j++;
      derived().receive(pid, index, b);
    }
    return true;
  }
 
  virtual int
  get_required_buffer_size(const Interface& interface)
  {
    return sizeof(char) * (interface.size()/8 + (interface.size()%8==0?0:1) );
  }
 
};
 
 
 
// StdArrayCommunicationScheme
//------------------------------
template<typename TArray>
class StdArrayCommunicationScheme : public CommunicationScheme<StdArrayCommunicationScheme<TArray>,
  typename TArray::value_type >
{
  typedef typename TArray::value_type value_type;
public:
  StdArrayCommunicationScheme(TArray &t) : m_arr(t)
  {
 
  }
 
  inline const value_type &send(int pid, size_t index) const
  {
    return m_arr[index];
  }
 
  inline void receive(int pid, size_t index, value_type &v)
  {
    m_arr[index] = v;
  }
 
  inline int get_element_size() const
  {
    if(block_traits<value_type>::is_static) return sizeof(value_type);
    else
      return -1;
  }
private:
  TArray &m_arr;
};
 
 
template<typename TCommunicationScheme>
void CommunicateOnInterfaces(pcl::InterfaceCommunicator<IndexLayout> &communicator,
    const IndexLayout &sendingLayout, const IndexLayout &receivingLayout, TCommunicationScheme &scheme)
{
  AMG_PROFILE_FUNC();
  int i=0;
 
  for(IndexLayout::const_iterator it = sendingLayout.begin(); it != sendingLayout.end(); ++it, ++i)
  {
    BinaryBuffer buff;
    const IndexLayout::Interface &interface = sendingLayout.interface(it);
    scheme.collect(buff, interface);
    // todo: don't use parallelcommunicator send_raw
    // todo: reserve and reuse buff
    communicator.send_raw(interface.get_target_proc(), buff.buffer(), buff.write_pos(),
        scheme.get_required_buffer_size(interface) != -1);
  }
 
  int receivingLayoutSize=0;
  for(IndexLayout::const_iterator it = receivingLayout.begin(); it != receivingLayout.end(); ++it)
    receivingLayoutSize++;
 
  stdvector< BinaryBuffer > bufs(receivingLayoutSize);
  i=0;
  for(IndexLayout::const_iterator it = receivingLayout.begin(); it != receivingLayout.end(); ++it, ++i)
    communicator.receive_raw(receivingLayout.proc_id(it), bufs[i],
        scheme.get_required_buffer_size(receivingLayout.interface(it)));
 
  communicator.communicate();
  i=0;
  for(IndexLayout::const_iterator it = receivingLayout.begin(); it != receivingLayout.end(); ++it, ++i)
    scheme.extract(bufs[i], receivingLayout.interface(it));
}
 
template<typename TCommunicationScheme>
void CommunicateFromSlaveToMaster(HorizontalAlgebraLayouts &layouts, TCommunicationScheme &scheme)
{
  CommunicateOnInterfaces(layouts.proc_comm(), layouts.slave(), layouts.master(), scheme);
}
template<typename TCommunicationScheme>
void CommunicateFromMasterToSlave(HorizontalAlgebraLayouts &layouts, TCommunicationScheme &scheme)
{
  CommunicateOnInterfaces(layouts.proc_comm(), layouts.master(), layouts.slave(), scheme);
}
 
template<typename TSendingScheme, typename TPIDs>
void SendOnInterfaces(pcl::InterfaceCommunicator<IndexLayout> &communicator, TPIDs &pids,
    IndexLayout &layout, TSendingScheme &sender)
{
  AMG_PROFILE_FUNC();
 
  for(size_t i=0; i<pids.size(); i++)
  {
    int pid = pids[i];
    IndexLayout::Interface &interface = layout.interface(pid);
 
    BinaryBuffer buff;
    sender.collect(buff, interface);
    // todo: don't use parallelcommunicator send_raw
    // todo: reserve and reuse buff
    communicator.send_raw(pid, buff.buffer(), buff.write_pos(),
        sender.get_required_buffer_size(interface) != -1);
  }
  communicator.communicate();
 
}
 
template<typename TPIDs, typename TReceiveScheme>
void ReceiveOnInterfaces(pcl::InterfaceCommunicator<IndexLayout> &communicator, TPIDs &pids,
    IndexLayout &layout, TReceiveScheme &receiver)
{
  stdvector< BinaryBuffer > bufs(pids.size());
  for(size_t i=0; i<pids.size(); i++)
    communicator.receive_raw(pids[i], bufs[i], receiver.get_required_buffer_size(layout.interface(pids[i])));
 
  communicator.communicate();
 
  for(size_t i=0; i<pids.size(); i++)
    receiver.extract(bufs[i], layout.interface(pids[i]));
}
 
 
 
 
 
#if 0
template<typename TDerived, typename TValue>
class SparseCommunicationScheme : public pcl::ICommunicationPolicy<IndexLayout>
{
public:
  TDerived& derived() {
         return static_cast<TDerived&>(*this);
  }
 
  virtual bool
  collect(ug::BinaryBuffer& buff, const Interface& interface)
 
  {
    int pid = interface.get_target_proc();
    for(IndexLayout::Interface::const_iterator iter = interface.begin(); iter != interface.end(); ++iter)
    {
      int index;
      char a = 0;
      if(derived().need_send(pid, index))
        a = 1;
      Serialize(buff, a);
      if(a) Serialize(buff, derived().send(pid, index);
    }
  }
 
 
  virtual bool
  extract(ug::BinaryBuffer& buff, const Interface& interface)
  {
    int pid = interface.get_target_proc();
    TValue val;
    for(IndexLayout::Interface::const_iterator iter = interface.begin(); iter != interface.end(); ++iter)
    {
      char a;
      Deserialize(buff, a);
      if(a)
      {
        Deserialize(buff, val);
        derived().receive(pid, interface.get_element(iter), val);
      }
    }
  }
 
  virtual int
  get_required_buffer_size(const Interface& interface)
  {
    int s = derived().get_element_size();
    if(s == -1)
      return -1;
    else
      return s * interface.size();
  }
};
#endif
 
// end group lib_algebra_parallelization_scheme
 
} // namespace ug
#endif /* SEND_INTERFACE_H_ */
 
#endif /* UG_PARALLEL */