working version

Author: tanneberger

external/proto/message.pb.c

@@ -1,5 +1,5 @@
 /* Automatically generated nanopb constant definitions */
-/* Generated by nanopb-0.4.9 */
+/* Generated by nanopb-0.4.9.1 */
 
 #include "./message.pb.h"
 #if PB_PROTO_HEADER_VERSION != 40

external/proto/message.pb.h

@@ -1,5 +1,5 @@
 /* Automatically generated nanopb header */
-/* Generated by nanopb-0.4.9 */
+/* Generated by nanopb-0.4.9.1 */
 
 #ifndef PB_EXTERNAL_PROTO_MESSAGE_PB_H_INCLUDED
 #define PB_EXTERNAL_PROTO_MESSAGE_PB_H_INCLUDED
@@ -58,7 +58,7 @@ typedef struct _StartTimeRequest {
 
 /* The response to a StartTimeRequest, telling the transient federate the start time of the federation. */
 typedef struct _StartTimeResponse {
-    int64_t current_logical_time;
+    int64_t elapsed_logical_time;
     int64_t federation_start_time;
 } StartTimeResponse;
 
@@ -218,7 +218,7 @@ extern "C" {
 #define StartupHandshakeResponse_state_tag       1
 #define StartTimeProposal_time_tag               1
 #define StartTimeProposal_step_tag               2
-#define StartTimeResponse_current_logical_time_tag 1
+#define StartTimeResponse_elapsed_logical_time_tag 1
 #define StartTimeResponse_federation_start_time_tag 2
 #define JoiningTimeAnnouncement_joining_time_tag 1
 #define StartupCoordination_startup_handshake_request_tag 1
@@ -282,7 +282,7 @@ X(a, STATIC,   REQUIRED, UINT32,   step,              2)
 #define StartTimeRequest_DEFAULT NULL
 
 #define StartTimeResponse_FIELDLIST(X, a) \
-X(a, STATIC,   REQUIRED, INT64,    current_logical_time,   1) \
+X(a, STATIC,   REQUIRED, INT64,    elapsed_logical_time,   1) \
 X(a, STATIC,   REQUIRED, INT64,    federation_start_time,   2)
 #define StartTimeResponse_CALLBACK NULL
 #define StartTimeResponse_DEFAULT NULL

external/proto/message.proto

@@ -42,7 +42,7 @@ message StartTimeRequest{ }
 
 // The response to a StartTimeRequest, telling the transient federate the start time of the federation.
 message StartTimeResponse {
-  required int64 current_logical_time = 1;
+  required int64 elapsed_logical_time = 1;
   required int64 federation_start_time = 2;
 }
 

include/reactor-uc/startup_coordinator.h

@@ -15,6 +15,7 @@ enum JoiningPolicy { JOIN_IMMEDIATELY = 0, JOIN_ALIGNED_WITH_SHORT_TIMER = 1, JO
 
 /** Represents the state of a neighbor. */
 typedef struct {
+  bool core_federate;                   // Whether this federate can be not available
   bool handshake_response_received;     // Whether a handshake response has been received from this neighbor.
   bool handshake_request_received;      // Whether a handshake response has been sent to this neighbor.
   bool handshake_response_sent;         // Whether a handshake response has been sent to this neighbor.

src/schedulers/dynamic/scheduler.c

@@ -100,11 +100,12 @@ void Scheduler_register_for_cleanup(Scheduler *untyped_self, Trigger *trigger) {
 void Scheduler_prepare_timestep(Scheduler *untyped_self, tag_t tag) {
   DynamicScheduler *self = (DynamicScheduler *)untyped_self;
 
-  LF_DEBUG(SCHED, "Preparing timestep for tag " PRINTF_TAG, tag);
   // Before setting `current_tag` we must lock because it is read from async and channel context.
   MUTEX_LOCK(self->mutex);
   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);
 }
 
@@ -355,6 +356,9 @@ 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);

src/startup_coordinator.c

@@ -59,7 +59,9 @@ static lf_ret_t StartupCoordinator_connect_to_neighbors_blocking(StartupCoordina
 
 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);
   }
@@ -385,6 +387,10 @@ static void StartupCoordinator_handle_start_time_request(StartupCoordinator *sel
       do {
         ret = chan->send_blocking(chan, &self->msg);
       } while (ret != LF_OK);
+
+      // We now schedule a system event here, because otherwise we will never detect no other federates responding
+      StartupCoordinator_schedule_system_self_event(self, self->env->get_physical_time(self->env) + MSEC(250),
+                                                StartupCoordination_start_time_response_tag);
     }
 
   } else {
@@ -394,13 +400,13 @@ static void StartupCoordinator_handle_start_time_request(StartupCoordinator *sel
       NetworkChannel *chan = env->net_bundles[payload->neighbor_index]->net_channel;
       msg->which_message = FederateMessage_startup_coordination_tag;
       msg->message.startup_coordination.which_message = StartupCoordination_start_time_response_tag;
-      msg->message.startup_coordination.message.start_time_response.current_logical_time =
+      msg->message.startup_coordination.message.start_time_response.elapsed_logical_time =
           self->env->get_elapsed_logical_time(self->env);
       msg->message.startup_coordination.message.start_time_response.federation_start_time = self->start_time_proposal;
       chan->send_blocking(chan, msg);
-        LF_INFO(FED, "SENDING TIME start_tag: " PRINTF_TIME " current_time: " PRINTF_TIME,
+        LF_INFO(FED, "SENDING TIME start_tag: " PRINTF_TIME " elapsed_time: " PRINTF_TIME,
           msg->message.startup_coordination.message.start_time_response.federation_start_time,
-          msg->message.startup_coordination.message.start_time_response.current_logical_time);
+          msg->message.startup_coordination.message.start_time_response.elapsed_logical_time);
       break;
     }
     default:;
@@ -411,43 +417,71 @@ static void StartupCoordinator_handle_start_time_request(StartupCoordinator *sel
 
 
 static void StartupCoordinator_handle_start_time_response(StartupCoordinator *self, StartupEvent *payload) {
+  if (self->start_time_proposal > 0) {
+    return;
+  }
+
   if (payload->neighbor_index == NEIGHBOR_INDEX_SELF) {
-    // SHOULD only come from other federates
-  } else {
-    self->neighbor_state[payload->neighbor_index].start_time_proposals_received++;
+    bool got_no_response = true;
+    for (size_t i = 0; i < self->num_neighbours; i++) {
+      got_no_response = got_no_response && (self->neighbor_state[i].current_logical_time == 0);
+    }
 
-    if (self->start_time_proposal > 0) {
+    if (got_no_response) {
+      LF_ERR(FED, "No other federate responded to the start time request! Shutting Down!");
+      self->env->request_shutdown(self->env);
+    }
+  }
+
+  const instant_t start_time = payload->msg.message.start_time_response.federation_start_time;
+  const interval_t elapsed_logical = payload->msg.message.start_time_response.elapsed_logical_time;
+  const instant_t current_logical_time = start_time + elapsed_logical;
+  validate(start_time < current_logical_time);
+
+  self->neighbor_state[payload->neighbor_index].current_logical_time = current_logical_time;
+  self->neighbor_state[payload->neighbor_index].start_time_proposals_received++;
+
+  // checking if we got a response from all core federates
+  for (size_t i = 0; i < self->num_neighbours; i++) {
+    if (self->neighbor_state[i].core_federate && self->neighbor_state[i].current_logical_time == 0) {
       return;
     }
+  }
 
-    const instant_t start_time = payload->msg.message.start_time_response.federation_start_time;
-    const instant_t current_logical = payload->msg.message.start_time_response.current_logical_time;
-    self->start_time_proposal = start_time;
-    self->state = StartupCoordinationState_RUNNING;
-
-    instant_t joining_time = 0;
-
-    if (self->joining_policy == JOIN_IMMEDIATELY) {
-        joining_time = current_logical + MSEC(50);
-        tag_t start_tag = {.time = joining_time, .microstep = 0};
-        self->env->scheduler->prepare_timestep(self->env->scheduler, start_tag);
-        self->env->scheduler->set_and_schedule_start_tag(self->env->scheduler, joining_time);
-        LF_INFO(FED, "Policy: IMMEDIATELY Scheduling start_tag: " PRINTF_TIME " join_time: " PRINTF_TIME, start_time, joining_time);
-        StartupCoordinator_schedule_startups(self, start_tag);
-        StartupCoordinator_schedule_timers(self, self->env->main, start_tag);
-    } else if (self->joining_policy == JOIN_ALIGNED_WITH_SHORT_TIMER) {
-        joining_time = current_logical + MSEC(50);
-        tag_t start_tag = {.time = joining_time, .microstep = 0};
-        self->env->scheduler->prepare_timestep(self->env->scheduler, start_tag);
-        start_tag.time = joining_time + 1;
-        self->env->scheduler->set_and_schedule_start_tag(self->env->scheduler, joining_time + 1);
-        LF_INFO(FED, "Policy: Timer Aligned Scheduling start_tag: " PRINTF_TIME "join_time: " PRINTF_TIME, start_time, joining_time);
-        StartupCoordinator_schedule_startups(self, start_tag);
-        StartupCoordinator_schedule_timers_joining(self, self->env->main, start_time, joining_time + 1);
-    } else {
-      validate(false);
+  // calculating the maximum logical time from all neighbors that responded
+  interval_t max_logical_time = 0;
+  for (size_t i = 0; i < self->num_neighbours; i++) {
+    if (self->neighbor_state[i].current_logical_time > max_logical_time) {
+      max_logical_time = self->neighbor_state[i].current_logical_time;
     }
   }
+
+  self->start_time_proposal = start_time;
+  self->state = StartupCoordinationState_RUNNING;
+
+  instant_t joining_time = 0;
+
+  if (self->joining_policy == JOIN_IMMEDIATELY) {
+      joining_time = max_logical_time + MSEC(50);
+      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);
+      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);
+      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);
+  } else {
+    validate(false);
+  }
 }
 
 /** Invoked by scheduler when handling any system event destined for StartupCoordinator. */