#410: Dependent Epochs rewritten

src/vt/collective/reduce/reduce_manager.cc

@@ -46,12 +46,12 @@
 
 namespace vt { namespace collective { namespace reduce {
 
+static std::unique_ptr<Reduce> makeReduceScope(detail::ReduceScope const& scope) {
+  return std::make_unique<Reduce>(scope);
+}
+
 ReduceManager::ReduceManager()
-  : reducers_( // default cons reducer for non-group
-      [](detail::ReduceScope const& scope) {
-        return std::make_unique<Reduce>(scope);
-      }
-    )
+  : reducers_(makeReduceScope)
 {
   // insert the default reducer scope
   reducers_.make(

src/vt/context/runnable_context/td.h

@@ -104,6 +104,13 @@ struct TD {
    */
   void resume();
 
+  /**
+   * \brief Get epoch for this context
+   *
+   * \return the epoch
+   */
+  EpochType getEpoch() const { return ep_; }
+
 private:
   EpochType ep_ = no_epoch;                    /**< The epoch for the task */
 #if vt_check_enabled(fcontext)

src/vt/epoch/epoch.h

@@ -115,7 +115,7 @@ static constexpr BitCountType const epoch_category_num_bits = 2;
  */
 enum struct eEpochCategory : int8_t {
   NoCategoryEpoch       = 0x0,
-  InsertEpoch           = 0x1,
+  DependentEpoch        = 0x1,
   DijkstraScholtenEpoch = 0x2
 };
 

src/vt/epoch/epoch_manip.cc

@@ -55,12 +55,6 @@ namespace vt { namespace epoch {
 
 static EpochType const arch_epoch_coll = makeEpochZero();
 
-EpochManip::EpochManip()
-  : terminated_collective_epochs_(
-      std::make_unique<EpochWindow>(arch_epoch_coll)
-    )
-{ }
-
 /*static*/ EpochType EpochManip::generateEpoch(
   bool const& is_rooted, NodeType const& root_node,
   eEpochCategory const& category
@@ -123,22 +117,19 @@ EpochType EpochManip::getArchetype(EpochType epoch) const {
 
 EpochWindow* EpochManip::getTerminatedWindow(EpochType epoch) {
   auto const is_rooted = isRooted(epoch);
-  if (is_rooted and epoch != term::any_epoch_sentinel) {
-    auto const& arch_epoch = getArchetype(epoch);
-    auto iter = terminated_epochs_.find(arch_epoch);
-    if (iter == terminated_epochs_.end()) {
-      terminated_epochs_.emplace(
-        std::piecewise_construct,
-        std::forward_as_tuple(arch_epoch),
-        std::forward_as_tuple(std::make_unique<EpochWindow>(arch_epoch))
-      );
-      iter = terminated_epochs_.find(arch_epoch);
-    }
-    return iter->second.get();
-  } else {
-    vtAssertExpr(terminated_collective_epochs_ != nullptr);
-    return terminated_collective_epochs_.get();
+  auto& container = is_rooted and epoch != term::any_epoch_sentinel ?
+    terminated_epochs_ : terminated_collective_epochs_;
+  auto const& arch_epoch = getArchetype(epoch);
+  auto iter = container.find(arch_epoch);
+  if (iter == container.end()) {
+    container.emplace(
+      std::piecewise_construct,
+      std::forward_as_tuple(arch_epoch),
+      std::forward_as_tuple(std::make_unique<EpochWindow>(arch_epoch))
+    );
+    iter = container.find(arch_epoch);
   }
+  return iter->second.get();
 }
 
 /*static*/ bool EpochManip::isRooted(EpochType const& epoch) {
@@ -148,6 +139,29 @@ EpochWindow* EpochManip::getTerminatedWindow(EpochType epoch) {
   return BitPackerType::boolGetField<field,size,ImplType>(*epoch);
 }
 
+/*static*/ bool EpochManip::isDS(EpochType epoch) {
+  using T = typename std::underlying_type<epoch::eEpochCategory>::type;
+  if (isRooted(epoch)) {
+    auto const ds_bit = epoch::eEpochCategory::DijkstraScholtenEpoch;
+    auto const cat = category(epoch);
+    bool const is_ds = static_cast<T>(cat) & static_cast<T>(ds_bit);
+    return is_ds;
+  } else {
+    return false;
+  }
+}
+
+/*static*/ bool EpochManip::isDep(EpochType epoch) {
+  using T = typename std::underlying_type<epoch::eEpochCategory>::type;
+  if (epoch == no_epoch or epoch == term::any_epoch_sentinel) {
+    return false;
+  }
+  auto const dep_bit = epoch::eEpochCategory::DependentEpoch;
+  auto const cat = epoch::EpochManip::category(epoch);
+  bool const is_dep = static_cast<T>(cat) & static_cast<T>(dep_bit);
+  return is_dep;
+}
+
 /*static*/ eEpochCategory EpochManip::category(EpochType const& epoch) {
   return BitPackerType::getField<
     eEpochRoot::rEpochCategory, epoch_category_num_bits, eEpochCategory
@@ -190,6 +204,14 @@ void EpochManip::setCategory(EpochType& epoch, eEpochCategory const cat) {
   >(*epoch,cat);
 }
 
+/*static*/ eEpochCategory EpochManip::makeCat(
+  eEpochCategory c1, eEpochCategory c2
+) {
+  using T = typename std::underlying_type<eEpochCategory>::type;
+  auto ret = static_cast<T>(c1) | static_cast<T>(c2);
+  return static_cast<eEpochCategory>(ret);
+}
+
 /*static*/
 void EpochManip::setNode(EpochType& epoch, NodeType const node) {
   vtAssert(isRooted(epoch), "Must be rooted to manipulate the node");

src/vt/epoch/epoch_manip.h

@@ -72,7 +72,7 @@ namespace vt { namespace epoch {
 struct EpochManip : runtime::component::Component<EpochManip> {
   using CapturedContextType = term::ParentEpochCapture;
 
-  EpochManip();
+  EpochManip() = default;
 
   std::string name() override { return "EpochManip"; }
 
@@ -89,6 +89,24 @@ struct EpochManip : runtime::component::Component<EpochManip> {
    */
   static bool isRooted(EpochType const& epoch);
 
+  /**
+   * \brief Gets whether an epoch is DS or onot
+   *
+   * \param[in] epoch the epoch
+   *
+   * \return whether it is DS
+   */
+  static bool isDS(EpochType epoch);
+
+  /**
+   * \brief Gets whether an epoch is dependent or onot
+   *
+   * \param[in] epoch the epoch
+   *
+   * \return whether it is dependent
+   */
+  static bool isDep(EpochType epoch);
+
   /**
    * \brief Gets the \c eEpochCategory of a given epoch
    *
@@ -152,6 +170,14 @@ struct EpochManip : runtime::component::Component<EpochManip> {
    */
   static void setSeq(EpochType& epoch, EpochType::ImplType const seq);
 
+  /**
+   * \brief Combine eEpochCategory elements
+   *
+   * \param[in] c1 category 1
+   * \param[in] c2 category 2
+   */
+  static eEpochCategory makeCat(eEpochCategory c1, eEpochCategory c2);
+
   /*
    * General (stateless) methods for creating a epoch with certain properties
    * based on a current sequence number
@@ -250,10 +276,12 @@ struct EpochManip : runtime::component::Component<EpochManip> {
   }
 
 private:
-  // epoch window container for specific archetyped epochs
-  std::unordered_map<EpochType,std::unique_ptr<EpochWindow>> terminated_epochs_;
-  // epoch window for basic collective epochs
-  std::unique_ptr<EpochWindow> terminated_collective_epochs_ = nullptr;
+  /// epoch window container for specific archetyped epochs
+  std::unordered_map<EpochType, std::unique_ptr<EpochWindow>> terminated_epochs_;
+  /// epoch window for basic collective epochs
+  std::unordered_map<
+    EpochType, std::unique_ptr<EpochWindow>
+  > terminated_collective_epochs_;
 };
 
 }} /* end namespace vt::epoch */

src/vt/messaging/active.cc

@@ -859,7 +859,7 @@ void ActiveMessenger::finishPendingDataMsgAsyncRecv(InProgressDataIRecv* irecv)
       theTerm()->consume(term::any_epoch_sentinel,1,sender);
       theTerm()->hangDetectRecv();
     };
-    theSched()->enqueue(irecv->priority, run);
+    theSched()->enqueueLambda(irecv->priority, run);
   }
 }
 
@@ -934,12 +934,13 @@ void ActiveMessenger::prepareActiveMsgToRun(
   using MsgType = ShortMessage;
   auto msg = base.to<MsgType>().get();
 
-  auto const is_term = envelopeIsTerm(msg->env);
-  auto const is_bcast = envelopeIsBcast(msg->env);
-  auto const dest = envelopeGetDest(msg->env);
-  auto const handler = envelopeGetHandler(msg->env);
-  auto const epoch = envelopeIsEpochType(msg->env) ?
+  auto const is_term   = envelopeIsTerm(msg->env);
+  auto const is_bcast  = envelopeIsBcast(msg->env);
+  auto const dest      = envelopeGetDest(msg->env);
+  auto const handler   = envelopeGetHandler(msg->env);
+  auto const epoch     = envelopeIsEpochType(msg->env) ?
     envelopeGetEpoch(msg->env) : term::any_epoch_sentinel;
+
   auto const from_node = is_bcast ? dest : in_from_node;
 
   if (!is_term || vt_check_enabled(print_term_msgs)) {

src/vt/messaging/active.h

@@ -268,7 +268,7 @@ struct BufferedActiveMsg {
 
   BufferedActiveMsg(
     MessageType const& in_buffered_msg, NodeType const& in_from_node,
-    ActionType in_cont
+    ActionType in_cont = nullptr
   ) : buffered_msg(in_buffered_msg), from_node(in_from_node), cont(in_cont)
   { }
 

src/vt/runnable/runnable.cc

@@ -246,6 +246,14 @@ void RunnableNew::resume(TimeType time) {
 #endif
 }
 
+EpochType RunnableNew::getEpoch() const {
+  if (contexts_.has_td) {
+    return contexts_.td.getEpoch();
+  } else {
+    return no_epoch;
+  }
+}
+
 void RunnableNew::send(elm::ElementIDStruct elm, MsgSizeType bytes) {
   if (contexts_.has_lb) contexts_.lb.send(elm, bytes);
 }

src/vt/runnable/runnable.h

@@ -353,6 +353,13 @@ struct RunnableNew {
    */
   static void operator delete(void* ptr);
 
+  /**
+   * \brief Get the epoch for a runnable
+   *
+   * \return the epoch
+   */
+  EpochType getEpoch() const;
+
 private:
   detail::Contexts contexts_;               /**< The contexts  */
   MsgSharedPtr<BaseMsgType> msg_ = nullptr; /**< The associated message */

src/vt/runtime/runtime.cc

@@ -444,7 +444,7 @@ bool Runtime::initialize(bool const force_now) {
       printStartupBanner();
       // Enqueue a check for later in case arguments are modified before work
       // actually executes
-      theSched->enqueue([this]{
+      theSched->enqueueLambda([this]{
         this->checkForArgumentErrors();
       });
 

src/vt/runtime/runtime_banner.cc

@@ -939,7 +939,7 @@ void Runtime::printStartupBanner() {
 
   // Enqueue a check for later in case arguments are modified before work
   // actually executes
-  theSched->enqueue([this]{
+  theSched->enqueueLambda([this]{
     this->checkForArgumentErrors();
   });
 }

src/vt/scheduler/scheduler.cc

@@ -384,6 +384,17 @@ void Scheduler::resume(ThreadIDType tid) {
   suspended_.resumeRunnable(tid);
 }
 
+void Scheduler::releaseEpoch(EpochType ep) {
+  auto iter = pending_work_.find(ep);
+  if (iter != pending_work_.end()) {
+    auto& container = iter->second;
+    while (container.size() > 0) {
+      work_queue_.emplace(container.pop());
+    }
+    pending_work_.erase(iter);
+  }
+}
+
 #if vt_check_enabled(fcontext)
 ThreadManager* Scheduler::getThreadManager() {
   return thread_manager_.get();

src/vt/scheduler/scheduler.h

@@ -234,6 +234,24 @@ struct Scheduler : runtime::component::Component<Scheduler> {
   template <typename RunT>
   void enqueue(RunT r);
 
+  /**
+   * \brief Enqueue a callable to execute later with the default priority
+   * \c default_priority
+   *
+   * \param[in] r action to execute
+   */
+  template <typename Callable>
+  void enqueueLambda(Callable&& c);
+
+  /**
+   * \brief Enqueue a callable to execute later with a priority
+   *
+   * \param[in] priority the priority of the action
+   * \param[in] r action to execute
+   */
+  template <typename Callable>
+  void enqueueLambda(PriorityType priority, Callable&& c);
+
   /**
    * \brief Enqueue an runnable with a priority to execute later
    *
@@ -352,6 +370,13 @@ struct Scheduler : runtime::component::Component<Scheduler> {
   ThreadManager* getThreadManager();
 #endif
 
+  /**
+   * \brief Release an epoch to run
+   *
+   * \param[in] ep the epoch to release
+   */
+  void releaseEpoch(EpochType ep);
+
   template <typename SerializerT>
   void serialize(SerializerT& s) {
     s | work_queue_
@@ -381,7 +406,8 @@ struct Scheduler : runtime::component::Component<Scheduler> {
       | vtLiveTime
       | schedLoopTime
       | idleTime
-      | idleTimeMinusTerm;
+      | idleTimeMinusTerm
+      | pending_work_;
   }
 
 private:
@@ -419,6 +445,9 @@ struct Scheduler : runtime::component::Component<Scheduler> {
   Queue<UnitType> work_queue_;
 # endif
 
+  /// Unreleased work pending an epoch release
+  std::unordered_map<EpochType, Queue<UnitType>> pending_work_;
+
 #if vt_check_enabled(fcontext)
   std::unique_ptr<ThreadManager> thread_manager_ = nullptr;
 #endif