communicator.h

Go to the documentation of this file.

// The TIMPI Message-Passing Parallelism Library.
// Copyright (C) 2002-2025 Benjamin S. Kirk, John W. Peterson, Roy H. Stogner

// This library is free software; you can redistribute it and/or
// modify it under the terms of the GNU Lesser General Public
// License as published by the Free Software Foundation; either
// version 2.1 of the License, or (at your option) any later version.

// This library 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.

// You should have received a copy of the GNU Lesser General Public
// License along with this library; if not, write to the Free Software
// Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA


#ifndef TIMPI_COMMUNICATOR_H
#define TIMPI_COMMUNICATOR_H

// TIMPI includes
#include "timpi/message_tag.h"
#include "timpi/packing.h"
#include "timpi/request.h"
#include "timpi/standard_type.h"
#include "timpi/status.h"
#include "timpi/timpi_config.h"
#include "timpi/timpi_macros.h"

// C++ includes
#include <cstdint>
#include <map>
#include <memory> // shared_ptr
#include <string>
#include <vector>
#include <type_traits>

// C++ includes needed for parallel_communicator_specializations
//
// These could be forward declarations if only that wasn't illegal
#include <complex> // for specializations
#include <set>
#include <unordered_map>

namespace TIMPI
{

using libMesh::Parallel::Packing;
using libMesh::Parallel::Has_buffer_type;

// Define processor id storage type.
#if TIMPI_PROCESSOR_ID_BYTES == 1
typedef uint8_t processor_id_type;
#elif TIMPI_PROCESSOR_ID_BYTES == 2
typedef uint16_t processor_id_type;
#elif TIMPI_PROCESSOR_ID_BYTES == 4
typedef uint32_t processor_id_type; // default
#elif TIMPI_PROCESSOR_ID_BYTES == 8
typedef uint64_t processor_id_type;
#else
// We should not get here: it's not currently possible to set any
// other size.
DIE A HORRIBLE DEATH HERE...
#endif


#ifdef TIMPI_HAVE_MPI

//-------------------------------------------------------------------
typedef MPI_Comm communicator;

typedef MPI_Info info;

const unsigned int any_source =
  static_cast<unsigned int>(MPI_ANY_SOURCE);

#else

// These shouldn't actually be needed, but must be
// unique types for function overloading to work
// properly.
typedef int communicator; // Must match petsc-nompi definition

typedef int info;

const unsigned int any_source=0;

#endif // TIMPI_HAVE_MPI

//-------------------------------------------------------------------
class Communicator
{
  // Basic operations:
public:

  Communicator ();

  /*
   * Constructor from MPI_Comm
   */
  explicit Communicator (const communicator & comm);

  /*
   * Don't use copy construction or assignment, just copy by reference
   * or pointer - it's too hard to keep a common used_tag_values if
   * each communicator is shared by more than one Communicator
   */
  Communicator (const Communicator &) = delete;
  Communicator & operator= (const Communicator &) = delete;

  /*
   * Move constructor and assignment operator
   */
  Communicator (Communicator &&) = default;
  Communicator & operator= (Communicator &&) = default;

  /*
   * NON-VIRTUAL destructor
   */
  ~Communicator ();

  /*
   * Create a new communicator between some subset of \p this,
   * based on specified "color"
   */
  void split(int color, int key, Communicator & target) const;

  /*
   * Create a new communicator between some subset of \p this,
   * based on specified split "type" (e.g. MPI_COMM_TYPE_SHARED)
   */
  void split_by_type(int split_type, int key, info i, Communicator & target) const;

  /*
   * Make \p this a new duplicate of the Communicator \p comm -
   * sharing the same processes but with a new communication context
   * and without sharing unique MessageTag assignment.
   */
  void duplicate(const Communicator & comm);

  /*
   * Make \p this a new duplicate of the MPI communicator \p comm -
   * sharing the same processes but with a new communication context.
   */
  void duplicate(const communicator & comm);

  communicator & get() { return _communicator; }

  const communicator & get() const { return _communicator; }

  MessageTag get_unique_tag(int tagvalue = MessageTag::invalid_tag) const;

  void reference_unique_tag(int tagvalue) const;

  void dereference_unique_tag(int tagvalue) const;

  void clear();

  Communicator & operator= (const communicator & comm);

  processor_id_type rank() const { return _rank; }

#ifdef TIMPI_HAVE_MPI
  processor_id_type size() const { return _size; }
#else
  // Help the compiler optimize in serial
  constexpr processor_id_type size() const { return 1; }
#endif

