Adding support of transient federates

core/federated/RTI/main.c

@@ -121,6 +121,8 @@ void usage(int argc, const char* argv[]) {
   lf_print("   The ID of the federation that this RTI will control.\n");
   lf_print("  -n, --number_of_federates <n>");
   lf_print("   The number of federates in the federation that this RTI will control.\n");
+  lf_print("  -nt, --number_of_transient_federates <n>");
+  lf_print("   The number of transient federates in the federation that this RTI will control.\n");
   lf_print("  -p, --port <n>");
   lf_print("   The port number to use for the RTI. Must be larger than 0 and smaller than %d. Default is %d.\n",
            UINT16_MAX, DEFAULT_PORT);
@@ -233,6 +235,21 @@ int process_args(int argc, const char* argv[]) {
       }
       rti.base.number_of_scheduling_nodes = (int32_t)num_federates; // FIXME: Loses numbers on 64-bit machines
       lf_print("RTI: Number of federates: %d", rti.base.number_of_scheduling_nodes);
+    } else if (strcmp(argv[i], "-nt") == 0 || strcmp(argv[i], "--number_of_transient_federates") == 0) {
+      if (argc < i + 2) {
+        lf_print_error("--number_of_transient_federates needs a valid positive argument.");
+        usage(argc, argv);
+        return 0;
+      }
+      i++;
+      long num_transient_federates = strtol(argv[i], NULL, 10);
+      if (num_transient_federates == LONG_MAX || num_transient_federates == LONG_MIN) {
+        lf_print_error("--number_of_transient_federates needs a valid positive or null integer argument.");
+        usage(argc, argv);
+        return 0;
+      }
+      rti.number_of_transient_federates = (int32_t)num_transient_federates; // FIXME: Loses numbers on 64-bit machines
+      lf_print("RTI: Number of transient federates: %d", rti.number_of_transient_federates);
     } else if (strcmp(argv[i], "-p") == 0 || strcmp(argv[i], "--port") == 0) {
       if (argc < i + 2) {
         lf_print_error("--port needs a short unsigned integer argument ( > 0 and < %d).", UINT16_MAX);
@@ -275,6 +292,16 @@ int process_args(int argc, const char* argv[]) {
       return 0;
     }
   }
+  if (rti.base.number_of_scheduling_nodes == 0) {
+    lf_print_error("--number_of_federates needs a valid positive integer argument.");
+    usage(argc, argv);
+    return 0;
+  }
+  if (rti.number_of_transient_federates > rti.base.number_of_scheduling_nodes) {
+    lf_print_error("--number_of_transient_federates cannot be higher than the number of federates.");
+    usage(argc, argv);
+    return 0;
+  }
   return 1;
 }
 int main(int argc, const char* argv[]) {
@@ -314,8 +341,8 @@ int main(int argc, const char* argv[]) {
     lf_print("Tracing the RTI execution in %s file.", rti_trace_file_name);
   }
 
-  lf_print("Starting RTI for %d federates in federation ID %s.", rti.base.number_of_scheduling_nodes,
-           rti.federation_id);
+  lf_print("Starting RTI for a total of %d federates, with %d being transient, in federation ID %s",
+           rti.base.number_of_scheduling_nodes, rti.number_of_transient_federates, rti.federation_id);
   assert(rti.base.number_of_scheduling_nodes < UINT16_MAX);
 
   // Allocate memory for the federates

core/federated/RTI/rti_common.c

@@ -42,6 +42,7 @@ void invalidate_min_delays() {
       node->flags = 0; // All flags cleared because they get set lazily.
     }
     free(rti_common->min_delays);
+    rti_common->min_delays = NULL;
   }
 }
 
@@ -101,6 +102,8 @@ tag_t earliest_future_incoming_message_tag(scheduling_node_t* e) {
     if (lf_tag_compare(rti_common->min_delays[i * n + e->id], FOREVER_TAG) != 0) {
       // Node i is upstream of e with min delay rti_common->min_delays[i * n + e->id]
       scheduling_node_t* upstream = rti_common->scheduling_nodes[i];
+      if (upstream->state == NOT_CONNECTED)
+        continue;
       // If we haven't heard from the upstream node, then assume it can send an event at the start time.
       if (lf_tag_compare(upstream->next_event, NEVER_TAG) == 0) {
         tag_t start_tag = {.time = start_time, .microstep = 0};
@@ -163,6 +166,9 @@ tag_t eimt_strict(scheduling_node_t* e) {
 tag_advance_grant_t tag_advance_grant_if_safe(scheduling_node_t* e) {
   tag_advance_grant_t result = {.tag = NEVER_TAG, .is_provisional = false};
 
+  // Check how many upstream federates are connected
+  int num_connected_upstream = 0;
+
   // Find the earliest LTC of upstream scheduling_nodes (M).
   tag_t min_upstream_completed = FOREVER_TAG;
 
@@ -172,6 +178,7 @@ tag_advance_grant_t tag_advance_grant_if_safe(scheduling_node_t* e) {
     // Ignore this enclave/federate if it is not connected.
     if (upstream->state == NOT_CONNECTED)
       continue;
+    num_connected_upstream++;
 
     // Adjust by the "after" delay.
     // Note that "no delay" is encoded as NEVER,
@@ -184,8 +191,15 @@ tag_advance_grant_t tag_advance_grant_if_safe(scheduling_node_t* e) {
   }
   LF_PRINT_LOG("RTI: Minimum upstream LTC for federate/enclave %d is " PRINTF_TAG "(adjusted by after delay).", e->id,
                min_upstream_completed.time - start_time, min_upstream_completed.microstep);
-  if (lf_tag_compare(min_upstream_completed, e->last_granted) > 0 &&
-      lf_tag_compare(min_upstream_completed, e->next_event) >= 0 // The enclave has to advance its tag
+
+  if (num_connected_upstream == 0) {
+    // When none of the upstream federates is connected (case of transients),
+    if (lf_tag_compare(e->next_event, FOREVER_TAG) != 0) {
+      result.tag = e->next_event;
+      return result;
+    }
+  } else if (lf_tag_compare(min_upstream_completed, e->last_granted) > 0 &&
+             lf_tag_compare(min_upstream_completed, e->next_event) >= 0 // The enclave has to advance its tag
   ) {
     result.tag = min_upstream_completed;
     return result;