]> https://gitweb.dealii.org/ - dealii.git/commitdiff
Switch the consensus algorithms interface to using function objects. 13272/head
authorWolfgang Bangerth <bangerth@colostate.edu>
Thu, 20 Jan 2022 15:57:14 +0000 (08:57 -0700)
committerWolfgang Bangerth <bangerth@colostate.edu>
Tue, 25 Jan 2022 02:40:14 +0000 (19:40 -0700)
include/deal.II/base/mpi_consensus_algorithms.h
include/deal.II/base/mpi_consensus_algorithms.templates.h
source/base/mpi_consensus_algorithms.cc

index 6768a6c06da5ca8e213e416bec1e3ac0642d1ca4..03f7edbb5703f47478f2f93f9d4938721f486614 100644 (file)
@@ -200,12 +200,24 @@ namespace Utilities
       class Interface
       {
       public:
+        /**
+         * Default constructor.
+         */
+        Interface();
+
         /**
          * Constructor. @p process is an object that provides information
          * about what processes the current process wants to communicate with,
          * and the data to be sent/received. @p comm is the communicator on
          * which this communication is to happen.
+         *
+         * @deprecated This constructor stores the Process object and the
+         *   communicator so that one can later call the run() function
+         *   without arguments. This approach is deprecated. Instead, use
+         *   the default constructor of this class along with the run()
+         *   function that takes an argument.
          */
+        DEAL_II_DEPRECATED
         Interface(Process<T1, T2> &process, const MPI_Comm &comm);
 
         /**
@@ -215,21 +227,82 @@ namespace Utilities
         virtual ~Interface() = default;
 
         /**
-         * Run the consensus algorithm and return the requesting processes.
+         * Run the consensus algorithm and return a vector of process ranks
+         * that have requested answers from the current process.
+         *
+         * @deprecated This function is deprecated. It can be called
+         *   if the Process object and communicator to be used have previously
+         *   been provided to the non-default constructor. Use the run()
+         *   functions taking arguments instead.
+         */
+        DEAL_II_DEPRECATED
+        std::vector<unsigned int>
+        run();
+
+        /**
+         * Run the consensus algorithm and return a vector of process ranks
+         * that have requested answers from the current process.
+         *
+         * This version of the run() function simply unpacks the functions
+         * packaged in `process` and calls the version of the run() function
+         * that takes a number of `std::function` arguments.
+         */
+        std::vector<unsigned int>
+        run(Process<T1, T2> &process, const MPI_Comm &comm);
+
+        /**
+         * Run the consensus algorithm and return a vector of process ranks
+         * that have requested answers from the current process.
+         *
+         * @param[in] targets A vector that contains the ranks of processes
+         *   to which requests should be sent and from which answers need
+         *   to be received.
+         * @param[in] create_request A function object that takes the rank
+         *   of a target process as argument and returns the message that
+         *   forms the request to this target.
+         * @param[in] answer_request A function that takes as arguments the
+         *   rank of the process that has sent a request to us, along with
+         *   the message of the request, and returns the message that forms
+         *   the answer that should be sent back to the requesting process.
+         * @param[in] process_answer A function object that takes as argument
+         *   the rank of a process from which we have received an answer
+         *   to a previously sent request, along with the message that
+         *   forms this answer. This function is used to describe what
+         *   the caller of the consensus algorithm wants to do with the
+         *   received answer.
+         * @param[in] comm The MPI communicator on which the whole algorithm
+         *   is to be performed.
          */
         virtual std::vector<unsigned int>
-        run() = 0;
+        run(const std::vector<unsigned int> &targets,
+            const std::function<std::vector<T1>(const unsigned int)>
+              &create_request,
+            const std::function<std::vector<T2>(const unsigned int,
+                                                const std::vector<T1> &)>
+              &                                                 answer_request,
+            const std::function<void(const unsigned int,
+                                     const std::vector<T2> &)> &process_answer,
+            const MPI_Comm &                                    comm) = 0;
 
-      protected:
+      private:
         /**
          * Reference to the process provided by the user.
+         *
+         * This member variable is only used in the deprecated constructor
+         * and the run() function without argument. It is a `nullptr`
+         * otherwise
          */
-        Process<T1, T2> &process;
+        DEAL_II_DEPRECATED
+        Process<T1, T2> *process;
 
         /**
          * MPI communicator.
+         *
+         * This member variable is only used in the deprecated constructor
+         * and the run() function without argument.
          */
-        const MPI_Comm &comm;
+        DEAL_II_DEPRECATED
+        MPI_Comm comm;
       };
 
 