  enum SendMode { DEFAULT=0, SYNCHRONOUS };

  enum SyncType { NBX, ALLTOALL_COUNTS, SENDRECEIVE };


private:

  void assign(const communicator & comm);

  communicator  _communicator;
  processor_id_type _rank, _size;
  SendMode _send_mode;
  SyncType _sync_type;

  // mutable used_tag_values and tag_queue - not thread-safe, but then
  // TIMPI:: isn't thread-safe in general.
  mutable std::map<int, unsigned int> used_tag_values;
  mutable int _next_tag;

  int _max_tag;

  // Keep track of duplicate/split operations so we know when to free
  bool _I_duped_it;

  template <typename Map,
            typename std::enable_if<std::is_base_of<DataType, StandardType<typename Map::key_type>>::value &&
                                    std::is_base_of<DataType, StandardType<typename Map::mapped_type>>::value,
                                    int>::type = 0>
  void map_sum(Map & data) const;

  template <typename Map,
            typename std::enable_if<!(std::is_base_of<DataType, StandardType<typename Map::key_type>>::value &&
                                      std::is_base_of<DataType, StandardType<typename Map::mapped_type>>::value),
                                    int>::type = 0>
  void map_sum(Map & data) const;

  template <typename Map,
            typename std::enable_if<std::is_base_of<DataType, StandardType<typename Map::key_type>>::value &&
                                    std::is_base_of<DataType, StandardType<typename Map::mapped_type>>::value,
                                    int>::type = 0>
  void map_broadcast(Map & data,
                     const unsigned int root_id,
                     const bool identical_sizes) const;

  template <typename Map,
            typename std::enable_if<!(std::is_base_of<DataType, StandardType<typename Map::key_type>>::value &&
                                      std::is_base_of<DataType, StandardType<typename Map::mapped_type>>::value),
                                    int>::type = 0>
  void map_broadcast(Map & data,
                     const unsigned int root_id,
                     const bool identical_sizes) const;

  template <typename Map,
            typename std::enable_if<std::is_base_of<DataType, StandardType<typename Map::key_type>>::value &&
                                    std::is_base_of<DataType, StandardType<typename Map::mapped_type>>::value,
                                    int>::type = 0>
  void map_max(Map & data) const;

  template <typename Map,
            typename std::enable_if<!(std::is_base_of<DataType, StandardType<typename Map::key_type>>::value &&
                                      std::is_base_of<DataType, StandardType<typename Map::mapped_type>>::value),
                                    int>::type = 0>
  void map_max(Map & data) const;

  // Utility function for determining size for buffering of
  // vector<vector<T>> into vector<char> via MPI_Pack*
  template <typename T, typename A1, typename A2>
  std::size_t packed_size_of(const std::vector<std::vector<T,A1>,A2> & buf,
                             const DataType & type) const;

  // Communication operations:
public:

  void send_mode (const SendMode sm) { _send_mode = sm; }

  SendMode send_mode() const { return _send_mode; }

  void sync_type (const SyncType st) { _sync_type = st; }

  void sync_type (const std::string & st);

  SyncType sync_type() const { return _sync_type; }

  void barrier () const;

  void nonblocking_barrier (Request & req) const;

  template <typename T>
  timpi_pure
  inline
  bool verify(const T & r) const;

  template <typename T>
  timpi_pure
  inline
  bool semiverify(const T * r) const;

  template <typename T>
  inline
  void min(const T & r, T & o, Request & req) const;

  template <typename T>
  inline
  void min(T & r) const;

  template <typename T>
  inline
  void minloc(T & r,
              unsigned int & min_id) const;

  template <typename T, typename A1, typename A2>
  inline
  void minloc(std::vector<T,A1> & r,
              std::vector<unsigned int,A2> & min_id) const;

  template <typename T>
  inline
  void max(const T & r, T & o, Request & req) const;

  template <typename T>
  inline
  void max(T & r) const;

  template <typename T>
  inline
  void maxloc(T & r,
              unsigned int & max_id) const;

  template <typename T, typename A1, typename A2>
  inline
  void maxloc(std::vector<T,A1> & r,
              std::vector<unsigned int,A2> & max_id) const;

  template <typename T>
  inline
  void sum(T & r) const;

  template <typename T>
  inline
  void sum(const T & r, T & o, Request & req) const;

  template <typename T>
  inline
  void set_union(T & data, const unsigned int root_id) const;

