diff --git a/src/synchronizer/communication_buffer.hh b/src/synchronizer/communication_buffer.hh
index 5e0b0da8d..54586466c 100644
--- a/src/synchronizer/communication_buffer.hh
+++ b/src/synchronizer/communication_buffer.hh
@@ -1,182 +1,190 @@
/**
* @file communication_buffer.hh
*
* @author Guillaume Anciaux <guillaume.anciaux@epfl.ch>
* @author Nicolas Richart <nicolas.richart@epfl.ch>
*
* @date creation: Fri Jun 18 2010
* @date last modification: Wed Nov 08 2017
*
* @brief Buffer for packing and unpacking data
*
* @section LICENSE
*
* Copyright (©) 2010-2018 EPFL (Ecole Polytechnique Fédérale de Lausanne)
* Laboratory (LSMS - Laboratoire de Simulation en Mécanique des Solides)
*
* 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_array.hh"
#include "aka_common.hh"
#include "element.hh"
#ifndef __AKANTU_COMMUNICATION_BUFFER_HH__
#define __AKANTU_COMMUNICATION_BUFFER_HH__
namespace akantu {
template <bool is_static = true> class CommunicationBufferTemplated {
/* ------------------------------------------------------------------------ */
/* Constructors/Destructors */
/* ------------------------------------------------------------------------ */
public:
explicit CommunicationBufferTemplated(UInt size) : buffer(size, 1) {
ptr_pack = buffer.storage();
ptr_unpack = buffer.storage();
};
CommunicationBufferTemplated() : CommunicationBufferTemplated(0) {}
CommunicationBufferTemplated(const CommunicationBufferTemplated & other) {
buffer = other.buffer;
ptr_pack = buffer.storage();
ptr_unpack = buffer.storage();
}
CommunicationBufferTemplated &
operator=(const CommunicationBufferTemplated & other) {
if (this != &other) {
buffer = other.buffer;
ptr_pack = buffer.storage();
ptr_unpack = buffer.storage();
}
return *this;
}
virtual ~CommunicationBufferTemplated() = default;
/* ------------------------------------------------------------------------ */
/* Methods */
/* ------------------------------------------------------------------------ */
public:
/// reset to "empty"
inline void reset();
- /// resize the internal buffer
+ /// resize the internal buffer do not allocate on dynamic buffers
inline void resize(UInt size);
+ /// resize the internal buffer allocate always
+ inline void reserve(UInt size);
+
/// clear buffer context
inline void clear();
private:
inline void packResize(UInt size);
/* ------------------------------------------------------------------------ */
/* Accessors */
/* ------------------------------------------------------------------------ */
public:
inline char * storage() { return buffer.storage(); };
inline const char * storage() const { return buffer.storage(); };
/* ------------------------------------------------------------------------ */
/* Operators */
/* ------------------------------------------------------------------------ */
public:
/// printing tool
template <typename T> inline std::string extractStream(UInt packet_size);
/// packing data
template <typename T>
inline CommunicationBufferTemplated & operator<<(const T & to_pack);
template <typename T>
inline CommunicationBufferTemplated & operator<<(const Vector<T> & to_pack);
template <typename T>
inline CommunicationBufferTemplated & operator<<(const Matrix<T> & to_pack);
template <typename T>
inline CommunicationBufferTemplated &
operator<<(const std::vector<T> & to_pack);
/// unpacking data
template <typename T>
inline CommunicationBufferTemplated & operator>>(T & to_unpack);
template <typename T>
inline CommunicationBufferTemplated & operator>>(Vector<T> & to_unpack);
template <typename T>
inline CommunicationBufferTemplated & operator>>(Matrix<T> & to_unpack);
template <typename T>
inline CommunicationBufferTemplated & operator>>(std::vector<T> & to_unpack);
inline CommunicationBufferTemplated & operator<<(const std::string & to_pack);
inline CommunicationBufferTemplated & operator>>(std::string & to_unpack);
private:
template <typename T> inline void packIterable(T & to_pack);
template <typename T> inline void unpackIterable(T & to_pack);
/* ------------------------------------------------------------------------ */
/* Accessor */
/* ------------------------------------------------------------------------ */
public:
template <typename T> static inline UInt sizeInBuffer(const T & data);
template <typename T> static inline UInt sizeInBuffer(const Vector<T> & data);
template <typename T> static inline UInt sizeInBuffer(const Matrix<T> & data);
template <typename T>
static inline UInt sizeInBuffer(const std::vector<T> & data);
static inline UInt sizeInBuffer(const std::string & data);
/// return the size in bytes of the stored values
inline UInt getPackedSize() const { return ptr_pack - buffer.storage(); };
/// return the size in bytes of data left to be unpacked
inline UInt getLeftToUnpack() const {
return buffer.size() - (ptr_unpack - buffer.storage());
};
/// return the global size allocated
inline UInt size() const { return buffer.size(); };
+ /// is the buffer empty
+ inline bool empty() const {
+ return (getPackedSize() == 0) and (getLeftToUnpack() == 0);
+ }
+
/* ------------------------------------------------------------------------ */
/* Class Members */
/* ------------------------------------------------------------------------ */
private:
/// current position for packing
char * ptr_pack;
/// current position for unpacking
char * ptr_unpack;
/// storing buffer
Array<char> buffer;
};
/* -------------------------------------------------------------------------- */
/* inline functions */
/* -------------------------------------------------------------------------- */
#if defined(AKANTU_INCLUDE_INLINE_IMPL)
#include "communication_buffer_inline_impl.cc"
#endif
using CommunicationBuffer = CommunicationBufferTemplated<true>;
using DynamicCommunicationBuffer = CommunicationBufferTemplated<false>;
} // namespace akantu
#endif /* __AKANTU_COMMUNICATION_BUFFER_HH__ */
diff --git a/src/synchronizer/communication_buffer_inline_impl.cc b/src/synchronizer/communication_buffer_inline_impl.cc
index d242a4b42..144e07d98 100644
--- a/src/synchronizer/communication_buffer_inline_impl.cc
+++ b/src/synchronizer/communication_buffer_inline_impl.cc
@@ -1,318 +1,327 @@
/**
* @file communication_buffer_inline_impl.cc
*
* @author Guillaume Anciaux <guillaume.anciaux@epfl.ch>
* @author Nicolas Richart <nicolas.richart@epfl.ch>
*
* @date creation: Thu Apr 14 2011
* @date last modification: Wed Nov 08 2017
*
* @brief CommunicationBuffer inline implementation
*
* @section LICENSE
*
* Copyright (©) 2010-2018 EPFL (Ecole Polytechnique Fédérale de Lausanne)
* Laboratory (LSMS - Laboratoire de Simulation en Mécanique des Solides)
*
* 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/>.
*
*/
/* -------------------------------------------------------------------------- */
template <bool is_static>
template <typename T>
inline UInt CommunicationBufferTemplated<is_static>::sizeInBuffer(const T &) {
return sizeof(T);
}
template <bool is_static>
template <typename T>
inline UInt
CommunicationBufferTemplated<is_static>::sizeInBuffer(const Vector<T> & data) {
UInt size = data.size() * sizeof(T);
return size;
}
template <bool is_static>
template <typename T>
inline UInt
CommunicationBufferTemplated<is_static>::sizeInBuffer(const Matrix<T> & data) {
UInt size = data.size() * sizeof(T);
return size;
}
template <bool is_static>
template <typename T>
inline UInt CommunicationBufferTemplated<is_static>::sizeInBuffer(
const std::vector<T> & data) {
UInt size = data.size() * sizeof(T) + sizeof(size_t);
return size;
}
template <bool is_static>
inline UInt CommunicationBufferTemplated<is_static>::sizeInBuffer(
const std::string & data) {
UInt size = data.size() * sizeof(std::string::value_type) + sizeof(size_t);
return size;
}
/* -------------------------------------------------------------------------- */
template <bool is_static>
inline void CommunicationBufferTemplated<is_static>::packResize(UInt size) {
- if (!is_static) {
+ if (not is_static) {
char * values = buffer.storage();
- buffer.resize(buffer.size() + size);
- ptr_pack = buffer.storage() + (ptr_pack - values);
+ auto nb_packed = ptr_pack - values;
+ if(buffer.size() > nb_packed + size) return;
+
+ buffer.resize(nb_packed + size);
+ ptr_pack = buffer.storage() + nb_packed;
ptr_unpack = buffer.storage() + (ptr_unpack - values);
}
}
/* -------------------------------------------------------------------------- */
template <bool is_static>
template <typename T>
inline CommunicationBufferTemplated<is_static> &
CommunicationBufferTemplated<is_static>::operator<<(const T & to_pack) {
UInt size = sizeInBuffer(to_pack);
packResize(size);
AKANTU_DEBUG_ASSERT(
(buffer.storage() + buffer.size()) >= (ptr_pack + size),
"Packing too much data in the CommunicationBufferTemplated");
memcpy(ptr_pack, reinterpret_cast<const char *>(&to_pack), size);
ptr_pack += size;
return *this;
}
/* -------------------------------------------------------------------------- */
template <bool is_static>
template <typename T>
inline CommunicationBufferTemplated<is_static> &
CommunicationBufferTemplated<is_static>::operator>>(T & to_unpack) {
UInt size = sizeInBuffer(to_unpack);
alignas(alignof(T)) std::array<char, sizeof(T)> aligned_ptr;
memcpy(aligned_ptr.data(), ptr_unpack, size);
auto * tmp = reinterpret_cast<T *>(aligned_ptr.data());
AKANTU_DEBUG_ASSERT(
(buffer.storage() + buffer.size()) >= (ptr_unpack + size),
"Unpacking too much data in the CommunicationBufferTemplated");
to_unpack = *tmp;
// memcpy(reinterpret_cast<char *>(&to_unpack), ptr_unpack, size);
ptr_unpack += size;
return *this;
}
/* -------------------------------------------------------------------------- */
/* Specialization */
/* -------------------------------------------------------------------------- */
/**
* Vector
*/
/* -------------------------------------------------------------------------- */
template <bool is_static>
template <typename T>
inline CommunicationBufferTemplated<is_static> &
CommunicationBufferTemplated<is_static>::operator<<(const Vector<T> & to_pack) {
UInt size = sizeInBuffer(to_pack);
packResize(size);
AKANTU_DEBUG_ASSERT(
(buffer.storage() + buffer.size()) >= (ptr_pack + size),
"Packing too much data in the CommunicationBufferTemplated");
memcpy(ptr_pack, to_pack.storage(), size);
ptr_pack += size;
return *this;
}
/* -------------------------------------------------------------------------- */
template <bool is_static>
template <typename T>
inline CommunicationBufferTemplated<is_static> &
CommunicationBufferTemplated<is_static>::operator>>(Vector<T> & to_unpack) {
UInt size = sizeInBuffer(to_unpack);
AKANTU_DEBUG_ASSERT(
(buffer.storage() + buffer.size()) >= (ptr_unpack + size),
"Unpacking too much data in the CommunicationBufferTemplated");
memcpy(to_unpack.storage(), ptr_unpack, size);
ptr_unpack += size;
return *this;
}
/**
* Matrix
*/
/* -------------------------------------------------------------------------- */
template <bool is_static>
template <typename T>
inline CommunicationBufferTemplated<is_static> &
CommunicationBufferTemplated<is_static>::operator<<(const Matrix<T> & to_pack) {
UInt size = sizeInBuffer(to_pack);
packResize(size);
AKANTU_DEBUG_ASSERT(
(buffer.storage() + buffer.size()) >= (ptr_pack + size),
"Packing too much data in the CommunicationBufferTemplated");
memcpy(ptr_pack, to_pack.storage(), size);
ptr_pack += size;
return *this;
}
/* -------------------------------------------------------------------------- */
template <bool is_static>
template <typename T>
inline CommunicationBufferTemplated<is_static> &
CommunicationBufferTemplated<is_static>::operator>>(Matrix<T> & to_unpack) {
UInt size = sizeInBuffer(to_unpack);
AKANTU_DEBUG_ASSERT(
(buffer.storage() + buffer.size()) >= (ptr_unpack + size),
"Unpacking too much data in the CommunicationBufferTemplated");
memcpy(to_unpack.storage(), ptr_unpack, size);
ptr_unpack += size;
return *this;
}
/* -------------------------------------------------------------------------- */
template <bool is_static>
template <typename T>
inline void CommunicationBufferTemplated<is_static>::packIterable(T & to_pack) {
operator<<(size_t(to_pack.size()));
auto it = to_pack.begin();
auto end = to_pack.end();
for (; it != end; ++it)
operator<<(*it);
}
/* -------------------------------------------------------------------------- */
template <bool is_static>
template <typename T>
inline void
CommunicationBufferTemplated<is_static>::unpackIterable(T & to_unpack) {
size_t size;
operator>>(size);
to_unpack.resize(size);
auto it = to_unpack.begin();
auto end = to_unpack.end();
for (; it != end; ++it)
operator>>(*it);
}
/**
* std::vector<T>
*/
/* -------------------------------------------------------------------------- */
template <bool is_static>
template <typename T>
inline CommunicationBufferTemplated<is_static> &
CommunicationBufferTemplated<is_static>::
operator<<(const std::vector<T> & to_pack) {
packIterable(to_pack);
return *this;
}
/* -------------------------------------------------------------------------- */
template <bool is_static>
template <typename T>
inline CommunicationBufferTemplated<is_static> &
CommunicationBufferTemplated<is_static>::
operator>>(std::vector<T> & to_unpack) {
unpackIterable(to_unpack);
return *this;
}
/**
* std::string
*/
/* -------------------------------------------------------------------------- */
template <bool is_static>
inline CommunicationBufferTemplated<is_static> &
CommunicationBufferTemplated<is_static>::
operator<<(const std::string & to_pack) {
packIterable(to_pack);
return *this;
}
/* -------------------------------------------------------------------------- */
template <bool is_static>
inline CommunicationBufferTemplated<is_static> &
CommunicationBufferTemplated<is_static>::operator>>(std::string & to_unpack) {
unpackIterable(to_unpack);
return *this;
}
/* -------------------------------------------------------------------------- */
template <bool is_static>
template <typename T>
inline std::string
CommunicationBufferTemplated<is_static>::extractStream(UInt block_size) {
std::stringstream str;
auto * ptr = reinterpret_cast<T *>(buffer.storage());
UInt sz = buffer.size() / sizeof(T);
UInt sz_block = block_size / sizeof(T);
UInt n_block = 0;
for (UInt i = 0; i < sz; ++i) {
if (i % sz_block == 0) {
str << std::endl << n_block << " ";
++n_block;
}
str << *ptr << " ";
++ptr;
}
return str.str();
}
/* -------------------------------------------------------------------------- */
template <bool is_static>
inline void CommunicationBufferTemplated<is_static>::resize(UInt size) {
if (!is_static) {
buffer.resize(0);
} else {
buffer.resize(size);
}
reset();
#ifndef AKANTU_NDEBUG
clear();
#endif
}
+/* -------------------------------------------------------------------------- */
+template <bool is_static>
+inline void CommunicationBufferTemplated<is_static>::reserve(UInt size) {
+ packResize(size);
+}
+
/* -------------------------------------------------------------------------- */
template <bool is_static>
inline void CommunicationBufferTemplated<is_static>::clear() {
buffer.clear();
}
/* -------------------------------------------------------------------------- */
template <bool is_static>
inline void CommunicationBufferTemplated<is_static>::reset() {
ptr_pack = buffer.storage();
ptr_unpack = buffer.storage();
}
/* -------------------------------------------------------------------------- */
// template<bool is_static>
// inline CommunicationBufferTemplated<is_static> &
// CommunicationBufferTemplated<is_static>::packMeshData (const MeshData &
// to_pack, const ElementType & type) {
// UInt size = to_pack.size();
// operator<<(size);
// typename std::vector<T>::iterator it = to_pack.begin();
// typename std::vector<T>::iterator end = to_pack.end();
// for(;it != end; ++it) operator<<(*it);
// return *this;
//}
diff --git a/src/synchronizer/communicator.hh b/src/synchronizer/communicator.hh
index e3ac8f462..c6486c3ac 100644
--- a/src/synchronizer/communicator.hh
+++ b/src/synchronizer/communicator.hh
@@ -1,436 +1,445 @@
/**
* @file communicator.hh
*
* @author Nicolas Richart <nicolas.richart@epfl.ch>
*
* @date creation: Fri Jun 18 2010
* @date last modification: Wed Nov 15 2017
*
* @brief Class handling the parallel communications
*
* @section LICENSE
*
* Copyright (©) 2010-2018 EPFL (Ecole Polytechnique Fédérale de Lausanne)
* Laboratory (LSMS - Laboratoire de Simulation en Mécanique des Solides)
*
* 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_array.hh"
#include "aka_common.hh"
#include "aka_event_handler_manager.hh"
#include "communication_buffer.hh"
#include "communication_request.hh"
#include "communicator_event_handler.hh"
/* -------------------------------------------------------------------------- */
#ifndef __AKANTU_STATIC_COMMUNICATOR_HH__
#define __AKANTU_STATIC_COMMUNICATOR_HH__
namespace akantu {
namespace debug {
class CommunicationException : public Exception {
public:
CommunicationException()
: Exception("An exception happen during a communication process.") {}
};
} // namespace debug
/// @enum SynchronizerOperation reduce operation that the synchronizer can
/// perform
enum class SynchronizerOperation {
_sum,
_min,
_max,
_prod,
_land,
_band,
_lor,
_bor,
_lxor,
_bxor,
_min_loc,
_max_loc,
_null
};
enum class CommunicationMode { _auto, _synchronous, _ready };
namespace {
int _any_source = -1;
}
} // namespace akantu
namespace akantu {
struct CommunicatorInternalData {
virtual ~CommunicatorInternalData() = default;
};
/* -------------------------------------------------------------------------- */
/* -------------------------------------------------------------------------- */
class Communicator : public EventHandlerManager<CommunicatorEventHandler> {
struct private_member {};
/* ------------------------------------------------------------------------ */
/* Constructors/Destructors */
/* ------------------------------------------------------------------------ */
public:
Communicator(int & argc, char **& argv, const private_member &);
~Communicator() override;
/* ------------------------------------------------------------------------ */
/* Methods */
/* ------------------------------------------------------------------------ */
public:
/* ------------------------------------------------------------------------ */
/* Point to Point */
+ /* ------------------------------------------------------------------------ */
+ template <typename T>
+ void probe(Int sender, Int tag, CommunicationStatus & status) const;
+
+ template <typename T>
+ bool asyncProbe(Int sender, Int tag, CommunicationStatus & status) const;
+
/* ------------------------------------------------------------------------ */
template <typename T>
inline void receive(Array<T> & values, Int sender, Int tag) const {
return this->receiveImpl(
values.storage(), values.size() * values.getNbComponent(), sender, tag);
}
template <typename Tensor>
inline void
receive(Tensor & values, Int sender, Int tag,
std::enable_if_t<is_tensor<Tensor>::value> * = nullptr) const {
return this->receiveImpl(values.storage(), values.size(), sender, tag);
}
- template <bool is_static>
- inline void receive(CommunicationBufferTemplated<is_static> & values,
+
+ inline void receive(CommunicationBufferTemplated<true> & values,
Int sender, Int tag) const {
return this->receiveImpl(values.storage(), values.size(), sender, tag);
}
+
+ inline void receive(CommunicationBufferTemplated<false> & values,
+ Int sender, Int tag) const {
+ CommunicationStatus status;
+ this->probe<char>(sender, tag, status);
+ values.reserve(status.size());
+ return this->receiveImpl(values.storage(), values.size(), sender, tag);
+ }
+
template <typename T>
inline void
receive(T & values, Int sender, Int tag,
std::enable_if_t<std::is_arithmetic<T>::value> * = nullptr) const {
return this->receiveImpl(&values, 1, sender, tag);
}
/* ------------------------------------------------------------------------ */
template <typename T>
inline void
send(const Array<T> & values, Int receiver, Int tag,
const CommunicationMode & mode = CommunicationMode::_auto) const {
return this->sendImpl(values.storage(),
values.size() * values.getNbComponent(), receiver,
tag, mode);
}
template <typename Tensor>
inline void
send(const Tensor & values, Int receiver, Int tag,
const CommunicationMode & mode = CommunicationMode::_auto,
std::enable_if_t<is_tensor<Tensor>::value> * = nullptr) const {
return this->sendImpl(values.storage(), values.size(), receiver, tag, mode);
}
template <bool is_static>
inline void
send(const CommunicationBufferTemplated<is_static> & values, Int receiver,
Int tag,
const CommunicationMode & mode = CommunicationMode::_auto) const {
return this->sendImpl(values.storage(), values.size(), receiver, tag, mode);
}
template <typename T>
inline void
send(const T & values, Int receiver, Int tag,
const CommunicationMode & mode = CommunicationMode::_auto,
std::enable_if_t<std::is_arithmetic<T>::value> * = nullptr) const {
return this->sendImpl(&values, 1, receiver, tag, mode);
}
/* ------------------------------------------------------------------------ */
template <typename T>
inline CommunicationRequest
asyncSend(const Array<T> & values, Int receiver, Int tag,
const CommunicationMode & mode = CommunicationMode::_auto) const {
return this->asyncSendImpl(values.storage(),
values.size() * values.getNbComponent(),
receiver, tag, mode);
}
template <typename T>
inline CommunicationRequest
asyncSend(const std::vector<T> & values, Int receiver, Int tag,
const CommunicationMode & mode = CommunicationMode::_auto) const {
return this->asyncSendImpl(values.data(), values.size(), receiver, tag,
mode);
}
template <typename Tensor>
inline CommunicationRequest
asyncSend(const Tensor & values, Int receiver, Int tag,
const CommunicationMode & mode = CommunicationMode::_auto,
std::enable_if_t<is_tensor<Tensor>::value> * = nullptr) const {
return this->asyncSendImpl(values.storage(), values.size(), receiver, tag,
mode);
}
template <bool is_static>
inline CommunicationRequest
asyncSend(const CommunicationBufferTemplated<is_static> & values,
Int receiver, Int tag,
const CommunicationMode & mode = CommunicationMode::_auto) const {
return this->asyncSendImpl(values.storage(), values.size(), receiver, tag,
mode);
}
template <typename T>
inline CommunicationRequest
asyncSend(const T & values, Int receiver, Int tag,
const CommunicationMode & mode = CommunicationMode::_auto,
std::enable_if_t<std::is_arithmetic<T>::value> * = nullptr) const {
return this->asyncSendImpl(&values, 1, receiver, tag, mode);
}
/* ------------------------------------------------------------------------ */
template <typename T>
inline CommunicationRequest asyncReceive(Array<T> & values, Int sender,
Int tag) const {
return this->asyncReceiveImpl(
values.storage(), values.size() * values.getNbComponent(), sender, tag);
}
template <typename T>
inline CommunicationRequest asyncReceive(std::vector<T> & values, Int sender,
Int tag) const {
return this->asyncReceiveImpl(values.data(), values.size(), sender, tag);
}
template <typename Tensor,
typename = std::enable_if_t<is_tensor<Tensor>::value>>
inline CommunicationRequest asyncReceive(Tensor & values, Int sender,
Int tag) const {
return this->asyncReceiveImpl(values.storage(), values.size(), sender, tag);
}
template <bool is_static>
inline CommunicationRequest
asyncReceive(CommunicationBufferTemplated<is_static> & values, Int sender,
Int tag) const {
return this->asyncReceiveImpl(values.storage(), values.size(), sender, tag);
}
- /* ------------------------------------------------------------------------ */
- template <typename T>
- void probe(Int sender, Int tag, CommunicationStatus & status) const;
-
- template <typename T>
- bool asyncProbe(Int sender, Int tag, CommunicationStatus & status) const;
-
/* ------------------------------------------------------------------------ */
/* Collectives */
/* ------------------------------------------------------------------------ */
template <typename T>
inline void allReduce(Array<T> & values,
const SynchronizerOperation & op) const {
this->allReduceImpl(values.storage(),
values.size() * values.getNbComponent(), op);
}
template <typename Tensor>
inline void
allReduce(Tensor & values, const SynchronizerOperation & op,
std::enable_if_t<is_tensor<Tensor>::value> * = nullptr) const {
this->allReduceImpl(values.storage(), values.size(), op);
}
template <typename T>
inline void
allReduce(T & values, const SynchronizerOperation & op,
std::enable_if_t<std::is_arithmetic<T>::value> * = nullptr) const {
this->allReduceImpl(&values, 1, op);
}
/* ------------------------------------------------------------------------ */
template <typename T> inline void allGather(Array<T> & values) const {
AKANTU_DEBUG_ASSERT(UInt(psize) == values.size(),
"The array size is not correct");
this->allGatherImpl(values.storage(), values.getNbComponent());
}
template <typename Tensor,
typename = std::enable_if_t<is_tensor<Tensor>::value>>
inline void allGather(Tensor & values) const {
AKANTU_DEBUG_ASSERT(values.size() / UInt(psize) > 0,
"The vector size is not correct");
this->allGatherImpl(values.storage(), values.size() / UInt(psize));
}
/* ------------------------------------------------------------------------ */
template <typename T>
inline void allGatherV(Array<T> & values, const Array<Int> & sizes) const {
this->allGatherVImpl(values.storage(), sizes.storage());
}
/* ------------------------------------------------------------------------ */
template <typename T>
inline void reduce(Array<T> & values, const SynchronizerOperation & op,
int root = 0) const {
this->reduceImpl(values.storage(), values.size() * values.getNbComponent(),
op, root);
}
/* ------------------------------------------------------------------------ */
template <typename Tensor>
inline void
gather(Tensor & values, int root = 0,
std::enable_if_t<is_tensor<Tensor>::value> * = nullptr) const {
this->gatherImpl(values.storage(), values.getNbComponent(), root);
}
template <typename T>
inline void
gather(T values, int root = 0,
std::enable_if_t<std::is_arithmetic<T>::value> * = nullptr) const {
this->gatherImpl(&values, 1, root);
}
/* ------------------------------------------------------------------------ */
template <typename Tensor, typename T>
inline void
gather(Tensor & values, Array<T> & gathered,
std::enable_if_t<is_tensor<Tensor>::value> * = nullptr) const {
AKANTU_DEBUG_ASSERT(values.size() == gathered.getNbComponent(),
"The array size is not correct");
gathered.resize(psize);
this->gatherImpl(values.data(), values.size(), gathered.storage(),
gathered.getNbComponent());
}
template <typename T>
inline void
gather(T values, Array<T> & gathered,
std::enable_if_t<std::is_arithmetic<T>::value> * = nullptr) const {
this->gatherImpl(&values, 1, gathered.storage(), 1);
}
/* ------------------------------------------------------------------------ */
template <typename T>
inline void gatherV(Array<T> & values, const Array<Int> & sizes,
int root = 0) const {
this->gatherVImpl(values.storage(), sizes.storage(), root);
}
/* ------------------------------------------------------------------------ */
template <typename T>
inline void broadcast(Array<T> & values, int root = 0) const {
this->broadcastImpl(values.storage(),
values.size() * values.getNbComponent(), root);
}
template <bool is_static>
inline void broadcast(CommunicationBufferTemplated<is_static> & values,
int root = 0) const {
this->broadcastImpl(values.storage(), values.size(), root);
}
template <typename T> inline void broadcast(T & values, int root = 0) const {
this->broadcastImpl(&values, 1, root);
}
/* ------------------------------------------------------------------------ */
void barrier() const;
CommunicationRequest asyncBarrier() const;
/* ------------------------------------------------------------------------ */
/* Request handling */
/* ------------------------------------------------------------------------ */
bool test(CommunicationRequest & request) const;
bool testAll(std::vector<CommunicationRequest> & request) const;
void wait(CommunicationRequest & request) const;
void waitAll(std::vector<CommunicationRequest> & requests) const;
UInt waitAny(std::vector<CommunicationRequest> & requests) const;
inline void freeCommunicationRequest(CommunicationRequest & request) const;
inline void
freeCommunicationRequest(std::vector<CommunicationRequest> & requests) const;
template <typename T, typename MsgProcessor>
inline void
receiveAnyNumber(std::vector<CommunicationRequest> & send_requests,
MsgProcessor && processor, Int tag) const;
protected:
template <typename T>
void
sendImpl(const T * buffer, Int size, Int receiver, Int tag,
const CommunicationMode & mode = CommunicationMode::_auto) const;
template <typename T>
void receiveImpl(T * buffer, Int size, Int sender, Int tag) const;
template <typename T>
CommunicationRequest asyncSendImpl(
const T * buffer, Int size, Int receiver, Int tag,
const CommunicationMode & mode = CommunicationMode::_auto) const;
template <typename T>
CommunicationRequest asyncReceiveImpl(T * buffer, Int size, Int sender,
Int tag) const;
template <typename T>
void allReduceImpl(T * values, int nb_values,
const SynchronizerOperation & op) const;
template <typename T> void allGatherImpl(T * values, int nb_values) const;
template <typename T> void allGatherVImpl(T * values, int * nb_values) const;
template <typename T>
void reduceImpl(T * values, int nb_values, const SynchronizerOperation & op,
int root = 0) const;
template <typename T>
void gatherImpl(T * values, int nb_values, int root = 0) const;
template <typename T>
void gatherImpl(T * values, int nb_values, T * gathered,
int nb_gathered = 0) const;
template <typename T>
void gatherVImpl(T * values, int * nb_values, int root = 0) const;
template <typename T>
void broadcastImpl(T * values, int nb_values, int root = 0) const;
/* ------------------------------------------------------------------------ */
/* Accessors */
/* ------------------------------------------------------------------------ */
public:
Int getNbProc() const { return psize; };
Int whoAmI() const { return prank; };
static Communicator & getStaticCommunicator();
static Communicator & getStaticCommunicator(int & argc, char **& argv);
int getMaxTag() const;
int getMinTag() const;
AKANTU_GET_MACRO(CommunicatorData, (*communicator_data), decltype(auto));
/* ------------------------------------------------------------------------ */
/* Class Members */
/* ------------------------------------------------------------------------ */
private:
static std::unique_ptr<Communicator> static_communicator;
protected:
Int prank{0};
Int psize{1};
std::unique_ptr<CommunicatorInternalData> communicator_data;
};
inline std::ostream & operator<<(std::ostream & stream,
const CommunicationRequest & _this) {
_this.printself(stream);
return stream;
}
} // namespace akantu
#include "communicator_inline_impl.hh"
#endif /* __AKANTU_STATIC_COMMUNICATOR_HH__ */