From: Peter Munch Date: Tue, 3 Mar 2020 10:10:10 +0000 (+0100) Subject: Move ConsensusAlgorithm related functions to mpi_consensus_algorithm.h X-Git-Tag: v9.2.0-rc1~458^2 X-Git-Url: https://gitweb.dealii.org/cgi-bin/gitweb.cgi?a=commitdiff_plain;h=refs%2Fpull%2F9600%2Fhead;p=dealii.git Move ConsensusAlgorithm related functions to mpi_consensus_algorithm.h --- diff --git a/include/deal.II/base/mpi.h b/include/deal.II/base/mpi.h index ecb6ff11da..6431e56ea1 100644 --- a/include/deal.II/base/mpi.h +++ b/include/deal.II/base/mpi.h @@ -975,461 +975,6 @@ namespace Utilities const T & object_to_send, const unsigned int root_process = 0); - /** - * An interface to be able to use the ConsensusAlgorithm classes. The main - * functionality of the implementations is to return a list of process ranks - * this process wants data from and to deal with the optional payload of the - * messages sent/received by the ConsensusAlgorithm classes. - * - * There are two kinds of messages: - * - send/request message: A message consisting of a data request - * which should be answered by another process. This message is - * considered as a request message by the receiving rank. - * - recv message: The answer to a send/request message. - * - * @tparam T1 the type of the elements of the vector to sent - * @tparam T2 the type of the elements of the vector to received - * - * @note Since the payloads of the messages are optional, users have - * to deal with buffers themselves. The ConsensusAlgorithm classes 1) - * deliver only references to empty vectors (of size 0) - * the data to be sent can be inserted to or read from, and - * 2) communicate these vectors blindly. - * - * @author Peter Munch, 2019 - */ - template - class ConsensusAlgorithmProcess - { - public: - /** - * Destructor. - */ - virtual ~ConsensusAlgorithmProcess() = default; - - /** - * @return A vector of ranks this process wants to send a request to. - * - * @note This is the only method which has to be implemented since the - * payloads of the messages are optional. - */ - virtual std::vector - compute_targets() = 0; - - /** - * Add to the request to the process with the specified rank a payload. - * - * @param[in] other_rank rank of the process - * @param[out] send_buffer data to be sent part of the request (optional) - * - * @note The buffer is empty. Before using it, you have to set its size. - */ - virtual void - create_request(const unsigned int other_rank, - std::vector & send_buffer); - - /** - * Prepare the buffer where the payload of the answer of the request to - * the process with the specified rank is saved in. The most obvious task - * is to resize the buffer, since it is empty when the function is called. - * - * @param[in] other_rank rank of the process - * @param[out] recv_buffer data to be sent part of the request (optional) - */ - virtual void - prepare_buffer_for_answer(const unsigned int other_rank, - std::vector & recv_buffer); - - /** - * Prepare the buffer where the payload of the answer of the request to - * the process with the specified rank is saved in. - * - * @param[in] other_rank rank of the process - * @param[in] buffer_recv received payload (optional) - * @param[out] request_buffer payload to be sent as part of the request - * (optional) - * - * @note The request_buffer is empty. Before using it, you have to set - * its size. - */ - virtual void - answer_request(const unsigned int other_rank, - const std::vector &buffer_recv, - std::vector & request_buffer); - - /** - * Process the payload of the answer of the request to the process with - * the specified rank. - * - * @param[in] other_rank rank of the process - * @param[in] recv_buffer data to be sent part of the request (optional) - */ - virtual void - read_answer(const unsigned int other_rank, - const std::vector &recv_buffer); - }; - - - - /** - * A base class for algorithms that implement the task of coming up with - * communication patterns to retrieve data from other processes in a - * dynamic-sparse way. In computer science, this is often called a - * consensus - * problem. - * - * Dynamic-sparse means in this context: - * - By the time this function is called, the other processes do - * not know yet that they have to answer requests. - * - Each process only has to communicate with a small subset of - * processes of the MPI communicator. - * - * Naturally, the user has to provide: - * - A communicator. - * - For each rank a list of ranks of processes this process should - * communicate to. - * - Functionality to pack/unpack data to be sent/received. - * - * This base class only introduces a basic interface to achieve - * these goals, while derived classes implement different algorithms - * to actually compute such communication patterns. - * The last two features of the list above this paragraph are implemented - * in classes derived from ConsensusAlgorithmProcess. - * - * @tparam T1 The type of the elements of the vector to be sent. - * @tparam T2 The type of the elements of the vector to be received. - * - * @author Peter Munch, 2019 - */ - template - class ConsensusAlgorithm - { - public: - ConsensusAlgorithm(ConsensusAlgorithmProcess &process, - const MPI_Comm & comm); - - /** - * Destructor. - */ - virtual ~ConsensusAlgorithm() = default; - - /** - * Run consensus algorithm. - */ - virtual void - run() = 0; - - protected: - /** - * Reference to the process provided by the user. - */ - ConsensusAlgorithmProcess &process; - - /** - * MPI communicator. - */ - const MPI_Comm &comm; - - /** - * Rank of this process. - */ - const unsigned int my_rank; - - /** - * Number of processes in the communicator. - */ - const unsigned int n_procs; - }; - - - /** - * This class implements a concrete algorithm for the ConsensusAlgorithm - * base class, using only point-to-point communications and a single - * IBarrier. - * - * @note This class closely follows the paper Hoefler, Siebert, Lumsdaine - * "Scalable Communication Protocols for Dynamic Sparse Data - * Exchange". Since the algorithm shown there is not considering - * payloads, the algorithm has been modified here in such a way that - * synchronous sends (Issend) have been replaced by equivalent - * Isend/Irecv, where Irecv receives the answer to a request (with - * payload). - * - * @tparam T1 The type of the elements of the vector to be sent. - * @tparam T2 The type of the elements of the vector to be received. - * - * @author Peter Munch, 2019 - */ - template - class ConsensusAlgorithm_NBX : public ConsensusAlgorithm - { - public: - /** - * Constructor. - * - * @param process Process to be run during consensus algorithm. - * @param comm MPI Communicator - */ - ConsensusAlgorithm_NBX(ConsensusAlgorithmProcess &process, - const MPI_Comm & comm); - - /** - * Destructor. - */ - virtual ~ConsensusAlgorithm_NBX() = default; - - /** - * @copydoc ConsensusAlgorithm::run() - */ - virtual void - run() override; - - private: -#ifdef DEAL_II_WITH_MPI - /** - * List of processes this process wants to send requests to. - */ - std::vector targets; - - /** - * Buffers for sending requests. - */ - std::vector> send_buffers; - - /** - * Requests for sending requests. - */ - std::vector send_requests; - - /** - * Buffers for receiving answers to requests. - */ - std::vector> recv_buffers; - - - /** - * Requests for receiving answers to requests. - */ - std::vector recv_requests; - - /** - * Buffers for sending answers to requests. - */ - std::vector>> request_buffers; - - /** - * Requests for sending answers to requests. - */ - std::vector> request_requests; - - // request for barrier - MPI_Request barrier_request; -#endif - -#ifdef DEBUG - /** - * List of processes who have made a request to this process. - */ - std::set requesting_processes; -#endif - - /** - * Check if all request answers have been received by this rank. - */ - bool - check_own_state(); - - /** - * Signal to all other ranks that this rank has received all request - * answers via entering IBarrier. - */ - void - signal_finish(); - - /** - * Check if all ranks have received all their request answers, i.e. - * all ranks have reached the IBarrier. - */ - bool - check_global_state(); - - /** - * A request message from another rank has been received: process the - * request and send an answer. - */ - void - answer_requests(); - - /** - * Start to send all requests via ISend and post IRecvs for the incoming - * answer messages. - */ - void - start_communication(); - - /** - * 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(); - }; - - /** - * This class implements a concrete algorithm for the ConsensusAlgorithm - * base class, using a two step approach. In - * the first step the source ranks are determined and in the second step - * a static sparse data exchange is performed. - * - * @note In contrast to ConsensusAlgorithm_NBX, this class splits the same - * task into two distinct steps. In the first step, all processes are - * identified who want to send a request to this process. In the - * second step, the data is exchanged. However, since - in the second - * step - now it is clear how many requests have to be answered, i.e. - * when this process can stop waiting for requests, no IBarrier is - * needed. - * - * @note The function - * Utilities::MPI::compute_point_to_point_communication_pattern() is - * used to determine the source processes, which implements a - * PEX-algorithm from Hoefner et al., "Scalable Communication - * Protocols for Dynamic Sparse Data Exchange". - * - * @tparam T1 The type of the elements of the vector to be sent. - * @tparam T2 The type of the elements of the vector to be received. - * - * @author Peter Munch, 2019 - */ - template - class ConsensusAlgorithm_PEX : public ConsensusAlgorithm - { - public: - /** - * Constructor. - * - * @param process Process to be run during consensus algorithm. - * @param comm MPI Communicator - */ - ConsensusAlgorithm_PEX(ConsensusAlgorithmProcess &process, - const MPI_Comm & comm); - - /** - * Destructor. - */ - virtual ~ConsensusAlgorithm_PEX() = default; - - /** - * @copydoc ConsensusAlgorithm::run() - */ - virtual void - run() override; - - private: -#ifdef DEAL_II_WITH_MPI - /** - * List of ranks of processes this processes wants to send a request to. - */ - std::vector targets; - - /** - * List of ranks of processes wanting to send a request to this process. - */ - std::vector sources; - - // data structures to send and receive requests - - /** - * Buffers for sending requests. - */ - std::vector> send_buffers; - - /** - * Buffers for receiving answers to requests. - */ - std::vector> recv_buffers; - - /** - * Requests for sending requests and receiving answers to requests. - */ - std::vector send_and_recv_buffers; - - /** - * Buffers for sending answers to requests. - */ - std::vector> requests_buffers; - - /** - * Requests for sending answers to requests. - */ - std::vector requests_answers; -#endif - - /** - * The ith request message from another rank has been received: process - * the request and send an answer. - */ - void - answer_requests(int index); - - /** - * Start to send all requests via ISend and post IRecvs for the incoming - * answer messages. - */ - unsigned int - start_communication(); - - /** - * After all answers have been exchanged, the MPI data structures can be - * freed and the received answers can be processed. - */ - void - clean_up_and_end_communication(); - }; - - /** - * A class which delegates its task to other ConsensusAlgorithm - * implementations depending on the number of processes in the - * MPI communicator. For a small number of processes it uses - * ConsensusAlgorithm_PEX and for a large number of processes - * ConsensusAlgorithm_NBX. The threshold depends if the program is - * compiled in debug or release mode. - * - * @tparam T1 The type of the elements of the vector to be sent. - * @tparam T2 The type of the elements of the vector to be received. - * - * @author Peter Munch, 2019 - */ - template - class ConsensusAlgorithmSelector : public ConsensusAlgorithm - { - public: - /** - * Constructor. - * - * @param process Process to be run during consensus algorithm. - * @param comm MPI Communicator. - */ - ConsensusAlgorithmSelector(ConsensusAlgorithmProcess &process, - const MPI_Comm & comm); - - /** - * Destructor. - */ - virtual ~ConsensusAlgorithmSelector() = default; - - /** - * @copydoc ConsensusAlgorithm::run() - * - * @note The function call is delegated to another ConsensusAlgorithm implementation. - */ - virtual void - run() override; - - private: - // Pointer to the actual ConsensusAlgorithm implementation. - std::shared_ptr> consensus_algo; - }; - /** * Given a partitioned index set space, compute the owning MPI process rank * of each element of a second index set according to the partitioned index diff --git a/include/deal.II/base/mpi.templates.h b/include/deal.II/base/mpi.templates.h index 5cc190f05f..1088896367 100644 --- a/include/deal.II/base/mpi.templates.h +++ b/include/deal.II/base/mpi.templates.h @@ -413,559 +413,6 @@ namespace Utilities internal::all_reduce(MPI_MIN, values, mpi_communicator, minima); } - - - template - void - ConsensusAlgorithmProcess::answer_request(const unsigned int, - const std::vector &, - std::vector &) - { - // nothing to do - } - - - - template - void - ConsensusAlgorithmProcess::create_request(const unsigned int, - std::vector &) - { - // nothing to do - } - - - - template - void - ConsensusAlgorithmProcess::prepare_buffer_for_answer( - const unsigned int, - std::vector &) - { - // nothing to do - } - - - - template - void - ConsensusAlgorithmProcess::read_answer(const unsigned int, - const std::vector &) - { - // nothing to do - } - - - - template - ConsensusAlgorithm::ConsensusAlgorithm( - ConsensusAlgorithmProcess &process, - const MPI_Comm & comm) - : process(process) - , comm(comm) - , my_rank(this_mpi_process(comm)) - , n_procs(n_mpi_processes(comm)) - {} - - - - template - ConsensusAlgorithm_NBX::ConsensusAlgorithm_NBX( - ConsensusAlgorithmProcess &process, - const MPI_Comm & comm) - : ConsensusAlgorithm(process, comm) - {} - - - - template - void - ConsensusAlgorithm_NBX::run() - { - static CollectiveMutex mutex; - CollectiveMutex::ScopedLock lock(mutex, this->comm); - - // 1) send requests and start receiving the answers - start_communication(); - - // 2) answer requests and check if all requests of this process have been - // answered - while (!check_own_state()) - answer_requests(); - - // 3) signal to all other processes that all requests of this process have - // been answered - signal_finish(); - - // 4) nevertheless, this process has to keep on answering (potential) - // incoming requests until all processes have received the - // answer to all requests - while (!check_global_state()) - answer_requests(); - - // 5) process the answer to all requests - clean_up_and_end_communication(); - } - - - - template - bool - ConsensusAlgorithm_NBX::check_own_state() - { -#ifdef DEAL_II_WITH_MPI - int all_receive_requests_are_done; - const auto ierr = MPI_Testall(recv_requests.size(), - recv_requests.data(), - &all_receive_requests_are_done, - MPI_STATUSES_IGNORE); - AssertThrowMPI(ierr); - - return all_receive_requests_are_done; -#else - return true; -#endif - } - - - - template - void - ConsensusAlgorithm_NBX::signal_finish() - { -#ifdef DEAL_II_WITH_MPI -# if DEAL_II_MPI_VERSION_GTE(3, 0) - const auto ierr = MPI_Ibarrier(this->comm, &barrier_request); - AssertThrowMPI(ierr); -# else - AssertThrow( - false, - ExcMessage( - "ConsensusAlgorithm_NBX uses MPI 3.0 features. You should compile with at least MPI 3.0.")); -# endif -#endif - } - - - - template - bool - ConsensusAlgorithm_NBX::check_global_state() - { -#ifdef DEAL_II_WITH_MPI - int all_ranks_reached_barrier; - const auto ierr = MPI_Test(&barrier_request, - &all_ranks_reached_barrier, - MPI_STATUSES_IGNORE); - AssertThrowMPI(ierr); - return all_ranks_reached_barrier; -#else - return true; -#endif - } - - - - template - void - ConsensusAlgorithm_NBX::answer_requests() - { -#ifdef DEAL_II_WITH_MPI - - const int tag_request = - Utilities::MPI::internal::Tags::consensus_algorithm_nbx_answer_request; - const int tag_deliver = - Utilities::MPI::internal::Tags::consensus_algorithm_nbx_process_deliver; - - // check if there is a request pending - MPI_Status status; - int request_is_pending; - const auto ierr = MPI_Iprobe( - MPI_ANY_SOURCE, tag_request, this->comm, &request_is_pending, &status); - AssertThrowMPI(ierr); - - if (request_is_pending) // request is pending - { - // get rank of requesting process - const auto other_rank = status.MPI_SOURCE; - -# ifdef DEBUG - Assert(requesting_processes.find(other_rank) == - requesting_processes.end(), - ExcMessage("Process is requesting a second time!")); - requesting_processes.insert(other_rank); -# endif - - std::vector buffer_recv; - // get size of of incoming message - int number_amount; - auto ierr = MPI_Get_count(&status, MPI_BYTE, &number_amount); - AssertThrowMPI(ierr); - - // allocate memory for incoming message - Assert(number_amount % sizeof(T1) == 0, ExcInternalError()); - buffer_recv.resize(number_amount / sizeof(T1)); - ierr = MPI_Recv(buffer_recv.data(), - number_amount, - MPI_BYTE, - other_rank, - tag_request, - this->comm, - &status); - AssertThrowMPI(ierr); - - // allocate memory for answer message - request_buffers.emplace_back( - std_cxx14::make_unique>()); - request_requests.emplace_back(std_cxx14::make_unique()); - - // process request - auto &request_buffer = *request_buffers.back(); - this->process.answer_request(other_rank, buffer_recv, request_buffer); - - // start to send answer back - ierr = MPI_Isend(request_buffer.data(), - request_buffer.size() * sizeof(T2), - MPI_BYTE, - other_rank, - tag_deliver, - this->comm, - request_requests.back().get()); - AssertThrowMPI(ierr); - } -#endif - } - - - - template - void - ConsensusAlgorithm_NBX::start_communication() - { -#ifdef DEAL_II_WITH_MPI - // 1) - targets = this->process.compute_targets(); - const auto n_targets = targets.size(); - - const int tag_request = - Utilities::MPI::internal::Tags::consensus_algorithm_nbx_answer_request; - const int tag_deliver = - Utilities::MPI::internal::Tags::consensus_algorithm_nbx_process_deliver; - - // 2) allocate memory - recv_buffers.resize(n_targets); - recv_requests.resize(n_targets); - send_requests.resize(n_targets); - send_buffers.resize(n_targets); - - { - // 4) send and receive - for (unsigned int i = 0; i < n_targets; i++) - { - const unsigned int rank = targets[i]; - const unsigned int index = i; - - // translate index set to a list of pairs - auto &send_buffer = send_buffers[index]; - this->process.create_request(rank, send_buffer); - - // start to send data - auto ierr = MPI_Isend(send_buffer.data(), - send_buffer.size() * sizeof(T1), - MPI_BYTE, - rank, - tag_request, - this->comm, - &send_requests[index]); - AssertThrowMPI(ierr); - - // start to receive data - auto &recv_buffer = recv_buffers[index]; - this->process.prepare_buffer_for_answer(rank, recv_buffer); - ierr = MPI_Irecv(recv_buffer.data(), - recv_buffer.size() * sizeof(T2), - MPI_BYTE, - rank, - tag_deliver, - this->comm, - &recv_requests[index]); - AssertThrowMPI(ierr); - } - } -#endif - } - - - - template - void - ConsensusAlgorithm_NBX::clean_up_and_end_communication() - { -#ifdef DEAL_II_WITH_MPI - // clean up - { - if (send_requests.size() > 0) - { - const int ierr = MPI_Waitall(send_requests.size(), - send_requests.data(), - MPI_STATUSES_IGNORE); - AssertThrowMPI(ierr); - } - - if (recv_requests.size() > 0) - { - const int ierr = MPI_Waitall(recv_requests.size(), - recv_requests.data(), - MPI_STATUSES_IGNORE); - AssertThrowMPI(ierr); - } - - - const int ierr = MPI_Wait(&barrier_request, MPI_STATUS_IGNORE); - AssertThrowMPI(ierr); - - for (auto &i : request_requests) - { - const auto ierr = MPI_Wait(i.get(), MPI_STATUS_IGNORE); - AssertThrowMPI(ierr); - } - -# ifdef DEBUG - // note: IBarrier seems to make problem during testing, this additional - // Barrier seems to help - MPI_Barrier(this->comm); -# endif - } - - // unpack data - { - for (unsigned int i = 0; i < targets.size(); i++) - this->process.read_answer(targets[i], recv_buffers[i]); - } -#endif - } - - - - template - ConsensusAlgorithm_PEX::ConsensusAlgorithm_PEX( - ConsensusAlgorithmProcess &process, - const MPI_Comm & comm) - : ConsensusAlgorithm(process, comm) - {} - - - - template - void - ConsensusAlgorithm_PEX::run() - { - static CollectiveMutex mutex; - CollectiveMutex::ScopedLock lock(mutex, this->comm); - - // 1) send requests and start receiving the answers - // especially determine how many requests are expected - const unsigned int n_requests = start_communication(); - - // 2) answer requests - for (unsigned int request = 0; request < n_requests; request++) - answer_requests(request); - - // 3) process answers - clean_up_and_end_communication(); - } - - - - template - void - ConsensusAlgorithm_PEX::answer_requests(int index) - { -#ifdef DEAL_II_WITH_MPI - const int tag_request = - Utilities::MPI::internal::Tags::consensus_algorithm_pex_answer_request; - const int tag_deliver = - Utilities::MPI::internal::Tags::consensus_algorithm_pex_process_deliver; - - MPI_Status status; - auto ierr = MPI_Probe(MPI_ANY_SOURCE, tag_request, this->comm, &status); - AssertThrowMPI(ierr); - - // get rank of incoming message - const auto other_rank = status.MPI_SOURCE; - - std::vector buffer_recv; - - // get size of incoming message - int number_amount; - ierr = MPI_Get_count(&status, MPI_BYTE, &number_amount); - AssertThrowMPI(ierr); - - // allocate memory for incoming message - Assert(number_amount % sizeof(T1) == 0, ExcInternalError()); - buffer_recv.resize(number_amount / sizeof(T1)); - ierr = MPI_Recv(buffer_recv.data(), - number_amount, - MPI_BYTE, - other_rank, - tag_request, - this->comm, - &status); - AssertThrowMPI(ierr); - - // process request - auto &request_buffer = requests_buffers[index]; - this->process.answer_request(other_rank, buffer_recv, request_buffer); - - // start to send answer back - ierr = MPI_Isend(request_buffer.data(), - request_buffer.size() * sizeof(T2), - MPI_BYTE, - other_rank, - tag_deliver, - this->comm, - &requests_answers[index]); - AssertThrowMPI(ierr); -#else - (void)index; -#endif - } - - - - template - unsigned int - ConsensusAlgorithm_PEX::start_communication() - { -#ifdef DEAL_II_WITH_MPI - // 1) determine with which processes this process wants to communicate - targets = this->process.compute_targets(); - - const int tag_request = - Utilities::MPI::internal::Tags::consensus_algorithm_pex_answer_request; - const int tag_deliver = - Utilities::MPI::internal::Tags::consensus_algorithm_pex_process_deliver; - - // 2) determine who wants to communicate with this process - sources = - compute_point_to_point_communication_pattern(this->comm, targets); - - const auto n_targets = targets.size(); - const auto n_sources = sources.size(); - - // 2) allocate memory - recv_buffers.resize(n_targets); - send_buffers.resize(n_targets); - send_and_recv_buffers.resize(2 * n_targets); - - requests_answers.resize(n_sources); - requests_buffers.resize(n_sources); - - // 4) send and receive - for (unsigned int i = 0; i < n_targets; i++) - { - const unsigned int rank = targets[i]; - - // pack data which should be sent - auto &send_buffer = send_buffers[i]; - this->process.create_request(rank, send_buffer); - - // start to send data - auto ierr = MPI_Isend(send_buffer.data(), - send_buffer.size() * sizeof(T1), - MPI_BYTE, - rank, - tag_request, - this->comm, - &send_and_recv_buffers[n_targets + i]); - AssertThrowMPI(ierr); - - // start to receive data - auto &recv_buffer = recv_buffers[i]; - this->process.prepare_buffer_for_answer(rank, recv_buffer); - ierr = MPI_Irecv(recv_buffer.data(), - recv_buffer.size() * sizeof(T2), - MPI_BYTE, - rank, - tag_deliver, - this->comm, - &send_and_recv_buffers[i]); - AssertThrowMPI(ierr); - } - - return sources.size(); -#else - return 0; -#endif - } - - - - template - void - ConsensusAlgorithm_PEX::clean_up_and_end_communication() - { -#ifdef DEAL_II_WITH_MPI - // finalize all MPI_Requests - if (send_and_recv_buffers.size() > 0) - { - auto ierr = MPI_Waitall(send_and_recv_buffers.size(), - send_and_recv_buffers.data(), - MPI_STATUSES_IGNORE); - AssertThrowMPI(ierr); - } - - if (requests_answers.size() > 0) - { - auto ierr = MPI_Waitall(requests_answers.size(), - requests_answers.data(), - MPI_STATUSES_IGNORE); - AssertThrowMPI(ierr); - } - - // unpack received data - for (unsigned int i = 0; i < targets.size(); i++) - this->process.read_answer(targets[i], recv_buffers[i]); -#endif - } - - - - template - ConsensusAlgorithmSelector::ConsensusAlgorithmSelector( - ConsensusAlgorithmProcess &process, - const MPI_Comm & comm) - : ConsensusAlgorithm(process, comm) - { - // Depending on the number of processes we switch between implementations. - // We reduce the threshold for debug mode to be able to test also the - // non-blocking implementation. This feature is tested by: - // tests/multigrid/transfer_matrix_free_06.with_mpi=true.with_p4est=true.with_trilinos=true.mpirun=15.output -#ifdef DEAL_II_WITH_MPI -# if DEAL_II_MPI_VERSION_GTE(3, 0) -# ifdef DEBUG - if (Utilities::MPI::n_mpi_processes(comm) > 14) -# else - if (Utilities::MPI::n_mpi_processes(comm) > 99) -# endif - consensus_algo.reset(new ConsensusAlgorithm_NBX(process, comm)); - else -# endif -#endif - consensus_algo.reset(new ConsensusAlgorithm_PEX(process, comm)); - } - - - - template - void - ConsensusAlgorithmSelector::run() - { - consensus_algo->run(); - } - - - } // end of namespace MPI } // end of namespace Utilities diff --git a/include/deal.II/base/mpi_compute_index_owner_internal.h b/include/deal.II/base/mpi_compute_index_owner_internal.h index 7ecce658dc..d853cd8739 100644 --- a/include/deal.II/base/mpi_compute_index_owner_internal.h +++ b/include/deal.II/base/mpi_compute_index_owner_internal.h @@ -19,6 +19,7 @@ #include #include +#include DEAL_II_NAMESPACE_OPEN diff --git a/include/deal.II/base/mpi_consensus_algorithm.h b/include/deal.II/base/mpi_consensus_algorithm.h index e9c13eb799..bfd7b3a274 100644 --- a/include/deal.II/base/mpi_consensus_algorithm.h +++ b/include/deal.II/base/mpi_consensus_algorithm.h @@ -28,6 +28,461 @@ namespace Utilities { namespace MPI { + /** + * An interface to be able to use the ConsensusAlgorithm classes. The main + * functionality of the implementations is to return a list of process ranks + * this process wants data from and to deal with the optional payload of the + * messages sent/received by the ConsensusAlgorithm classes. + * + * There are two kinds of messages: + * - send/request message: A message consisting of a data request + * which should be answered by another process. This message is + * considered as a request message by the receiving rank. + * - recv message: The answer to a send/request message. + * + * @tparam T1 the type of the elements of the vector to sent + * @tparam T2 the type of the elements of the vector to received + * + * @note Since the payloads of the messages are optional, users have + * to deal with buffers themselves. The ConsensusAlgorithm classes 1) + * deliver only references to empty vectors (of size 0) + * the data to be sent can be inserted to or read from, and + * 2) communicate these vectors blindly. + * + * @author Peter Munch, 2019 + */ + template + class ConsensusAlgorithmProcess + { + public: + /** + * Destructor. + */ + virtual ~ConsensusAlgorithmProcess() = default; + + /** + * @return A vector of ranks this process wants to send a request to. + * + * @note This is the only method which has to be implemented since the + * payloads of the messages are optional. + */ + virtual std::vector + compute_targets() = 0; + + /** + * Add to the request to the process with the specified rank a payload. + * + * @param[in] other_rank rank of the process + * @param[out] send_buffer data to be sent part of the request (optional) + * + * @note The buffer is empty. Before using it, you have to set its size. + */ + virtual void + create_request(const unsigned int other_rank, + std::vector & send_buffer); + + /** + * Prepare the buffer where the payload of the answer of the request to + * the process with the specified rank is saved in. The most obvious task + * is to resize the buffer, since it is empty when the function is called. + * + * @param[in] other_rank rank of the process + * @param[out] recv_buffer data to be sent part of the request (optional) + */ + virtual void + prepare_buffer_for_answer(const unsigned int other_rank, + std::vector & recv_buffer); + + /** + * Prepare the buffer where the payload of the answer of the request to + * the process with the specified rank is saved in. + * + * @param[in] other_rank rank of the process + * @param[in] buffer_recv received payload (optional) + * @param[out] request_buffer payload to be sent as part of the request + * (optional) + * + * @note The request_buffer is empty. Before using it, you have to set + * its size. + */ + virtual void + answer_request(const unsigned int other_rank, + const std::vector &buffer_recv, + std::vector & request_buffer); + + /** + * Process the payload of the answer of the request to the process with + * the specified rank. + * + * @param[in] other_rank rank of the process + * @param[in] recv_buffer data to be sent part of the request (optional) + */ + virtual void + read_answer(const unsigned int other_rank, + const std::vector &recv_buffer); + }; + + + + /** + * A base class for algorithms that implement the task of coming up with + * communication patterns to retrieve data from other processes in a + * dynamic-sparse way. In computer science, this is often called a + * consensus + * problem. + * + * Dynamic-sparse means in this context: + * - By the time this function is called, the other processes do + * not know yet that they have to answer requests. + * - Each process only has to communicate with a small subset of + * processes of the MPI communicator. + * + * Naturally, the user has to provide: + * - A communicator. + * - For each rank a list of ranks of processes this process should + * communicate to. + * - Functionality to pack/unpack data to be sent/received. + * + * This base class only introduces a basic interface to achieve + * these goals, while derived classes implement different algorithms + * to actually compute such communication patterns. + * The last two features of the list above this paragraph are implemented + * in classes derived from ConsensusAlgorithmProcess. + * + * @tparam T1 The type of the elements of the vector to be sent. + * @tparam T2 The type of the elements of the vector to be received. + * + * @author Peter Munch, 2019 + */ + template + class ConsensusAlgorithm + { + public: + ConsensusAlgorithm(ConsensusAlgorithmProcess &process, + const MPI_Comm & comm); + + /** + * Destructor. + */ + virtual ~ConsensusAlgorithm() = default; + + /** + * Run consensus algorithm. + */ + virtual void + run() = 0; + + protected: + /** + * Reference to the process provided by the user. + */ + ConsensusAlgorithmProcess &process; + + /** + * MPI communicator. + */ + const MPI_Comm &comm; + + /** + * Rank of this process. + */ + const unsigned int my_rank; + + /** + * Number of processes in the communicator. + */ + const unsigned int n_procs; + }; + + + /** + * This class implements a concrete algorithm for the ConsensusAlgorithm + * base class, using only point-to-point communications and a single + * IBarrier. + * + * @note This class closely follows the paper Hoefler, Siebert, Lumsdaine + * "Scalable Communication Protocols for Dynamic Sparse Data + * Exchange". Since the algorithm shown there is not considering + * payloads, the algorithm has been modified here in such a way that + * synchronous sends (Issend) have been replaced by equivalent + * Isend/Irecv, where Irecv receives the answer to a request (with + * payload). + * + * @tparam T1 The type of the elements of the vector to be sent. + * @tparam T2 The type of the elements of the vector to be received. + * + * @author Peter Munch, 2019 + */ + template + class ConsensusAlgorithm_NBX : public ConsensusAlgorithm + { + public: + /** + * Constructor. + * + * @param process Process to be run during consensus algorithm. + * @param comm MPI Communicator + */ + ConsensusAlgorithm_NBX(ConsensusAlgorithmProcess &process, + const MPI_Comm & comm); + + /** + * Destructor. + */ + virtual ~ConsensusAlgorithm_NBX() = default; + + /** + * @copydoc ConsensusAlgorithm::run() + */ + virtual void + run() override; + + private: +#ifdef DEAL_II_WITH_MPI + /** + * List of processes this process wants to send requests to. + */ + std::vector targets; + + /** + * Buffers for sending requests. + */ + std::vector> send_buffers; + + /** + * Requests for sending requests. + */ + std::vector send_requests; + + /** + * Buffers for receiving answers to requests. + */ + std::vector> recv_buffers; + + + /** + * Requests for receiving answers to requests. + */ + std::vector recv_requests; + + /** + * Buffers for sending answers to requests. + */ + std::vector>> request_buffers; + + /** + * Requests for sending answers to requests. + */ + std::vector> request_requests; + + // request for barrier + MPI_Request barrier_request; +#endif + +#ifdef DEBUG + /** + * List of processes who have made a request to this process. + */ + std::set requesting_processes; +#endif + + /** + * Check if all request answers have been received by this rank. + */ + bool + check_own_state(); + + /** + * Signal to all other ranks that this rank has received all request + * answers via entering IBarrier. + */ + void + signal_finish(); + + /** + * Check if all ranks have received all their request answers, i.e. + * all ranks have reached the IBarrier. + */ + bool + check_global_state(); + + /** + * A request message from another rank has been received: process the + * request and send an answer. + */ + void + answer_requests(); + + /** + * Start to send all requests via ISend and post IRecvs for the incoming + * answer messages. + */ + void + start_communication(); + + /** + * 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(); + }; + + /** + * This class implements a concrete algorithm for the ConsensusAlgorithm + * base class, using a two step approach. In + * the first step the source ranks are determined and in the second step + * a static sparse data exchange is performed. + * + * @note In contrast to ConsensusAlgorithm_NBX, this class splits the same + * task into two distinct steps. In the first step, all processes are + * identified who want to send a request to this process. In the + * second step, the data is exchanged. However, since - in the second + * step - now it is clear how many requests have to be answered, i.e. + * when this process can stop waiting for requests, no IBarrier is + * needed. + * + * @note The function + * Utilities::MPI::compute_point_to_point_communication_pattern() is + * used to determine the source processes, which implements a + * PEX-algorithm from Hoefner et al., "Scalable Communication + * Protocols for Dynamic Sparse Data Exchange". + * + * @tparam T1 The type of the elements of the vector to be sent. + * @tparam T2 The type of the elements of the vector to be received. + * + * @author Peter Munch, 2019 + */ + template + class ConsensusAlgorithm_PEX : public ConsensusAlgorithm + { + public: + /** + * Constructor. + * + * @param process Process to be run during consensus algorithm. + * @param comm MPI Communicator + */ + ConsensusAlgorithm_PEX(ConsensusAlgorithmProcess &process, + const MPI_Comm & comm); + + /** + * Destructor. + */ + virtual ~ConsensusAlgorithm_PEX() = default; + + /** + * @copydoc ConsensusAlgorithm::run() + */ + virtual void + run() override; + + private: +#ifdef DEAL_II_WITH_MPI + /** + * List of ranks of processes this processes wants to send a request to. + */ + std::vector targets; + + /** + * List of ranks of processes wanting to send a request to this process. + */ + std::vector sources; + + // data structures to send and receive requests + + /** + * Buffers for sending requests. + */ + std::vector> send_buffers; + + /** + * Buffers for receiving answers to requests. + */ + std::vector> recv_buffers; + + /** + * Requests for sending requests and receiving answers to requests. + */ + std::vector send_and_recv_buffers; + + /** + * Buffers for sending answers to requests. + */ + std::vector> requests_buffers; + + /** + * Requests for sending answers to requests. + */ + std::vector requests_answers; +#endif + + /** + * The ith request message from another rank has been received: process + * the request and send an answer. + */ + void + answer_requests(int index); + + /** + * Start to send all requests via ISend and post IRecvs for the incoming + * answer messages. + */ + unsigned int + start_communication(); + + /** + * After all answers have been exchanged, the MPI data structures can be + * freed and the received answers can be processed. + */ + void + clean_up_and_end_communication(); + }; + + /** + * A class which delegates its task to other ConsensusAlgorithm + * implementations depending on the number of processes in the + * MPI communicator. For a small number of processes it uses + * ConsensusAlgorithm_PEX and for a large number of processes + * ConsensusAlgorithm_NBX. The threshold depends if the program is + * compiled in debug or release mode. + * + * @tparam T1 The type of the elements of the vector to be sent. + * @tparam T2 The type of the elements of the vector to be received. + * + * @author Peter Munch, 2019 + */ + template + class ConsensusAlgorithmSelector : public ConsensusAlgorithm + { + public: + /** + * Constructor. + * + * @param process Process to be run during consensus algorithm. + * @param comm MPI Communicator. + */ + ConsensusAlgorithmSelector(ConsensusAlgorithmProcess &process, + const MPI_Comm & comm); + + /** + * Destructor. + */ + virtual ~ConsensusAlgorithmSelector() = default; + + /** + * @copydoc ConsensusAlgorithm::run() + * + * @note The function call is delegated to another ConsensusAlgorithm implementation. + */ + virtual void + run() override; + + private: + // Pointer to the actual ConsensusAlgorithm implementation. + std::shared_ptr> consensus_algo; + }; + /** * This class implements Utilities::MPI::ConsensusAlgorithmProcess, * using user-provided function wrappers. diff --git a/include/deal.II/base/mpi_consensus_algorithm.templates.h b/include/deal.II/base/mpi_consensus_algorithm.templates.h new file mode 100644 index 0000000000..3b39939e38 --- /dev/null +++ b/include/deal.II/base/mpi_consensus_algorithm.templates.h @@ -0,0 +1,597 @@ +// --------------------------------------------------------------------- +// +// Copyright (C) 2011 - 2019 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_templates_h +#define dealii_mpi_consensus_algorithm_templates_h + +#include + +#include +#include +#include +#include +#include + +#include +#include +#include +#include + +#include + +DEAL_II_NAMESPACE_OPEN + +namespace Utilities +{ + namespace MPI + { + template + void + ConsensusAlgorithmProcess::answer_request(const unsigned int, + const std::vector &, + std::vector &) + { + // nothing to do + } + + + + template + void + ConsensusAlgorithmProcess::create_request(const unsigned int, + std::vector &) + { + // nothing to do + } + + + + template + void + ConsensusAlgorithmProcess::prepare_buffer_for_answer( + const unsigned int, + std::vector &) + { + // nothing to do + } + + + + template + void + ConsensusAlgorithmProcess::read_answer(const unsigned int, + const std::vector &) + { + // nothing to do + } + + + + template + ConsensusAlgorithm::ConsensusAlgorithm( + ConsensusAlgorithmProcess &process, + const MPI_Comm & comm) + : process(process) + , comm(comm) + , my_rank(this_mpi_process(comm)) + , n_procs(n_mpi_processes(comm)) + {} + + + + template + ConsensusAlgorithm_NBX::ConsensusAlgorithm_NBX( + ConsensusAlgorithmProcess &process, + const MPI_Comm & comm) + : ConsensusAlgorithm(process, comm) + {} + + + + template + void + ConsensusAlgorithm_NBX::run() + { + static CollectiveMutex mutex; + CollectiveMutex::ScopedLock lock(mutex, this->comm); + + // 1) send requests and start receiving the answers + start_communication(); + + // 2) answer requests and check if all requests of this process have been + // answered + while (!check_own_state()) + answer_requests(); + + // 3) signal to all other processes that all requests of this process have + // been answered + signal_finish(); + + // 4) nevertheless, this process has to keep on answering (potential) + // incoming requests until all processes have received the + // answer to all requests + while (!check_global_state()) + answer_requests(); + + // 5) process the answer to all requests + clean_up_and_end_communication(); + } + + + + template + bool + ConsensusAlgorithm_NBX::check_own_state() + { +#ifdef DEAL_II_WITH_MPI + int all_receive_requests_are_done; + const auto ierr = MPI_Testall(recv_requests.size(), + recv_requests.data(), + &all_receive_requests_are_done, + MPI_STATUSES_IGNORE); + AssertThrowMPI(ierr); + + return all_receive_requests_are_done; +#else + return true; +#endif + } + + + + template + void + ConsensusAlgorithm_NBX::signal_finish() + { +#ifdef DEAL_II_WITH_MPI +# if DEAL_II_MPI_VERSION_GTE(3, 0) + const auto ierr = MPI_Ibarrier(this->comm, &barrier_request); + AssertThrowMPI(ierr); +# else + AssertThrow( + false, + ExcMessage( + "ConsensusAlgorithm_NBX uses MPI 3.0 features. You should compile with at least MPI 3.0.")); +# endif +#endif + } + + + + template + bool + ConsensusAlgorithm_NBX::check_global_state() + { +#ifdef DEAL_II_WITH_MPI + int all_ranks_reached_barrier; + const auto ierr = MPI_Test(&barrier_request, + &all_ranks_reached_barrier, + MPI_STATUSES_IGNORE); + AssertThrowMPI(ierr); + return all_ranks_reached_barrier; +#else + return true; +#endif + } + + + + template + void + ConsensusAlgorithm_NBX::answer_requests() + { +#ifdef DEAL_II_WITH_MPI + + const int tag_request = + Utilities::MPI::internal::Tags::consensus_algorithm_nbx_answer_request; + const int tag_deliver = + Utilities::MPI::internal::Tags::consensus_algorithm_nbx_process_deliver; + + // check if there is a request pending + MPI_Status status; + int request_is_pending; + const auto ierr = MPI_Iprobe( + MPI_ANY_SOURCE, tag_request, this->comm, &request_is_pending, &status); + AssertThrowMPI(ierr); + + if (request_is_pending) // request is pending + { + // get rank of requesting process + const auto other_rank = status.MPI_SOURCE; + +# ifdef DEBUG + Assert(requesting_processes.find(other_rank) == + requesting_processes.end(), + ExcMessage("Process is requesting a second time!")); + requesting_processes.insert(other_rank); +# endif + + std::vector buffer_recv; + // get size of of incoming message + int number_amount; + auto ierr = MPI_Get_count(&status, MPI_BYTE, &number_amount); + AssertThrowMPI(ierr); + + // allocate memory for incoming message + Assert(number_amount % sizeof(T1) == 0, ExcInternalError()); + buffer_recv.resize(number_amount / sizeof(T1)); + ierr = MPI_Recv(buffer_recv.data(), + number_amount, + MPI_BYTE, + other_rank, + tag_request, + this->comm, + &status); + AssertThrowMPI(ierr); + + // allocate memory for answer message + request_buffers.emplace_back( + std_cxx14::make_unique>()); + request_requests.emplace_back(std_cxx14::make_unique()); + + // process request + auto &request_buffer = *request_buffers.back(); + this->process.answer_request(other_rank, buffer_recv, request_buffer); + + // start to send answer back + ierr = MPI_Isend(request_buffer.data(), + request_buffer.size() * sizeof(T2), + MPI_BYTE, + other_rank, + tag_deliver, + this->comm, + request_requests.back().get()); + AssertThrowMPI(ierr); + } +#endif + } + + + + template + void + ConsensusAlgorithm_NBX::start_communication() + { +#ifdef DEAL_II_WITH_MPI + // 1) + targets = this->process.compute_targets(); + const auto n_targets = targets.size(); + + const int tag_request = + Utilities::MPI::internal::Tags::consensus_algorithm_nbx_answer_request; + const int tag_deliver = + Utilities::MPI::internal::Tags::consensus_algorithm_nbx_process_deliver; + + // 2) allocate memory + recv_buffers.resize(n_targets); + recv_requests.resize(n_targets); + send_requests.resize(n_targets); + send_buffers.resize(n_targets); + + { + // 4) send and receive + for (unsigned int i = 0; i < n_targets; i++) + { + const unsigned int rank = targets[i]; + const unsigned int index = i; + + // translate index set to a list of pairs + auto &send_buffer = send_buffers[index]; + this->process.create_request(rank, send_buffer); + + // start to send data + auto ierr = MPI_Isend(send_buffer.data(), + send_buffer.size() * sizeof(T1), + MPI_BYTE, + rank, + tag_request, + this->comm, + &send_requests[index]); + AssertThrowMPI(ierr); + + // start to receive data + auto &recv_buffer = recv_buffers[index]; + this->process.prepare_buffer_for_answer(rank, recv_buffer); + ierr = MPI_Irecv(recv_buffer.data(), + recv_buffer.size() * sizeof(T2), + MPI_BYTE, + rank, + tag_deliver, + this->comm, + &recv_requests[index]); + AssertThrowMPI(ierr); + } + } +#endif + } + + + + template + void + ConsensusAlgorithm_NBX::clean_up_and_end_communication() + { +#ifdef DEAL_II_WITH_MPI + // clean up + { + if (send_requests.size() > 0) + { + const int ierr = MPI_Waitall(send_requests.size(), + send_requests.data(), + MPI_STATUSES_IGNORE); + AssertThrowMPI(ierr); + } + + if (recv_requests.size() > 0) + { + const int ierr = MPI_Waitall(recv_requests.size(), + recv_requests.data(), + MPI_STATUSES_IGNORE); + AssertThrowMPI(ierr); + } + + + const int ierr = MPI_Wait(&barrier_request, MPI_STATUS_IGNORE); + AssertThrowMPI(ierr); + + for (auto &i : request_requests) + { + const auto ierr = MPI_Wait(i.get(), MPI_STATUS_IGNORE); + AssertThrowMPI(ierr); + } + +# ifdef DEBUG + // note: IBarrier seems to make problem during testing, this additional + // Barrier seems to help + MPI_Barrier(this->comm); +# endif + } + + // unpack data + { + for (unsigned int i = 0; i < targets.size(); i++) + this->process.read_answer(targets[i], recv_buffers[i]); + } +#endif + } + + + + template + ConsensusAlgorithm_PEX::ConsensusAlgorithm_PEX( + ConsensusAlgorithmProcess &process, + const MPI_Comm & comm) + : ConsensusAlgorithm(process, comm) + {} + + + + template + void + ConsensusAlgorithm_PEX::run() + { + static CollectiveMutex mutex; + CollectiveMutex::ScopedLock lock(mutex, this->comm); + + // 1) send requests and start receiving the answers + // especially determine how many requests are expected + const unsigned int n_requests = start_communication(); + + // 2) answer requests + for (unsigned int request = 0; request < n_requests; request++) + answer_requests(request); + + // 3) process answers + clean_up_and_end_communication(); + } + + + + template + void + ConsensusAlgorithm_PEX::answer_requests(int index) + { +#ifdef DEAL_II_WITH_MPI + const int tag_request = + Utilities::MPI::internal::Tags::consensus_algorithm_pex_answer_request; + const int tag_deliver = + Utilities::MPI::internal::Tags::consensus_algorithm_pex_process_deliver; + + MPI_Status status; + auto ierr = MPI_Probe(MPI_ANY_SOURCE, tag_request, this->comm, &status); + AssertThrowMPI(ierr); + + // get rank of incoming message + const auto other_rank = status.MPI_SOURCE; + + std::vector buffer_recv; + + // get size of incoming message + int number_amount; + ierr = MPI_Get_count(&status, MPI_BYTE, &number_amount); + AssertThrowMPI(ierr); + + // allocate memory for incoming message + Assert(number_amount % sizeof(T1) == 0, ExcInternalError()); + buffer_recv.resize(number_amount / sizeof(T1)); + ierr = MPI_Recv(buffer_recv.data(), + number_amount, + MPI_BYTE, + other_rank, + tag_request, + this->comm, + &status); + AssertThrowMPI(ierr); + + // process request + auto &request_buffer = requests_buffers[index]; + this->process.answer_request(other_rank, buffer_recv, request_buffer); + + // start to send answer back + ierr = MPI_Isend(request_buffer.data(), + request_buffer.size() * sizeof(T2), + MPI_BYTE, + other_rank, + tag_deliver, + this->comm, + &requests_answers[index]); + AssertThrowMPI(ierr); +#else + (void)index; +#endif + } + + + + template + unsigned int + ConsensusAlgorithm_PEX::start_communication() + { +#ifdef DEAL_II_WITH_MPI + // 1) determine with which processes this process wants to communicate + targets = this->process.compute_targets(); + + const int tag_request = + Utilities::MPI::internal::Tags::consensus_algorithm_pex_answer_request; + const int tag_deliver = + Utilities::MPI::internal::Tags::consensus_algorithm_pex_process_deliver; + + // 2) determine who wants to communicate with this process + sources = + compute_point_to_point_communication_pattern(this->comm, targets); + + const auto n_targets = targets.size(); + const auto n_sources = sources.size(); + + // 2) allocate memory + recv_buffers.resize(n_targets); + send_buffers.resize(n_targets); + send_and_recv_buffers.resize(2 * n_targets); + + requests_answers.resize(n_sources); + requests_buffers.resize(n_sources); + + // 4) send and receive + for (unsigned int i = 0; i < n_targets; i++) + { + const unsigned int rank = targets[i]; + + // pack data which should be sent + auto &send_buffer = send_buffers[i]; + this->process.create_request(rank, send_buffer); + + // start to send data + auto ierr = MPI_Isend(send_buffer.data(), + send_buffer.size() * sizeof(T1), + MPI_BYTE, + rank, + tag_request, + this->comm, + &send_and_recv_buffers[n_targets + i]); + AssertThrowMPI(ierr); + + // start to receive data + auto &recv_buffer = recv_buffers[i]; + this->process.prepare_buffer_for_answer(rank, recv_buffer); + ierr = MPI_Irecv(recv_buffer.data(), + recv_buffer.size() * sizeof(T2), + MPI_BYTE, + rank, + tag_deliver, + this->comm, + &send_and_recv_buffers[i]); + AssertThrowMPI(ierr); + } + + return sources.size(); +#else + return 0; +#endif + } + + + + template + void + ConsensusAlgorithm_PEX::clean_up_and_end_communication() + { +#ifdef DEAL_II_WITH_MPI + // finalize all MPI_Requests + if (send_and_recv_buffers.size() > 0) + { + auto ierr = MPI_Waitall(send_and_recv_buffers.size(), + send_and_recv_buffers.data(), + MPI_STATUSES_IGNORE); + AssertThrowMPI(ierr); + } + + if (requests_answers.size() > 0) + { + auto ierr = MPI_Waitall(requests_answers.size(), + requests_answers.data(), + MPI_STATUSES_IGNORE); + AssertThrowMPI(ierr); + } + + // unpack received data + for (unsigned int i = 0; i < targets.size(); i++) + this->process.read_answer(targets[i], recv_buffers[i]); +#endif + } + + + + template + ConsensusAlgorithmSelector::ConsensusAlgorithmSelector( + ConsensusAlgorithmProcess &process, + const MPI_Comm & comm) + : ConsensusAlgorithm(process, comm) + { + // Depending on the number of processes we switch between implementations. + // We reduce the threshold for debug mode to be able to test also the + // non-blocking implementation. This feature is tested by: + // tests/multigrid/transfer_matrix_free_06.with_mpi=true.with_p4est=true.with_trilinos=true.mpirun=15.output +#ifdef DEAL_II_WITH_MPI +# if DEAL_II_MPI_VERSION_GTE(3, 0) +# ifdef DEBUG + if (Utilities::MPI::n_mpi_processes(comm) > 14) +# else + if (Utilities::MPI::n_mpi_processes(comm) > 99) +# endif + consensus_algo.reset(new ConsensusAlgorithm_NBX(process, comm)); + else +# endif +#endif + consensus_algo.reset(new ConsensusAlgorithm_PEX(process, comm)); + } + + + + template + void + ConsensusAlgorithmSelector::run() + { + consensus_algo->run(); + } + + + + } // end of namespace MPI +} // end of namespace Utilities + + +DEAL_II_NAMESPACE_CLOSE + +#endif diff --git a/source/base/CMakeLists.txt b/source/base/CMakeLists.txt index be79de63f7..ed2f0f819f 100644 --- a/source/base/CMakeLists.txt +++ b/source/base/CMakeLists.txt @@ -45,6 +45,7 @@ SET(_unity_include_src logstream.cc hdf5.cc mpi.cc + mpi_consensus_algorithm.cc mu_parser_internal.cc multithread_info.cc named_selection.cc diff --git a/source/base/mpi.cc b/source/base/mpi.cc index 487461cdbe..799cbe1f81 100644 --- a/source/base/mpi.cc +++ b/source/base/mpi.cc @@ -1045,10 +1045,6 @@ namespace Utilities return owning_ranks; } - template class ConsensusAlgorithmSelector< - std::pair, - unsigned int>; - CollectiveMutex::CollectiveMutex() diff --git a/source/base/mpi_consensus_algorithm.cc b/source/base/mpi_consensus_algorithm.cc new file mode 100644 index 0000000000..b4ecfb91bf --- /dev/null +++ b/source/base/mpi_consensus_algorithm.cc @@ -0,0 +1,41 @@ +// --------------------------------------------------------------------- +// +// Copyright (C) 2005 - 2019 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. +// +// --------------------------------------------------------------------- + +#include + +DEAL_II_NAMESPACE_OPEN + + +namespace Utilities +{ + namespace MPI + { + template class ConsensusAlgorithmProcess; + + template class ConsensusAlgorithmSelector; + + + template class ConsensusAlgorithmProcess< + std::pair, + unsigned int>; + + template class ConsensusAlgorithmSelector< + std::pair, + unsigned int>; + + } // end of namespace MPI +} // end of namespace Utilities + +DEAL_II_NAMESPACE_CLOSE