fixing startup in non-federated

Author: tanneberger

include/reactor-uc/environment.h

@@ -150,4 +150,7 @@ struct Environment {
 void Environment_ctor(Environment *self, Reactor *main, Scheduler *scheduler, bool fast_mode);
 void Environment_free(Environment *self);
 
+void Environment_schedule_startups(const Environment *self, tag_t start_tag);
+void Environment_schedule_timers(Environment *self, const Reactor *reactor, tag_t start_tag);
+
 #endif

include/reactor-uc/startup_coordinator.h

@@ -4,6 +4,7 @@
 #include "reactor-uc/error.h"
 #include "reactor-uc/tag.h"
 #include "reactor-uc/event.h"
+#include "reactor-uc/reactor.h"
 #include "proto/message.pb.h"
 
 typedef struct StartupCoordinator StartupCoordinator;
@@ -53,4 +54,7 @@ void StartupCoordinator_ctor(StartupCoordinator *self, Environment *env, Neighbo
                              size_t payload_size, void *payload_buf, bool *payload_used_buf,
                              size_t payload_buf_capacity);
 
+void StartupCoordinator_schedule_startups(const StartupCoordinator *self, const tag_t start_tag);
+void StartupCoordinator_schedule_timers(StartupCoordinator *self, const Reactor *reactor, const tag_t start_tag);
+
 #endif // REACTOR_UC_STARTUP_COORDINATOR_H

src/environment.c

@@ -4,3 +4,32 @@
 #else
 #include "./environments/unfederated_environment.c"
 #endif
+
+#include "reactor-uc/timer.h"
+
+void Environment_schedule_startups(const Environment *self, const tag_t start_tag) {
+  if (self->startup) {
+    LF_DEBUG(FED, "Scheduling Startup Reactions at" PRINTF_TAG, start_tag);
+    Event event = EVENT_INIT(start_tag, &self->startup->super, NULL);
+    LF_INFO(FED, "Self: %p Scheduler: %p", self, self->scheduler);
+    lf_ret_t ret = self->scheduler->schedule_at(self->scheduler, &event);
+    validate(ret == LF_OK);
+  }
+}
+
+void Environment_schedule_timers(Environment *self, const Reactor *reactor, const tag_t start_tag) {
+  lf_ret_t ret;
+  for (size_t i = 0; i < reactor->triggers_size; i++) {
+    Trigger *trigger = reactor->triggers[i];
+    if (trigger->type == TRIG_TIMER) {
+      Timer *timer = (Timer *)trigger;
+      tag_t tag = {.time = start_tag.time + timer->offset, .microstep = start_tag.microstep};
+      Event event = EVENT_INIT(tag, trigger, NULL);
+      ret = self->scheduler->schedule_at(self->scheduler, &event);
+      validate(ret == LF_OK);
+    }
+  }
+  for (size_t i = 0; i < reactor->children_size; i++) {
+    Environment_schedule_timers(self, reactor->children[i], start_tag);
+  }
+}

src/environments/unfederated_environment.c

@@ -3,6 +3,7 @@
 #include "reactor-uc/reactor.h"
 #include "reactor-uc/scheduler.h"
 #include "reactor-uc/queues.h"
+#include "reactor-uc/startup_coordinator.h"
 #include <assert.h>
 #include <inttypes.h>
 
@@ -18,6 +19,11 @@ static void Environment_assemble(Environment *self) {
 static void Environment_start(Environment *self) {
   instant_t start_time = self->get_physical_time(self);
   LF_INFO(ENV, "Starting program at " PRINTF_TIME " nsec", start_time);
+  tag_t start_tag = {.time = start_time, .microstep = 0};
+  self->scheduler->prepare_timestep(self->scheduler, NEVER_TAG);
+  Environment_schedule_startups(self, start_tag);
+  Environment_schedule_timers(self, self->main, start_tag);
+  self->scheduler->prepare_timestep(self->scheduler, start_tag);
   self->scheduler->set_and_schedule_start_tag(self->scheduler, start_time);
   self->scheduler->run(self->scheduler);
 }

src/schedulers/dynamic/scheduler.c

@@ -105,7 +105,6 @@ void Scheduler_prepare_timestep(Scheduler *untyped_self, tag_t tag) {
   self->current_tag = tag;
   MUTEX_UNLOCK(self->mutex);
 
-  LF_DEBUG(SCHED, "Preparing timestep for tag " PRINTF_TAG " object %p", self->current_tag, self);
   self->reaction_queue->reset(self->reaction_queue);
 }
 
@@ -354,9 +353,7 @@ void Scheduler_run(Scheduler *untyped_self) {
 lf_ret_t Scheduler_schedule_at(Scheduler *super, Event *event) {
   DynamicScheduler *self = (DynamicScheduler *)super;
   lf_ret_t ret;
-  LF_DEBUG(SCHED, "Scheduler_schedule_at %p", self);
-  LF_DEBUG(SCHED, "scheduling at intended: " PRINTF_TAG, event->intended_tag);
-  LF_DEBUG(SCHED, "scheduling current tag: " PRINTF_TAG, self->current_tag);
+
   // This can be called from the async context and the channel context. It reads stop_tag, current_tag, start_time
   // and more and we lock the scheduler mutex before doing anything.
   MUTEX_LOCK(self->mutex);
@@ -369,7 +366,7 @@ lf_ret_t Scheduler_schedule_at(Scheduler *super, Event *event) {
     goto unlock_and_return;
   }
 
-  // Check if we are tring to schedule into the past
+  // Check if we are trying to schedule into the past
   if (lf_tag_compare(event->super.tag, self->current_tag) <= 0) {
     LF_WARN(SCHED, "Trying to schedule event at tag " PRINTF_TAG " which is before current tag " PRINTF_TAG,
             event->super.tag, self->current_tag);

src/startup_coordinator.c

@@ -57,33 +57,6 @@ static lf_ret_t StartupCoordinator_connect_to_neighbors_blocking(StartupCoordina
   return LF_OK;
 }
 
-void StartupCoordinator_schedule_startups(const StartupCoordinator *self, const tag_t start_tag) {
-  if (self->env->startup) {
-    LF_DEBUG(FED, "Scheduling Startup Reactions at" PRINTF_TAG, start_tag);
-    Event event = EVENT_INIT(start_tag, &self->env->startup->super, NULL);
-    LF_INFO(FED, "Self: %p Scheduler: %p", self, self->env->scheduler);
-    lf_ret_t ret = self->env->scheduler->schedule_at(self->env->scheduler, &event);
-    validate(ret == LF_OK);
-  }
-}
-
-void StartupCoordinator_schedule_timers(StartupCoordinator *self, const Reactor *reactor, const tag_t start_tag) {
-  lf_ret_t ret;
-  for (size_t i = 0; i < reactor->triggers_size; i++) {
-    Trigger *trigger = reactor->triggers[i];
-    if (trigger->type == TRIG_TIMER) {
-      Timer *timer = (Timer *)trigger;
-      tag_t tag = {.time = start_tag.time + timer->offset, .microstep = start_tag.microstep};
-      Event event = EVENT_INIT(tag, trigger, NULL);
-      ret = self->env->scheduler->schedule_at(self->env->scheduler, &event);
-      validate(ret == LF_OK);
-    }
-  }
-  for (size_t i = 0; i < reactor->children_size; i++) {
-    StartupCoordinator_schedule_timers(self, reactor->children[i], start_tag);
-  }
-}
-
 void StartupCoordinator_schedule_timers_joining(StartupCoordinator *self, Reactor *reactor,
                                                 interval_t federation_start_time, interval_t join_time) {
   lf_ret_t ret;
@@ -368,8 +341,8 @@ static void StartupCoordinator_handle_start_time_proposal(StartupCoordinator *se
       self->state = StartupCoordinationState_RUNNING;
       self->env->scheduler->set_and_schedule_start_tag(self->env->scheduler, self->start_time_proposal);
       tag_t start_tag = {.time = self->start_time_proposal, .microstep = 0};
-      StartupCoordinator_schedule_startups(self, start_tag);
-      StartupCoordinator_schedule_timers(self, self->env->main, start_tag);
+      Environment_schedule_startups(self->env, start_tag);
+      Environment_schedule_timers(self->env, self->env->main, start_tag);
     } else {
       self->start_time_proposal_step++;
       send_start_time_proposal(self, self->start_time_proposal, self->start_time_proposal_step);
@@ -465,16 +438,16 @@ static void StartupCoordinator_handle_start_time_response(StartupCoordinator *se
     tag_t start_tag = {.time = joining_time, .microstep = 0};
     LF_INFO(FED, "Policy: IMMEDIATELY Scheduling join_time: " PRINTF_TIME, joining_time);
     self->env->scheduler->prepare_timestep(self->env->scheduler, NEVER_TAG);
-    StartupCoordinator_schedule_startups(self, start_tag);
-    StartupCoordinator_schedule_timers(self, self->env->main, start_tag);
+    Environment_schedule_startups(self->env, start_tag);
+    Environment_schedule_timers(self->env, self->env->main, start_tag);
     self->env->scheduler->prepare_timestep(self->env->scheduler, start_tag);
     self->env->scheduler->set_and_schedule_start_tag(self->env->scheduler, joining_time);
   } else if (self->joining_policy == JOIN_ALIGNED_WITH_SHORT_TIMER) {
     joining_time = max_logical_time + MSEC(50);
     tag_t start_tag = {.time = joining_time, .microstep = 0};
     LF_INFO(FED, "Policy: Timer Aligned Scheduling join_time: " PRINTF_TIME, joining_time);
     self->env->scheduler->prepare_timestep(self->env->scheduler, NEVER_TAG);
-    StartupCoordinator_schedule_startups(self, start_tag);
+    Environment_schedule_startups(self->env, start_tag);
     StartupCoordinator_schedule_timers_joining(self, self->env->main, start_time, joining_time);
     self->env->scheduler->prepare_timestep(self->env->scheduler, start_tag);
     self->env->scheduler->set_and_schedule_start_tag(self->env->scheduler, joining_time);
@@ -542,6 +515,8 @@ void StartupCoordinator_ctor(StartupCoordinator *self, Environment *env, Neighbo
   self->start_time_proposal = NEVER;
   self->joining_policy = joining_policy;
   for (size_t i = 0; i < self->num_neighbours; i++) {
+    self->neighbor_state[i].core_federate = true;
+    self->neighbor_state[i].current_logical_time = 0;
     self->neighbor_state[i].handshake_response_received = false;
     self->neighbor_state[i].handshake_request_received = false;
     self->neighbor_state[i].handshake_response_sent = false;