Add .bcast()

include/kamping/collectives/alltoall.hpp

@@ -59,34 +59,39 @@ class Alltoall : public CRTPHelper<Communicator, Alltoall> {
             internal::has_parameter_type<internal::ParameterType::send_buf, Args...>(),
             "Missing required parameter send_buf.");
 
-        auto& send_buf_param       = internal::select_parameter_type<internal::ParameterType::send_buf>(args...);
-        auto  send_buf             = send_buf_param.get();
-        using send_value_type      = typename std::remove_reference_t<decltype(send_buf)>::value_type;
-        MPI_Datatype mpi_send_type = mpi_datatype<send_value_type>();
+        const auto& send_buf  = internal::select_parameter_type<internal::ParameterType::send_buf>(args...).get();
+        using send_value_type = typename std::remove_reference_t<decltype(send_buf)>::value_type;
+        using default_recv_value_type = std::remove_const_t<send_value_type>;
+        MPI_Datatype mpi_send_type    = mpi_datatype<send_value_type>();
 
-        using default_recv_buf_type = decltype(kamping::recv_buf(NewContainer<std::vector<send_value_type>>{}));
+        using default_recv_buf_type = decltype(kamping::recv_buf(NewContainer<std::vector<default_recv_value_type>>{}));
         auto&& recv_buf =
             internal::select_parameter_type_or_default<internal::ParameterType::recv_buf, default_recv_buf_type>(
                 std::tuple(), args...);
         using recv_value_type      = typename std::remove_reference_t<decltype(recv_buf)>::value_type;
         MPI_Datatype mpi_recv_type = mpi_datatype<recv_value_type>();
 
+        static_assert(
+            std::is_same_v<std::remove_const_t<send_value_type>, recv_value_type>,
+            "Types of send and receive buffers do not match.");
+        static_assert(!std::is_const_v<recv_value_type>, "The receive buffer must not have a const value_type.");
+        KASSERT(
+            mpi_send_type == mpi_recv_type, "The MPI receive type does not match the MPI send type.", assert::light);
+
         // Get the send and receive counts
-        int send_count = throwing_cast<int>(send_buf.size / asserting_cast<size_t>(this->underlying().size()));
-        KASSERT(mpi_send_type == mpi_recv_type, "The specified receive type does not match the send type.");
+        int send_count = throwing_cast<int>(send_buf.size() / asserting_cast<size_t>(this->underlying().size()));
 
-        size_t recv_buf_size = send_buf.size;
+        size_t recv_buf_size = send_buf.size();
         int    recv_count    = throwing_cast<int>(recv_buf_size / asserting_cast<size_t>(this->underlying().size()));
         KASSERT(send_count == recv_count, assert::light);
+        recv_buf.resize(recv_buf_size);
 
         // These KASSERTs are required to avoid a false warning from g++ in release mode
-        auto* send_buf_ptr = send_buf.ptr;
-        KASSERT(send_buf_ptr != nullptr);
-        auto* recv_buf_ptr = recv_buf.get_ptr(recv_buf_size);
-        KASSERT(recv_buf_ptr != nullptr);
+        KASSERT(send_buf.data() != nullptr, assert::light);
+        KASSERT(recv_buf.data() != nullptr, assert::light);
 
         [[maybe_unused]] int err = MPI_Alltoall(
-            send_buf.ptr, send_count, mpi_send_type, recv_buf.get_ptr(recv_buf_size), recv_count, mpi_recv_type,
+            send_buf.data(), send_count, mpi_send_type, recv_buf.data(), recv_count, mpi_recv_type,
             this->underlying().mpi_communicator());
 
         THROW_IF_MPI_ERROR(err, MPI_Alltoall);

include/kamping/collectives/bcast.hpp

