#973: implement default proxy able to send, broadcast and reduce

examples/callback/callback_context.cc

@@ -108,8 +108,9 @@ int main(int argc, char** argv) {
     auto cb = vt::theCB()->makeFunc<DataMsg,MyContext>(
       vt::pipe::LifetimeEnum::Once, &my_global_ctx, callbackFn
     );
-    auto msg = vt::makeMessage<CallbackMsg>(cb);
-    vt::theMsg()->sendMsg<CallbackMsg,handler>(1, msg);
+
+    auto const default_proxy = vt::theObjGroup()->makeCollective();
+    default_proxy[1].send<CallbackMsg, handler>(cb);
   }
 
   vt::finalize();

src/vt/objgroup/manager.h

@@ -110,6 +110,14 @@ struct ObjGroupManager : runtime::component::Component<ObjGroupManager> {
    * communicator
    */
 
+  /**
+   * \brief Construct a special proxy instance that allows sending, broadcasting
+   * and reducing without actual object group.
+   *
+   * \return proxy to the object group
+   */
+  proxy::DefaultProxyType makeCollective() { return proxy::DefaultProxyType{}; }
+
   /**
    * \brief Collectively construct a new object group. Allocates and constructs
    * the object on each node by forwarding constructor arguments.

src/vt/objgroup/proxy/proxy_objgroup.h

@@ -59,6 +59,7 @@
 #include "vt/rdmahandle/handle.fwd.h"
 #include "vt/rdmahandle/handle_set.fwd.h"
 #include "vt/messaging/pending_send.h"
+#include "vt/collective/collective_alg.h"
 
 namespace vt { namespace objgroup { namespace proxy {
 
@@ -341,6 +342,96 @@ struct Proxy {
   ObjGroupProxyType proxy_ = no_obj_group; /**< The raw proxy ID bits */
 };
 
+template <>
+struct Proxy<void> {
+  /**
+   * \brief Index the proxy to get the element proxy for a particular node
+   *
+   * \param[in] node the desired node
+   *
+   * \return an indexed proxy to that node
+   */
+  DefaultProxyElm operator[](NodeType node) const {
+    return DefaultProxyElm{node};
+  }
+
+  /**
+   * \brief Broadcast a message.
+   *
+   * \note Creates message from given args
+   *
+   * \param[in] args the arguments used to make a message
+   *
+   * \return the \c PendingSend for the sent message
+   */
+  template <typename MsgT, ActiveTypedFnType<MsgT>* f, typename... Args>
+  messaging::PendingSend broadcast(Args&&... args) const {
+    return broadcastMsg<MsgT, f>(
+      makeMessage<MsgT>(std::forward<Args>(args)...)
+    );
+  }
+
+  /**
+   * \brief Broadcast a message.
+   *
+   * \note Takes ownership of the supplied message.
+   *
+   * \param[in] msg the message to broadcast
+   * \param[in] tag the tag to put on the message
+   *
+   * \return the \c PendingSend for the sent message
+   */
+  template <typename MsgT, ActiveTypedFnType<MsgT>* f>
+  messaging::PendingSend
+  broadcastMsg(messaging::MsgPtrThief<MsgT> msg, TagType tag = no_tag) const {
+    return theMsg()->broadcastMsg<MsgT, f>(msg, tag);
+  }
+
+  /**
+   * \brief Reduce a message up the tree, possibly delayed through a pending
+   * send
+   *
+   * \note Creates message from given args
+   *
+   * \param[in] root the root node where the final handler provides the result
+   * \param[in] args the arguments used to make a message
+   *
+   * \return the pending send corresponding to the reduce
+   */
+  template <
+    typename OpT,
+    typename FunctorT,
+    typename MsgT,
+    typename... Args,
+    ActiveTypedFnType<MsgT>* f = MsgT::template msgHandler<MsgT, OpT, FunctorT>
+  >
+  messaging::PendingSend reduce(NodeType root, Args&&... args) const {
+    auto const msg = makeMessage<MsgT>(std::forward<Args>(args)...);
+    return reduceMsg<OpT, FunctorT, MsgT, f>(root, msg.get());
+  }
+
+  /**
+   * \brief Reduce a message up the tree, possibly delayed through a pending
+   * send
+   *
+   * \param[in] root the root node where the final handler provides the result
+   * \param[in] msg the message to reduce on this node
+   *
+   * \return the pending send corresponding to the reduce
+   */
+  template <
+    typename OpT,
+    typename FunctorT,
+    typename MsgT,
+    ActiveTypedFnType<MsgT>* f = MsgT::template msgHandler<MsgT, OpT, FunctorT>
+  >
+  messaging::PendingSend reduceMsg(NodeType root, MsgT* const msg) const {
+    return theCollective()->global()->reduce<OpT, FunctorT>(root, msg);
+  }
+};
+
+using DefaultProxyType = Proxy<void>;
+
 }}} /* end namespace vt::objgroup::proxy */
 
 #endif /*INCLUDED_VT_OBJGROUP_PROXY_PROXY_OBJGROUP_H*/

