fix startup for cyclic enclave programs

Author: cmnrd

include/reactor-cpp/connection.hh

@@ -197,12 +197,11 @@ public:
   inline auto acquire_tag(const Tag& tag, std::unique_lock<std::mutex>& lock, std::condition_variable& cv,
                           const std::function<bool(void)>& abort_waiting) -> bool override {
     // Since this is a delayed connection, we can go back in time and need to
-    // acquire the latest tag upstream tag that can create an event at the given
+    // acquire the latest upstream tag that can create an event at the given
     // tag. We also need to consider that given a delay d and a tag g=(t, n),
     // for any value of n, g + d = (t, 0). Hence, we need to quire a tag with
     // the highest possible microstep value.
-    auto time_point = tag.time_point() - this->min_delay();
-    auto upstream_tag = Tag::max_for_timepoint(time_point);
+    auto upstream_tag = tag.subtract(this->min_delay());
     return EnclaveConnection<T>::acquire_tag(upstream_tag, lock, cv, abort_waiting);
   }
 };

include/reactor-cpp/environment.hh

@@ -55,7 +55,7 @@ private:
 
   Scheduler scheduler_;
   Phase phase_{Phase::Construction};
-  TimePoint start_time_{};
+  Tag start_tag_{};
 
   const Duration timeout_{};
 
@@ -96,7 +96,7 @@ public:
   auto scheduler() noexcept -> Scheduler* { return &scheduler_; }
 
   [[nodiscard]] auto logical_time() const noexcept -> const LogicalTime& { return scheduler_.logical_time(); }
-  [[nodiscard]] auto start_time() const noexcept -> const TimePoint& { return start_time_; }
+  [[nodiscard]] auto start_tag() const noexcept -> const Tag& { return start_tag_; }
   [[nodiscard]] auto timeout() const noexcept -> const Duration& { return timeout_; }
 
   static auto physical_time() noexcept -> TimePoint { return get_physical_time(); }

include/reactor-cpp/logical_time.hh

@@ -41,6 +41,8 @@ public:
   [[nodiscard]]static auto max_for_timepoint(TimePoint time_point) noexcept -> Tag;
 
   [[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;
 };
 
 // define all the comparison operators

lib/action.cc

@@ -41,11 +41,11 @@ void Timer::startup() {
     return;
   }
 
-  Tag tag_zero = Tag::from_physical_time(environment()->start_time());
+  const Tag& start_tag = environment()->start_tag();
   if (offset_ != Duration::zero()) {
-    environment()->scheduler()->schedule_sync(this, tag_zero.delay(offset_));
+    environment()->scheduler()->schedule_sync(this, start_tag.delay(offset_));
   } else {
-    environment()->scheduler()->schedule_sync(this, tag_zero);
+    environment()->scheduler()->schedule_sync(this, start_tag);
   }
 }
 

lib/environment.cc

@@ -254,8 +254,7 @@ void Environment::calculate_indexes() {
 
 auto Environment::startup() -> std::thread {
   validate(this == top_environment_, "startup() may only be called on the top environment");
-  auto start_time = get_physical_time();
-  return startup(start_time);
+  return startup(get_physical_time());
 }
 
 auto Environment::startup(const TimePoint& start_time) -> std::thread {
@@ -264,7 +263,7 @@ auto Environment::startup(const TimePoint& start_time) -> std::thread {
   log_.debug() << "Starting the execution";
   phase_ = Phase::Startup;
 
-  start_time_ = start_time;
+  this->start_tag_ = Tag::from_physical_time(start_time);
   // start up initialize all reactors
   for (auto* reactor : top_level_reactors_) {
     reactor->startup();
@@ -273,12 +272,12 @@ auto Environment::startup(const TimePoint& start_time) -> std::thread {
   // start processing events
   phase_ = Phase::Execution;
 
-  return std::thread([this]() {
+  return std::thread([this, start_time]() {
     std::vector<std::thread> threads;
     threads.reserve(contained_environments_.size());
     // startup all contained environments recursively
     for (auto* env : contained_environments_) {
-      threads.emplace_back(env->startup(start_time_));
+      threads.emplace_back(env->startup(start_time));
     }
     // start the local scheduler and wait until it returns
     this->scheduler_.start();

lib/logical_time.cc

@@ -28,15 +28,26 @@ auto Tag::from_logical_time(const LogicalTime& logical_time) noexcept -> Tag {
   return {logical_time.time_point(), logical_time.micro_step()};
 }
 
-auto Tag::max_for_timepoint(TimePoint time_point) noexcept -> Tag {
-  return {time_point, std::numeric_limits<mstep_t>::max()};
+auto Tag::delay(Duration offset) const noexcept -> Tag {
+  if (offset == Duration::zero()) {
+    validate(this->micro_step_ != std::numeric_limits<mstep_t>::max(), "Microstep overflow detected!");
+    return {this->time_point_, this->micro_step_ + 1};
+  }
+  return {this->time_point_ + offset, 0};
 }
 
-auto Tag::delay(Duration offset) const noexcept -> Tag {
+auto Tag::subtract(Duration offset) const noexcept -> Tag {
   if (offset == Duration::zero()) {
-    return Tag{this->time_point_, this->micro_step_ + 1};
+    return decrement();
+  }
+  return {time_point_ - offset, std::numeric_limits<mstep_t>::max()};
+}
+
+auto Tag::decrement() const noexcept -> Tag {
+  if (micro_step_ == 0) {
+    return {time_point_ - Duration{1}, std::numeric_limits<mstep_t>::max()};
   }
-  return Tag{this->time_point_ + offset, 0};
+  return {time_point_, micro_step_ - 1};
 }
 
 void LogicalTime::advance_to(const Tag& tag) {
@@ -46,7 +57,7 @@ void LogicalTime::advance_to(const Tag& tag) {
 }
 
 void LogicalTime::advance_to(const LogicalTime& time) {
-  reactor_assert(*this <= Tag::from_logical_time(time));
+  reactor_assert(*this < Tag::from_logical_time(time));
   time_point_ = time.time_point();
   micro_step_ = time.micro_step();
 }

lib/reactor.cc

@@ -169,11 +169,11 @@ auto Reactor::get_tag() const noexcept -> Tag {
 }
 
 auto Reactor::get_elapsed_logical_time() const noexcept -> Duration {
-  return get_logical_time() - environment()->start_time();
+  return get_logical_time() - environment()->start_tag().time_point();
 }
 
 auto Reactor::get_elapsed_physical_time() const noexcept -> Duration {
-  return get_physical_time() - environment()->start_time();
+  return get_physical_time() - environment()->start_tag().time_point();
 }
 
 } // namespace reactor

lib/scheduler.cc

@@ -202,6 +202,11 @@ auto Scheduler::schedule_ready_reactions() -> bool {
 void Scheduler::start() {
   log_.debug() << "Starting the scheduler...";
 
+  // Initialize our logical time to the value right before the start tag. This
+  // is important for usage with enclaves/federates, to indicate, that no events
+  // before the start tag ca be generated.
+  logical_time_.advance_to(environment_->start_tag().decrement());
+
   auto num_workers = environment_->num_workers();
   // initialize the reaction queue, set ports vector, and triggered reactions
   // vector