@@ -250,12 +323,24 @@ namespace Utilities
       class NBX : public Interface<T1, T2>
       {
       public:
+        /**
+         * Default constructor.
+         */
+        NBX() = default;
+
         /**
          * Constructor.
          *
          * @param process Process to be run during consensus algorithm.
          * @param comm MPI Communicator
+         *
+         * @deprecated This constructor stores the Process object and the
+         *   communicator so that one can later call the run() function
+         *   without arguments. This approach is deprecated. Instead, use
+         *   the default constructor of this class along with the run()
+         *   function that takes an argument.
          */
+        DEAL_II_DEPRECATED
         NBX(Process<T1, T2> &process, const MPI_Comm &comm);
 
         /**
@@ -263,19 +348,25 @@ namespace Utilities
          */
         virtual ~NBX() = default;
 
+        // Import the declarations from the base class.
+        using Interface<T1, T2>::run;
+
         /**
          * @copydoc Interface::run()
          */
         virtual std::vector<unsigned int>
-        run() override;
+        run(const std::vector<unsigned int> &targets,
+            const std::function<std::vector<T1>(const unsigned int)>
+              &create_request,
+            const std::function<std::vector<T2>(const unsigned int,
+                                                const std::vector<T1> &)>
+              &                                                 answer_request,
+            const std::function<void(const unsigned int,
+                                     const std::vector<T2> &)> &process_answer,
+            const MPI_Comm &                                    comm) override;
 
       private:
 #ifdef DEAL_II_WITH_MPI
-        /**
-         * List of processes this process wants to send requests to.
-         */
-        std::vector<unsigned int> targets;
-
         /**
          * Buffers for sending requests.
          */
@@ -320,14 +411,17 @@ namespace Utilities
          * have been satisfied.
          */
         bool
-        all_locally_originated_receives_are_completed();
+        all_locally_originated_receives_are_completed(
+          const std::function<void(const unsigned int, const std::vector<T2> &)>
+            &             process_answer,
+          const MPI_Comm &comm);
 
         /**
          * Signal to all other ranks that this rank has received all request
          * answers via entering IBarrier.
          */
         void
-        signal_finish();
+        signal_finish(const MPI_Comm &comm);
 
         /**
          * Check whether all of the requests for answers that were created by
@@ -343,21 +437,29 @@ namespace Utilities
          * answer.
          */
         void
-        maybe_answer_one_request();
+        maybe_answer_one_request(
+          const std::function<std::vector<T2>(const unsigned int,
+                                              const std::vector<T1> &)>
+            &             answer_request,
+          const MPI_Comm &comm);
 
         /**
          * Start to send all requests via ISend and post IRecvs for the incoming
          * answer messages.
          */
         void
-        start_communication();
+        start_communication(
+          const std::vector<unsigned int> &targets,
+          const std::function<std::vector<T1>(const unsigned int)>
+            &             create_request,
+          const MPI_Comm &comm);
 
         /**
          * After all rank has received all answers, the MPI data structures can
          * be freed and the received answers can be processed.
          */
         void
-        clean_up_and_end_communication();
+        clean_up_and_end_communication(const MPI_Comm &comm);
       };
 
       /**
@@ -389,12 +491,25 @@ namespace Utilities
       class PEX : public Interface<T1, T2>
       {
       public:
+        /**
+         * Default constructor.
+         */
+        PEX() = default;
+
+
         /**
          * Constructor.
          *
          * @param process Process to be run during consensus algorithm.
          * @param comm MPI Communicator
+         *
+         * @deprecated This constructor stores the Process object and the
+         *   communicator so that one can later call the run() function
+         *   without arguments. This approach is deprecated. Instead, use
+         *   the default constructor of this class along with the run()
+         *   function that takes an argument.
          */
+        DEAL_II_DEPRECATED
         PEX(Process<T1, T2> &process, const MPI_Comm &comm);
 
         /**
@@ -402,21 +517,25 @@ namespace Utilities
          */
         virtual ~PEX() = default;
 
+        // Import the declarations from the base class.
+        using Interface<T1, T2>::run;
+
         /**
          * @copydoc Interface::run()
          */
         virtual std::vector<unsigned int>
-        run() override;
+        run(const std::vector<unsigned int> &targets,
+            const std::function<std::vector<T1>(const unsigned int)>
+              &create_request,
+            const std::function<std::vector<T2>(const unsigned int,
+                                                const std::vector<T1> &)>
+              &                                                 answer_request,
+            const std::function<void(const unsigned int,
+                                     const std::vector<T2> &)> &process_answer,
+            const MPI_Comm &                                    comm) override;
 
       private:
 #ifdef DEAL_II_WITH_MPI
-        /**
-         * List of ranks of processes this processes wants to send a request to.
-         */
-        std::vector<unsigned int> targets;
-
-        // data structures to send and receive requests
-
         /**
          * Buffers for sending requests.
          */
@@ -452,21 +571,34 @@ namespace Utilities
          * answer messages.
          */
         unsigned int