@@ -0,0 +1,137 @@
+// This file is part of KaMPI.ng.
+//
+// Copyright 2022 The KaMPI.ng Authors
+//
+// KaMPI.ng is free software : you can redistribute it and/or modify it under the terms of the GNU Lesser General Public
+// License as published by the Free Software Foundation, either version 3 of the License, or (at your option) any later
+// version. KaMPI.ng is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY; without even the
+// implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU Lesser General Public License
+// for more details.
+//
+// You should have received a copy of the GNU Lesser General Public License along with KaMPI.ng.  If not, see
+// <https://www.gnu.org/licenses/>.
+
+#pragma once
+
+#include "kamping/checking_casts.hpp"
+#include "kamping/mpi_datatype.hpp"
+#include "kamping/mpi_function_wrapper_helpers.hpp"
+#include "kamping/named_parameter_selection.hpp"
+#include "kamping/parameter_factories.hpp"
+#include "kamping/parameter_objects.hpp"
+#include "kamping/parameter_type_definitions.hpp"
+#include <mpi.h>
+
+namespace kamping::internal {
+
+/// @brief CRTP mixin class for \c MPI_Bcast.
+///
+/// This class is only to be used as a super class of kamping::Communicator
+template <typename Communicator>
+class Bcast : public CRTPHelper<Communicator, Bcast> {
+public:
+    /// @brief Wrapper for \c MPI_Bcast
+    ///
+    /// This wrapper for \c MPI_Bcast sends data from the root to all other ranks.
+    /// The following buffers are required:
+    /// - \ref kamping::send_buf() containing the data that is sent to the other ranks.
+    /// The following parameters are optional:
+    /// - \ref kamping::root() specifying an alternative root. If not present, the default root of the \c Communicator
+    /// is used, see root().
+    /// - \ref kamping::recv_buf() containing a buffer for the output. Afterwards, at all other ranks, this buffer will
+    /// contain the data from the root send buffer.
+    /// @todo Describe what happens at the root
+    /// @tparam Args Automatically deducted template parameters.
+    /// @param args All required and any number of the optional buffers described above.
+    /// @return Result type wrapping the output buffer if not specified as input parameter.
+    template <typename... Args>
+    auto bcast(Args&&... args) {
+        static_assert(
+            all_parameters_are_rvalues<Args...>,
+            "All parameters have to be passed in as rvalue references, meaning that you must not hold a variable "
+            "returned by the named parameter helper functions like recv_buf().");
+
+        // Check and get all parameters.
+
+        // The parameter send_recv_buf() is required on all processes.
+        static_assert(
+            internal::has_parameter_type<internal::ParameterType::send_recv_buf, Args...>(),
+            "Missing required parameter send_recv_buf.");
+
+        const auto& send_recv_buf =
+            internal::select_parameter_type<internal::ParameterType::send_recv_buf>(args...).get();
+        using value_type = typename std::remove_reference_t<decltype(send_recv_buf)>::value_type;
+
+        auto&& root = internal::select_parameter_type_or_default<internal::ParameterType::root, internal::Root>(
+            std::tuple(this->underlying().root()), args...);
+
+        auto mpi_value_type = mpi_datatype<value_type>();
+
+        // Conduct some validity check on the parmeters.
+        KASSERT(this->underlying().is_valid_rank(root.rank()), "Invalid rank as root.", assert::light);
+
+        KASSERT(
+            this->underlying().rank() != root.rank() || send_recv_buf.size() > 0,
+            "The send_recv_buf() on root is empty.", assert::light);
+
+        KASSERT(
+            recv_buf_large_enough_on_all_processes(send_recv_buf, root.rank()),
+            "The receive buffer is too small on at least one rank.", assert::light_communication);
+
+        // Perform the broadcast.
+        // int size = 0;
+        // void* buffer = nullptr;
+        // if constexpr (internal:: ))
+
+        // The error code is unused if KTHROW is removed at compile time.
+        // KASSERT(size != 0, assert::light);
+        // KASSERT(buffer != nullptr, assert::light);
+        [[maybe_unused]] int err = MPI_Bcast(
+            send_recv_buf.data(),                      // buffer*
+            asserting_cast<int>(send_recv_buf.size()), // count
+            mpi_value_type,                            // datatype
+            root.rank(),                               // root
+            this->underlying().mpi_communicator()      // MPI_Comm comm
+        );
+        THROW_IF_MPI_ERROR(err, MPI_Bcast);
+
+        return MPIResult(
+            std::move(send_recv_buf), internal::BufferCategoryNotUsed{}, internal::BufferCategoryNotUsed{},
+            internal::BufferCategoryNotUsed{});
+    }
+
+protected:
+    Bcast() {}
+
+private:
+    /// @brief Checks if the receive buffer is large enough to receive all elements on all ranks.
+    ///
+    /// Broadcasts the size of the send buffer (which is equal to the recv_buf) from the root rank,
+    /// performs local comparison and collects the result using an allreduce.
+    /// @param send_recv_buf The send buffer on root, the receive buffer on all other ranks.
+    /// @param root The rank of the root process.
+    template <typename RecvBuf>
+    bool recv_buf_large_enough_on_all_processes(RecvBuf const& send_recv_buf, int const root) const {
+        uint64_t size = send_recv_buf.size();
+        MPI_Bcast(
+            &size,                                // src/dest buffer
+            1,                                    // size
+            mpi_datatype<decltype(size)>(),       // datatype
+            root,                                 // root
+            this->underlying().mpi_communicator() // communicator
+        );
+        bool const local_buffer_large_enough = size <= send_recv_buf.size();
+        bool       every_buffer_large_enough;
+        MPI_Allreduce(
+            &local_buffer_large_enough,           // src buffer
+            &every_buffer_large_enough,           // dest buffer
+            1,                                    // count
+            mpi_datatype<bool>(),                 // datatype
+            MPI_LAND,                             // operation
+            this->underlying().mpi_communicator() // communicator
+        );
+        return every_buffer_large_enough;
+    }
+}; // class Bcast
+
+} // namespace kamping::internal