src/vt/objgroup/proxy/proxy_objgroup_elm.h

@@ -49,6 +49,8 @@
 #include "vt/objgroup/common.h"
 #include "vt/objgroup/proxy/proxy_bits.h"
 #include "vt/objgroup/active_func/active_func.h"
+#include "vt/activefn/activefn.h"
+#include "vt/messaging/active.h"
 #include "vt/messaging/message/smart_ptr.h"
 
 namespace vt { namespace objgroup { namespace proxy {
@@ -179,6 +181,28 @@ struct ProxyElm {
   NodeType node_           = uninitialized_destination; /**< The indexed node */
 };
 
+template <>
+struct ProxyElm<void> {
+  explicit ProxyElm(NodeType in_node) : node_{in_node} {}
+
+  /**
+   * \brief Send a message to the node indexed by this proxy to be
+   * delivered to the local object instance
+   *
+   * \param[in] args args to pass to the message constructor
+   */
+  template <typename MsgT, ActiveTypedFnType<MsgT>* fn, typename... Args>
+  void send(Args&&... args) const {
+    vt::theMsg()->sendMsg<MsgT, fn>(
+      node_, vt::makeMessage<MsgT>(std::forward<Args>(args)...));
+  }
+
+private:
+  NodeType node_ = uninitialized_destination; /**< The indexed node */
+};
+
+using DefaultProxyElm = ProxyElm<void>;
+
 }}} /* end namespace vt::objgroup::proxy */
 
 #endif /*INCLUDED_VT_OBJGROUP_PROXY_PROXY_OBJGROUP_ELM_H*/

tests/unit/collectives/test_collectives_reduce.extended.cc

@@ -107,10 +107,12 @@ TEST_F(TestReduce, test_reduce_vec_int_msg) {
   vecOfInt.push_back(3);
 
   auto const root = 0;
-  auto msg = makeMessage<ReduceVecMsg<int>>(vecOfInt);
-  theCollective()->global()->reduce<
-    PlusOp<std::vector<int>>, Verify<ReduceOP::Plus>
-  >(root, msg.get());
+  auto const default_proxy = theObjGroup()->makeCollective();
+  default_proxy.reduce<
+    PlusOp<std::vector<int>>,
+    Verify<ReduceOP::Plus>,
+    ReduceVecMsg<int>
+  >(root, vecOfInt);
 }
 
 }}} // end namespace vt::tests::unit

tests/unit/test_parallel_harness.h

@@ -92,7 +92,6 @@ struct TestParallelHarnessAny : TestHarnessAny<TestBase> {
 
     TestHarnessAny<TestBase>::SetUp();
 
-    // TODO (STRZ) - do we want to use additional args here as well?
     if (mpi_singleton == nullptr) {
       mpi_singleton =
         std::make_unique<MPISingletonMultiTest>(test_argc, test_argv);

tutorial/tutorial_1d.h

@@ -74,20 +74,26 @@ static inline void activeMessageBroadcast() {
   (void)num_nodes;  // don't warn about unused variable
 
   /*
-   * The theMsg()->broadcastMsg(..) will send the message to every node in the
+   * default_proxy.broadcast(..) will send the message to every node in the
    * system. Every node will include all the nodes that VT has depending on the
    * MPI communicator passed in or the size attained (number of ranks) when
    * executing MPI init directly in non-interoperability mode.
    *
    * -- Message Ownership --
-   * As with sendMsg, sending with broadcastMsg relinquishes ownership of
-   * the message. Most calls to VT that supply a message are expected
-   * to relinquish ownership.
+   * default_proxy.broadcast(..) will create message out of args passed
+   * into it. There's an alternative - default_proxy.broadcastMsg(..) - which
+   * relinquishes ownership of the message passed to:
+   *
+   *   auto msg = ::vt::makeMessage<MyDataMsg>(1.0, 2.0, 3.0);
+   *   auto const default_proxy = theObjGroup()->makeCollective();
+   *   default_proxy.broadcastMsg<MyDataMsg, msgHandlerX>(msg);
+   *
+   *  Most calls to VT that supply a message are expected to relinquish ownership.
    */
 
   if (this_node == 0) {
-    auto msg = ::vt::makeMessage<MyDataMsg>(1.0,2.0,3.0);
-    ::vt::theMsg()->broadcastMsg<MyDataMsg,msgHandlerX>(msg);
+    auto const default_proxy = theObjGroup()->makeCollective();
+    default_proxy.broadcast<MyDataMsg, msgHandlerX>(1.0, 2.0, 3.0);
   }
 }
 

tutorial/tutorial_1h.h

@@ -90,7 +90,8 @@ static inline void activeMessageReduce() {
   // Get a reference to the value to set it in this reduce msg
   reduce_msg->getVal() = 50;
 
-  ::vt::theCollective()->global()->reduce<ReduceOp,ReduceResult>(
+  auto const default_proxy = theObjGroup()->makeCollective();
+  default_proxy.reduceMsg<ReduceOp, ReduceResult>(
     root_reduce_node, reduce_msg.get()
   );
 }