-        start_communication();
+        start_communication(
+          const std::vector<unsigned int> &targets,
+          const std::function<std::vector<T1>(const unsigned int)>
+            &             create_request,
+          const MPI_Comm &comm);
 
         /**
-         * The ith request message from another rank has been received: process
-         * the request and send an answer.
+         * The `index`th request message from another rank has been received:
+         * process the request and send an answer.
          */
         void
-        answer_one_request(const unsigned int index);
+        answer_one_request(
+          const unsigned int index,
+          const std::function<std::vector<T2>(const unsigned int,
+                                              const std::vector<T1> &)>
+            &             answer_request,
+          const MPI_Comm &comm);
 
         /**
          * Receive and process all of the incoming responses to the
          * requests we sent.
          */
         void
-        process_incoming_answers();
+        process_incoming_answers(
+          const unsigned int n_targets,
+          const std::function<void(const unsigned int, const std::vector<T2> &)>
+            &             process_answer,
+          const MPI_Comm &comm);
 
         /**
          * After all answers have been exchanged, the MPI data structures can be
@@ -486,19 +618,42 @@ namespace Utilities
       class Serial : public Interface<T1, T2>
       {
       public:
+        /**
+         * Default constructor.
+         */
+        Serial() = default;
+
         /**
          * Constructor.
          *
          * @param process Process to be run during consensus algorithm.
          * @param comm MPI Communicator (ignored)
+         *
+         * @deprecated This constructor stores the Process object and the
+         *   communicator so that one can later call the run() function
+         *   without arguments. This approach is deprecated. Instead, use
+         *   the default constructor of this class along with the run()
+         *   function that takes an argument.
          */
+        DEAL_II_DEPRECATED
         Serial(Process<T1, T2> &process, const MPI_Comm &comm);
 
+        // Import the declarations from the base class.
+        using Interface<T1, T2>::run;
+
         /**
          * @copydoc Interface::run()
          */
         virtual std::vector<unsigned int>
-        run() override;
+        run(const std::vector<unsigned int> &targets,
+            const std::function<std::vector<T1>(const unsigned int)>
+              &create_request,
+            const std::function<std::vector<T2>(const unsigned int,
+                                                const std::vector<T1> &)>
+              &                                                 answer_request,
+            const std::function<void(const unsigned int,
+                                     const std::vector<T2> &)> &process_answer,
+            const MPI_Comm &                                    comm) override;
       };
 
 
