Merge pull request #2136 from DARMA-tasking/276-automatically-synthesize-message-types-for-sends-to-arbitrary-handlers-cb

276 Callback parameterization

Author: lifflander

examples/callback/callback.cc

@@ -141,19 +141,19 @@ int main(int argc, char** argv) {
     auto cb_functor = vt::theCB()->makeSend<CallbackFunctor>(dest);
     bounceCallback(cb_functor);
 
-    auto cb_func = vt::theCB()->makeSend<TestMsg,callbackFunc>(dest);
+    auto cb_func = vt::theCB()->makeSend<callbackFunc>(dest);
     bounceCallback(cb_func);
 
-    auto cb_obj = vt::theCB()->makeSend<MyObj,TestMsg,&MyObj::handler>(obj[dest]);
+    auto cb_obj = vt::theCB()->makeSend<&MyObj::handler>(obj[dest]);
     bounceCallback(cb_obj);
 
-    auto cb_obj_bcast = vt::theCB()->makeBcast<MyObj,TestMsg,&MyObj::handler>(obj);
+    auto cb_obj_bcast = vt::theCB()->makeBcast<&MyObj::handler>(obj);
     bounceCallback(cb_obj_bcast);
 
-    auto cb_col = vt::theCB()->makeSend<MyCol,TestMsg,colHan>(col[5]);
+    auto cb_col = vt::theCB()->makeSend<colHan>(col[5]);
     bounceCallback(cb_col);
 
-    auto cb_col_bcast = vt::theCB()->makeBcast<MyCol,TestMsg,colHan>(col);
+    auto cb_col_bcast = vt::theCB()->makeBcast<colHan>(col);
     bounceCallback(cb_col_bcast);
   }
 

examples/collection/jacobi1d_vt.cc

@@ -73,9 +73,7 @@ static constexpr double const default_tol = 1.0e-02;
 
 struct NodeObj {
   bool is_finished_ = false;
-  struct WorkFinishedMsg : vt::Message {};
-
-  void workFinishedHandler(WorkFinishedMsg*) { is_finished_ = true; }
+  void workFinishedHandler() { is_finished_ = true; }
   bool isWorkFinished() { return is_finished_; }
 };
 using NodeObjProxy = vt::objgroup::proxy::Proxy<NodeObj>;
@@ -119,19 +117,12 @@ struct LinearPb1DJacobi : vt::Collection<LinearPb1DJacobi,vt::Index1D> {
 
   };
 
