Homec4science
Diffusion akantu (bugfixes/64bit-integer-corrections) Browse

rAKA/src/synchronizer/communicator_mpi_inline_impl.hh
8a2ff789cf15bugfixes/64bit-integer…

Last Change
Raw File

communicator_mpi_inline_impl.hh
View Options

/**
* Copyright (©) 2017-2023 EPFL (Ecole Polytechnique Fédérale de Lausanne)
* Laboratory (LSMS - Laboratoire de Simulation en Mécanique des Solides)
*
* This file is part of Akantu
*
* Akantu 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.
*
* Akantu 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 Akantu. If not, see <http://www.gnu.org/licenses/>.
*/
/* -------------------------------------------------------------------------- */
#include "aka_iterators.hh"
#include "communicator.hh"
#include "mpi_communicator_data.hh"
/* -------------------------------------------------------------------------- */
#include <memory>
#include <type_traits>
#include <unordered_map>
#include <vector>
/* -------------------------------------------------------------------------- */
#include <mpi.h>
/* -------------------------------------------------------------------------- */
#if (defined(__GNUC__) || defined(__GNUG__))
#if AKA_GCC_VERSION < 60000
namespace std {
template <> struct hash<akantu::SynchronizerOperation> {
using argument_type = akantu::SynchronizerOperation;
size_t operator()(const argument_type & e) const noexcept {
auto ue = underlying_type_t<argument_type>(e);
return uh(ue);
}
private:
const hash<underlying_type_t<argument_type>> uh{};
};
} // namespace std
#endif
#endif
namespace akantu {
class CommunicationRequestMPI : public InternalCommunicationRequest {
public:
CommunicationRequestMPI(Idx source, Idx dest)
: InternalCommunicationRequest(source, dest),
request(std::make_unique<MPI_Request>()) {}
MPI_Request & getMPIRequest() { return *request; };
private:
std::unique_ptr<MPI_Request> request;
};
namespace {
template <typename T> inline MPI_Datatype getMPIDatatype();
MPI_Op getMPISynchronizerOperation(SynchronizerOperation op) {
std::unordered_map<SynchronizerOperation, MPI_Op> _operations{
{SynchronizerOperation::_sum, MPI_SUM},
{SynchronizerOperation::_min, MPI_MIN},
{SynchronizerOperation::_max, MPI_MAX},
{SynchronizerOperation::_prod, MPI_PROD},
{SynchronizerOperation::_land, MPI_LAND},
{SynchronizerOperation::_band, MPI_BAND},
{SynchronizerOperation::_lor, MPI_LOR},
{SynchronizerOperation::_bor, MPI_BOR},
{SynchronizerOperation::_lxor, MPI_LXOR},
{SynchronizerOperation::_bxor, MPI_BXOR},
{SynchronizerOperation::_min_loc, MPI_MINLOC},
{SynchronizerOperation::_max_loc, MPI_MAXLOC},
{SynchronizerOperation::_null, MPI_OP_NULL}};
return _operations[op];
}
template <typename T> MPI_Datatype inline getMPIDatatype() {
return MPI_DATATYPE_NULL;
}
#define SPECIALIZE_MPI_DATATYPE(type, mpi_type) \
template <> MPI_Datatype inline getMPIDatatype<type>() { return mpi_type; }
#define COMMA ,
SPECIALIZE_MPI_DATATYPE(char, MPI_CHAR)
SPECIALIZE_MPI_DATATYPE(std::uint8_t, MPI_UINT8_T)
SPECIALIZE_MPI_DATATYPE(float, MPI_FLOAT)
SPECIALIZE_MPI_DATATYPE(double, MPI_DOUBLE)
SPECIALIZE_MPI_DATATYPE(long double, MPI_LONG_DOUBLE)
SPECIALIZE_MPI_DATATYPE(signed int, MPI_INT)
SPECIALIZE_MPI_DATATYPE(unsigned int, MPI_UNSIGNED)
SPECIALIZE_MPI_DATATYPE(signed long int, MPI_LONG)
SPECIALIZE_MPI_DATATYPE(unsigned long int, MPI_UNSIGNED_LONG)
SPECIALIZE_MPI_DATATYPE(signed long long int, MPI_LONG_LONG)
SPECIALIZE_MPI_DATATYPE(unsigned long long int, MPI_UNSIGNED_LONG_LONG)
SPECIALIZE_MPI_DATATYPE(SCMinMaxLoc<double COMMA int>, MPI_DOUBLE_INT)
SPECIALIZE_MPI_DATATYPE(SCMinMaxLoc<float COMMA int>, MPI_FLOAT_INT)
SPECIALIZE_MPI_DATATYPE(bool, MPI_CXX_BOOL)
template <> MPI_Datatype inline getMPIDatatype<NodeFlag>() {
return getMPIDatatype<std::underlying_type_t<NodeFlag>>();
}
inline int getMPISource(int src) {
if (src == _any_source) {
return MPI_ANY_SOURCE;
}
return src;
}
decltype(auto) convertRequests(std::vector<CommunicationRequest> & requests) {
std::vector<MPI_Request> mpi_requests(requests.size());
for (auto && request_pair : zip(requests, mpi_requests)) {
auto && req = std::get<0>(request_pair);
auto && mpi_req = std::get<1>(request_pair);
mpi_req = aka::as_type<CommunicationRequestMPI>(req.getInternal())
.getMPIRequest();
}
return mpi_requests;
}
} // namespace
// this is ugly but shorten the code a lot
#define MPIDATA \
(*reinterpret_cast<MPICommunicatorData *>(communicator_data.get()))
/* -------------------------------------------------------------------------- */
template <class Integer, class Function, class Function_c>
requires std::is_integral_v<std::decay_t<Integer>>
static void dispatchLargeFunctions(Function && function,
Function_c && function_c [[maybe_unused]],
Integer && size) {
#if MPI_VERSION >= 4
if constexpr (sizeof(Integer) == 4) {
function(size);
} else {
function_c(size);
}
#else
int size_ = size;
AKANTU_DEBUG_ASSERT(
(size >= 0) and (size < std::numeric_limits<int>::max()),
"In debug you will most probably never see this message, but your "
"MPI message is to big for int32_t");
function(size_);
if constexpr (not std::is_const_v<std::remove_reference_t<Integer>>) {
size = size_;
}
#endif
}
/* -------------------------------------------------------------------------- */
/* Implementation */
/* -------------------------------------------------------------------------- */
/* -------------------------------------------------------------------------- */
Communicator::Communicator(int & /*argc*/, char **& /*argv*/,
const private_member & m)
: Communicator(m) {}
/* -------------------------------------------------------------------------- */
Communicator::Communicator(const private_member & /*unused*/)
: communicator_data(std::make_unique<MPICommunicatorData>()) {}
/* -------------------------------------------------------------------------- */
template <typename T, typename integer>
requires std::signed_integral<integer>
void Communicator::sendImpl(const T * buffer, integer size, int receiver,
int tag, const CommunicationMode & mode) const {
MPI_Comm communicator = MPIDATA.getMPICommunicator();
MPI_Datatype type = getMPIDatatype<T>();
dispatchLargeFunctions(
[&](auto && size) {
switch (mode) {
case CommunicationMode::_auto:
MPI_Send(buffer, size, type, receiver, tag, communicator);
break;
case CommunicationMode::_synchronous:
MPI_Ssend(buffer, size, type, receiver, tag, communicator);
break;
case CommunicationMode::_ready:
MPI_Rsend(buffer, size, type, receiver, tag, communicator);
break;
}
},
[&](auto && size) {
switch (mode) {
case CommunicationMode::_auto:
MPI_Send_c(buffer, size, type, receiver, tag, communicator);
break;
case CommunicationMode::_synchronous:
MPI_Ssend_c(buffer, size, type, receiver, tag, communicator);
break;
case CommunicationMode::_ready:
MPI_Rsend_c(buffer, size, type, receiver, tag, communicator);
break;
}
},
size);
}
/* -------------------------------------------------------------------------- */
template <typename T, typename integer>
requires std::signed_integral<integer>
void Communicator::receiveImpl(T * buffer, integer size, int sender,
int tag) const {
MPI_Comm communicator = MPIDATA.getMPICommunicator();
MPI_Status status;
MPI_Datatype type = getMPIDatatype<T>();
dispatchLargeFunctions(
[&](auto && size) {
MPI_Recv(buffer, size, type, getMPISource(sender), tag, communicator,
&status);
},
[&](auto && size) {
MPI_Recv_c(buffer, size, type, getMPISource(sender), tag, communicator,
&status);
},
size);
}
/* -------------------------------------------------------------------------- */
template <typename T, typename integer>
requires std::signed_integral<integer>
CommunicationRequest
Communicator::asyncSendImpl(const T * buffer, integer size, int receiver,
int tag, const CommunicationMode & mode) const {
MPI_Comm communicator = MPIDATA.getMPICommunicator();
auto * request = new CommunicationRequestMPI(whoAmI(), receiver);
MPI_Request & req = request->getMPIRequest();
MPI_Datatype type = getMPIDatatype<T>();
dispatchLargeFunctions(
[&](auto && size) {
switch (mode) {
case CommunicationMode::_auto:
MPI_Isend(buffer, size, type, receiver, tag, communicator, &req);
break;
case CommunicationMode::_synchronous:
MPI_Issend(buffer, size, type, receiver, tag, communicator, &req);
break;
case CommunicationMode::_ready:
MPI_Irsend(buffer, size, type, receiver, tag, communicator, &req);
break;
}
},
[&](auto && size) {
switch (mode) {
case CommunicationMode::_auto:
MPI_Isend_c(buffer, size, type, receiver, tag, communicator, &req);
break;
case CommunicationMode::_synchronous:
MPI_Issend_c(buffer, size, type, receiver, tag, communicator, &req);
break;
case CommunicationMode::_ready:
MPI_Irsend_c(buffer, size, type, receiver, tag, communicator, &req);
break;
}
},
size);
return std::shared_ptr<InternalCommunicationRequest>(request);
}
/* -------------------------------------------------------------------------- */
template <typename T, typename integer>
requires std::signed_integral<integer>
CommunicationRequest Communicator::asyncReceiveImpl(T * buffer, integer size,
int sender, int tag) const {
MPI_Comm communicator = MPIDATA.getMPICommunicator();
auto * request = new CommunicationRequestMPI(sender, whoAmI());
MPI_Datatype type = getMPIDatatype<T>();
MPI_Request & req = request->getMPIRequest();
dispatchLargeFunctions(
[&](auto && size) {
MPI_Irecv(buffer, size, type, getMPISource(sender), tag, communicator,
&req);
},
[&](auto && size) {
MPI_Irecv(buffer, size, type, getMPISource(sender), tag, communicator,
&req);
},
size);
return std::shared_ptr<InternalCommunicationRequest>(request);
}
/* -------------------------------------------------------------------------- */
template <typename T, typename integer>
requires std::signed_integral<integer>
void Communicator::reduceImpl(T * values, integer nb_values,
SynchronizerOperation op, int root) const {
MPI_Comm communicator = MPIDATA.getMPICommunicator();
MPI_Datatype type = getMPIDatatype<T>();
dispatchLargeFunctions(
[&](auto && nb_values) {
MPI_Reduce(MPI_IN_PLACE, values, nb_values, type,
getMPISynchronizerOperation(op), root, communicator);
},
[&](auto && nb_values) {
MPI_Reduce_c(MPI_IN_PLACE, values, nb_values, type,
getMPISynchronizerOperation(op), root, communicator);
},
nb_values);
}
/* -------------------------------------------------------------------------- */
template <typename T, typename integer>
requires std::signed_integral<integer>
void Communicator::allReduceImpl(T * values, integer nb_values,
SynchronizerOperation op) const {
MPI_Comm communicator = MPIDATA.getMPICommunicator();
MPI_Datatype type = getMPIDatatype<T>();
dispatchLargeFunctions(
[&](auto && nb_values) {
MPI_Allreduce(MPI_IN_PLACE, values, nb_values, type,
getMPISynchronizerOperation(op), communicator);
},
[&](auto && nb_values) {
MPI_Allreduce_c(MPI_IN_PLACE, values, nb_values, type,
getMPISynchronizerOperation(op), communicator);
},
nb_values);
}
/* -------------------------------------------------------------------------- */
template <typename T, typename integer>
requires std::signed_integral<integer>
void Communicator::scanImpl(T * values, T * result, integer nb_values,
SynchronizerOperation op) const {
MPI_Comm communicator = MPIDATA.getMPICommunicator();
MPI_Datatype type = getMPIDatatype<T>();
if (values == result) {
values = reinterpret_cast<T *>(MPI_IN_PLACE);
}
dispatchLargeFunctions(
[&](auto && nb_values) {
MPI_Scan(values, result, nb_values, type,
getMPISynchronizerOperation(op), communicator);
},
[&](auto && nb_values) {
MPI_Scan_c(values, result, nb_values, type,
getMPISynchronizerOperation(op), communicator);
},
nb_values);
}
/* -------------------------------------------------------------------------- */
template <typename T, typename integer>
requires std::signed_integral<integer>
void Communicator::exclusiveScanImpl(T * values, T * result, integer nb_values,
SynchronizerOperation op) const {
MPI_Comm communicator = MPIDATA.getMPICommunicator();
MPI_Datatype type = getMPIDatatype<T>();
if (values == result) {
values = reinterpret_cast<T *>(MPI_IN_PLACE);
}
dispatchLargeFunctions(
[&](auto && nb_values) {
MPI_Exscan(values, result, nb_values, type,
getMPISynchronizerOperation(op), communicator);
},
[&](auto && nb_values) {
MPI_Exscan_c(values, result, nb_values, type,
getMPISynchronizerOperation(op), communicator);
},
nb_values);
if (whoAmI() == 0) {
result[0] = T();
}
}
/* -------------------------------------------------------------------------- */
template <typename T, typename integer>
requires std::signed_integral<integer>
void Communicator::allGatherImpl(T * values, integer nb_values) const {
MPI_Comm communicator = MPIDATA.getMPICommunicator();
MPI_Datatype type = getMPIDatatype<T>();
dispatchLargeFunctions(
[&](auto && nb_values) {
MPI_Allgather(MPI_IN_PLACE, nb_values, type, values, nb_values, type,
communicator);
},
[&](auto && nb_values) {
MPI_Allgather_c(MPI_IN_PLACE, nb_values, type, values, nb_values, type,
communicator);
},
nb_values);
}
/* -------------------------------------------------------------------------- */
template <typename T, typename integer>
requires std::signed_integral<integer>
void Communicator::allGatherVImpl(T * values, const integer * nb_values) const {
MPI_Comm communicator = MPIDATA.getMPICommunicator();
#if MPI_VERSION >= 4
std::vector<integer> displs(getNbProc());
#else
std::vector<int> displs(getNbProc());
#endif
for (int i = 1; i < getNbProc(); ++i) {
displs[i] = displs[i - 1] + nb_values[i - 1];
}
MPI_Datatype type = getMPIDatatype<T>();
dispatchLargeFunctions(
[&](auto && size) {
#if MPI_VERSION >= 4
int * nb_values_ = nb_values;
#else
std::vector<int> nb_values_vect(nb_values, nb_values + displs.size());
int * nb_values_ = nb_values_vect.data();
#endif
MPI_Allgatherv(MPI_IN_PLACE, size, type, values, nb_values_,
displs.data(), type, communicator);
},
[&](auto && size) {
MPI_Allgatherv_c(MPI_IN_PLACE, size, type, values, nb_values,
displs.data(), type, communicator);
},
*nb_values);
}
/* -------------------------------------------------------------------------- */
template <typename T, typename integer>
requires std::signed_integral<integer>
void Communicator::gatherImpl(T * values, integer nb_values, int root) const {
MPI_Comm communicator = MPIDATA.getMPICommunicator();
auto prank = whoAmI();
T * send_buf = (prank == root) ? static_cast<T *>(MPI_IN_PLACE) : values;
T * recv_buf = (prank == root) ? values : nullptr;
MPI_Datatype type = getMPIDatatype<T>();
dispatchLargeFunctions(
[&](auto && nb_values) {
MPI_Gather(send_buf, nb_values, type, recv_buf, nb_values, type, root,
communicator);
},
[&](auto && nb_values) {
MPI_Gather_c(send_buf, nb_values, type, recv_buf, nb_values, type, root,
communicator);
},
nb_values);
}
/* -------------------------------------------------------------------------- */
template <typename T, typename integer>
requires std::signed_integral<integer>
void Communicator::gatherImpl(T * values, integer nb_values, T * gathered,
integer nb_gathered) const {
MPI_Comm communicator = MPIDATA.getMPICommunicator();
T * send_buf = values;
T * recv_buf = gathered;
if (nb_gathered == 0) {
nb_gathered = nb_values;
}
MPI_Datatype type = getMPIDatatype<T>();
dispatchLargeFunctions(
[&](auto && nb_values) {
MPI_Gather(send_buf, nb_values, type, recv_buf, nb_gathered, type,
whoAmI(), communicator);
},
[&](auto && nb_values) {
MPI_Gather_c(send_buf, nb_values, type, recv_buf, nb_gathered, type,
whoAmI(), communicator);
},
nb_values);
}
/* -------------------------------------------------------------------------- */
template <typename T, typename integer>
requires std::signed_integral<integer>
void Communicator::gatherVImpl(T * values, integer * nb_values,
int root) const {
MPI_Comm communicator = MPIDATA.getMPICommunicator();
auto psize = getNbProc();
auto prank = whoAmI();
#if MPI_VERSION >= 4
std::vector<integer> displs;
#else
std::vector<int> displs;
#endif
if (prank == root) {
displs.resize(psize);
for (auto i : arange(1, psize)) {
displs[i] = displs[i - 1] + nb_values[i - 1];
}
}
T * send_buf = (prank == root) ? static_cast<T *>(MPI_IN_PLACE) : values;
T * recv_buf = (prank == root) ? values : nullptr;
MPI_Datatype type = getMPIDatatype<T>();
dispatchLargeFunctions(
[&](auto && size) {
#if MPI_VERSION >= 4
int * nb_values_ = nb_values;
#else
std::vector<int> nb_values_vect(nb_values, nb_values + displs.size());
int * nb_values_ = nb_values_vect.data();
#endif
MPI_Gatherv(send_buf, size, type, recv_buf, nb_values_, displs.data(),
type, root, communicator);
},
[&](auto && size) {
MPI_Gatherv_c(send_buf, size, type, recv_buf, nb_values, displs.data(),
type, root, communicator);
},
*nb_values);
}
/* -------------------------------------------------------------------------- */
template <typename T, typename integer>
requires std::signed_integral<integer>
void Communicator::broadcastImpl(T * values, integer nb_values,
int root) const {
MPI_Comm communicator = MPIDATA.getMPICommunicator();
MPI_Datatype type = getMPIDatatype<T>();
dispatchLargeFunctions(
[&](auto && nb_values) {
MPI_Bcast(values, nb_values, type, root, communicator);
},
[&](auto && nb_values) {
MPI_Bcast_c(values, nb_values, type, root, communicator);
},
nb_values);
}
/* -------------------------------------------------------------------------- */
template <typename T>
void Communicator::probe(int sender, int tag,
CommunicationStatus & status) const {
MPI_Comm communicator = MPIDATA.getMPICommunicator();
MPI_Status mpi_status;
MPI_Probe(getMPISource(sender), tag, communicator, &mpi_status);
MPI_Datatype type = getMPIDatatype<T>();
Int count{};
dispatchLargeFunctions(
[&](auto && count) { MPI_Get_count(&mpi_status, type, &count); },
[&](auto && count) { MPI_Get_count_x(&mpi_status, type, &count); },
count);
status.setSource(mpi_status.MPI_SOURCE);
status.setTag(mpi_status.MPI_TAG);
status.setSize(count);
}
/* -------------------------------------------------------------------------- */
template <typename T>
bool Communicator::asyncProbe(int sender, int tag,
CommunicationStatus & status) const {
MPI_Comm communicator = MPIDATA.getMPICommunicator();
MPI_Status mpi_status;
int test;
MPI_Iprobe(getMPISource(sender), tag, communicator, &test, &mpi_status);
if (not test) {
return false;
}
MPI_Datatype type = getMPIDatatype<T>();
Int count{};
dispatchLargeFunctions(
[&](auto && count) { MPI_Get_count(&mpi_status, type, &count); },
[&](auto && count) { MPI_Get_count_x(&mpi_status, type, &count); },
count);
status.setSource(mpi_status.MPI_SOURCE);
status.setTag(mpi_status.MPI_TAG);
status.setSize(count);
return true;
}
/* -------------------------------------------------------------------------- */
bool Communicator::test(CommunicationRequest & request) {
MPI_Status status;
int flag;
auto & req_mpi = aka::as_type<CommunicationRequestMPI>(request.getInternal());
MPI_Request & req = req_mpi.getMPIRequest();
MPI_Test(&req, &flag, &status);
return flag != 0;
}
/* -------------------------------------------------------------------------- */
bool Communicator::testAll(std::vector<CommunicationRequest> & requests) {
for (auto & request : requests) {
if (not test(request)) {
return false;
}
}
return true;
}
/* -------------------------------------------------------------------------- */
void Communicator::wait(CommunicationRequest & request) {
MPI_Status status;
auto & req_mpi = aka::as_type<CommunicationRequestMPI>(request.getInternal());
MPI_Request & req = req_mpi.getMPIRequest();
MPI_Wait(&req, &status);
}
/* -------------------------------------------------------------------------- */
void Communicator::waitAll(std::vector<CommunicationRequest> & requests) {
auto && mpi_requests = convertRequests(requests);
MPI_Waitall(mpi_requests.size(), mpi_requests.data(), MPI_STATUSES_IGNORE);
}
/* -------------------------------------------------------------------------- */
int Communicator::waitAny(std::vector<CommunicationRequest> & requests) {
auto && mpi_requests = convertRequests(requests);
int pos;
MPI_Waitany(mpi_requests.size(), mpi_requests.data(), &pos,
MPI_STATUSES_IGNORE);
if (pos != MPI_UNDEFINED) {
return pos;
}
return -1;
}
/* -------------------------------------------------------------------------- */
void Communicator::barrier() const {
MPI_Comm communicator = MPIDATA.getMPICommunicator();
MPI_Barrier(communicator);
}
/* -------------------------------------------------------------------------- */
CommunicationRequest Communicator::asyncBarrier() const {
MPI_Comm communicator = MPIDATA.getMPICommunicator();
auto * request = new CommunicationRequestMPI(0, 0);
MPI_Request & req = request->getMPIRequest();
MPI_Ibarrier(communicator, &req);
return std::shared_ptr<InternalCommunicationRequest>(request);
}
/* -------------------------------------------------------------------------- */
int Communicator::getMaxTag() const { return MPIDATA.getMaxTag(); }
int Communicator::
getMinTag() // NOLINT(readability-convert-member-functions-to-static)
const {
return 0;
}
/* -------------------------------------------------------------------------- */
int Communicator::getNbProc() const { return MPIDATA.size(); }
int Communicator::whoAmI() const { return MPIDATA.rank(); }
} // namespace akantu
c4science · Help