Add support of transient federates under decentralized coordination and physical connections

core/src/main/java/org/lflang/federated/extensions/CExtension.java

@@ -489,11 +489,12 @@ public String getNetworkBufferType() {
   /** Put the C preamble in a `include/_federate.name + _preamble.h` file. */
   protected final void writePreambleFile(
       FederateInstance federate,
+      List<FederateInstance> allFederates,
       FederationFileConfig fileConfig,
       RtiConfig rtiConfig,
       MessageReporter messageReporter)
       throws IOException {
-    String cPreamble = makePreamble(federate, rtiConfig, messageReporter);
+    String cPreamble = makePreamble(federate, allFederates, rtiConfig, messageReporter);
     String relPath = getPreamblePath(federate);
     Path fedPreamblePath = fileConfig.getSrcPath().resolve(relPath);
     Files.createDirectories(fedPreamblePath.getParent());
@@ -509,11 +510,12 @@ protected final void writePreambleFile(
   @Override
   public String generatePreamble(
       FederateInstance federate,
+      List<FederateInstance> allFederates,
       FederationFileConfig fileConfig,
       RtiConfig rtiConfig,
       MessageReporter messageReporter)
       throws IOException {
-    writePreambleFile(federate, fileConfig, rtiConfig, messageReporter);
+    writePreambleFile(federate, allFederates, fileConfig, rtiConfig, messageReporter);
     var includes = new CodeBuilder();
     includes.pr(
         """
@@ -539,7 +541,10 @@ public String generatePreamble(
 
   /** Generate the preamble to setup federated execution in C. */
   protected String makePreamble(
-      FederateInstance federate, RtiConfig rtiConfig, MessageReporter messageReporter) {
+      FederateInstance federate,
+      List<FederateInstance> allFederates,
+      RtiConfig rtiConfig,
+      MessageReporter messageReporter) {
 
     var code = new CodeBuilder();
 
@@ -609,7 +614,7 @@ protected String makePreamble(
             """
                 .formatted(numOfSTAAOffsets)));
 
-    code.pr(generateExecutablePreamble(federate, rtiConfig, messageReporter));
+    code.pr(generateExecutablePreamble(federate, allFederates, rtiConfig, messageReporter));
 
     code.pr(generateSTAAInitialization(federate));
 
@@ -663,12 +668,15 @@ private String generateInitializeTriggers(
 
   /** Generate code for an executed preamble. */
   private String generateExecutablePreamble(
-      FederateInstance federate, RtiConfig rtiConfig, MessageReporter messageReporter) {
+      FederateInstance federate,
+      List<FederateInstance> allFederates,
+      RtiConfig rtiConfig,
+      MessageReporter messageReporter) {
     CodeBuilder code = new CodeBuilder();
 
     code.pr(generateCodeForPhysicalActions(federate, messageReporter));
 
-    code.pr(generateCodeToInitializeFederate(federate, rtiConfig, messageReporter));
+    code.pr(generateCodeToInitializeFederate(federate, allFederates, rtiConfig, messageReporter));
     return """
            void _lf_executable_preamble(environment_t* env) {
            %s
@@ -698,7 +706,10 @@ void staa_initialization() {
    * @return The generated code
    */
   private String generateCodeToInitializeFederate(
-      FederateInstance federate, RtiConfig rtiConfig, MessageReporter messageReporter) {
+      FederateInstance federate,
+      List<FederateInstance> allFederates,
+      RtiConfig rtiConfig,
+      MessageReporter messageReporter) {
     CodeBuilder code = new CodeBuilder();
     code.pr("// ***** Start initializing the federated execution. */");
     code.pr(
@@ -758,18 +769,30 @@ else if (globalSTP instanceof CodeExprImpl)
     // thread without requiring a mutex lock.
     var numberOfInboundConnections = federate.inboundP2PConnections.size();
     var numberOfOutboundConnections = federate.outboundP2PConnections.size();
+    var numberOfInboundConnectionsToTransients =
+        (int) federate.inboundP2PConnections.stream().filter(f -> f.isTransient).count();
+    var numberOfOutboundConnectionsToTransients =
+        (int) federate.outboundP2PConnections.stream().filter(f -> f.isTransient).count();
 
     code.pr(
         String.join(
             "\n",
             "_fed.number_of_inbound_p2p_connections = " + numberOfInboundConnections + ";",
-            "_fed.number_of_outbound_p2p_connections = " + numberOfOutboundConnections + ";"));
+            "_fed.number_of_outbound_p2p_connections = " + numberOfOutboundConnections + ";",
+            "_fed.number_of_inbound_p2p_transients = "
+                + numberOfInboundConnectionsToTransients
+                + ";",
+            "_fed.number_of_outbound_p2p_transients = "
+                + numberOfOutboundConnectionsToTransients
+                + ";"));
+
     code.pr(
         String.join(
             "\n",
             "// Initialize the array of network abstractions for incoming connections to -1.",
             "for (int i = 0; i < NUMBER_OF_FEDERATES; i++) {",
             "    _fed.net_for_inbound_p2p_connections[i] = NULL;",
+            "    _fed.inbound_p2p_connection_is_transient[i] = false;",
             "}"));
     code.pr(
         String.join(
@@ -778,6 +801,7 @@ else if (globalSTP instanceof CodeExprImpl)
             "for (int i = 0; i < NUMBER_OF_FEDERATES; i++) {",
             "    _fed.net_for_outbound_p2p_connections[i] = NULL;",
             "}"));
+
     var clockSyncOptions = federate.targetConfig.getOrDefault(ClockSyncOptionsProperty.INSTANCE);
     // If a test clock offset has been specified, insert code to set it here.
     if (clockSyncOptions.testOffset != null) {
@@ -819,13 +843,21 @@ else if (globalSTP instanceof CodeExprImpl)
               "// This is done in a separate thread because this thread will call",
               "// lf_connect_to_federate for each outbound connection at the same",
               "// time that the new thread is listening for such connections for inbound",
-              "// connections. The thread will live until all connections have been established.",
+              "// connections. The thread will live until all connections have been established,",
+              "// or if outbound connections are transient, until the end of the execution.",
               "lf_thread_create(&_fed.inbound_p2p_handling_thread_id,"
                   + " lf_handle_p2p_connections_from_federates, env);"));
     }
 
     for (FederateInstance remoteFederate : federate.outboundP2PConnections) {
-      code.pr("lf_connect_to_federate(" + remoteFederate.id + ");");
+      code.pr(
+          "lf_connect_to_federate(" + remoteFederate.id + ", " + remoteFederate.isTransient + ");");
+      code.pr(
+          "_fed.outbound_p2p_connection_is_transient["
+              + remoteFederate.id
+              + "] = "
+              + remoteFederate.isTransient
+              + ";");
     }
     return code.getCode();
   }

core/src/main/java/org/lflang/federated/extensions/FedTargetExtension.java

@@ -115,12 +115,14 @@ default void annotateReaction(Reaction reaction) {}
    * Add preamble to the source to set up federated execution.
    *
    * @param federate The federate to which the generated setup code will correspond.
+   * @param allFederates The list of all federates in the federation.
    * @param fileConfig The federation file configuration.
    * @param rtiConfig The settings of the RTI.
    * @param messageReporter Used to report errors and warnings.
    */
   String generatePreamble(
       FederateInstance federate,
+      List<FederateInstance> allFederates,
       FederationFileConfig fileConfig,
       RtiConfig rtiConfig,
       MessageReporter messageReporter)

core/src/main/java/org/lflang/federated/extensions/PythonExtension.java

@@ -1,6 +1,7 @@
 package org.lflang.federated.extensions;
 
 import java.io.IOException;
+import java.util.List;
 import org.lflang.InferredType;
 import org.lflang.MessageReporter;
 import org.lflang.ast.ASTUtils;
@@ -261,11 +262,12 @@ public void annotateReaction(Reaction reaction) {
   @Override
   public String generatePreamble(
       FederateInstance federate,
+      List<FederateInstance> allFederates,
       FederationFileConfig fileConfig,
       RtiConfig rtiConfig,
       MessageReporter messageReporter)
       throws IOException {
-    writePreambleFile(federate, fileConfig, rtiConfig, messageReporter);
+    writePreambleFile(federate, allFederates, fileConfig, rtiConfig, messageReporter);
     return "";
   }
 }

core/src/main/java/org/lflang/federated/extensions/TSExtension.java

@@ -162,6 +162,7 @@ public String getNetworkBufferType() {
   @Override
   public String generatePreamble(
       FederateInstance federate,
+      List<FederateInstance> allFederates,
       FederationFileConfig fileConfig,
       RtiConfig rtiConfig,
       MessageReporter messageReporter) {

core/src/main/java/org/lflang/federated/generator/FedEmitter.java

@@ -37,7 +37,10 @@ public FedEmitter(
 
   /** Generate a .lf file for federate `federate`. */
   Map<Path, CodeMap> generateFederate(
-      LFGeneratorContext context, FederateInstance federate, List<String> federateNames)
+      LFGeneratorContext context,
+      FederateInstance federate,
+      List<FederateInstance> allFederates,
+      List<String> federateNames)
       throws IOException {
     String fedName = federate.name;
     Files.createDirectories(fileConfig.getSrcPath());
@@ -57,7 +60,7 @@ Map<Path, CodeMap> generateFederate(
                     context, federateNames, federate, fileConfig, messageReporter, rtiConfig),
             new FedImportEmitter().generateImports(federate, fileConfig),
             new FedPreambleEmitter()
-                .generatePreamble(federate, fileConfig, rtiConfig, messageReporter),
+                .generatePreamble(federate, allFederates, fileConfig, rtiConfig, messageReporter),
             new FedReactorEmitter().generateReactorDefinitions(federate),
             new FedMainEmitter()
                 .generateMainReactor(federate, originalMainReactor, messageReporter));

core/src/main/java/org/lflang/federated/generator/FedGenerator.java

@@ -181,6 +181,7 @@ public boolean doGenerate(Resource resource, LFGeneratorContext context) throws
           fedEmitter.generateFederate(
               context,
               federate,
+              federates,
               federates.stream().map(fed -> fed.name).collect(Collectors.toList())));
     }
 

core/src/main/java/org/lflang/federated/generator/FedPreambleEmitter.java

@@ -3,6 +3,7 @@
 import static org.lflang.ast.ASTUtils.toText;
 
 import java.io.IOException;
+import java.util.List;
 import org.lflang.MessageReporter;
 import org.lflang.ast.ASTUtils;
 import org.lflang.federated.extensions.FedTargetExtensionFactory;
@@ -26,6 +27,7 @@ public FedPreambleEmitter() {}
    */
   String generatePreamble(
       FederateInstance federate,
+      List<FederateInstance> allFederates,
       FederationFileConfig fileConfig,
       RtiConfig rtiConfig,
       MessageReporter messageReporter)
@@ -53,7 +55,8 @@ String generatePreamble(
         =}"""
             .formatted(
                 FedTargetExtensionFactory.getExtension(federate.targetConfig.target)
-                    .generatePreamble(federate, fileConfig, rtiConfig, messageReporter)));
+                    .generatePreamble(
+                        federate, allFederates, fileConfig, rtiConfig, messageReporter)));
 
     return preambleCode.getCode();
   }