Fix timeout so shutdown is invoked only once

This fixes lf-lang/lingua-franca#1955 and makes the
overall implementation more robust.

Author: cmnrd

include/reactor-cpp/environment.hh

@@ -58,6 +58,7 @@ private:
   Tag start_tag_{};
 
   const Duration timeout_{};
+  Tag timeout_tag_{};
 
   Graph<BasePort*, ConnectionProperties> graph_{};
   Graph<BasePort*, ConnectionProperties> optimized_graph_{};
@@ -118,6 +119,7 @@ public:
   [[nodiscard]] auto logical_time() const noexcept -> const LogicalTime& { return scheduler_.logical_time(); }
   [[nodiscard]] auto start_tag() const noexcept -> const Tag& { return start_tag_; }
   [[nodiscard]] auto timeout() const noexcept -> const Duration& { return timeout_; }
+  [[nodiscard]] auto timeout_tag() const noexcept -> const Tag& { return timeout_tag_; }
 
   static auto physical_time() noexcept -> TimePoint { return get_physical_time(); }
 

include/reactor-cpp/logical_time.hh

@@ -43,6 +43,8 @@ public:
   [[nodiscard]] auto delay(Duration offset = Duration::zero()) const noexcept -> Tag;
   [[nodiscard]] auto subtract(Duration offset = Duration::zero()) const noexcept -> Tag;
   [[nodiscard]] auto decrement() const noexcept -> Tag;
+
+  [[nodiscard]] static auto max() noexcept -> Tag { return {TimePoint::max(), std::numeric_limits<mstep_t>::max()}; }
 };
 
 // define all the comparison operators

lib/action.cc

@@ -71,16 +71,11 @@ void Timer::cleanup() noexcept {
 ShutdownTrigger::ShutdownTrigger(const std::string& name, Reactor* container)
     : Timer(name, container, Duration::zero(), container->environment()->timeout()) {}
 
-void ShutdownTrigger::setup() noexcept {
-  BaseAction::setup();
-  environment()->sync_shutdown();
-}
+void ShutdownTrigger::setup() noexcept { BaseAction::setup(); }
 
 void ShutdownTrigger::shutdown() {
-  if (!is_present()) {
-    Tag tag = Tag::from_logical_time(environment()->logical_time()).delay();
-    environment()->scheduler()->schedule_sync(this, tag);
-  }
+  Tag tag = Tag::from_logical_time(environment()->logical_time()).delay();
+  environment()->scheduler()->schedule_sync(this, tag);
 }
 
 auto Action<void>::schedule_at(const Tag& tag) -> bool {

lib/environment.cc

@@ -330,6 +330,12 @@ auto Environment::startup(const TimePoint& start_time) -> std::thread {
   phase_ = Phase::Startup;
 
   this->start_tag_ = Tag::from_physical_time(start_time);
+  if (this->timeout_ == Duration::max()) {
+    this->timeout_tag_ = Tag::max();
+  } else {
+    this->timeout_tag_ = this->start_tag_.delay(this->timeout_);
+  }
+
   // start up initialize all reactors
   for (auto* reactor : top_level_reactors_) {
     reactor->startup();

lib/scheduler.cc

@@ -91,17 +91,17 @@ void Scheduler::schedule() noexcept {
   bool found_ready_reactions = schedule_ready_reactions();
 
   while (!found_ready_reactions) {
-    log_.debug() << "call next()";
-    next();
-    reaction_queue_pos_ = 0;
-
-    found_ready_reactions = schedule_ready_reactions();
-
     if (!continue_execution_ && !found_ready_reactions) {
       // let all workers know that they should terminate
       terminate_all_workers();
       break;
     }
+
+    log_.debug() << "call next()";
+    next();
+    reaction_queue_pos_ = 0;
+
+    found_ready_reactions = schedule_ready_reactions();
   }
 }
 
@@ -354,8 +354,8 @@ void Scheduler::next() { // NOLINT
         log_.debug() << "Shutting down the scheduler";
         Tag t_next = Tag::from_logical_time(logical_time_).delay();
         if (!event_queue_.empty() && t_next == event_queue_.next_tag()) {
-          log_.debug() << "Schedule the last round of reactions including all "
-                          "termination reactions";
+          log_.debug() << "Trigger the last round of reactions including all "
+                          "shutdown reactions";
           triggered_actions_ = event_queue_.extract_next_event();
           advance_logical_time_to(t_next);
         } else {
@@ -364,6 +364,13 @@ void Scheduler::next() { // NOLINT
       } else {
         auto t_next = event_queue_.next_tag();
 
+        if (t_next == environment_->timeout_tag()) {
+          continue_execution_ = false;
+          log_.debug() << "Shutting down the scheduler due to timeout";
+          log_.debug() << "Trigger the last round of reactions including all "
+                          "shutdwon reactions";
+        }
+
         // synchronize with physical time if not in fast forward mode
         if (!environment_->fast_fwd_execution()) {
           log_.debug() << "acquire tag " << t_next << " from physical time barrier";