mpi_consensus_algorithms.h

Go to the documentation of this file.

// ---------------------------------------------------------------------
//
// Copyright (C) 2020 - 2021 by the deal.II authors
//
// This file is part of the deal.II library.
//
// The deal.II library is free software; you can use it, 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.
// The full text of the license can be found in the file LICENSE.md at
// the top level directory of deal.II.
//
// ---------------------------------------------------------------------
 
#ifndef dealii_mpi_consensus_algorithm_h
#define dealii_mpi_consensus_algorithm_h
 
#include <deal.II/base/config.h>
 
#include <deal.II/base/mpi.h>
#include <deal.II/base/mpi.templates.h>
 
DEAL_II_NAMESPACE_OPEN
 
 
namespace Utilities
{
  namespace MPI
  {
    namespace ConsensusAlgorithms
    {
      template <typename T1, typename T2>
      class Process
      {
      public:
        virtual ~Process() = default;
 
        virtual std::vector<unsigned int>
        compute_targets() = 0;
 
        virtual void
        create_request(const unsigned int other_rank,
                       std::vector<T1> &  send_buffer);
 
        virtual void
        prepare_buffer_for_answer(const unsigned int other_rank,
                                  std::vector<T2> &  recv_buffer);
 
        virtual void
        answer_request(const unsigned int     other_rank,
                       const std::vector<T1> &buffer_recv,
                       std::vector<T2> &      request_buffer);
 
        virtual void
        read_answer(const unsigned int     other_rank,
                    const std::vector<T2> &recv_buffer);
      };
 
 
 
      template <typename T1, typename T2>
      class Interface
      {
      public:
        Interface(Process<T1, T2> &process, const MPI_Comm &comm);
 
        virtual ~Interface() = default;
 
        virtual std::vector<unsigned int>
        run() = 0;
 
      protected:
        Process<T1, T2> &process;
 
        const MPI_Comm &comm;
 
        const bool job_supports_mpi;
 
        const unsigned int my_rank;
 
        const unsigned int n_procs;
      };
 
 
      template <typename T1, typename T2>
      class NBX : public Interface<T1, T2>
      {
      public:
        NBX(Process<T1, T2> &process, const MPI_Comm &comm);
 
        virtual ~NBX() = default;
 
        virtual std::vector<unsigned int>
        run() override;
 
      private:
#ifdef DEAL_II_WITH_MPI
        std::vector<unsigned int> targets;
 
        std::vector<std::vector<T1>> send_buffers;
 
        std::vector<MPI_Request> send_requests;
 
        std::vector<std::vector<T2>> recv_buffers;
 
 
        std::vector<MPI_Request> recv_requests;
 
        std::vector<std::unique_ptr<std::vector<T2>>> request_buffers;
 
        std::vector<std::unique_ptr<MPI_Request>> request_requests;
 
        // request for barrier
        MPI_Request barrier_request;
#endif
 
        std::set<unsigned int> requesting_processes;
 
        bool
        check_own_state();
 
        void
        signal_finish();
 
        bool
        check_global_state();
 
        void
        answer_requests();
 
        void
        start_communication();
 
        void
        clean_up_and_end_communication();
      };
 
      template <typename T1, typename T2>
      class PEX : public Interface<T1, T2>
      {
      public:
        PEX(Process<T1, T2> &process, const MPI_Comm &comm);
 
        virtual ~PEX() = default;
 
        virtual std::vector<unsigned int>
        run() override;
 
      private:
#ifdef DEAL_II_WITH_MPI
        std::vector<unsigned int> targets;
 
        std::vector<unsigned int> sources;
 
        // data structures to send and receive requests
 
        std::vector<std::vector<T1>> send_buffers;
 
        std::vector<std::vector<T2>> recv_buffers;
 
        std::vector<MPI_Request> send_and_recv_buffers;
 
        std::vector<std::vector<T2>> requests_buffers;
 
        std::vector<MPI_Request> requests_answers;
#endif
        std::set<unsigned int> requesting_processes;
 
        void
        answer_requests(int index);
 
        unsigned int
        start_communication();
 
        void
        clean_up_and_end_communication();
      };
 
      template <typename T1, typename T2>
      class Serial : public Interface<T1, T2>
      {
      public:
        Serial(Process<T1, T2> &process, const MPI_Comm &comm);
 
        virtual std::vector<unsigned int>
        run() override;
      };
 
      template <typename T1, typename T2>
      class Selector : public Interface<T1, T2>
      {
      public:
        Selector(Process<T1, T2> &process, const MPI_Comm &comm);
 
        virtual ~Selector() = default;
 
        virtual std::vector<unsigned int>
        run() override;
 
