diff --git a/src/vt/collective/reduce/allreduce/rabenseifner_group.cc b/src/vt/collective/reduce/allreduce/rabenseifner_group.cc deleted file mode 100644 index ce4ca81d11..0000000000 --- a/src/vt/collective/reduce/allreduce/rabenseifner_group.cc +++ /dev/null @@ -1,366 +0,0 @@ -/* -//@HEADER -// ***************************************************************************** -// -// rabenseifner_group.impl.h -// DARMA/vt => Virtual Transport -// -// Copyright 2019-2021 National Technology & Engineering Solutions of Sandia, LLC -// (NTESS). Under the terms of Contract DE-NA0003525 with NTESS, the U.S. -// Government retains certain rights in this software. -// -// Redistribution and use in source and binary forms, with or without -// modification, are permitted provided that the following conditions are met: -// -// * Redistributions of source code must retain the above copyright notice, -// this list of conditions and the following disclaimer. -// -// * Redistributions in binary form must reproduce the above copyright notice, -// this list of conditions and the following disclaimer in the documentation -// and/or other materials provided with the distribution. -// -// * Neither the name of the copyright holder nor the names of its -// contributors may be used to endorse or promote products derived from this -// software without specific prior written permission. -// -// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" -// AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE -// IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE -// ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE -// LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR -// CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF -// SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS -// INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN -// CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) -// ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE -// POSSIBILITY OF SUCH DAMAGE. -// -// Questions? Contact darma@sandia.gov -// -// ***************************************************************************** -//@HEADER -*/ - -#include "rabenseifner_group.h" -#include "vt/collective/reduce/allreduce/data_handler.h" -#include "vt/config.h" -#include "vt/context/context.h" -#include "vt/messaging/active.h" - -#include -#include -#include - -namespace vt::collective::reduce::allreduce { - -RabenseifnerGroup::RabenseifnerGroup( - GroupType group, std::vector const& nodes) : - group_(group), - nodes_(nodes), - num_nodes_(nodes.size()), - num_steps_(static_cast(log2(num_nodes_))), - nprocs_pof2_(1 << num_steps_), - nprocs_rem_(num_nodes_ - nprocs_pof2_) { - - auto it = std::find(nodes.begin(), nodes.end(), theContext()->getNode()); - - vtAssert(it != nodes.end(), "This node was not found in group nodes!"); - vtAssert(std::is_sorted(nodes.begin(), nodes.end()), "Has to be sorted!"); - - // index in group list - this_node_ = it - nodes.begin(); - is_even_ = this_node_ % 2 == 0; - is_part_of_adjustment_group_ = this_node_ < (2 * nprocs_rem_); - if (is_part_of_adjustment_group_) { - if (is_even_) { - vrt_node_ = this_node_ / 2; - } else { - vrt_node_ = -1; - } - } else { - vrt_node_ = this_node_ - nprocs_rem_; - } - - vt_debug_print( - terse, allreduce, - "RabenseifnerGroup constructor: this_node_={} vrt_node_={} " - "is_part_of_adjustment_group_={} \n", - this_node_, vrt_node_, is_part_of_adjustment_group_ - ); - // initialize(generateNewId(), std::forward(data)...); -} - -void RabenseifnerGroup::executeFinalHan(size_t ) { -// auto& state = states_.at(id); -// vt_debug_print(terse, allreduce, "Rabenseifner executing final handler ID = {}\n", id); - -// // TODO convert -// // parent_proxy_[this_node_].template invoke(state.val_); -// state.completed_ = true; -// } - - -// void RabenseifnerGroup::allreduce(size_t) { -// if (is_part_of_adjustment_group_) { -// adjustForPowerOfTwo(id); -// } else { -// scatterReduceIter(id); -// } -} - - -void RabenseifnerGroup::adjustForPowerOfTwo(size_t ) { - // if (is_part_of_adjustment_group_) { - // auto& state = states_.at(id); - // auto const partner = is_even_ ? this_node_ + 1 : this_node_ - 1; - - // vt_debug_print( - // terse, allreduce, "Rabenseifner AdjustInitial (To {}): ID = {}\n", partner, id - // ); - - // if (is_even_) { - // theMsg()->sendMsg<&RabenseifnerGroup::adjustForPowerOfTwoRightHalf>( - // partner, DataHelperT::createMessage( - // state.val_, state.size_ / 2, state.size_ - (state.size_ / 2), id - // ) - // ); - - // if(state.left_adjust_message_ != nullptr){ - // adjustForPowerOfTwoLeftHalf(state.left_adjust_message_.get()); - // } - // } else { - // theMsg()->sendMsg<&RabenseifnerGroup::adjustForPowerOfTwoLeftHalf>( - // partner, DataHelperT::createMessage(state.val_, 0, state.size_ / 2, id) - // ); - - // if(state.right_adjust_message_ != nullptr){ - // adjustForPowerOfTwoRightHalf(state.right_adjust_message_.get()); - // } - // } - // } -} - -bool RabenseifnerGroup::scatterAllMessagesReceived(size_t) { - return false; - // auto const& state = states_.at(id); - - // return std::all_of( - // state.scatter_steps_recv_.cbegin(), state.scatter_steps_recv_.cbegin() + state.scatter_step_, - // [](auto const val) { return val; }); -} - - -bool RabenseifnerGroup::scatterIsDone(size_t) { - return false; - //auto const& state = states_.at(id); - //return (state.scatter_step_ == num_steps_) and (state.scatter_num_recv_ == num_steps_); -} - - -bool RabenseifnerGroup::scatterIsReady(size_t) { - return false; - // auto const& state = states_.at(id); - // return ((is_part_of_adjustment_group_ and state.finished_adjustment_part_) and - // state.scatter_step_ == 0) or - // ((state.scatter_mask_ < nprocs_pof2_) and scatterAllMessagesReceived(id)); -} - - -void RabenseifnerGroup::scatterTryReduce( - size_t, int32_t ) { - // auto& state = states_.at(id); - - // auto do_reduce = (step < state.scatter_step_) and - // not state.scatter_steps_reduced_[step] and - // state.scatter_steps_recv_[step] and - // std::all_of(state.scatter_steps_reduced_.cbegin(), - // state.scatter_steps_reduced_.cbegin() + step, - // [](auto const val) { return val; }); - - // vt_debug_print( - // verbose, allreduce, "Rabenseifner ScatterTryReduce (Step = {} ID = {}): {}\n", - // step, id, do_reduce - // ); - - // if (do_reduce) { - // auto& in_msg = state.scatter_messages_.at(step); - // DataHelperT::template reduce(state.val_, state.r_index_[in_msg->step_], in_msg.get()); - - // state.scatter_steps_reduced_[step] = true; - // } -} - - -void RabenseifnerGroup::scatterReduceIter(size_t) { - // if (not scatterIsReady(id)) { - // return; - // } - - // auto& state = states_.at(id); - // auto vdest = vrt_node_ ^ state.scatter_mask_; - // auto dest = (vdest < nprocs_rem_) ? vdest * 2 : vdest + nprocs_rem_; - - // vt_debug_print( - // terse, allreduce, - // "Rabenseifner Scatter (Send step {} to {}): Starting with idx = {} and " - // "count " - // "{} ID = {}\n", - // state.scatter_step_, dest, state.s_index_[state.scatter_step_], - // state.s_count_[state.scatter_step_], id - // ); - - // theMsg()->sendMsg<&RabenseifnerGroup::scatterReduceIterHandler>(dest, - // DataHelperT::createMessage(state.val_, state.s_index_[state.scatter_step_], state.s_count_[state.scatter_step_], id, state.scatter_step_) - // ); - - // state.scatter_mask_ <<= 1; - // state.scatter_step_++; - - // scatterTryReduce(id, state.scatter_step_ - 1); - - // if (scatterIsDone(id)) { - // state.finished_scatter_part_ = true; - // gatherIter(id); - // } else { - // scatterReduceIter(id); - // } -} - -bool RabenseifnerGroup::gatherAllMessagesReceived(size_t) { - return false; - // auto& state = states_.at(id); - // return std::all_of( - // state.gather_steps_recv_.cbegin() + state.gather_step_ + 1, state.gather_steps_recv_.cend(), - // [](auto const val) { return val; }); -} - - -bool RabenseifnerGroup::gatherIsDone(size_t) { - return false; - // auto& state = states_.at(id); - // return (state.gather_step_ < 0) and (state.gather_num_recv_ == num_steps_); -} - - -bool RabenseifnerGroup::gatherIsReady(size_t) { - return false; - // auto& state = states_.at(id); - // return (state.gather_step_ == num_steps_ - 1) or gatherAllMessagesReceived(id); -} - - -void RabenseifnerGroup::gatherTryReduce( - size_t, int32_t ) { - // auto& state = states_.at(id); - - // auto const doRed = (step > state.gather_step_) and - // not state.gather_steps_reduced_[step] and state.gather_steps_recv_[step] and - // std::all_of(state.gather_steps_reduced_.cbegin() + step + 1, - // state.gather_steps_reduced_.cend(), - // [](auto const val) { return val; }); - - // if (doRed) { - // auto& in_msg = state.gather_messages_.at(step); - // DataHelperT::copy(state.val_, state.s_index_[in_msg->step_], in_msg.get()); - - // state.gather_steps_reduced_[step] = true; - // } -} - - -void RabenseifnerGroup::gatherIter(size_t) { - // if (not gatherIsReady(id)) { - // return; - // } - - // auto& state = states_.at(id); - // auto vdest = vrt_node_ ^ state.gather_mask_; - // auto dest = (vdest < nprocs_rem_) ? vdest * 2 : vdest + nprocs_rem_; - - // vt_debug_print( - // terse, allreduce, - // "Rabenseifner Gather (step {}): Sending to Node {} starting with idx = {} and " - // "count " - // "{} ID = {}\n", - // state.gather_step_, dest, state.r_index_[state.gather_step_], - // state.r_count_[state.gather_step_], id - // ); - - // theMsg()->sendMsg<&RabenseifnerGroup::gatherIterHandler>(dest, - // DataHelperT::createMessage( - // state.val_, state.r_index_[state.gather_step_], - // state.r_count_[state.gather_step_], id, state.gather_step_ - // ) - // ); - - // state.gather_mask_ >>= 1; - // state.gather_step_--; - - // gatherTryReduce(id, state.gather_step_ + 1); - - // if (gatherIsDone(id)) { - // finalPart(id); - // } else if (gatherIsReady(id)) { - // gatherIter(id); - // } -} - -void RabenseifnerGroup::finalPart(size_t) { - // auto& state = states_.at(id); - // if (state.completed_) { - // return; - // } - - // if (nprocs_rem_) { - // sendToExcludedNodes(id); - // } - - // vt_debug_print( - // terse, allreduce, - // "RabenseifnerGroup::finalPart(): Executing final handler with size {} ID = {}\n", - // state.val_.size(), id - // ); - - // // if constexpr (ShouldUseView_v) { - // // parent_proxy_[this_node_].template invoke(state.val_); - // // } else { - // // parent_proxy_[this_node_].template invoke( - // // DataType::fromVec(state.val_) - // // ); - // // } - - // state.completed_ = true; - - // std::fill(state.scatter_messages_.begin(), state.scatter_messages_.end(), nullptr); - // std::fill(state.gather_messages_.begin(), state.gather_messages_.end(), nullptr); - - // state.scatter_steps_recv_.assign(num_steps_, false); - // state.gather_steps_recv_.assign(num_steps_, false); - - // state.scatter_steps_reduced_.assign(num_steps_, false); - // state.gather_steps_reduced_.assign(num_steps_, false); - - // state.r_index_.assign(num_steps_, 0); - // state.r_count_.assign(num_steps_, 0); - // state.s_index_.assign(num_steps_, 0); - // state.s_count_.assign(num_steps_, 0); -} - - -void RabenseifnerGroup::sendToExcludedNodes(size_t) { - // auto& state = states_.at(id); - // if (is_part_of_adjustment_group_ and is_even_) { - // vt_debug_print( - // terse, allreduce, "RabenseifnerGroup::sendToExcludedNodes(): Sending to Node {} ID = {}\n", - // this_node_ + 1, id - // ); - - // theMsg()->sendMsg<&RabenseifnerGroup::sendToExcludedNodesHandler>(this_node_ + 1, - // DataHelperT::createMessage(state.val_, 0, state.size_, id) - // ); - // } -} - - -} // namespace vt::collective::reduce::allreduce - diff --git a/src/vt/collective/reduce/allreduce/rabenseifner_group.h b/src/vt/collective/reduce/allreduce/rabenseifner_group.h deleted file mode 100644 index ff25ee0297..0000000000 --- a/src/vt/collective/reduce/allreduce/rabenseifner_group.h +++ /dev/null @@ -1,312 +0,0 @@ -/* -//@HEADER -// ***************************************************************************** -// -// rabenseifner_group.h -// DARMA/vt => Virtual Transport -// -// Copyright 2019-2021 National Technology & Engineering Solutions of Sandia, LLC -// (NTESS). Under the terms of Contract DE-NA0003525 with NTESS, the U.S. -// Government retains certain rights in this software. -// -// Redistribution and use in source and binary forms, with or without -// modification, are permitted provided that the following conditions are met: -// -// * Redistributions of source code must retain the above copyright notice, -// this list of conditions and the following disclaimer. -// -// * Redistributions in binary form must reproduce the above copyright notice, -// this list of conditions and the following disclaimer in the documentation -// and/or other materials provided with the distribution. -// -// * Neither the name of the copyright holder nor the names of its -// contributors may be used to endorse or promote products derived from this -// software without specific prior written permission. -// -// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" -// AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE -// IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE -// ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE -// LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR -// CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF -// SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS -// INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN -// CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) -// ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE -// POSSIBILITY OF SUCH DAMAGE. -// -// Questions? Contact darma@sandia.gov -// -// ***************************************************************************** -//@HEADER -*/ - -#if !defined INCLUDED_VT_COLLECTIVE_REDUCE_ALLREDUCE_RABENSEIFNER_GROUP_H -#define INCLUDED_VT_COLLECTIVE_REDUCE_ALLREDUCE_RABENSEIFNER_GROUP_H - -#include "vt/config.h" -#include "vt/context/context.h" -#include "vt/messaging/message/message.h" -#include "vt/objgroup/proxy/proxy_objgroup.h" -#include "vt/registry/auto/auto_registry.h" -#include "vt/pipe/pipe_manager.h" -#include "data_handler.h" -#include "type.h" -#include "rabenseifner_msg.h" -#include "helpers.h" -#include "vt/configs/types/types_type.h" - -#include -#include - -namespace vt::collective::reduce::allreduce { - -struct StateHolderBase { - virtual ~StateHolderBase() = default; -}; - -template class Op> -struct StateHolder : StateHolderBase { - using ReduceOp = Op; - static constexpr bool KokkosPaylod = ShouldUseView_v; - State state_ = {}; - - - // static HolderID storeState(State const& state) { - // auto key = - // } -}; - - -/** - * \struct Rabenseifner - * \brief Class implementing Rabenseifner's allreduce algorithm. - * - * This class performs an allreduce operation using Rabenseifner's method. The algorithm consists - * of several phases: adjustment for power-of-two processes, scatter-reduce, and gather-allgather. - * - * \tparam DataT Type of the data being reduced. - * \tparam Op Reduction operation (e.g., sum, max, min). - * \tparam ObjT Object type used for callback invocation. - * \tparam finalHandler Callback handler for the final result. - */ - -struct RabenseifnerGroup { - - using ID = size_t; - /// Each allreduce state holder per ID - std::unordered_map< ID, std::unique_ptr> state_holder_ = {}; -RabenseifnerGroup() = default; -RabenseifnerGroup( - GroupType group, std::vector const& nodes); - /** - * \brief Initialize the allreduce algorithm. - * - * This function sets up the necessary data structures and initial values for the reduction operation. - * - * \param args Additional arguments for initializing the data value. - */ - template class Op, typename ...Args> - void initialize(ID id, Args&&... args); - - template class Op> - void initializeState(ID id); - - ID generateNewId() { return id_++; } - - /** - * \brief Execute the final handler callback with the reduced result. - */ - void executeFinalHan(ID id); - - /** - * \brief Perform the allreduce operation. - * - * This function starts the allreduce operation, adjusting for non-power-of-two process counts if necessary. - */ - template class Op, typename ...Args> - void allreduce(size_t id, Args&&... args); - - /** - * \brief Adjust the process count to the nearest power-of-two. - * - * This function performs additional steps to handle non-power-of-two process counts, ensuring that the - * main scatter-reduce and gather-allgather phases can proceed with a power-of-two number of processes. - */ - void adjustForPowerOfTwo(size_t id); - - /** - * \brief Handler for adjusting the right half of the process group. - * - * This function handles the data received from the partner process and combines it using the reduction operation. - * - * \param msg Message containing the data from the partner process. - */ - template - static void adjustForPowerOfTwoRightHalf(RabenseifnerMsg* msg); - - /** - * \brief Handler for adjusting the left half of the process group. - * - * This function handles the data received from the partner process and combines it using the reduction operation. - * - * \param msg Message containing the data from the partner process. - */ - template - static void adjustForPowerOfTwoLeftHalf(RabenseifnerMsg* msg); - - /** - * \brief Final adjustment step for non-power-of-two process counts. - * - * This function handles the final step of the adjustment phase, combining the data and proceeding to the next phase. - * - * \param msg Message containing the data from the partner process. - */ - template - static void adjustForPowerOfTwoFinalPart(RabenseifnerMsg* msg); - - /** - * \brief Check if all scatter messages have been received. - * - * \return True if all scatter messages have been received, false otherwise. - */ - bool scatterAllMessagesReceived(size_t id); - - /** - * \brief Check if the scatter phase is complete. - * - * \return True if the scatter phase is complete, false otherwise. - */ - bool scatterIsDone(size_t id); - - /** - * \brief Check if the scatter phase is ready to proceed. - * - * \return True if the scatter phase is ready to proceed, false otherwise. - */ - bool scatterIsReady(size_t id); - - /** - * \brief Try to reduce the received scatter messages. - * - * \param step The current step in the scatter phase. - */ - void scatterTryReduce(size_t id, int32_t step); - - /** - * \brief Perform the scatter-reduce iteration. - * - * This function sends data to the appropriate partner process and proceeds to the next step in the scatter phase. - */ - void scatterReduceIter(size_t id); - - /** - * \brief Handler for receiving scatter-reduce messages. - * - * This function handles the data received during the scatter-reduce phase and combines it using the reduction operation. - * - * \param msg Message containing the data from the partner process. - */ - template - static void scatterReduceIterHandler(RabenseifnerMsg* msg); - - /** - * \brief Check if all gather messages have been received. - * - * \return True if all gather messages have been received, false otherwise. - */ - bool gatherAllMessagesReceived(size_t id); - - /** - * \brief Check if the gather phase is complete. - * - * \return True if the gather phase is complete, false otherwise. - */ - bool gatherIsDone(size_t id); - - /** - * \brief Check if the gather phase is ready to proceed. - * - * \return True if the gather phase is ready to proceed, false otherwise. - */ - bool gatherIsReady(size_t id); - - /** - * \brief Try to reduce the received gather messages. - * - * \param step The current step in the gather phase. - */ - void gatherTryReduce(size_t id, int32_t step); - - /** - * \brief Perform the gather iteration. - * - * This function sends data to the appropriate partner process and proceeds to the next step in the gather phase. - */ - void gatherIter(size_t id); - - /** - * \brief Handler for receiving gather messages. - * - * This function handles the data received during the gather phase and combines it using the reduction operation. - * - * \param msg Message containing the data from the partner process. - */ - template - static void gatherIterHandler(RabenseifnerMsg* msg); - - /** - * \brief Perform the final part of the allreduce operation. - * - * This function completes the allreduce operation, handling any remaining steps and invoking the final handler. - */ - void finalPart(size_t id); - - /** - * \brief Send the result to excluded nodes. - * - * This function handles the final step for non-power-of-two process counts, sending the reduced result to excluded nodes. - */ - void sendToExcludedNodes(size_t id); - - /** - * \brief Handler for receiving the final result on excluded nodes. - * - * This function handles the data received on excluded nodes and invokes the final handler. - * - * \param msg Message containing the final result. - */ - template - static void sendToExcludedNodesHandler(RabenseifnerMsg* msg); - - GroupType group_ = {}; - - size_t id_ = 0; - std::unordered_map> states_ = {}; - - /// (Sorted) list of Nodes that take part in this allreduce - std::vector nodes_ = {}; - - /// This rank's index inside 'nodes_' - NodeType this_node_ = {}; - - size_t num_nodes_ = {}; - - bool is_even_ = false; - int32_t num_steps_ = {}; - int32_t nprocs_pof2_ = {}; - int32_t nprocs_rem_ = {}; - - /// Temporary Node value (can be changed if we have non power of 2 'num_nodes_') - NodeType vrt_node_ = {}; - bool is_part_of_adjustment_group_ = false; - static inline const std::string name_ = "Rabenseifner"; - static inline const ReducerType type_ = ReducerType::Rabenseifner; - HandlerType hand_ = {}; -}; - -} // namespace vt::collective::reduce::allreduce - -#include "rabenseifner_group.impl.h" - -#endif /*INCLUDED_VT_COLLECTIVE_REDUCE_ALLREDUCE_RABENSEIFNER_GROUP_H*/ diff --git a/src/vt/collective/reduce/allreduce/rabenseifner_group.impl.h b/src/vt/collective/reduce/allreduce/rabenseifner_group.impl.h deleted file mode 100644 index 7ad47ac357..0000000000 --- a/src/vt/collective/reduce/allreduce/rabenseifner_group.impl.h +++ /dev/null @@ -1,348 +0,0 @@ -/* -//@HEADER -// ***************************************************************************** -// -// rabenseifner_group.impl.h -// DARMA/vt => Virtual Transport -// -// Copyright 2019-2021 National Technology & Engineering Solutions of Sandia, LLC -// (NTESS). Under the terms of Contract DE-NA0003525 with NTESS, the U.S. -// Government retains certain rights in this software. -// -// Redistribution and use in source and binary forms, with or without -// modification, are permitted provided that the following conditions are met: -// -// * Redistributions of source code must retain the above copyright notice, -// this list of conditions and the following disclaimer. -// -// * Redistributions in binary form must reproduce the above copyright notice, -// this list of conditions and the following disclaimer in the documentation -// and/or other materials provided with the distribution. -// -// * Neither the name of the copyright holder nor the names of its -// contributors may be used to endorse or promote products derived from this -// software without specific prior written permission. -// -// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" -// AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE -// IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE -// ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE -// LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR -// CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF -// SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS -// INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN -// CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) -// ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE -// POSSIBILITY OF SUCH DAMAGE. -// -// Questions? Contact darma@sandia.gov -// -// ***************************************************************************** -//@HEADER -*/ - -#if !defined INCLUDED_VT_COLLECTIVE_REDUCE_ALLREDUCE_RABENSEIFNER_GROUP_IMPL_H -#define INCLUDED_VT_COLLECTIVE_REDUCE_ALLREDUCE_RABENSEIFNER_GROUP_IMPL_H - -#include "vt/collective/reduce/allreduce/data_handler.h" -#include "vt/config.h" -#include "vt/context/context.h" -#include "vt/messaging/active.h" - -#include -#include -#include - -namespace vt::collective::reduce::allreduce { - -template class Op, typename ...Args> -void RabenseifnerGroup::initialize(size_t id, Args&&... data) { - using DataT = typename function_traits::template arg_type<0>; - using ScalarT = typename DataHandler::Scalar; - - using StateT = StateHolder; - state_holder_[id] = std::make_unique(); - - auto& state = dynamic_cast(state_holder_[id].get())->state_; - - DataHelper::assign(state.val_, std::forward(data)...); - - if(not state.initialized_){ - initializeState(id); - } - - int step = 0; - state.size_ = state.val_.size(); - auto size = state.size_; - for (int mask = 1; mask < nprocs_pof2_; mask <<= 1) { - auto vdest = vrt_node_ ^ mask; - auto dest = (vdest < nprocs_rem_) ? vdest * 2 : vdest + nprocs_rem_; - - if (this_node_ < dest) { - state.r_count_[step] = size / 2; - state.s_count_[step] = size - state.r_count_[step]; - state.s_index_[step] = state.r_index_[step] + state.r_count_[step]; - } else { - state.s_count_[step] = size / 2; - state.r_count_[step] = size - state.s_count_[step]; - state.r_index_[step] = state.s_index_[step] + state.s_count_[step]; - } - - if (step + 1 < num_steps_) { - state.r_index_[step + 1] = state.r_index_[step]; - state.s_index_[step + 1] = state.r_index_[step]; - size = state.r_count_[step]; - step++; - } - } - - vt_debug_print( - terse, allreduce, - "Rabenseifner initialize: size_ = {} num_steps_ = {} nprocs_pof2_ = {} nprocs_rem_ = " - "{} " - "is_part_of_adjustment_group_ = {} vrt_node_ = {} ID = {:x}\n", - state.size_, num_steps_, nprocs_pof2_, nprocs_rem_, is_part_of_adjustment_group_, - vrt_node_, id - ); -} - -template class Op> -void RabenseifnerGroup::initializeState(size_t id) { - using ScalarT = typename DataHandler::Scalar; - using StateT = StateHolder; - auto it = state_holder_.find(id); - - if(it == state_holder_.end()){ - state_holder_[id] = std::make_unique(); - } - - auto& state = dynamic_cast(state_holder_.at(id).get())->state_; - vt_debug_print(terse, allreduce, "Rabenseifner initializing state for ID = {:x}\n", id); - - state.scatter_messages_.resize(num_steps_, nullptr); - state.scatter_steps_recv_.resize(num_steps_, false); - state.scatter_steps_reduced_.resize(num_steps_, false); - - state.gather_messages_.resize(num_steps_, nullptr); - state.gather_steps_recv_.resize(num_steps_, false); - state.gather_steps_reduced_.resize(num_steps_, false); - - state.finished_adjustment_part_ = not is_part_of_adjustment_group_; - state.completed_ = false; - - state.scatter_mask_ = 1; - state.scatter_step_ = 0; - state.scatter_num_recv_ = 0; - state.finished_scatter_part_ = false; - - state.gather_step_ = num_steps_ - 1; - state.gather_mask_ = nprocs_pof2_ >> 1; - state.gather_num_recv_ = 0; - - state.r_index_.resize(num_steps_, 0); - state.r_count_.resize(num_steps_, 0); - state.s_index_.resize(num_steps_, 0); - state.s_count_.resize(num_steps_, 0); - - state.initialized_ = true; -} - -template class Op, typename ...Args> -void RabenseifnerGroup::allreduce(size_t id, Args &&... args){ - using DataT = typename function_traits::template arg_type<0>; - using ScalarT = typename DataHandler::Scalar; - using StateT = StateHolder; - initialize(id, std::forward(args)...); - - auto& state = dynamic_cast(state_holder_.at(id).get())->state_; - - if (is_part_of_adjustment_group_) { - adjustForPowerOfTwo(id); - } else { - scatterReduceIter(id); - } -} - -template -void RabenseifnerGroup::adjustForPowerOfTwoRightHalf( - RabenseifnerMsg* msg) { - - // auto& state = states_[msg->id_]; - - // if (DataHelperT::empty(state.val_)) { - // if (not state.initialized_) { - // vt_debug_print( - // verbose, allreduce, - // "Rabenseifner AdjustRightHalf (From {}): State not initialized ID {}!\n", - // theContext()->getFromNodeCurrentTask(), msg->id_ - // ); - - // initializeState(msg->id_); - // } - // state.right_adjust_message_ = promoteMsg(msg); - - // return; - // } - - // vt_debug_print( - // terse, allreduce, "Rabenseifner AdjustRightHalf (From {}): ID = {}\n", - // theContext()->getFromNodeCurrentTask(), msg->id_ - // ); - - // DataHelperT::template reduce(state.val_, state.size_ / 2, msg); - - // // Send to left node - // theMsg()->sendMsg<&RabenseifnerGroup::adjustForPowerOfTwoFinalPart>(theContext()->getNode() - 1, - // DataHelperT::createMessage(state.val_, state.size_ / 2, state.size_ - (state.size_ / 2), msg->id_) - // ); -} - -template -void RabenseifnerGroup::adjustForPowerOfTwoLeftHalf( - RabenseifnerMsg* msg) { - - // auto& state = states_[msg->id_]; - // if (DataHelperT::empty(state.val_)) { - // if (not state.initialized_) { - // vt_debug_print( - // verbose, allreduce, - // "Rabenseifner AdjustLeftHalf (From {}): State not initialized ID {}!\n", - // theContext()->getFromNodeCurrentTask(), msg->id_); - - // initializeState(msg->id_); - // } - // state.left_adjust_message_ = promoteMsg(msg); - - // return; - // } - - // vt_debug_print( - // terse, allreduce, "Rabenseifner AdjustLeftHalf (From {}): ID = {}\n", - // theContext()->getFromNodeCurrentTask(), msg->id_ - // ); - - // DataHelperT::template reduce(state.val_, 0, msg); -} - -template -void RabenseifnerGroup::adjustForPowerOfTwoFinalPart( - RabenseifnerMsg* msg) { - - // vt_debug_print( - // terse, allreduce, "Rabenseifner AdjustFinal (From {}): ID = {}\n", - // theContext()->getFromNodeCurrentTask(), msg->id_ - // ); - - // auto& state = states_[msg->id_]; - - // DataHelperT::copy(state.val_, state.size_ / 2, msg); - - // state.finished_adjustment_part_ = true; - - // scatterReduceIter(msg->id_); -} - - -template -void RabenseifnerGroup::scatterReduceIterHandler( - RabenseifnerMsg* msg) { - // auto& state = states_[msg->id_]; - - // if (DataHelperT::empty(state.val_)) { - // if (not state.initialized_) { - // vt_debug_print( - // verbose, allreduce, - // "Rabenseifner Scatter (Recv step {} from {}): State not initialized " - // "for ID = " - // "{}!\n", - // msg->step_, theContext()->getFromNodeCurrentTask(), msg->id_); - // initializeState(msg->id_); - // } - - // state.scatter_messages_[msg->step_] = promoteMsg(msg); - // state.scatter_steps_recv_[msg->step_] = true; - // state.scatter_num_recv_++; - - // return; - // } - - // vt_debug_print( - // terse, allreduce, - // "Rabenseifner Scatter (Recv step {} from {}): initialized = {} " - // "scatter_mask_= {} nprocs_pof2_ = {}: scatterAllMessagesReceived() = {} " - // "state.finished_adjustment_part_ = {}" - // "idx = {} ID = {}\n", - // msg->step_, theContext()->getFromNodeCurrentTask(), state.initialized_, - // state.scatter_mask_, nprocs_pof2_, scatterAllMessagesReceived(msg->id_), - // state.finished_adjustment_part_, state.r_index_[msg->step_], msg->id_ - // ); - - // state.scatter_messages_[msg->step_] = promoteMsg(msg); - // state.scatter_steps_recv_[msg->step_] = true; - // state.scatter_num_recv_++; - - // if (not state.finished_adjustment_part_) { - // return; - // } - - // scatterTryReduce(msg->id_, msg->step_); - - // if ((state.scatter_mask_ < nprocs_pof2_) and scatterAllMessagesReceived(msg->id_)) { - // scatterReduceIter(msg->id_); - // } else if (scatterIsDone(msg->id_)) { - // state.finished_scatter_part_ = true; - // gatherIter(msg->id_); - // } -} - -template -void RabenseifnerGroup::gatherIterHandler( - RabenseifnerMsg* msg) { - // auto& state = states_.at(msg->id_); - // vt_debug_print( - // terse, allreduce, "Rabenseifner Gather (Recv step {} from {}): idx = {} ID = {}\n", - // msg->step_, theContext()->getFromNodeCurrentTask(), state.s_index_[msg->step_], - // msg->id_ - // ); - - // state.gather_messages_[msg->step_] = promoteMsg(msg); - // state.gather_steps_recv_[msg->step_] = true; - // state.gather_num_recv_++; - - // if (not state.finished_scatter_part_) { - // return; - // } - - // gatherTryReduce(msg->id_, msg->step_); - - // if (state.gather_mask_ > 0 and gatherIsReady(msg->id_)) { - // gatherIter(msg->id_); - // } else if (gatherIsDone(msg->id_)) { - // finalPart(msg->id_); - // } -} - -template -void RabenseifnerGroup::sendToExcludedNodesHandler( - RabenseifnerMsg* msg) { - // auto& state = states_.at(msg->id_); - // vt_debug_print( - // terse, allreduce, - // "RabenseifnerGroup::sendToExcludedNodesHandler(): Received allreduce result " - // "with ID = {}\n", - // msg->id_ - // ); - - // // if constexpr (ShouldUseView_v) { - // // parent_proxy_[this_node_].template invoke(msg->val_); - // // } else { - // // parent_proxy_[this_node_].template invoke( - // // DataType::fromMemory(msg->val_, msg->size_)); - // // } - - // state.completed_ = true; -} - -} // namespace vt::collective::reduce::allreduce - -#endif /*INCLUDED_VT_COLLECTIVE_REDUCE_ALLREDUCE_RABENSEIFNER_GROUP_IMPL_H*/ diff --git a/src/vt/collective/reduce/operators/functors/plus_op.h b/src/vt/collective/reduce/operators/functors/plus_op.h index 37994c8714..98a75ff486 100644 --- a/src/vt/collective/reduce/operators/functors/plus_op.h +++ b/src/vt/collective/reduce/operators/functors/plus_op.h @@ -47,6 +47,10 @@ #include "vt/config.h" #include "vt/collective/reduce/operators/functors/tuple_op_helper.h" +#if MAGISTRATE_KOKKOS_ENABLED +#include +#endif + namespace vt { namespace collective { namespace reduce { namespace operators { template @@ -66,6 +70,7 @@ struct PlusOp> { }; #if MAGISTRATE_KOKKOS_ENABLED + template struct PlusOp> { void operator()( diff --git a/src/vt/group/collective/group_info_collective.cc b/src/vt/group/collective/group_info_collective.cc index e112e72cab..4351229b91 100644 --- a/src/vt/group/collective/group_info_collective.cc +++ b/src/vt/group/collective/group_info_collective.cc @@ -41,7 +41,6 @@ //@HEADER */ -#include "vt/collective/reduce/allreduce/rabenseifner_group.h" #include "vt/collective/reduce/operators/functors/plus_op.h" #include "vt/config.h" #include "vt/configs/types/types_type.h" diff --git a/src/vt/group/group_manager.h b/src/vt/group/group_manager.h index 043d46dbc7..bc062cd599 100644 --- a/src/vt/group/group_manager.h +++ b/src/vt/group/group_manager.h @@ -63,7 +63,6 @@ #include "vt/collective/collective_scope.h" #include "vt/runtime/component/component_pack.h" #include "vt/utils/fntraits/fntraits.h" -#include "vt/collective/reduce/allreduce/rabenseifner_group.h" #include "vt/configs/types/types_type.h" #include @@ -458,7 +457,6 @@ struct GroupManager : runtime::component::Component { CollectiveScopeType collective_scope_; std::unordered_map temporary_groups_ = {}; - std::unordered_map reducers_ = {}; }; /**