-  struct ReduxMsg : vt::collective::ReduceTMsg<double> {
-    ReduxMsg() = default;
-    explicit ReduxMsg(double in_val) : ReduceTMsg<double>(in_val) { }
-  };
-
-  void checkCompleteCB(ReduxMsg* msg) {
+  void checkCompleteCB(double normRes) {
     //
     // Only one object for the reduction will visit
     // this function
     //
 
-    double normRes = msg->getConstVal();
-
     auto const iter_max_reached = iter_ > maxIter_;
     auto const norm_res_done = normRes < default_tol;
 
@@ -143,7 +134,7 @@ struct LinearPb1DJacobi : vt::Collection<LinearPb1DJacobi,vt::Index1D> {
       fmt::print(to_print);
 
       // Notify all nodes that computation is finished
-      objProxy_.broadcast<NodeObj::WorkFinishedMsg, &NodeObj::workFinishedHandler>();
+      objProxy_.broadcast<&NodeObj::workFinishedHandler>();
     } else {
       fmt::print(" ## ITER {} >> Residual Norm = {} \n", iter_, normRes);
     }
@@ -184,11 +175,9 @@ struct LinearPb1DJacobi : vt::Collection<LinearPb1DJacobi,vt::Index1D> {
     }
 
     auto proxy = this->getCollectionProxy();
-    auto cb = vt::theCB()->makeSend<
-      LinearPb1DJacobi,ReduxMsg,&LinearPb1DJacobi::checkCompleteCB
-    >(proxy[0]);
-    auto msg2 = vt::makeMessage<ReduxMsg>(maxNorm);
-    proxy.reduce<vt::collective::MaxOp<double>>(msg2.get(),cb);
+    proxy.reduce<&LinearPb1DJacobi::checkCompleteCB, vt::collective::MaxOp>(
+      proxy[0], maxNorm
+    );
 
   }
 

examples/collection/jacobi2d_vt.cc

@@ -80,9 +80,7 @@ static constexpr double const default_tol = 1.0e-02;
 
 struct NodeObj {
   bool is_finished_ = false;
-  struct WorkFinishedMsg : vt::Message {};
-
-  void workFinishedHandler(WorkFinishedMsg*) { is_finished_ = true; }
+  void workFinishedHandler() { is_finished_ = true; }
   bool isWorkFinished() { return is_finished_; }
 };
 using NodeObjProxy = vt::objgroup::proxy::Proxy<NodeObj>;
@@ -129,19 +127,12 @@ struct LinearPb2DJacobi : vt::Collection<LinearPb2DJacobi,vt::Index2D> {
   };
 
 
-  struct ReduxMsg : vt::collective::ReduceTMsg<double> {
-    ReduxMsg() = default;
-    explicit ReduxMsg(double in_val) : ReduceTMsg<double>(in_val) { }
-  };
-
-
-  void checkCompleteCB(ReduxMsg* msg) {
+  void checkCompleteCB(double const normRes) {
     //
     // Only one object for the reduction will visit
     // this function
     //
 
-    const double normRes = msg->getConstVal();
     auto const iter_max_reached = iter_ > maxIter_;
     auto const norm_res_done = normRes < default_tol;
 
@@ -153,7 +144,7 @@ struct LinearPb2DJacobi : vt::Collection<LinearPb2DJacobi,vt::Index2D> {
       fmt::print(to_print);
 
       // Notify all nodes that computation is finished
-      objProxy_.broadcast<NodeObj::WorkFinishedMsg, &NodeObj::workFinishedHandler>();
+      objProxy_.broadcast<&NodeObj::workFinishedHandler>();
     } else {
       fmt::print(" ## ITER {} >> Residual Norm = {} \n", iter_, normRes);
     }
@@ -222,11 +213,9 @@ struct LinearPb2DJacobi : vt::Collection<LinearPb2DJacobi,vt::Index2D> {
     }
 
     auto proxy = this->getCollectionProxy();
-    auto cb = vt::theCB()->makeSend<
-      LinearPb2DJacobi,ReduxMsg,&LinearPb2DJacobi::checkCompleteCB
-    >(proxy(0,0));
-    auto msg2 = vt::makeMessage<ReduxMsg>(maxNorm);
-    proxy.reduce<vt::collective::MaxOp<double>>(msg2.get(),cb);
+    proxy.reduce<&LinearPb2DJacobi::checkCompleteCB, vt::collective::MaxOp>(
+      proxy(0,0), maxNorm
+    );
   }
 
   struct VecMsg : vt::CollectionMessage<LinearPb2DJacobi> {

examples/collection/reduce_integral.cc

@@ -101,21 +101,17 @@ struct Integration1D : vt::Collection<Integration1D, vt::Index1D> {
       numPartsPerObject_(default_nparts_object)
   { }
 
-  struct CheckIntegral {
-
-    void operator()(ReduceMsg* msg) {
-      fmt::print(" >> The integral over [0, 1] is {}\n", msg->getConstVal());
-      fmt::print(
-        " >> The absolute error is {}\n",
-        std::fabs(msg->getConstVal() - exactIntegral)
-      );
-      //
-      // Set the 'root_reduce_finished' variable to true.
-      //
-      root_reduce_finished = true;
-    }
-
-  };
+  void checkIntegral(double val) {
+    fmt::print(" >> The integral over [0, 1] is {}\n", val);
+    fmt::print(
+      " >> The absolute error is {}\n",
+      std::fabs(val - exactIntegral)
+    );
+    //
+    // Set the 'root_reduce_finished' variable to true.
+    //
+    root_reduce_finished = true;
+  }
 
   struct InitMsg : vt::CollectionMessage<Integration1D> {
 
@@ -177,9 +173,9 @@ struct Integration1D : vt::Collection<Integration1D, vt::Index1D> {
     //
 
     auto proxy = this->getCollectionProxy();
-    auto msgCB = vt::makeMessage<ReduceMsg>(quadsum);
-    auto cback = vt::theCB()->makeSend<CheckIntegral>(reduce_root_node);
-    proxy.reduce<vt::collective::PlusOp<double>>(msgCB.get(),cback);
+    proxy.reduce<&Integration1D::checkIntegral, vt::collective::PlusOp>(
+      proxy[0], quadsum
+    );
   }
 
 };

examples/collection/transpose.cc

@@ -58,8 +58,6 @@ struct RequestDataMsg : vt::CollectionMessage<ColT> {
   vt::NodeType node_;
 };
 
-struct InitMsg : vt::collective::ReduceNoneMsg { };
-
 struct DataMsg : vt::Message {
   using MessageParentType = vt::Message;
   vt_msg_serialize_required(); // by payload_
@@ -170,7 +168,7 @@ struct Block : vt::Collection<Block, vt::Index1D> {
     );
   }
 
-  void doneInit(InitMsg* msg) {
+  void doneInit() {
     if (getIndex().x() == 0) {
       auto proxy = this->getCollectionProxy();
       auto proxy_msg = vt::makeMessage<ProxyMsg>(proxy.getProxy());
@@ -183,9 +181,7 @@ struct Block : vt::Collection<Block, vt::Index1D> {
     initialize();
     // Wait for all initializations to complete
     auto proxy = this->getCollectionProxy();
-    auto cb = vt::theCB()->makeBcast<Block, InitMsg, &Block::doneInit>(proxy);
-    auto empty = vt::makeMessage<InitMsg>();
-    proxy.reduce(empty.get(), cb);
+    proxy.allreduce<&Block::doneInit>();
   }
 
 private:

examples/hello_world/CMakeLists.txt

@@ -11,6 +11,7 @@ set(
   hello_world_virtual_context_remote
   ring
   objgroup
+  hello_reduce
 )
 
 foreach(EXAMPLE_NAME ${HELLO_WORLD_EXAMPLES})

examples/hello_world/hello_reduce.cc

@@ -0,0 +1,62 @@
+/*
+//@HEADER
+// *****************************************************************************
+//
+//                               hello_reduce.cc
+//                       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 <vt/transport.h>
+
+void reduceResult(int result, double result2) {
+  auto num_nodes = vt::theContext()->getNumNodes();
+  fmt::print("reduction value={}, {}\n", result, result2);
+  vtAssert(num_nodes * 50 == result, "Must be equal");
+}
+
+int main(int argc, char** argv) {
+  vt::initialize(argc, argv);
+
+  vt::NodeType const root = 0;
+
+  auto r = vt::theCollective()->global();
+  r->reduce<reduceResult, vt::collective::PlusOp>(vt::Node{root}, 50, 52.334);
+
+  vt::finalize();
+  return 0;
+}

examples/hello_world/hello_world_collection_reduce.cc

@@ -45,24 +45,19 @@
 
 /// [Hello world reduce collection]
 struct Hello : vt::Collection<Hello, vt::Index1D> {
-  using ReduceMsg = vt::collective::ReduceTMsg<int>;
-
-  void done(ReduceMsg* msg) {
-    fmt::print("Reduce complete at {} value {}\n", this->getIndex(), msg->getVal());
+  void done(int val, double val2) {
+    fmt::print("Reduce complete at {} values {} {}\n", getIndex(), val, val2);
   }
 
   void doWork() {
-    fmt::print("Hello from {}\n", this->getIndex());
+    fmt::print("Hello from {}\n", getIndex());
 
     // Get the proxy for the collection
-    auto proxy = this->getCollectionProxy();
-
-    // Create a callback for when the reduction finishes
-    auto cb = vt::theCB()->makeSend<Hello,ReduceMsg,&Hello::done>(proxy(2));
+    auto proxy = getCollectionProxy();
 
-    // Create and send the reduction message holding an int
-    auto red_msg = vt::makeMessage<ReduceMsg>(this->getIndex().x());
-    proxy.reduce<vt::collective::PlusOp<int>>(red_msg.get(),cb);
+    auto val = getIndex().x();
+    auto val2 = 2.4;
+    proxy.allreduce<&Hello::done, vt::collective::PlusOp>(val, val2);
   }
 };
 

src/vt/collective/collective_scope.cc

@@ -83,7 +83,7 @@ TagType CollectiveScope::mpiCollectiveAsync(ActionType action) {
   NodeType collective_root = 0;
 
   using CollectiveMsg = CollectiveAlg::CollectiveMsg;
-  auto cb = theCB()->makeBcast<CollectiveMsg,&CollectiveAlg::runCollective>();
+  auto cb = theCB()->makeBcast<&CollectiveAlg::runCollective>();
   auto msg = makeMessage<CollectiveMsg>(is_user_tag_, scope_, tag, collective_root);
 
   // The tag for the reduce is a combination of the scope and seq tag.