@@ -519,12 +674,24 @@ namespace Utilities
       class Selector : public Interface<T1, T2>
       {
       public:
+        /**
+         * Default constructor.
+         */
+        Selector() = default;
+
         /**
          * Constructor.
          *
          * @param process Process to be run during consensus algorithm.
          * @param comm MPI Communicator.
+         *
+         * @deprecated This constructor stores the Process object and the
+         *   communicator so that one can later call the run() function
+         *   without arguments. This approach is deprecated. Instead, use
+         *   the default constructor of this class along with the run()
+         *   function that takes an argument.
          */
+        DEAL_II_DEPRECATED
         Selector(Process<T1, T2> &process, const MPI_Comm &comm);
 
         /**
@@ -532,13 +699,24 @@ namespace Utilities
          */
         virtual ~Selector() = default;
 
+        // Import the declarations from the base class.
+        using Interface<T1, T2>::run;
+
         /**
          * @copydoc Interface::run()
          *
          * @note The function call is delegated to another ConsensusAlgorithms::Interface implementation.
          */
         virtual std::vector<unsigned int>
-        run() override;
+        run(const std::vector<unsigned int> &targets,
+            const std::function<std::vector<T1>(const unsigned int)>
+              &create_request,
+            const std::function<std::vector<T2>(const unsigned int,
+                                                const std::vector<T1> &)>
+              &                                                 answer_request,
+            const std::function<void(const unsigned int,
+                                     const std::vector<T2> &)> &process_answer,
+            const MPI_Comm &                                    comm) override;
 
       private:
         // Pointer to the actual ConsensusAlgorithms::Interface implementation.
index 174465c933fabf1cdd3f1f388162bfb8cb014eb9..d1b5b9766ef0711c86af90bf026cba4d0ad2ce3d 100644 (file)
@@ -104,12 +104,64 @@ namespace Utilities
       template <typename T1, typename T2>
       Interface<T1, T2>::Interface(Process<T1, T2> &process,
                                    const MPI_Comm & comm)
-        : process(process)
+        : process(&process)
         , comm(comm)
       {}
 
 
 
+      template <typename T1, typename T2>
+      Interface<T1, T2>::Interface()
+        : process(nullptr)
+        , comm(MPI_COMM_NULL)
+      {}
+
+
+
+      template <typename T1, typename T2>
+      std::vector<unsigned int>
+      Interface<T1, T2>::run()
+      {
+        Assert(process != nullptr,
+               ExcMessage("This function can only be called if the "
+                          "deprecated non-default constructor of this class "
+                          "has previously been called to set the Process "
+                          "object and a communicator."));
+        return run(*process, comm);
+      }
+
+
+
+      template <typename T1, typename T2>
+      std::vector<unsigned int>
+      Interface<T1, T2>::run(Process<T1, T2> &process, const MPI_Comm &comm)
+      {
+        // Unpack the 'process' object and call the function that takes
+        // function objects for all operations.
+        return run(
+          process.compute_targets(),
+          /* create_request: */
+          [&process](const unsigned int target) {
+            std::vector<T1> request;
+            process.create_request(target, request);
+            return request;
+          },
+          /* answer_request: */
+          [&process](const unsigned int     source,
+                     const std::vector<T1> &request) {
+            std::vector<T2> answer;
+            process.answer_request(source, request, answer);
+            return answer;
+          },
+          /* process_answer: */
+          [&process](const unsigned int target, const std::vector<T2> &answer) {
+            process.read_answer(target, answer);
+          },
+          comm);
+      }
+
+
+
       template <typename T1, typename T2>
       NBX<T1, T2>::NBX(Process<T1, T2> &process, const MPI_Comm &comm)
         : Interface<T1, T2>(process, comm)
@@ -119,14 +171,28 @@ namespace Utilities
 
       template <typename T1, typename T2>
       std::vector<unsigned int>
-      NBX<T1, T2>::run()
+      NBX<T1, T2>::run(
+        const std::vector<unsigned int> &targets,
+        const std::function<std::vector<T1>(const unsigned int)>
+          &create_request,
+        const std::function<std::vector<T2>(const unsigned int,
+                                            const std::vector<T1> &)>
+          &answer_request,
+        const std::function<void(const unsigned int, const std::vector<T2> &)>
+          &             process_answer,
+        const MPI_Comm &comm)
       {
+        Assert(has_unique_elements(targets),
+               ExcMessage("The consensus algorithms expect that each process "
+                          "only sends a single message to another process, "
+                          "but the targets provided include duplicates."));
+
         static CollectiveMutex      mutex;
-        CollectiveMutex::ScopedLock lock(mutex, this->comm);
+        CollectiveMutex::ScopedLock lock(mutex, comm);
 
         // 1) Send data to identified targets and start receiving
         //    the answers from these very same processes.
-        start_communication();
+        start_communication(targets, create_request, comm);
 
         // 2) Until all posted receive operations are known to have completed,
         //    answer requests and keep checking whether all requests of
@@ -142,21 +208,22 @@ namespace Utilities
         //    That's ok: We will get around to dealing with all remaining
         //    message later. We just want to move on to the next step
         //    as early as possible.
-        while (all_locally_originated_receives_are_completed() == false)
-          maybe_answer_one_request();
+        while (all_locally_originated_receives_are_completed(process_answer,
+                                                             comm) == false)
+          maybe_answer_one_request(answer_request, comm);
 
         // 3) Signal to all other processes that all requests of this process
         //    have been answered
-        signal_finish();
+        signal_finish(comm);
 
         // 4) Nevertheless, this process has to keep on answering (potential)
         //    incoming requests until all processes have received the
         //    answer to all requests
         while (all_remotely_originated_receives_are_completed() == false)
-          maybe_answer_one_request();
+          maybe_answer_one_request(answer_request, comm);
 
         // 5) process the answer to all requests
-        clean_up_and_end_communication();
+        clean_up_and_end_communication(comm);
 
         return std::vector<unsigned int>(requesting_processes.begin(),
                                          requesting_processes.end());
@@ -166,16 +233,14 @@ namespace Utilities
 
       template <typename T1, typename T2>
       void
-      NBX<T1, T2>::start_communication()
+      NBX<T1, T2>::start_communication(
+        const std::vector<unsigned int> &targets,
+        const std::function<std::vector<T1>(const unsigned int)>
+          &             create_request,
+        const MPI_Comm &comm)
       {
 #ifdef DEAL_II_WITH_MPI
         // 1)
-        targets = this->process.compute_targets();
-        Assert(has_unique_elements(targets),
-               ExcMessage("The consensus algorithms expect that each process "
-                          "only sends a single message to another process, "
-                          "but the targets provided include duplicates."));
-
         const auto n_targets = targets.size();
 
         const int tag_request = Utilities::MPI::internal::Tags::
@@ -190,11 +255,10 @@ namespace Utilities
           for (unsigned int index = 0; index < n_targets; ++index)
             {
               const unsigned int rank = targets[index];
-              AssertIndexRange(rank,
-                               Utilities::MPI::n_mpi_processes(this->comm));
+              AssertIndexRange(rank, Utilities::MPI::n_mpi_processes(comm));
 
               auto &send_buffer = send_buffers[index];
-              this->process.create_request(rank, send_buffer);
+              send_buffer       = create_request(rank);
 
               // Post a request to send data
               auto ierr = MPI_Isend(send_buffer.data(),
@@ -202,7 +266,7 @@ namespace Utilities
                                     MPI_BYTE,
                                     rank,
                                     tag_request,
-                                    this->comm,
+                                    comm,
                                     &send_requests[index]);
               AssertThrowMPI(ierr);
             }
@@ -211,6 +275,10 @@ namespace Utilities
           // a request to:
           n_outstanding_answers = n_targets;
         }
+#else
+        (void)targets;
+        (void)create_request;
+        (void)comm;
 #endif
       }
 
@@ -218,7 +286,10 @@ namespace Utilities
 
       template <typename T1, typename T2>
       bool
-      NBX<T1, T2>::all_locally_originated_receives_are_completed()
+      NBX<T1, T2>::all_locally_originated_receives_are_completed(
+        const std::function<void(const unsigned int, const std::vector<T2> &)>
+          &             process_answer,
+        const MPI_Comm &comm)
       {
 #ifdef DEAL_II_WITH_MPI
         // We know that all requests have come in when we have pending
@@ -237,11 +308,8 @@ namespace Utilities
 
             int        request_is_pending;
             MPI_Status status;
-            const auto ierr = MPI_Iprobe(MPI_ANY_SOURCE,
-                                         tag_deliver,
-                                         this->comm,
-                                         &request_is_pending,
-                                         &status);
+            const auto ierr = MPI_Iprobe(
+              MPI_ANY_SOURCE, tag_deliver, comm, &request_is_pending, &status);
             AssertThrowMPI(ierr);
 
             // If there is no pending message with this tag,
@@ -276,12 +344,12 @@ namespace Utilities
                                             MPI_BYTE,
                                             target,
                                             tag_deliver,
-                                            this->comm,
+                                            comm,
                                             MPI_STATUS_IGNORE);
                   AssertThrowMPI(ierr);
                 }
 
