Adding support of transient federates

core/CMakeLists.txt

@@ -181,6 +181,7 @@ define(SCHEDULER)
 define(LF_SOURCE_DIRECTORY)
 define(LF_SOURCE_GEN_DIRECTORY)
 define(LF_PACKAGE_DIRECTORY)
+define(LF_FEDERATES_BIN_DIRECTORY)
 define(LF_FILE_SEPARATOR)
 define(WORKERS_NEEDED_FOR_FEDERATE)
 define(LF_ENCLAVES)

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 federates that are transient; this must be strictly less than the number of federates.\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);
@@ -232,6 +234,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 < 0L || 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);
@@ -272,8 +289,19 @@ 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 or equal to the number of federates.");
+    usage(argc, argv);
+    return 0;
+  }
   return 1;
 }
+
 int main(int argc, const char* argv[]) {
   initialize_lf_thread_id();
   initialize_RTI(&rti);
@@ -311,8 +339,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

@@ -155,6 +155,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;
 
@@ -164,6 +167,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,
@@ -176,8 +180,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;