Transient federates

core/CMakeLists.txt

@@ -122,4 +122,5 @@ define(USER_THREADS)
 define(SCHEDULER)
 define(LF_SOURCE_DIRECTORY)
 define(LF_PACKAGE_DIRECTORY)
+define(LF_FEDERATES_BIN_DIRECTORY)
 define(LF_FILE_SEPARATOR)

core/federated/RTI/enclave.c

@@ -63,6 +63,9 @@ void logical_tag_complete(enclave_t* enclave, tag_t completed) {
 tag_advance_grant_t tag_advance_grant_if_safe(enclave_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 enclaves (M).
     tag_t min_upstream_completed = FOREVER_TAG;
 
@@ -72,6 +75,8 @@ tag_advance_grant_t tag_advance_grant_if_safe(enclave_t* e) {
         // Ignore this enclave if it no longer connected.
         if (upstream->state == NOT_CONNECTED) continue;
 
+        num_connected_upstream++;
+
         // Adjust by the "after" delay.
         // Note that "no delay" is encoded as NEVER,
         // whereas one microstep delay is encoded as 0LL.
@@ -81,16 +86,25 @@ tag_advance_grant_t tag_advance_grant_if_safe(enclave_t* e) {
             min_upstream_completed = candidate;
         }
     }
+
     LF_PRINT_LOG("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
+
+    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;
-    }
+    } 
 
     // Can't make progress based only on upstream LTCs.
     // If all (transitive) upstream enclaves of the enclave
@@ -257,6 +271,7 @@ tag_advance_grant_t next_event_tag(enclave_t* e, tag_t next_event_tag) {
 
 void notify_advance_grant_if_safe(enclave_t* e) {
     tag_advance_grant_t grant = tag_advance_grant_if_safe(e);
+
     if (lf_tag_compare(grant.tag, NEVER_TAG) != 0) {
         if (grant.is_provisional) {
             notify_provisional_tag_advance_grant(e, grant.tag);

core/federated/RTI/rti.c

@@ -113,10 +113,14 @@ 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 a total of %d federates, with %d being transient, in federation ID %s", \
+        _f_rti->number_of_enclaves, 
+        _f_rti->number_of_transient_federates, 
+        _f_rti->federation_id);
 
-    lf_print("Starting RTI for %d federates in federation ID %s.",  _f_rti->number_of_enclaves, _f_rti->federation_id);
     assert(_f_rti->number_of_enclaves < UINT16_MAX);
-
+    assert(_f_rti->number_of_transient_federates < UINT16_MAX);
+
     // Allocate memory for the federates
     _f_rti->enclaves = (federate_t**)calloc(_f_rti->number_of_enclaves, sizeof(federate_t*));
     for (uint16_t i = 0; i < _f_rti->number_of_enclaves; i++) {
@@ -129,6 +133,5 @@ int main(int argc, const char* argv[]) {
 
     int socket_descriptor = start_rti_server(_f_rti->user_specified_port);
     wait_for_federates(socket_descriptor);
-    lf_print("RTI is exiting.");
     return 0;
 }