-                this->process.read_answer(target, recv_buffer);
+                process_answer(target, recv_buffer);
 
                 // Finally, remove this rank from the list of outstanding
                 // targets:
@@ -300,6 +368,9 @@ namespace Utilities
           }
 
 #else
+        (void)process_answer;
+        (void)comm;
+
         return true;
 #endif
       }
@@ -308,7 +379,11 @@ namespace Utilities
 
       template <typename T1, typename T2>
       void
-      NBX<T1, T2>::maybe_answer_one_request()
+      NBX<T1, T2>::maybe_answer_one_request(
+        const std::function<std::vector<T2>(const unsigned int,
+                                            const std::vector<T1> &)>
+          &             answer_request,
+        const MPI_Comm &comm)
       {
 #ifdef DEAL_II_WITH_MPI
 
@@ -326,11 +401,8 @@ namespace Utilities
         // only answer one.
         MPI_Status status;
         int        request_is_pending;
-        const auto ierr = MPI_Iprobe(MPI_ANY_SOURCE,
-                                     tag_request,
-                                     this->comm,
-                                     &request_is_pending,
-                                     &status);
+        const auto ierr = MPI_Iprobe(
+          MPI_ANY_SOURCE, tag_request, comm, &request_is_pending, &status);
         AssertThrowMPI(ierr);
 
         if (request_is_pending != 0)
@@ -357,7 +429,7 @@ namespace Utilities
                             MPI_BYTE,
                             other_rank,
                             tag_request,
-                            this->comm,
+                            comm,
                             MPI_STATUS_IGNORE);
             AssertThrowMPI(ierr);
 
@@ -365,9 +437,7 @@ namespace Utilities
             // and ask the 'process' object to produce an answer:
             request_buffers.emplace_back(std::make_unique<std::vector<T2>>());
             auto &request_buffer = *request_buffers.back();
-            this->process.answer_request(other_rank,
-                                         buffer_recv,
-                                         request_buffer);
+            request_buffer       = answer_request(other_rank, buffer_recv);
 
             // Then initiate sending the answer back to the requester.
             request_requests.emplace_back(std::make_unique<MPI_Request>());
@@ -376,10 +446,13 @@ namespace Utilities
                              MPI_BYTE,
                              other_rank,
                              tag_deliver,
-                             this->comm,
+                             comm,
                              request_requests.back().get());
             AssertThrowMPI(ierr);
           }
+#else
+        (void)answer_request;
+        (void)comm;
 #endif
       }
 
@@ -387,18 +460,20 @@ namespace Utilities
 
       template <typename T1, typename T2>
       void