  template <typename T>
  inline
  void set_union(T & data) const;

  status probe (const unsigned int src_processor_id,
                const MessageTag & tag=any_tag) const;

  template <typename T>
  inline
  Status packed_range_probe (const unsigned int src_processor_id,
                             const MessageTag & tag,
                             bool & flag) const;

  template <typename T>
  inline
  void send (const unsigned int dest_processor_id,
             const T & buf,
             const MessageTag & tag=no_tag) const;

  template <typename T>
  inline
  void send (const unsigned int dest_processor_id,
             const T & buf,
             Request & req,
             const MessageTag & tag=no_tag) const;

  template <typename T>
  inline
  void send (const unsigned int dest_processor_id,
             const T & buf,
             const DataType & type,
             const MessageTag & tag=no_tag) const;

  template <typename T>
  inline
  void send (const unsigned int dest_processor_id,
             const T & buf,
             const DataType & type,
             Request & req,
             const MessageTag & tag=no_tag) const;

  template <typename T>
  inline
  void send (const unsigned int dest_processor_id,
             const T & buf,
             const NotADataType & type,
             Request & req,
             const MessageTag & tag=no_tag) const;

  template <typename T>
  inline
  Status receive (const unsigned int dest_processor_id,
                  T & buf,
                  const MessageTag & tag=any_tag) const;

  template <typename T>
  inline
  void receive (const unsigned int dest_processor_id,
                T & buf,
                Request & req,
                const MessageTag & tag=any_tag) const;

  template <typename T>
  inline
  Status receive (const unsigned int dest_processor_id,
                  T & buf,
                  const DataType & type,
                  const MessageTag & tag=any_tag) const;

  template <typename T>
  inline
  void receive (const unsigned int dest_processor_id,
                T & buf,
                const DataType & type,
                Request & req,
                const MessageTag & tag=any_tag) const;

  template <typename T, typename A,
            typename std::enable_if<std::is_base_of<DataType, StandardType<T>>::value, int>::type = 0>
  inline
  bool possibly_receive (unsigned int & src_processor_id,
                         std::vector<T,A> & buf,
                         Request & req,
                         const MessageTag & tag) const;

  template <typename T, typename A,
            typename std::enable_if<Has_buffer_type<Packing<T>>::value, int>::type = 0>
  inline
  bool possibly_receive (unsigned int & src_processor_id,
                         std::vector<T,A> & buf,
                         Request & req,
                         const MessageTag & tag) const;


  template <typename T, typename A, typename std::enable_if<std::is_base_of<DataType, StandardType<T>>::value, int>::type = 0>
  inline
  bool possibly_receive (unsigned int & src_processor_id,
                         std::vector<T,A> & buf,
                         const DataType & type,
                         Request & req,
                         const MessageTag & tag) const;

  template <typename T, typename A,
            typename std::enable_if<Has_buffer_type<Packing<T>>::value, int>::type = 0>
  inline
  bool possibly_receive (unsigned int & src_processor_id,
                         std::vector<T,A> & buf,
                         const NotADataType & type,
                         Request & req,
                         const MessageTag & tag) const;

  template <typename Context, typename OutputIter, typename T>
  bool possibly_receive_packed_range (unsigned int & src_processor_id,
                                      Context * context,
                                      OutputIter out,
                                      const T * output_type,
                                      Request & req,
                                      const MessageTag & tag) const;

  template <typename Context, typename Iter>
  inline
  void send_packed_range (const unsigned int dest_processor_id,
                          const Context * context,
                          Iter range_begin,
                          const Iter range_end,
                          const MessageTag & tag=no_tag,
                          std::size_t approx_buffer_size = 1000000) const;

  template <typename Context, typename Iter>
  inline
  void send_packed_range (const unsigned int dest_processor_id,
                          const Context * context,
                          Iter range_begin,
                          const Iter range_end,
                          Request & req,
                          const MessageTag & tag=no_tag,
                          std::size_t approx_buffer_size = 1000000) const;

  template <typename Context, typename Iter>
  inline
  void nonblocking_send_packed_range (const unsigned int dest_processor_id,
                                      const Context * context,
                                      Iter range_begin,
                                      const Iter range_end,
                                      Request & req,
                                      const MessageTag & tag=no_tag) const;


  template <typename Context, typename Iter>
  inline
  void nonblocking_send_packed_range (const unsigned int dest_processor_id,
                                      const Context * context,
                                      Iter range_begin,
                                      const Iter range_end,
                                      Request & req,
                                      std::shared_ptr<std::vector<typename TIMPI::Packing<typename std::iterator_traits<Iter>::value_type>::buffer_type>> & buffer,
                                      const MessageTag & tag=no_tag) const;

