hoo2/PDS
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

HW2: RC3b - A MinMax-MPIexchange pipeline and small changes

Browse Source
This commit is contained in:
hoo2
2025-01-05 21:26:56 +02:00
parent c485d0db2d
commit 3bf4522448
35 changed files with 286 additions and 101 deletions
Download Patch File Download Diff File Expand all files Collapse all files
homework_2/include/config.h
+9 -1
View File
@@ -25,7 +25,14 @@
#endif
// Default Data size (in case -q <N> is not present)
#define DEFAULT_DATA_SIZE (1 << 16)
static constexpr size_t DEFAULT_DATA_SIZE = 1 << 16;
// The maximum MPI size we support
static constexpr size_t MAX_MPI_SIZE = 1024UL;
// The maximum pipeline size we support
static constexpr size_t MAX_PIPELINE_SIZE = 64UL;
/*!
* Value type selection
@@ -46,6 +53,7 @@ using distValue_t = uint32_t;
*/
struct config_t {
size_t arraySize{DEFAULT_DATA_SIZE}; //!< The array size of the local data to sort.
size_t pipeline{1UL}; //!< Pipeline stages
bool validation{false}; //!< Request a full validation at the end, performed by process rank 0.
bool ndebug{false}; //!< Skips debug trap on DEBUG builds.
bool perf{false}; //!< Enable performance timing measurements and prints.
homework_2/include/distsort.hpp
+71 -19
View File
@@ -233,24 +233,24 @@ void elbowSort(ShadowedDataT& data, bool ascending) noexcept {
/*!
* Takes two sorted sequences where one is in increasing and the other is in decreasing order
* and selects either the larger or the smaller items in one-to-one comparison between them.
* The result is a bitonic sequence.
* Takes two sequences and selects either the larger or the smaller items
* in one-to-one comparison between them. If the initial sequences are bitonic, then
* the result is a bitonic sequence too!
*
* @tparam RangeT A range type with random access iterator
* @tparam ValueT The underlying type of the sequences
*
* @param local [RangeT] Reference to the local sequence
* @param remote [const RangeT] Reference to the remote sequence (copied locally by MPI)
* @param local [ValueT*] Pointer to the local sequence
* @param remote [const ValueT*] Pointer to the remote sequence (copied locally by MPI)
* @param count [size_t] The number of items to process
* @param keepSmall [bool] Flag to indicate if we keep the small items in local sequence
*/
template<typename RangeT>
void keepMinOrMax(RangeT& local, const RangeT& remote, bool keepSmall) noexcept {
using value_t = typename RangeT::value_type;
template<typename ValueT>
void keepMinOrMax(ValueT* local, const ValueT* remote, size_t count, bool keepSmall) noexcept {
std::transform(
local.begin(), local.end(),
remote.begin(),
local.begin(),
[&keepSmall](const value_t& a, const value_t& b){
local, local + count,
remote,
local,
[&keepSmall](const ValueT& a, const ValueT& b){
return (keepSmall) ? std::min(a, b) : std::max(a, b);
});
}
@@ -259,6 +259,60 @@ void keepMinOrMax(RangeT& local, const RangeT& remote, bool keepSmall) noexcept
* ============================== Sort algorithms ==============================
*/
/*!
* A small tag generator tool to provide consistent encoding to tag communication
*
* @param depth The current algorithmic depth[bitonic] of the communication, if any
* @param step The current step on the current depth
* @param stage The stage of the pipeline.
* @return The tag to use.
*
* @note
* In case we call this function outside of the pipeline loop, we can ommit
* @c stage argument and use the return value as starting tag for every communication
* of the pipeline loop. We need to increase the tags for each communication of
* the pipeline loop though!
*/
size_t tagGenerator(size_t depth, size_t step, size_t stage = 0);
/*!
* A pipeline loop for mixing min-max process with mpi data exchange
*
* @tparam ShadowedDataT A Shadowed buffer type with random access iterator.
*
* @param data [ShadowedDataT&] Reference to the data to exchange
* @param partner [mpi_id_t] The partner for the exchange
* @param keepSmall [bool] Flag to indicate if we keep the small values
* @param tag [int] The init tag to use for the loop.
*
* @note
* The @c tag is increased inside the pipeline loop for each different data exchange
*/
template<typename ShadowedDataT>
void exchangePipeline(ShadowedDataT& data, mpi_id_t partner, bool keepSmall, int tag) {
using Value_t = typename ShadowedDataT::value_type;
// Init counters and pointers
size_t count = data.size() / config.pipeline;
Value_t* active = data.getActive().data();
Value_t* shadow = data.getShadow().data();
// Pipeline
Texchange.start();
mpi.exchange_start(active, shadow, count, partner, tag);
for (size_t stage = 0 ; stage < config.pipeline ; active += count, shadow += count) {
// Wait previous chunk
mpi.exchange_wait(); Texchange.stop();
if (++stage < config.pipeline) {
// Start next chunk if there is a next one
Texchange.start();
mpi.exchange_start(active + count, shadow + count, count, partner, ++tag);
}
// process the arrived data
timeCall(Tminmax, keepMinOrMax, active, shadow, count, keepSmall);
}
}
/*!
* A distributed version of the Bubbletonic sort algorithm.
*
@@ -284,9 +338,8 @@ void distBubbletonic(ShadowedDataT& data, mpi_id_t Processes, mpi_id_t rank) {
if ( isActive(rank, Processes) &&
isActive(part, Processes) ) {
// Exchange with partner, keep nim-or-max and sort - O(N)
int tag = static_cast<int>(step);
timeCall(Texchange, mpi.exchange_data, data.getActive(), data.getShadow(), part, tag);
timeCall(Tminmax, keepMinOrMax, data.getActive(), data.getShadow(), ks);
int tag = static_cast<int>(tagGenerator(0, step));
exchangePipeline(data, part, ks, tag);
timeCall(TelbowSort, elbowSort, data, ascending<SortMode::Bubbletonic>(rank, Processes));
}
}
@@ -324,9 +377,8 @@ void distBitonic(ShadowedDataT& data, mpi_id_t Processes, mpi_id_t rank) {
auto part = partner<SortMode::Bitonic>(rank, step);
auto ks = keepSmall<SortMode::Bitonic>(rank, part, depth);
// Exchange with partner, keep nim-or-max
int tag = static_cast<int>( (2*p*depth) + step );
timeCall(Texchange, mpi.exchange_data, data.getActive(), data.getShadow(), part, tag);
timeCall(Tminmax, keepMinOrMax, data.getActive(), data.getShadow(), ks);
int tag = static_cast<int>(tagGenerator(depth, step));
exchangePipeline(data, part, ks, tag);
}
// sort - O(N)
timeCall(TelbowSort, elbowSort, data, ascending<SortMode::Bitonic>(rank, depth));
homework_2/include/utils.hpp
+58 -42
View File
@@ -65,6 +65,10 @@ struct MPI_t {
mpi_throw(err, "(MPI) MPI_Comm_rank() - ");
size_ = static_cast<ID_t>(size_value);
rank_ = static_cast<ID_t>(rank_value);
if (size_ > static_cast<ID_t>(MAX_MPI_SIZE))
throw std::runtime_error(
"(MPI) size - Not supported number of nodes [over " + std::to_string(MAX_MPI_SIZE) + "]\n"
);
// Get the name of the processor
char processor_name[MPI_MAX_PROCESSOR_NAME];
@@ -74,63 +78,56 @@ struct MPI_t {
name_ = std::string (processor_name, name_len);
}
/*!
* Exchange data with partner as part of the sorting network of both bubbletonic or bitonic
* sorting algorithms.
* Initiate a data exchange data with partner using non-blocking Isend-Irecv, as part of the
* sorting network of both bubbletonic or bitonic sorting algorithms.
*
* This function matches a transmit and a receive in order for fully exchanged data between
* current node and partner.
* @note
* This call MUST paired with exchange_wait() for each MPI_t object.
* Calling 2 consecutive exchange_start() for the same MPI_t object is undefined.
*
* @tparam T The inner valur type used in buffer
* @tparam ValueT The underlying value type used in buffers
*
* @param ldata [std::vector<T>] Reference to local data to send
* @param rdata [std::vector<T>] Reference to buffer to receive data from partner
* @param partner [mpi_id_t] The partner for the exchange
* @param tag [int] The tag to use for the MPI communication
* @param ldata [const ValueT*] Pointer to local data to send
* @param rdata [ValueT*] Pointer to buffer to receive data from partner
* @param count [size_t] The number of data to exchange
* @param partner [mpi_id_t] The partner for the exchange
* @param tag [int] The tag to use for the MPI communication
*/
template<typename T>
void exchange_data(const std::vector<T>& ldata, std::vector<T>& rdata, ID_t partner, int tag) {
template<typename ValueT>
void exchange_start(const ValueT* ldata, ValueT* rdata, size_t count, ID_t partner, int tag) {
if (tag < 0)
throw std::runtime_error("(MPI) exchange_data() [tag] - Out of bound");
MPI_Datatype datatype = MPI_TypeMapper<T>::getType();
int count = static_cast<int>(ldata.size());
MPI_Status status;
MPI_Datatype datatype = MPI_TypeMapper<ValueT>::getType();
int err;
if ((err = MPI_Sendrecv(
ldata.data(), count, datatype, partner, tag,
rdata.data(), count, datatype, partner, tag,
MPI_COMM_WORLD, &status
)) != MPI_SUCCESS)
mpi_throw(err, "(MPI) MPI_Sendrecv() [data] - ");
err = MPI_Isend(ldata, count, datatype, partner, tag, MPI_COMM_WORLD, &handle_tx);
if (err != MPI_SUCCESS)
mpi_throw(err, "(MPI) MPI_Isend() - ");
err = MPI_Irecv(rdata, count, datatype, partner, tag, MPI_COMM_WORLD, &handle_rx);
if (err != MPI_SUCCESS)
mpi_throw(err, "(MPI) MPI_Irecv() - ");
}
/*!
* Exchange a data object with partner as part of the sorting network of both bubbletonic
* or bitonic sorting algorithms.
* Block wait for the completion of the previously called exchange_start()
*
* This function matches a transmit and a receive in order for fully exchanged the data object
* between current node and partner.
*
* @tparam T The object type
*
* @param local [const T&] Reference to the local object to send
* @param remote [T&] Reference to the object to receive data from partner
* @param partner [mpi_id_t] The partner for the exchange
* @param tag [int] The tag to use for the MPI communication
* @note
* This call MUST paired with exchange_start() for each MPI_t object.
* Calling 2 consecutive exchange_wait() for the same MPI_t object is undefined.
*/
template<typename T>
void exchange_it(const T& local, T& remote, ID_t partner, int tag) {
if (tag < 0)
throw std::runtime_error("(MPI) exchange_it() [tag] - Out of bound");
void exchange_wait() {
MPI_Status status;
int err;
if ((err = MPI_Sendrecv(
&local, sizeof(T), MPI_BYTE, partner, tag,
&remote, sizeof(T), MPI_BYTE, partner, tag,
MPI_COMM_WORLD, &status
)) != MPI_SUCCESS)
mpi_throw(err, "(MPI) MPI_Sendrecv() [item] - ");
if ((err = MPI_Wait(&handle_tx, &status)) != MPI_SUCCESS)
mpi_throw(err, "(MPI) MPI_Wait() [send] - ");
if ((err = MPI_Wait(&handle_rx, &status)) != MPI_SUCCESS)
mpi_throw(err, "(MPI) MPI_Wait() [recv] - ");
}
// Accessors
@@ -181,6 +178,8 @@ private:
ID_t size_{}; //!< MPI total size of the execution
std::string name_{}; //!< The name of the local machine
bool initialized_{}; //!< RAII helper flag
MPI_Request handle_tx{}; //!< MPI async exchange handler for Transmission
MPI_Request handle_rx{}; //!< MPI async exchange handler for Receptions
};
/*
@@ -377,9 +376,13 @@ struct Timing {
else if (std::chrono::duration_cast<milliseconds>(duration_).count() < 10000)
std::cout << "[Timing] (Rank " << rank << ") " << what << ": "
<< std::to_string(std::chrono::duration_cast<milliseconds>(duration_).count()) << " [msec]\n";
else
std::cout << "[Timing] (Rank " << rank << ") " << what << ": "
<< std::to_string(std::chrono::duration_cast<seconds>(duration_).count()) << " [sec]\n";
else {
char stime[26]; // fit ulong
auto sec = std::chrono::duration_cast<seconds>(duration_).count();
auto msec = (std::chrono::duration_cast<milliseconds>(duration_).count() % 1000) / 10; // keep 2 digit
std::sprintf(stime, "%ld.%1ld", sec, msec);
std::cout << "[Timing] (Rank " << rank << ") " << what << ": " << stime << " [sec]\n";
}
}
@@ -402,4 +405,17 @@ private:
Tim.stop(); \
/*!
* A utility to check if a number is power of two
*
* @tparam Integral The integral type of the number to check
* @param x The number to check
* @return True if it is power of 2, false otherwise
*/
template <typename Integral>
constexpr inline bool isPowerOfTwo(Integral x) noexcept {
return (!(x & (x - 1)) && x);
}
#endif /* UTILS_HPP_ */