-      NBX<T1, T2>::signal_finish()
+      NBX<T1, T2>::signal_finish(const MPI_Comm &comm)
       {
 #ifdef DEAL_II_WITH_MPI
 #  if DEAL_II_MPI_VERSION_GTE(3, 0)
-        const auto ierr = MPI_Ibarrier(this->comm, &barrier_request);
+        const auto ierr = MPI_Ibarrier(comm, &barrier_request);
         AssertThrowMPI(ierr);
 #  else
-        AssertThrow(
-          false,
-          ExcMessage(
-            "ConsensusAlgorithms::NBX uses MPI 3.0 features. You should compile with at least MPI 3.0."));
+        AssertThrow(false,
+                    ExcMessage(
+                      "ConsensusAlgorithms::NBX uses MPI 3.0 features. "
+                      "You should compile with at least MPI 3.0."));
 #  endif
+#else
+        (void)comm;
 #endif
       }
 
@@ -424,7 +499,7 @@ namespace Utilities
 
       template <typename T1, typename T2>
       void
-      NBX<T1, T2>::clean_up_and_end_communication()
+      NBX<T1, T2>::clean_up_and_end_communication(const MPI_Comm &comm)
       {
 #ifdef DEAL_II_WITH_MPI
         // clean up
@@ -449,10 +524,12 @@ namespace Utilities
 #  ifdef DEBUG
           // note: IBarrier seems to make problem during testing, this
           // additional Barrier seems to help
-          ierr = MPI_Barrier(this->comm);
+          ierr = MPI_Barrier(comm);
           AssertThrowMPI(ierr);
 #  endif
         }
+#else
+        (void)comm;
 #endif
       }
 
@@ -467,22 +544,37 @@ namespace Utilities
 
       template <typename T1, typename T2>
       std::vector<unsigned int>
-      PEX<T1, T2>::run()
+      PEX<T1, T2>::run(
+        const std::vector<unsigned int> &targets,
+        const std::function<std::vector<T1>(const unsigned int)>
+          &create_request,
+        const std::function<std::vector<T2>(const unsigned int,
+                                            const std::vector<T1> &)>
+          &answer_request,
+        const std::function<void(const unsigned int, const std::vector<T2> &)>
+          &             process_answer,
+        const MPI_Comm &comm)
       {
+        Assert(has_unique_elements(targets),
+               ExcMessage("The consensus algorithms expect that each process "
+                          "only sends a single message to another process, "
+                          "but the targets provided include duplicates."));
+
         static CollectiveMutex      mutex;
-        CollectiveMutex::ScopedLock lock(mutex, this->comm);
+        CollectiveMutex::ScopedLock lock(mutex, comm);
 
         // 1) Send requests and start receiving the answers.
         //    In particular, determine how many requests we should expect
         //    on the current process.
-        const unsigned int n_requests = start_communication();
+        const unsigned int n_requests =
+          start_communication(targets, create_request, comm);
 
         // 2) Answer requests:
         for (unsigned int request = 0; request < n_requests; ++request)
-          answer_one_request(request);
+          answer_one_request(request, answer_request, comm);
 
         // 3) Process answers:
-        process_incoming_answers();
+        process_incoming_answers(targets.size(), process_answer, comm);
 
         // 4) Make sure all sends have successfully terminated:
         clean_up_and_end_communication();
@@ -495,7 +587,11 @@ namespace Utilities
 
       template <typename T1, typename T2>
       unsigned int
-      PEX<T1, T2>::start_communication()
+      PEX<T1, T2>::start_communication(
+        const std::vector<unsigned int> &targets,
+        const std::function<std::vector<T1>(const unsigned int)>
+          &             create_request,
+        const MPI_Comm &comm)
       {
 #ifdef DEAL_II_WITH_MPI
         const int tag_request = Utilities::MPI::internal::Tags::
@@ -503,16 +599,11 @@ namespace Utilities
 
         // 1) determine with which processes this process wants to communicate
         // with
-        targets = this->process.compute_targets();
-        Assert(has_unique_elements(targets),
-               ExcMessage("The consensus algorithms expect that each process "
-                          "only sends a single message to another process, "
-                          "but the targets provided include duplicates."));
         const unsigned int n_targets = targets.size();
 
         // 2) determine who wants to communicate with this process
         const unsigned int n_sources =
-          compute_n_point_to_point_communications(this->comm, targets);
+          compute_n_point_to_point_communications(comm, targets);
 
         // 2) allocate memory
         recv_buffers.resize(n_targets);
@@ -526,11 +617,11 @@ namespace Utilities
         for (unsigned int i = 0; i < n_targets; ++i)
           {
             const unsigned int rank = targets[i];
-            AssertIndexRange(rank, Utilities::MPI::n_mpi_processes(this->comm));
+            AssertIndexRange(rank, Utilities::MPI::n_mpi_processes(comm));
 
             // pack data which should be sent
             auto &send_buffer = send_buffers[i];
-            this->process.create_request(rank, send_buffer);
+            send_buffer       = create_request(rank);
 
             // start to send data
             auto ierr = MPI_Isend(send_buffer.data(),
@@ -538,13 +629,16 @@ namespace Utilities
                                   MPI_BYTE,
                                   rank,
                                   tag_request,
-                                  this->comm,
+                                  comm,
                                   &send_request_requests[i]);
             AssertThrowMPI(ierr);
           }
 
         return n_sources;
 #else
+        (void)targets;
+        (void)create_request;
+        (void)comm;
         return 0;
 #endif
       }
@@ -553,7 +647,12 @@ namespace Utilities
 
       template <typename T1, typename T2>
       void
-      PEX<T1, T2>::answer_one_request(const unsigned int index)
+      PEX<T1, T2>::answer_one_request(
+        const unsigned int index,
+        const std::function<std::vector<T2>(const unsigned int,
+                                            const std::vector<T1> &)>
+          &             answer_request,
+        const MPI_Comm &comm)
       {
 #ifdef DEAL_II_WITH_MPI
         const int tag_request = Utilities::MPI::internal::Tags::
@@ -564,7 +663,7 @@ namespace Utilities
         // Wait until we have a message ready for retrieval, though we don't
         // care which process it is from.
         MPI_Status status;
-        int ierr = MPI_Probe(MPI_ANY_SOURCE, tag_request, this->comm, &status);
+        int        ierr = MPI_Probe(MPI_ANY_SOURCE, tag_request, comm, &status);
         AssertThrowMPI(ierr);
 
         // Get rank of incoming message and verify that it makes sense
@@ -592,24 +691,26 @@ namespace Utilities
                         MPI_BYTE,
                         other_rank,
                         tag_request,
-                        this->comm,
+                        comm,
                         &status);
         AssertThrowMPI(ierr);
 
         // Process request by asking the user-provided function for
         // the answer and post a send for it.
         auto &request_buffer = requests_buffers[index];
-        this->process.answer_request(other_rank, buffer_recv, request_buffer);
+        request_buffer       = answer_request(other_rank, buffer_recv);
 
         ierr = MPI_Isend(request_buffer.data(),
                          request_buffer.size() * sizeof(T2),
                          MPI_BYTE,
                          other_rank,
                          tag_deliver,
-                         this->comm,
+                         comm,
                          &send_answer_requests[index]);
         AssertThrowMPI(ierr);
 #else
+        (void)answer_request;
+        (void)comm;
         (void)index;
 #endif
       }
