From eafa36496d99e366ef2e8830e85eb06f80fe64d7 Mon Sep 17 00:00:00 2001
From: Jacob Domagala <domagala.jacob@gmail.com>
Date: Wed, 17 Jul 2024 23:37:57 +0200
Subject: [PATCH] #2240: Add helpers and use Kokkos::View for internals of
 Rabenseifner when user's payload is View

---
 .../reduce/allreduce/data_handler.h           |  30 +--
 src/vt/collective/reduce/allreduce/helpers.h  | 195 ++++++++++++++++++
 .../reduce/allreduce/rabenseifner.h           | 117 +++--------
 .../reduce/allreduce/rabenseifner.impl.h      | 109 +++++-----
 .../reduce/allreduce/rabenseifner_msg.h       | 129 ++++++++++++
 tests/unit/objgroup/test_objgroup_common.h    |   2 +-
 6 files changed, 417 insertions(+), 165 deletions(-)
 create mode 100644 src/vt/collective/reduce/allreduce/helpers.h
 create mode 100644 src/vt/collective/reduce/allreduce/rabenseifner_msg.h

diff --git a/src/vt/collective/reduce/allreduce/data_handler.h b/src/vt/collective/reduce/allreduce/data_handler.h
index d614da1168..8afcb69e3d 100644
--- a/src/vt/collective/reduce/allreduce/data_handler.h
+++ b/src/vt/collective/reduce/allreduce/data_handler.h
@@ -66,15 +66,11 @@ class DataHandler<ScalarType, typename std::enable_if<std::is_arithmetic<ScalarT
 
   static std::vector<ScalarType> toVec(const ScalarType& data) { return std::vector<ScalarType>{data}; }
   static ScalarType fromVec(const std::vector<ScalarType>& data) { return data[0]; }
-  static ScalarType fromMemory(ScalarType* data, size_t) {
+  static ScalarType fromMemory(const ScalarType* data, size_t) {
     return *data;
   }
 
-  // static const ScalarType* data(const ScalarType& data) { return &data; }
   static size_t size(const ScalarType&) { return 1; }
-  // static ScalarType& at(ScalarType& data, size_t) { return data; }
-  // static void set(ScalarType& data, size_t, const ScalarType& value) { data = value; }
-  // static ScalarType split(ScalarType&, size_t, size_t) { return ScalarType{}; }
 };
 
 template <typename T>
@@ -84,20 +80,11 @@ class DataHandler<std::vector<T>> {
 
   static const std::vector<T>& toVec(const std::vector<T>& data) { return data; }
   static std::vector<T> fromVec(const std::vector<T>& data) { return data; }
-  static std::vector<T> fromMemory(T* data, size_t count) {
+  static std::vector<T> fromMemory(const T* data, size_t count) {
     return std::vector<T>(data, data + count);
   }
 
-  // static const T* data(const std::vector<T>& data) {return data.data(); }
   static size_t size(const std::vector<T>& data) { return data.size(); }
-  // static T at(const std::vector<T>& data, size_t idx) { return data[idx]; }
-  // static T& at(std::vector<T>& data, size_t idx) { return data[idx]; }
-  // static void set(std::vector<T>& data, size_t idx, const T& value) {
-  //   data[idx] = value;
-  // }
-  // static std::vector<T> split(std::vector<T>& data, size_t start, size_t end) {
-  //   return std::vector<T>{data.begin() + start, data.begin() + end};
-  // }
 };
 
 #if MAGISTRATE_KOKKOS_ENABLED
@@ -129,20 +116,7 @@ class DataHandler<Kokkos::View<T*, Kokkos::HostSpace, Props...>> {
     return view;
   }
 
-  // static const T* data(const ViewType& data) {return data.data(); }
   static size_t size(const ViewType& data) { return data.extent(0); }
-
-  // static T at(const ViewType& data, size_t idx) { return data(idx); }
-
-  // static T& at(ViewType& data, size_t idx) { return data(idx); }
-
-  // static void set(ViewType& data, size_t idx, const T& value) {
-  //   data(idx) = value;
-  // }
-
-  // static ViewType split(ViewType& data, size_t start, size_t end) {
-  //   return Kokkos::subview(data, std::make_pair(start, end));
-  // }
 };
 
 #endif // MAGISTRATE_KOKKOS_ENABLED