      private:
        // Pointer to the actual ConsensusAlgorithms::Interface implementation.
        std::shared_ptr<Interface<T1, T2>> consensus_algo;
      };
 
      template <typename T1, typename T2>
      class AnonymousProcess : public Process<T1, T2>
      {
      public:
        AnonymousProcess(
          const std::function<std::vector<unsigned int>()>
            &function_compute_targets,
          const std::function<void(const unsigned int, std::vector<T1> &)>
            &function_create_request =
              [](const unsigned int, std::vector<T1> &) {},
          const std::function<void(const unsigned int,
                                   const std::vector<T1> &,
                                   std::vector<T2> &)>
            &function_answer_request = [](const unsigned int,
                                          const std::vector<T1> &,
                                          std::vector<T2> &) {},
          const std::function<void(const unsigned int, std::vector<T2> &)>
            &function_prepare_buffer_for_answer =
              [](const unsigned int, std::vector<T2> &) {},
          const std::function<void(const unsigned int, const std::vector<T2> &)>
            &function_read_answer =
              [](const unsigned int, const std::vector<T2> &) {});
 
        std::vector<unsigned int>
        compute_targets() override;
 
        void
        create_request(const unsigned int other_rank,
                       std::vector<T1> &  send_buffer) override;
 
        void
        answer_request(const unsigned int     other_rank,
                       const std::vector<T1> &buffer_recv,
                       std::vector<T2> &      request_buffer) override;
 
        void
        prepare_buffer_for_answer(const unsigned int other_rank,
                                  std::vector<T2> &  recv_buffer) override;
 
        void
        read_answer(const unsigned int     other_rank,
                    const std::vector<T2> &recv_buffer) override;
 
      private:
        const std::function<std::vector<unsigned int>()>
          function_compute_targets;
        const std::function<void(const int, std::vector<T1> &)>
          function_create_request;
        const std::function<
          void(const unsigned int, const std::vector<T1> &, std::vector<T2> &)>
          function_answer_request;
        const std::function<void(const int, std::vector<T2> &)>
          function_prepare_buffer_for_answer;
        const std::function<void(const int, const std::vector<T2> &)>
          function_read_answer;
      };
 
 
 
      template <typename T1, typename T2>
      AnonymousProcess<T1, T2>::AnonymousProcess(
        const std::function<std::vector<unsigned int>()>
          &function_compute_targets,
        const std::function<void(const unsigned int, std::vector<T1> &)>
          &                                           function_create_request,
        const std::function<void(const unsigned int,
                                 const std::vector<T1> &,
                                 std::vector<T2> &)> &function_answer_request,
        const std::function<void(const unsigned int, std::vector<T2> &)>
          &function_prepare_buffer_for_answer,
        const std::function<void(const unsigned int, const std::vector<T2> &)>
          &function_read_answer)
        : function_compute_targets(function_compute_targets)
        , function_create_request(function_create_request)
        , function_answer_request(function_answer_request)
        , function_prepare_buffer_for_answer(function_prepare_buffer_for_answer)
        , function_read_answer(function_read_answer)
      {}
 
 
 
      template <typename T1, typename T2>
      std::vector<unsigned int>
      AnonymousProcess<T1, T2>::compute_targets()
      {
        return function_compute_targets();
      }
 
 
 
      template <typename T1, typename T2>
      void
      AnonymousProcess<T1, T2>::create_request(const unsigned int other_rank,
                                               std::vector<T1> &  send_buffer)
      {
        function_create_request(other_rank, send_buffer);
      }
 
 
 
      template <typename T1, typename T2>
      void
      AnonymousProcess<T1, T2>::answer_request(
        const unsigned int     other_rank,
        const std::vector<T1> &buffer_recv,
        std::vector<T2> &      request_buffer)
      {
        function_answer_request(other_rank, buffer_recv, request_buffer);
      }
 
 
 
      template <typename T1, typename T2>
      void
      AnonymousProcess<T1, T2>::prepare_buffer_for_answer(
        const unsigned int other_rank,
        std::vector<T2> &  recv_buffer)
      {
        function_prepare_buffer_for_answer(other_rank, recv_buffer);
      }
 
 
 
      template <typename T1, typename T2>
      void
      AnonymousProcess<T1, T2>::read_answer(const unsigned int     other_rank,
                                            const std::vector<T2> &recv_buffer)
      {
        function_read_answer(other_rank, recv_buffer);
      }
 
 
 
    } // namespace ConsensusAlgorithms
  }   // end of namespace MPI
} // end of namespace Utilities
 
 
DEAL_II_NAMESPACE_CLOSE
 
#endif