@@ -618,7 +719,11 @@ namespace Utilities
 
       template <typename T1, typename T2>
       void
-      PEX<T1, T2>::process_incoming_answers()
+      PEX<T1, T2>::process_incoming_answers(
+        const unsigned int n_targets,
+        const std::function<void(const unsigned int, const std::vector<T2> &)>
+          &             process_answer,
+        const MPI_Comm &comm)
       {
 #ifdef DEAL_II_WITH_MPI
         const int tag_deliver = Utilities::MPI::internal::Tags::
@@ -629,12 +734,12 @@ namespace Utilities
         // we need not process them in order -- rather, just see what
         // comes in and then look at message originators' ranks and
         // message sizes
-        for (unsigned int i = 0; i < targets.size(); ++i)
+        for (unsigned int i = 0; i < n_targets; ++i)
           {
             MPI_Status status;
             {
               const int ierr =
-                MPI_Probe(MPI_ANY_SOURCE, tag_deliver, this->comm, &status);
+                MPI_Probe(MPI_ANY_SOURCE, tag_deliver, comm, &status);
               AssertThrowMPI(ierr);
             }
 
@@ -656,13 +761,17 @@ namespace Utilities
                                         MPI_BYTE,
                                         other_rank,
                                         tag_deliver,
-                                        this->comm,
+                                        comm,
                                         MPI_STATUS_IGNORE);
               AssertThrowMPI(ierr);
             }
 
-            this->process.read_answer(other_rank, recv_buffer);
+            process_answer(other_rank, recv_buffer);
           }
+#else
+        (void)n_targets;
+        (void)process_answer;
+        (void)comm;
 #endif
       }
 
@@ -705,9 +814,23 @@ namespace Utilities
 
       template <typename T1, typename T2>
       std::vector<unsigned int>