diff --git a/src/vt/collective/reduce/allreduce/helpers.h b/src/vt/collective/reduce/allreduce/helpers.h
new file mode 100644
index 0000000000..ed848b4d93
--- /dev/null
+++ b/src/vt/collective/reduce/allreduce/helpers.h
@@ -0,0 +1,195 @@
+/*
+//@HEADER
+// *****************************************************************************
+//
+//                                  helpers.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_HELPERS_H
+#define INCLUDED_VT_COLLECTIVE_REDUCE_ALLREDUCE_HELPERS_H
+#include "data_handler.h"
+#include "rabenseifner_msg.h"
+#include "vt/messaging/message/shared_message.h"
+#include <vector>
+
+namespace vt::collective::reduce::allreduce {
+
+template <typename Scalar, typename DataT>
+struct DataHelper {
+  using DataType = DataHandler<DataT>;
+
+  template <typename... Args>
+  static void assign(std::vector<Scalar>& dest, Args&&... data) {
+    dest = DataHandler<DataT>::toVec(std::forward<Args>(data)...);
+  }
+
+  static MsgPtr<RabenseifnerMsg<Scalar, DataT>> createMessage(
+    const std::vector<Scalar>& payload, size_t begin, size_t count, size_t id,
+    int32_t step = 0) {
+    return vt::makeMessage<RabenseifnerMsg<Scalar, DataT>>(
+      payload.data() + begin, count, id, step);
+  }
+
+  static void copy(
+    std::vector<Scalar>& dest, size_t start_idx, RabenseifnerMsg<Scalar, DataT>* msg) {
+    for (uint32_t i = 0; i < msg->size_; i++) {
+      dest[start_idx + i] = msg->val_[i];
+    }
+  }
+
+  template <template <typename Arg> class Op>
+  static void reduce(
+    std::vector<Scalar>& dest, size_t start_idx, RabenseifnerMsg<Scalar, DataT>* msg) {
+    for (uint32_t i = 0; i < msg->size_; i++) {
+      Op<Scalar>()(dest[start_idx + i], msg->val_[i]);
+    }
+  }
+
+  static void invoke() { }
+
+  static bool empty(const std::vector<Scalar>& payload) {
+    return payload.empty();
+  }
+};
+
+#if MAGISTRATE_KOKKOS_ENABLED
+
+template <typename Scalar>
+struct DataHelper<Scalar, Kokkos::View<Scalar*, Kokkos::HostSpace>> {
+  using DataT = Kokkos::View<Scalar*, Kokkos::HostSpace>;
+  using DataType = DataHandler<DataT>;
+
+  template <typename... Args>
+  static void assign(DataT& dest, Args&&... data) {
+    dest = {std::forward<Args>(data)...};
+  }
+
+  static MsgPtr<RabenseifnerMsg<Scalar, DataT>> createMessage(
+    const DataT& payload, size_t begin, size_t count, size_t id,
+    int32_t step = 0) {
+    return vt::makeMessage<RabenseifnerMsg<Scalar, DataT>>(
+      Kokkos::subview(payload, std::make_pair(begin, begin + count)), id, step
+    );
+  }
+
+  static void
+  copy(DataT& dest, size_t start_idx, RabenseifnerMsg<Scalar, DataT>* msg) {
+    Kokkos::parallel_for(
+      "Rabenseifner::copy", msg->val_.extent(0),
+      KOKKOS_LAMBDA(const int i) { dest(start_idx + i) = msg->val_(i); }
+    );
+  }
+
+  template <template <typename Arg> class Op>
+  static void reduce(
+    DataT& dest, size_t start_idx, RabenseifnerMsg<Scalar, DataT>* msg) {
+    Kokkos::parallel_for(
+      "Rabenseifner::reduce", msg->val_.extent(0), KOKKOS_LAMBDA(const int i) {
+        Op<Scalar>()(dest(start_idx + i), msg->val_(i));
+      }
+    );
+  }
+
+  static void invoke() { }
+
+  static bool empty(const DataT& payload) {
+    return payload.extent(0) == 0;
+  }
+};
+
+#endif // MAGISTRATE_KOKKOS_ENABLED
+
+struct StateBase {
+  size_t size_ = {};
+
+  bool finished_adjustment_part_ = false;
+
+  int32_t mask_ = 1;
+  int32_t step_ = 0;
+  bool initialized_ = false;
+  bool completed_ = false;
+
+  // Scatter
+  int32_t scatter_mask_ = 1;
+  int32_t scatter_step_ = 0;
+  int32_t scatter_num_recv_ = 0;
+  std::vector<bool> scatter_steps_recv_ = {};
+  std::vector<bool> scatter_steps_reduced_ = {};
+
+  bool finished_scatter_part_ = false;
+
+  // Gather
+  int32_t gather_step_ = 0;
+  int32_t gather_mask_ = 1;
+  int32_t gather_num_recv_ = 0;
+  std::vector<bool> gather_steps_recv_ = {};
+  std::vector<bool> gather_steps_reduced_ = {};
+
+  std::vector<uint32_t> r_index_ = {};
+  std::vector<uint32_t> r_count_ = {};
+  std::vector<uint32_t> s_index_ = {};
+  std::vector<uint32_t> s_count_ = {};
+};
+
+template <typename Scalar, typename DataT>
+struct State : StateBase {
+  std::vector<Scalar> val_ = {};
+
+  MsgSharedPtr<RabenseifnerMsg<Scalar, DataT>> left_adjust_message_ = nullptr;
+  MsgSharedPtr<RabenseifnerMsg<Scalar, DataT>> right_adjust_message_ = nullptr;
+  std::vector<MsgSharedPtr<RabenseifnerMsg<Scalar, DataT>>> scatter_messages_ = {};
+  std::vector<MsgSharedPtr<RabenseifnerMsg<Scalar, DataT>>> gather_messages_ = {};
+};
+
+#if MAGISTRATE_KOKKOS_ENABLED
+template <typename Scalar>
+struct State<Scalar, Kokkos::View<Scalar*, Kokkos::HostSpace>> : StateBase {
+  using DataT = Kokkos::View<Scalar*, Kokkos::HostSpace>;
+
+  Kokkos::View<Scalar*, Kokkos::HostSpace> val_ = {};
+
+  MsgSharedPtr<RabenseifnerMsg<Scalar, DataT>> left_adjust_message_ = nullptr;
+  MsgSharedPtr<RabenseifnerMsg<Scalar, DataT>> right_adjust_message_ = nullptr;
+  std::vector<MsgSharedPtr<RabenseifnerMsg<Scalar, DataT>>> scatter_messages_ = {};
+  std::vector<MsgSharedPtr<RabenseifnerMsg<Scalar, DataT>>> gather_messages_ = {};
+};
+#endif //MAGISTRATE_KOKKOS_ENABLED
+
+} // namespace vt::collective::reduce::allreduce
+#endif /*INCLUDED_VT_COLLECTIVE_REDUCE_ALLREDUCE_HELPERS_H*/
diff --git a/src/vt/collective/reduce/allreduce/rabenseifner.h b/src/vt/collective/reduce/allreduce/rabenseifner.h
index a96dd4a2d2..58090cfef1 100644
--- a/src/vt/collective/reduce/allreduce/rabenseifner.h
+++ b/src/vt/collective/reduce/allreduce/rabenseifner.h
@@ -52,57 +52,30 @@
 #include "vt/pipe/pipe_manager.h"
 #include "data_handler.h"
 #include "type.h"