  template <typename Context, typename OutputIter, typename T>
  inline
  void receive_packed_range (const unsigned int dest_processor_id,
                             Context * context,
                             OutputIter out,
                             const T * output_type, // used only to infer T
                             const MessageTag & tag=any_tag) const;

  template <typename Context, typename OutputIter, typename T>
  inline
  void nonblocking_receive_packed_range (const unsigned int src_processor_id,
                                         Context * context,
                                         OutputIter out,
                                         const T * output_type,
                                         Request & req,
                                         Status & stat,
                                         const MessageTag & tag=any_tag) const;

  template <typename Context, typename OutputIter, typename T>
  inline
  void nonblocking_receive_packed_range (const unsigned int src_processor_id,
                                         Context * context,
                                         OutputIter out,
                                         const T * output_type,
                                         Request & req,
                                         Status & stat,
                                         std::shared_ptr<std::vector<typename TIMPI::Packing<T>::buffer_type>> & buffer,
                                         const MessageTag & tag=any_tag
                                         ) const;

  template <typename T1, typename T2,
            typename std::enable_if<std::is_base_of<DataType, StandardType<T1>>::value &&
                                    std::is_base_of<DataType, StandardType<T2>>::value,
                                    int>::type = 0>
  inline
  void send_receive(const unsigned int dest_processor_id,
                    const T1 & send_data,
                    const unsigned int source_processor_id,
                    T2 & recv_data,
                    const MessageTag & send_tag = no_tag,
                    const MessageTag & recv_tag = any_tag) const;

  template <typename Context1, typename RangeIter, typename Context2,
            typename OutputIter, typename T>
  inline
  void send_receive_packed_range(const unsigned int dest_processor_id,
                                 const Context1 * context1,
                                 RangeIter send_begin,
                                 const RangeIter send_end,
                                 const unsigned int source_processor_id,
                                 Context2 * context2,
                                 OutputIter out,
                                 const T * output_type, // used only to infer T
                                 const MessageTag & send_tag = no_tag,
                                 const MessageTag & recv_tag = any_tag,
                                 std::size_t approx_buffer_size = 1000000) const;

  template <typename T1, typename T2>
  inline
  void send_receive(const unsigned int dest_processor_id,
                    const T1 & send_data,
                    const DataType & type1,
                    const unsigned int source_processor_id,
                    T2 & recv_data,
                    const DataType & type2,
                    const MessageTag & send_tag = no_tag,
                    const MessageTag & recv_tag = any_tag) const;

  template <typename T, typename A>
  inline void gather(const unsigned int root_id,
                     const T & send_data,
                     std::vector<T,A> & recv) const;

  template <typename T, typename A>
  inline void gather(const unsigned int root_id,
                     const std::basic_string<T> & send_data,
                     std::vector<std::basic_string<T>,A> & recv_data,
                     const bool identical_buffer_sizes=false) const;

  template <typename T, typename A,
            typename std::enable_if<std::is_base_of<DataType, StandardType<T>>::value, int>::type = 0>
  inline void gather(const unsigned int root_id,
                     std::vector<T,A> & r) const;

  template <typename T, typename A,
            typename std::enable_if<Has_buffer_type<Packing<T>>::value, int>::type = 0>
  inline void gather(const unsigned int root_id,
                     std::vector<T,A> & r) const;

  template <typename T, typename A, typename std::enable_if<std::is_base_of<DataType, StandardType<T>>::value,
                                                            int>::type = 0>
  inline void allgather(const T & send_data,
                        std::vector<T,A> & recv_data) const;

  template <typename T, typename A, typename std::enable_if<Has_buffer_type<Packing<T>>::value,
                                                            int>::type = 0>
  inline void allgather(const T & send_data,
                        std::vector<T,A> & recv_data) const;

  template <typename T, typename A>
  inline void allgather(const std::basic_string<T> & send_data,
                        std::vector<std::basic_string<T>,A> & recv_data,
                        const bool identical_buffer_sizes=false) const;

  template <typename T, typename A,
            typename std::enable_if<std::is_base_of<DataType, StandardType<T>>::value, int>::type = 0>
  inline void allgather(std::vector<T,A> & r,
                        const bool identical_buffer_sizes = false) const;