-      Serial<T1, T2>::run()
+      Serial<T1, T2>::run(
+        const std::vector<unsigned int> &targets,
+        const std::function<std::vector<T1>(const unsigned int)>
+          &create_request,
+        const std::function<std::vector<T2>(const unsigned int,
+                                            const std::vector<T1> &)>
+          &answer_request,
+        const std::function<void(const unsigned int, const std::vector<T2> &)>
+          &             process_answer,
+        const MPI_Comm &comm)
       {
-        const auto targets = this->process.compute_targets();
+        (void)comm;
+        Assert((Utilities::MPI::job_supports_mpi() == false) ||
+                 (Utilities::MPI::n_mpi_processes(comm) == 1),
+               ExcMessage("You shouldn't use the 'Serial' class on "
+                          "communicators that have more than one process "
+                          "associated with it."));
 
         // The only valid target for a serial program is itself.
         if (targets.size() != 0)
@@ -722,14 +845,9 @@ namespace Utilities
             // Since the caller indicates that there is a target, and since we
             // know that it is the current process, let the process send
             // something to itself.
-            std::vector<T1> send_buffer;
-            std::vector<T2> recv_buffer;
-            std::vector<T2> request_buffer;
-
-            this->process.create_request(0, send_buffer);
-            this->process.answer_request(0, send_buffer, request_buffer);
-            recv_buffer = request_buffer;
-            this->process.read_answer(0, recv_buffer);
+            const std::vector<T1> request = create_request(0);
+            const std::vector<T2> answer  = answer_request(0, request);
+            process_answer(0, answer);
           }
 
         return targets; // nothing to do
@@ -740,6 +858,22 @@ namespace Utilities
       template <typename T1, typename T2>
       Selector<T1, T2>::Selector(Process<T1, T2> &process, const MPI_Comm &comm)
         : Interface<T1, T2>(process, comm)
+      {}
+
+
+
+      template <typename T1, typename T2>
+      std::vector<unsigned int>
+      Selector<T1, T2>::run(
+        const std::vector<unsigned int> &targets,
+        const std::function<std::vector<T1>(const unsigned int)>
+          &create_request,
+        const std::function<std::vector<T2>(const unsigned int,
+                                            const std::vector<T1> &)>
+          &answer_request,
+        const std::function<void(const unsigned int, const std::vector<T2> &)>
+          &             process_answer,
+        const MPI_Comm &comm)
       {
         // Depending on the number of processes we switch between
         // implementations. We reduce the threshold for debug mode to be
@@ -757,23 +891,17 @@ namespace Utilities
 #    else
         if (n_procs > 99)
 #    endif
-          consensus_algo.reset(new NBX<T1, T2>(process, comm));
+          consensus_algo.reset(new NBX<T1, T2>());
         else
 #  endif
 #endif
           if (n_procs > 1)
-          consensus_algo.reset(new PEX<T1, T2>(process, comm));
+          consensus_algo.reset(new PEX<T1, T2>());
         else
-          consensus_algo.reset(new Serial<T1, T2>(process, comm));
-      }
-
-
+          consensus_algo.reset(new Serial<T1, T2>());
 
-      template <typename T1, typename T2>
-      std::vector<unsigned int>
-      Selector<T1, T2>::run()
-      {
-        return consensus_algo->run();
+        return consensus_algo->run(
+          targets, create_request, answer_request, process_answer, comm);
       }
 
 
index 98ec56474fda78bb00ee0909da71da9f87d7c31b..c03d41141a20bf36729036726781eb4028c03d43 100644 (file)
@@ -25,6 +25,8 @@ namespace Utilities
     {
       template class Process<unsigned int, unsigned int>;
 
+      template class Interface<unsigned int, unsigned int>;
+
       template class NBX<unsigned int, unsigned int>;
 
       template class PEX<unsigned int, unsigned int>;
@@ -38,6 +40,10 @@ namespace Utilities
         std::pair<types::global_dof_index, types::global_dof_index>,
         unsigned int>;
 
+      template class Interface<
+        std::pair<types::global_dof_index, types::global_dof_index>,
+        unsigned int>;
+
       template class Selector<
         std::pair<types::global_dof_index, types::global_dof_index>,
         unsigned int>;
@@ -57,6 +63,8 @@ namespace Utilities
 #ifdef DEAL_II_WITH_64BIT_INDICES
       template class Process<types::global_dof_index, unsigned int>;
 
+      template class Interface<types::global_dof_index, unsigned int>;
+
       template class NBX<types::global_dof_index, unsigned int>;
 
       template class Serial<types::global_dof_index, unsigned int>;
@@ -68,6 +76,8 @@ namespace Utilities
 
       template class Process<char, char>;
 
+      template class Interface<char, char>;
+
       template class NBX<char, char>;
 
       template class PEX<char, char>;

In the beginning the Universe was created. This has made a lot of people very angry and has been widely regarded as a bad move.

Douglas Adams


Typeset in Trocchi and Trocchi Bold Sans Serif.