include/kamping/collectives/reduce.hpp

@@ -66,11 +66,11 @@ class Reduce : public CRTPHelper<Communicator, Reduce> {
         auto&& root = internal::select_parameter_type_or_default<internal::ParameterType::root, internal::Root>(
             std::tuple(this->underlying().root()), args...);
 
-        auto& send_buf_param  = internal::select_parameter_type<internal::ParameterType::send_buf>(args...);
-        auto  send_buf        = send_buf_param.get();
+        const auto& send_buf  = internal::select_parameter_type<internal::ParameterType::send_buf>(args...).get();
         using send_value_type = typename std::remove_reference_t<decltype(send_buf)>::value_type;
+        using default_recv_value_type = std::remove_const_t<send_value_type>;
 
-        using default_recv_buf_type = decltype(kamping::recv_buf(NewContainer<std::vector<send_value_type>>{}));
+        using default_recv_buf_type = decltype(kamping::recv_buf(NewContainer<std::vector<default_recv_value_type>>{}));
         auto&& recv_buf =
             internal::select_parameter_type_or_default<internal::ParameterType::recv_buf, default_recv_buf_type>(
                 std::tuple(), args...);
@@ -81,17 +81,19 @@ class Reduce : public CRTPHelper<Communicator, Reduce> {
 
         // Check parameters
         static_assert(
-            std::is_same_v<send_value_type, recv_value_type>, "Types of send and receive buffers do not match.");
+            std::is_same_v<std::remove_const_t<send_value_type>, recv_value_type>,
+            "Types of send and receive buffers do not match.");
         MPI_Datatype type = mpi_datatype<send_value_type>();
 
         KASSERT(this->underlying().is_valid_rank(root.rank()), "The provided root rank is invalid.");
 
         send_value_type* recv_buf_ptr = nullptr;
         if (this->underlying().rank() == root.rank()) {
-            recv_buf_ptr = recv_buf.get_ptr(send_buf.size);
+            recv_buf.resize(send_buf.size());
+            recv_buf_ptr = recv_buf.data();
         }
         [[maybe_unused]] int err = MPI_Reduce(
-            send_buf.ptr, recv_buf_ptr, asserting_cast<int>(send_buf.size), type, operation.op(), root.rank(),
+            send_buf.data(), recv_buf_ptr, asserting_cast<int>(send_buf.size()), type, operation.op(), root.rank(),
             this->underlying().mpi_communicator());
 
         THROW_IF_MPI_ERROR(err, MPI_Reduce);

include/kamping/communicator.hpp

@@ -18,14 +18,17 @@
 
 #include "error_handling.hpp"
 #include "kamping/collectives/alltoall.hpp"
+#include "kamping/collectives/bcast.hpp"
 #include "kamping/collectives/reduce.hpp"
 #include "kamping/kassert.hpp"
 
 namespace kamping {
 
 /// @brief Wrapper for MPI communicator providing access to \ref rank() and \ref size() of the communicator. The \ref
 /// Communicator is also access point to all MPI communications provided by KaMPI.ng.
-class Communicator : public internal::Alltoall<Communicator>, public internal::Reduce<Communicator> {
+class Communicator : public internal::Alltoall<Communicator>,
+                     public internal::Reduce<Communicator>,
+                     public internal::Bcast<Communicator> {
 public:
     /// @brief Default constructor not specifying any MPI communicator and using \c MPI_COMM_WORLD by default.
     Communicator() : Communicator(MPI_COMM_WORLD) {}

include/kamping/parameter_factories.hpp

@@ -55,6 +55,44 @@ auto send_buf(const Data& data) {
     }
 }
 
+
+/// @brief Generates a buffer wrapper encapsulating a buffer used for sending or receiving based on this processes rank
+/// and the root() of the operation.
+///
+/// For example when used as parameter to \c bcast, all processes provide this buffer; on the root process it
+/// acts as the send buffer, on all other processes as the receive buffer.
+///
+/// If the underlying container provides \c data(), it is assumed that it is a container and all elements in the
+/// container are considered for the operation. In this case, the container has to provide a \c size() member functions
+/// and expose the contained \c value_type. If no \c data() member function exists, a single element is wrapped in the
+/// send_recv buffer. For receiving, the buffer is automatically resized to the correct size and thus has to provide a
+/// \c resize() method.
+///
+/// @tparam Data Data type representing the element(s) to send/receive.
+/// @param data Data (either a container which contains the elements or the element directly) to send or the buffer to
+/// receive to.
+/// @return Object referring to the storage containing the data elements to send / the received elements.
+template <typename Data>
+auto send_recv_buf(Data& data) {
+    if constexpr (internal::has_data_member_v<Data>) {
+        return internal::UserAllocatedContainerBasedBuffer<Data, internal::ParameterType::send_recv_buf>(data);
+    } else {
+        return internal::SingleElementModifyableBuffer<Data, internal::ParameterType::send_recv_buf>(data);
+    }
+}
+
+/// @brief Generates buffer wrapper based on a container for the receive buffer, i.e. the underlying storage
+/// will contained the received elements when the \c MPI call has been completed.
+/// The storage is allocated by the library and encapsulated in a container of type Container.
+/// The underlying container must provide a \c data(), \c resize() and \c size() member function and expose the
+/// contained \c value_type
+/// @tparam Container Container type which contains the received elements.
+/// @return Object referring to the storage containing the send displacements.
+template <typename Container>
+auto send_recv_buf(NewContainer<Container>&&) {
+    return internal::LibAllocatedContainerBasedBuffer<Container, internal::ParameterType::send_recv_buf>();
+}
+
 /// @brief Generates buffer wrapper based on a container for the send counts, i.e. the underlying storage must contain
 /// the send counts to each relevant PE.
 ///
@@ -167,13 +205,13 @@ auto recv_counts_out(Container& container) {
     return internal::UserAllocatedContainerBasedBuffer<Container, internal::ParameterType::recv_counts>(container);
 }
 
-///@brief Generates buffer wrapper based on a container for the receive counts, i.e. the underlying storage
+/// @brief Generates buffer wrapper based on a container for the receive counts, i.e. the underlying storage
 /// will contained the receive counts when the \c MPI call has been completed.
 /// The storage is allocated by the library and encapsulated in a container of type Container.
 /// The underlying container must provide a \c data(), \c resize() and \c size() member function and expose the
 /// contained \c value_type
-///@tparam Container Container type which contains the send displacements.
-///@return Object referring to the storage containing the receive counts.
+/// @tparam Container Container type which contains the send displacements.
+/// @return Object referring to the storage containing the receive counts.
 template <typename Container>
 auto recv_counts_out(NewContainer<Container>&&) {
     return internal::LibAllocatedContainerBasedBuffer<Container, internal::ParameterType::recv_counts>();