+#include "rabenseifner_msg.h"
+#include "helpers.h"
 
 #include <cstdint>
 
 namespace vt::collective::reduce::allreduce {
 
+// Primary template
+template <typename Scalar, typename DataT>
+struct ShouldUseView {
+  static constexpr bool Value = false;
+};
+
+#if MAGISTRATE_KOKKOS_ENABLED
+// Partial specialization for Kokkos::View
 template <typename Scalar>
-struct AllreduceRbnRawMsg
-  : Message {
-    using MessageParentType = vt::Message;
-    vt_msg_serialize_required();
-
-
-  AllreduceRbnRawMsg() = default;
-  AllreduceRbnRawMsg(AllreduceRbnRawMsg const&) = default;
-  AllreduceRbnRawMsg(AllreduceRbnRawMsg&&) = default;
-  ~AllreduceRbnRawMsg() {
-    if (owning_) {
-      delete[] val_;
-    }
-  }
-
-  AllreduceRbnRawMsg(Scalar* in_val, size_t size, size_t id, int step = 0)
-    : MessageParentType(),
-      val_(in_val),
-      size_(size),
-      id_(id),
-      step_(step) { }
-
-  template <typename SerializeT>
-  void serialize(SerializeT& s) {
-      MessageParentType::serialize(s);
-
-      s | size_;
-
-      if (s.isUnpacking()) {
-        owning_ = true;
-        val_ = new Scalar[size_];
-      }
-
-      checkpoint::dispatch::serializeArray(s, val_, size_);
-
-      s | id_;
-      s | step_;
-  }
-
-  Scalar* val_ = {};
-  size_t size_ = {};
-  size_t id_ = {};
-  int32_t step_ = {};
-  bool owning_ = false;
+struct ShouldUseView<Scalar, Kokkos::View<Scalar*, Kokkos::HostSpace>> {
+  static constexpr bool Value = true;
 };
+#endif // MAGISTRATE_KOKKOS_ENABLED
+
+// Helper alias for cleaner usage
+template <typename Scalar, typename DataT>
+inline constexpr bool ShouldUseView_v = ShouldUseView<Scalar, DataT>::Value;
 
 /**
  * \struct Rabenseifner
@@ -122,6 +95,10 @@ template <
 struct Rabenseifner {
   using DataType = DataHandler<DataT>;
   using Scalar = typename DataType::Scalar;
+  using DataHelperT = DataHelper<Scalar, DataT>;
+  using StateT = State<Scalar, DataT>;
+
+  static constexpr bool KokkosPaylod = ShouldUseView_v<Scalar, DataT>;
 
   /**
    * \brief Constructor for Rabenseifner's allreduce algorithm.
@@ -175,7 +152,7 @@ struct Rabenseifner {
    *
    * \param msg Message containing the data from the partner process.
    */
-  void adjustForPowerOfTwoRightHalf(AllreduceRbnRawMsg<Scalar>* msg);
+  void adjustForPowerOfTwoRightHalf(RabenseifnerMsg<Scalar, DataT>* msg);
 
   /**
    * \brief Handler for adjusting the left half of the process group.
@@ -184,7 +161,7 @@ struct Rabenseifner {
    *
    * \param msg Message containing the data from the partner process.
    */
-  void adjustForPowerOfTwoLeftHalf(AllreduceRbnRawMsg<Scalar>* msg);
+  void adjustForPowerOfTwoLeftHalf(RabenseifnerMsg<Scalar, DataT>* msg);
 
   /**
    * \brief Final adjustment step for non-power-of-two process counts.
@@ -193,7 +170,7 @@ struct Rabenseifner {
    *
    * \param msg Message containing the data from the partner process.
    */
-  void adjustForPowerOfTwoFinalPart(AllreduceRbnRawMsg<Scalar>* msg);
+  void adjustForPowerOfTwoFinalPart(RabenseifnerMsg<Scalar, DataT>* msg);
 
   /**
    * \brief Check if all scatter messages have been received.
@@ -237,7 +214,7 @@ struct Rabenseifner {
    *
    * \param msg Message containing the data from the partner process.
    */
-  void scatterReduceIterHandler(AllreduceRbnRawMsg<Scalar>* msg);
+  void scatterReduceIterHandler(RabenseifnerMsg<Scalar, DataT>* msg);
 
   /**
    * \brief Check if all gather messages have been received.
@@ -281,7 +258,7 @@ struct Rabenseifner {
    *
    * \param msg Message containing the data from the partner process.
    */
-  void gatherIterHandler(AllreduceRbnRawMsg<Scalar>* msg);
+  void gatherIterHandler(RabenseifnerMsg<Scalar, DataT>* msg);
 
   /**
    * \brief Perform the final part of the allreduce operation.
@@ -304,51 +281,13 @@ struct Rabenseifner {
    *
    * \param msg Message containing the final result.
    */
-  void sendToExcludedNodesHandler(AllreduceRbnRawMsg<Scalar>* msg);
+  void sendToExcludedNodesHandler(RabenseifnerMsg<Scalar, DataT>* msg);
 
   vt::objgroup::proxy::Proxy<Rabenseifner> proxy_ = {};
   vt::objgroup::proxy::Proxy<ObjT> parent_proxy_ = {};
 
-  struct State {
-      std::vector<Scalar> val_ = {};
-      size_t size_ = {};
-
-      bool finished_adjustment_part_ = false;
-      MsgSharedPtr<AllreduceRbnRawMsg<Scalar>> left_adjust_message_ = nullptr;
-      MsgSharedPtr<AllreduceRbnRawMsg<Scalar>> right_adjust_message_ = nullptr;
-
-      int32_t mask_ = 1;
-      int32_t step_ = 0;
-      bool initialized_ = false;
-      bool completed_ = false;
-
-      // Scatter
-      int32_t scatter_mask_ = 1;
-      int32_t scatter_step_ = 0;
-      int32_t scatter_num_recv_ = 0;
-      std::vector<bool> scatter_steps_recv_ = {};
-      std::vector<bool> scatter_steps_reduced_ = {};
-      std::vector<MsgSharedPtr<AllreduceRbnRawMsg<Scalar>>> scatter_messages_ =
-        {};
-      bool finished_scatter_part_ = false;
-
-      // Gather
-      int32_t gather_step_ = 0;
-      int32_t gather_mask_ = 1;
-      int32_t gather_num_recv_ = 0;
-      std::vector<bool> gather_steps_recv_ = {};
-      std::vector<bool> gather_steps_reduced_ = {};
-      std::vector<MsgSharedPtr<AllreduceRbnRawMsg<Scalar>>> gather_messages_ =
-        {};
-
-      std::vector<uint32_t> r_index_ = {};
-      std::vector<uint32_t> r_count_ = {};
-      std::vector<uint32_t> s_index_ = {};
-      std::vector<uint32_t> s_count_ = {};
-  };
-
   size_t id_ = 0;
-  std::unordered_map<size_t, State> states_ = {};
+  std::unordered_map<size_t, StateT> states_ = {};
   NodeType num_nodes_ = {};
   NodeType this_node_ = {};
 
diff --git a/src/vt/collective/reduce/allreduce/rabenseifner.impl.h b/src/vt/collective/reduce/allreduce/rabenseifner.impl.h
index 858a17abd7..5e7c42e71c 100644
--- a/src/vt/collective/reduce/allreduce/rabenseifner.impl.h
+++ b/src/vt/collective/reduce/allreduce/rabenseifner.impl.h
@@ -41,12 +41,15 @@
 //@HEADER
 */
 
+#include "vt/collective/reduce/allreduce/rabenseifner.h"
 #if !defined INCLUDED_VT_COLLECTIVE_REDUCE_ALLREDUCE_RABENSEIFNER_IMPL_H
 #define INCLUDED_VT_COLLECTIVE_REDUCE_ALLREDUCE_RABENSEIFNER_IMPL_H
 
 #include "vt/collective/reduce/allreduce/data_handler.h"
 #include "vt/config.h"
 
+#include <type_traits>
+
 namespace vt::collective::reduce::allreduce {
 
 template <
@@ -119,7 +122,8 @@ template <
 template <typename ...Args>
 void Rabenseifner<DataT, Op, ObjT, finalHandler>::initialize(size_t id, Args&&... data) {
   auto& state = states_[id];
-  state.val_ = DataType::toVec(std::forward<Args>(data)...);
+
+  DataHelper<Scalar, DataT>::assign(state.val_, std::forward<Args>(data)...);
 
   if(not state.initialized_){
     initializeState(id);
@@ -167,6 +171,7 @@ void Rabenseifner<DataT, Op, ObjT, finalHandler>::executeFinalHan(size_t id) {
   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<finalHandler>(state.val_);
   state.completed_ = true;
 }
@@ -196,15 +201,20 @@ void Rabenseifner<DataT, Op, ObjT, finalHandler>::adjustForPowerOfTwo(size_t id)
 
     if (is_even_) {
       proxy_[partner]
-        .template send<&Rabenseifner::adjustForPowerOfTwoRightHalf>(
-          state.val_.data() + (state.size_ / 2), state.size_ - (state.size_ / 2), id);
+        .template sendMsg<&Rabenseifner::adjustForPowerOfTwoRightHalf>(
+          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 {
-      proxy_[partner].template send<&Rabenseifner::adjustForPowerOfTwoLeftHalf>(
-        state.val_.data(), state.size_ / 2, id);
+      proxy_[partner]
+        .template sendMsg<&Rabenseifner::adjustForPowerOfTwoLeftHalf>(
+          DataHelperT::createMessage(state.val_, 0, state.size_ / 2, id)
+        );
 
       if(state.right_adjust_message_ != nullptr){
         adjustForPowerOfTwoRightHalf(state.right_adjust_message_.get());
@@ -217,11 +227,11 @@ template <
   typename DataT, template <typename Arg> class Op, typename ObjT, auto finalHandler
 >
 void Rabenseifner<DataT, Op, ObjT, finalHandler>::adjustForPowerOfTwoRightHalf(
-  AllreduceRbnRawMsg<Scalar>* msg) {
+  RabenseifnerMsg<Scalar, DataT>* msg) {
 
   auto& state = states_[msg->id_];
 
-  if (state.val_.empty()) {
+  if (DataHelperT::empty(state.val_)) {
     if (not state.initialized_) {
       vt_debug_print(
         verbose, allreduce,
@@ -241,24 +251,23 @@ void Rabenseifner<DataT, Op, ObjT, finalHandler>::adjustForPowerOfTwoRightHalf(
     theContext()->getFromNodeCurrentTask(), msg->id_
   );
 
-  for (uint32_t i = 0; i < msg->size_; i++) {
-    Op<Scalar>()(state.val_[(state.size_ / 2) + i], msg->val_[i]);
-  }
+  DataHelperT::template reduce<Op>(state.val_, state.size_ / 2, msg);
 
   // Send to left node
   proxy_[theContext()->getNode() - 1]
-    .template send<&Rabenseifner::adjustForPowerOfTwoFinalPart>(
-      state.val_.data() + (state.size_ / 2), state.size_ - (state.size_ / 2), msg->id_);
+    .template sendMsg<&Rabenseifner::adjustForPowerOfTwoFinalPart>(
+      DataHelperT::createMessage(state.val_, state.size_ / 2, state.size_ - (state.size_ / 2), msg->id_)
+    );
 }
 
 template <
   typename DataT, template <typename Arg> class Op, typename ObjT, auto finalHandler
 >
 void Rabenseifner<DataT, Op, ObjT, finalHandler>::adjustForPowerOfTwoLeftHalf(
-  AllreduceRbnRawMsg<Scalar>* msg) {
+  RabenseifnerMsg<Scalar, DataT>* msg) {
 
   auto& state = states_[msg->id_];
-  if (state.val_.empty()) {
+  if (DataHelperT::empty(state.val_)) {
     if (not state.initialized_) {
       vt_debug_print(
         verbose, allreduce,
@@ -277,16 +286,14 @@ void Rabenseifner<DataT, Op, ObjT, finalHandler>::adjustForPowerOfTwoLeftHalf(
     theContext()->getFromNodeCurrentTask(), msg->id_
   );
 
-  for (uint32_t i = 0; i < msg->size_; i++) {
-    Op<Scalar>()(state.val_[i], msg->val_[i]);
-  }
+  DataHelperT::template reduce<Op>(state.val_, 0, msg);
 }
 
 template <
   typename DataT, template <typename Arg> class Op, typename ObjT, auto finalHandler
 >
 void Rabenseifner<DataT, Op, ObjT, finalHandler>::adjustForPowerOfTwoFinalPart(
-  AllreduceRbnRawMsg<Scalar>* msg) {
+  RabenseifnerMsg<Scalar, DataT>* msg) {
 
   vt_debug_print(
     terse, allreduce, "Rabenseifner AdjustFinal (From {}): ID = {}\n",
@@ -295,9 +302,7 @@ void Rabenseifner<DataT, Op, ObjT, finalHandler>::adjustForPowerOfTwoFinalPart(
 
   auto& state = states_[msg->id_];
 
-  for (uint32_t i = 0; i < msg->size_; i++) {
-    state.val_[(state.size_ / 2) + i] = msg->val_[i];
-  }
+  DataHelperT::copy(state.val_, state.size_ / 2, msg);
 
   state.finished_adjustment_part_ = true;
 
@@ -354,10 +359,7 @@ void Rabenseifner<DataT, Op, ObjT, finalHandler>::scatterTryReduce(
 
   if (do_reduce) {
     auto& in_msg = state.scatter_messages_.at(step);
-    auto& in_val = in_msg->val_;
-    for (uint32_t i = 0; i < in_msg->size_; i++) {
-      Op<Scalar>()(state.val_[state.r_index_[in_msg->step_] + i], in_val[i]);
-    }
+    DataHelperT::template reduce<Op>(state.val_, state.r_index_[in_msg->step_], in_msg.get());
 
     state.scatter_steps_reduced_[step] = true;
   }
@@ -384,9 +386,8 @@ void Rabenseifner<DataT, Op, ObjT, finalHandler>::scatterReduceIter(size_t id) {
     state.s_count_[state.scatter_step_], id
   );
 
-  proxy_[dest].template send<&Rabenseifner::scatterReduceIterHandler>(
-    state.val_.data() + state.s_index_[state.scatter_step_],
-    state.s_count_[state.scatter_step_], id, state.scatter_step_
+  proxy_[dest].template sendMsg<&Rabenseifner::scatterReduceIterHandler>(
+    DataHelperT::createMessage(state.val_, state.s_index_[state.scatter_step_], state.s_count_[state.scatter_step_], id, state.scatter_step_)
   );
 
   state.scatter_mask_ <<= 1;
@@ -406,10 +407,10 @@ template <
   typename DataT, template <typename Arg> class Op, typename ObjT, auto finalHandler
 >
 void Rabenseifner<DataT, Op, ObjT, finalHandler>::scatterReduceIterHandler(
-  AllreduceRbnRawMsg<Scalar>* msg) {
+  RabenseifnerMsg<Scalar, DataT>* msg) {
   auto& state = states_[msg->id_];
 
-  if (state.val_.empty()) {
+  if (DataHelperT::empty(state.val_)) {
     if (not state.initialized_) {
       vt_debug_print(
         verbose, allreduce,
@@ -497,10 +498,7 @@ void Rabenseifner<DataT, Op, ObjT, finalHandler>::gatherTryReduce(
 
   if (doRed) {
     auto& in_msg = state.gather_messages_.at(step);
-    auto& in_val = in_msg->val_;
-    for (uint32_t i = 0; i < in_msg->size_; i++) {
-      state.val_[state.s_index_[in_msg->step_] + i] = in_val[i];
-    }
+    DataHelperT::copy(state.val_, state.s_index_[in_msg->step_], in_msg.get());
 
     state.gather_steps_reduced_[step] = true;
   }
@@ -527,8 +525,11 @@ void Rabenseifner<DataT, Op, ObjT, finalHandler>::gatherIter(size_t id) {
     state.r_count_[state.gather_step_], id
   );
 
-  proxy_[dest].template send<&Rabenseifner::gatherIterHandler>(
-    state.val_.data() + state.r_index_[state.gather_step_], state.r_count_[state.gather_step_], id, state.gather_step_
+  proxy_[dest].template sendMsg<&Rabenseifner::gatherIterHandler>(
+    DataHelperT::createMessage(
+      state.val_, state.r_index_[state.gather_step_],
+      state.r_count_[state.gather_step_], id, state.gather_step_
+    )
   );
 
   state.gather_mask_ >>= 1;
@@ -547,7 +548,7 @@ template <
   typename DataT, template <typename Arg> class Op, typename ObjT, auto finalHandler
 >
 void Rabenseifner<DataT, Op, ObjT, finalHandler>::gatherIterHandler(
-  AllreduceRbnRawMsg<Scalar>* msg) {
+  RabenseifnerMsg<Scalar, DataT>* msg) {
   auto& state = states_.at(msg->id_);
   vt_debug_print(
     terse, allreduce, "Rabenseifner Gather (Recv step {} from {}): idx = {} ID = {}\n",
@@ -587,12 +588,17 @@ void Rabenseifner<DataT, Op, ObjT, finalHandler>::finalPart(size_t id) {
 
   vt_debug_print(
     terse, allreduce,
-    "Rabenseifner::finalPart(): Executing final handler with size {} ID = {}\n", state.val_.size(), id
+    "Rabenseifner::finalPart(): Executing final handler with size {} ID = {}\n",
+    state.val_.size(), id
   );
 
-  parent_proxy_[this_node_].template invoke<finalHandler>(
-    DataType::fromVec(state.val_)
-  );
+  if constexpr (ShouldUseView_v<Scalar, DataT>) {
+    parent_proxy_[this_node_].template invoke<finalHandler>(state.val_);
+  } else {
+    parent_proxy_[this_node_].template invoke<finalHandler>(
+      DataType::fromVec(state.val_)
+    );
+  }
 
   state.completed_ = true;
 
@@ -621,8 +627,11 @@ void Rabenseifner<DataT, Op, ObjT, finalHandler>::sendToExcludedNodes(size_t id)
       terse, allreduce, "Rabenseifner::sendToExcludedNodes(): Sending to Node {} ID = {}\n",
       this_node_ + 1, id
     );
+
     proxy_[this_node_ + 1]
-      .template send<&Rabenseifner::sendToExcludedNodesHandler>(state.val_.data(), state.size_, id);
+      .template sendMsg<&Rabenseifner::sendToExcludedNodesHandler>(
+        DataHelperT::createMessage(state.val_, 0, state.size_, id)
+      );
   }
 }
 
@@ -630,16 +639,22 @@ template <
   typename DataT, template <typename Arg> class Op, typename ObjT, auto finalHandler
 >
 void Rabenseifner<DataT, Op, ObjT, finalHandler>::sendToExcludedNodesHandler(
-  AllreduceRbnRawMsg<Scalar>* msg) {
+  RabenseifnerMsg<Scalar, DataT>* msg) {
   auto& state = states_.at(msg->id_);
   vt_debug_print(
     terse, allreduce,
-    "Rabenseifner::sendToExcludedNodesHandler(): Received allreduce result with size {} ID = {}\n", msg->size_, msg->id_
+    "Rabenseifner::sendToExcludedNodesHandler(): Received allreduce result "
+    "with ID = {}\n",
+    msg->id_
   );
 
-  parent_proxy_[this_node_].template invoke<finalHandler>(
-    DataType::fromMemory(msg->val_, msg->size_)
-  );
+  if constexpr (ShouldUseView_v<Scalar, DataT>) {
+    parent_proxy_[this_node_].template invoke<finalHandler>(msg->val_);
+  } else {
+    parent_proxy_[this_node_].template invoke<finalHandler>(
+      DataType::fromMemory(msg->val_, msg->size_));
+  }
+
   state.completed_ = true;
 }
 
diff --git a/src/vt/collective/reduce/allreduce/rabenseifner_msg.h b/src/vt/collective/reduce/allreduce/rabenseifner_msg.h
new file mode 100644
index 0000000000..9bf2653d7b
--- /dev/null
+++ b/src/vt/collective/reduce/allreduce/rabenseifner_msg.h
@@ -0,0 +1,129 @@
+/*
+//@HEADER
+// *****************************************************************************
+//
+//                              rabenseifner_msg.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_MSG_H
+#define INCLUDED_VT_COLLECTIVE_REDUCE_ALLREDUCE_RABENSEIFNER_MSG_H
+#include "vt/config.h"
+#include "vt/messaging/active.h"
+
+namespace vt::collective::reduce::allreduce {
+
+template <typename Scalar, typename DataT>
+struct RabenseifnerMsg : Message {
+  using MessageParentType = vt::Message;
+  vt_msg_serialize_required();
+
+  RabenseifnerMsg() = default;
+  RabenseifnerMsg(RabenseifnerMsg const&) = default;
+  RabenseifnerMsg(RabenseifnerMsg&&) = default;
+  ~RabenseifnerMsg() {
+    if (owning_) {
+      delete[] val_;
+    }
+  }
+
+  RabenseifnerMsg(const Scalar* in_val, size_t size, size_t id, int step = 0)
+    : MessageParentType(),
+      val_(in_val),
+      size_(size),
+      id_(id),
+      step_(step) { }
+
+  template <typename SerializeT>
+  void serialize(SerializeT& s) {
+    MessageParentType::serialize(s);
+
+    s | size_;
+
+    if (s.isUnpacking()) {
+      owning_ = true;
+      val_ = new Scalar[size_];
+    }
+
+    checkpoint::dispatch::serializeArray(s, val_, size_);
+
+    s | id_;
+    s | step_;
+  }
+
+  const Scalar* val_ = {};
+  size_t size_ = {};
+  size_t id_ = {};
+  int32_t step_ = {};
+  bool owning_ = false;
+};
+
+#if MAGISTRATE_KOKKOS_ENABLED
+template <typename Scalar>
+struct RabenseifnerMsg<Scalar, Kokkos::View<Scalar*, Kokkos::HostSpace>> : Message {
+  using ViewT = Kokkos::View<Scalar*, Kokkos::HostSpace>;
+  using MessageParentType = vt::Message;
+  vt_msg_serialize_required();
+
+  RabenseifnerMsg() = default;
+  RabenseifnerMsg(RabenseifnerMsg const&) = default;
+  RabenseifnerMsg(RabenseifnerMsg&&) = default;
+
+  RabenseifnerMsg(const ViewT& in_val, size_t id, int step = 0)
+    : MessageParentType(),
+      val_(in_val),
+      id_(id),
+      step_(step) { }
+
+  template <typename SerializeT>
+  void serialize(SerializeT& s) {
+    MessageParentType::serialize(s);
+
+    s | val_;
+    s | id_;
+    s | step_;
+  }
+
+  ViewT val_ = {};
+  size_t id_ = {};
+  int32_t step_ = {};
+};
+#endif // MAGISTRATE_KOKKOS_ENABLED
+
+} // namespace vt::collective::reduce::allreduce
+#endif /*INCLUDED_VT_COLLECTIVE_REDUCE_ALLREDUCE_RABENSEIFNER_MSG_H*/
diff --git a/tests/unit/objgroup/test_objgroup_common.h b/tests/unit/objgroup/test_objgroup_common.h
index e6ba455239..2c394a84f5 100644
--- a/tests/unit/objgroup/test_objgroup_common.h
+++ b/tests/unit/objgroup/test_objgroup_common.h
@@ -215,7 +215,7 @@ class DataHandler<tests::unit::VectorPayload> {
   static size_t size(const DataT& data) { return data.vec_.size(); }
   static const std::vector<Scalar>& toVec(const DataT& data) { return data.vec_; }
   static DataT fromVec(const std::vector<Scalar>& data) { return DataT{data}; }
-  static DataT fromMemory(Scalar* data, size_t count) {
+  static DataT fromMemory(const Scalar* data, size_t count) {
     return DataT{std::vector<Scalar>(data, data + count)};
   }
 };