  template <typename T, typename A1, typename A2,
            typename std::enable_if<std::is_base_of<DataType, StandardType<T>>::value, int>::type = 0>
  inline void allgather(const std::vector<T,A1> & send_data,
                        std::vector<std::vector<T,A1>, A2> & recv_data,
                        const bool identical_buffer_sizes = false) const;

  template <typename T, typename A1, typename A2,
            typename std::enable_if<Has_buffer_type<Packing<T>>::value, int>::type = 0>
  inline void allgather(const std::vector<T,A1> & send_data,
                        std::vector<std::vector<T,A1>, A2> & recv_data,
                        const bool identical_buffer_sizes = false) const;

  template <typename T, typename A,
            typename std::enable_if<Has_buffer_type<Packing<T>>::value, int>::type = 0>
  inline void allgather(std::vector<T,A> & r,
                        const bool identical_buffer_sizes = false) const;

  template <typename T, typename A>
  inline void allgather(std::vector<std::basic_string<T>,A> & r,
                        const bool identical_buffer_sizes = false) const;

  //-------------------------------------------------------------------
  template <typename T, typename A>
  inline void scatter(const std::vector<T,A> & data,
                      T & recv,
                      const unsigned int root_id=0) const;

  template <typename T, typename A>
  inline void scatter(const std::vector<T,A> & data,
                      std::vector<T,A> & recv,
                      const unsigned int root_id=0) const;

  template <typename T, typename A1, typename A2>
  inline void scatter(const std::vector<T,A1> & data,
                      const std::vector<CountType,A2> counts,
                      std::vector<T,A1> & recv,
                      const unsigned int root_id=0) const;

#ifdef TIMPI_HAVE_MPI
#if MPI_VERSION > 3

  template <typename T, typename A1, typename A2>
  inline void scatter(const std::vector<T,A1> & data,
                      const std::vector<int,A2> counts,
                      std::vector<T,A1> & recv,
                      const unsigned int root_id=0) const;
#endif
#endif

  template <typename T, typename A1, typename A2>
  inline void scatter(const std::vector<std::vector<T,A1>,A2> & data,
                      std::vector<T,A1> & recv,
                      const unsigned int root_id=0,
                      const bool identical_buffer_sizes=false) const;

  //-------------------------------------------------------------------
  template <typename Context, typename Iter, typename OutputIter>
  inline void gather_packed_range (const unsigned int root_id,
                                   Context * context,
                                   Iter range_begin,
                                   const Iter range_end,
                                   OutputIter out,
                                   std::size_t approx_buffer_size = 1000000) const;

  template <typename Context, typename Iter, typename OutputIter>
  inline void allgather_packed_range (Context * context,
                                      Iter range_begin,
                                      const Iter range_end,
                                      OutputIter out,
                                      std::size_t approx_buffer_size = 1000000) const;

  template <typename T, typename A>
  inline void alltoall(std::vector<T,A> & r) const;

  template <typename T
#ifdef TIMPI_HAVE_MPI
            ,
            typename std::enable_if<std::is_base_of<DataType, StandardType<T>>::value, int>::type = 0
#endif
            >
  inline void broadcast(T & data, const unsigned int root_id=0,
                        const bool identical_sizes=false) const;

#ifdef TIMPI_HAVE_MPI

  template <typename T,
            typename std::enable_if<Has_buffer_type<Packing<T>>::value, int>::type = 0>
  inline void broadcast(T & data, const unsigned int root_id=0,
                        const bool identical_sizes=false) const;
#endif

  template <typename Context, typename OutputContext, typename Iter, typename OutputIter>
  inline void broadcast_packed_range (const Context * context1,
                                      Iter range_begin,
                                      const Iter range_end,
                                      OutputContext * context2,
                                      OutputIter out,
                                      const unsigned int root_id = 0,
                                      std::size_t approx_buffer_size = 1000000) const;

#include "timpi/parallel_communicator_specializations"

}; // class Communicator


} // namespace TIMPI


// Backwards compatibility for libMesh forward declarations
namespace libMesh {
namespace Parallel {
class Communicator : public TIMPI::Communicator {
public:
  Communicator () = default;
  explicit Communicator (const TIMPI::communicator & comm) : TIMPI::Communicator(comm) {}
  Communicator (const Communicator &) = delete;
  Communicator & operator= (const Communicator &) = delete;
  Communicator (Communicator &&) = default;
  Communicator & operator= (Communicator &&) = default;
  ~Communicator () = default;
};
}
}

#endif // TIMPI_COMMUNICATOR_H