Merge pull request #2368 from Depetrol/startup-immediately

Immediate start of federates with STP offset under decentralized coordination & fix target code STP_offset

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

@@ -55,6 +55,7 @@
 import org.lflang.lf.Port;
 import org.lflang.lf.Reactor;
 import org.lflang.lf.VarRef;
+import org.lflang.lf.impl.CodeExprImpl;
 import org.lflang.target.Target;
 import org.lflang.target.property.ClockSyncOptionsProperty;
 import org.lflang.target.property.CoordinationOptionsProperty;
@@ -644,7 +645,7 @@ private String generateExecutablePreamble(
 
     code.pr(generateCodeForPhysicalActions(federate, messageReporter));
 
-    code.pr(generateCodeToInitializeFederate(federate, rtiConfig));
+    code.pr(generateCodeToInitializeFederate(federate, rtiConfig, messageReporter));
     return """
            void _lf_executable_preamble(environment_t* env) {
            %s
@@ -673,7 +674,8 @@ void staa_initialization() {
    * @param rtiConfig Information about the RTI's deployment.
    * @return The generated code
    */
-  private String generateCodeToInitializeFederate(FederateInstance federate, RtiConfig rtiConfig) {
+  private String generateCodeToInitializeFederate(
+      FederateInstance federate, RtiConfig rtiConfig, MessageReporter messageReporter) {
     CodeBuilder code = new CodeBuilder();
     code.pr("// ***** Start initializing the federated execution. */");
     code.pr(
@@ -701,7 +703,12 @@ private String generateCodeToInitializeFederate(FederateInstance federate, RtiCo
       if (stpParam.isPresent()) {
         var globalSTP = ASTUtils.initialValue(stpParam.get(), List.of(federate.instantiation));
         var globalSTPTV = ASTUtils.getLiteralTimeValue(globalSTP);
-        code.pr("lf_set_stp_offset(" + CTypes.getInstance().getTargetTimeExpr(globalSTPTV) + ");");
+        if (globalSTPTV != null)
+          code.pr(
+              "lf_set_stp_offset(" + CTypes.getInstance().getTargetTimeExpr(globalSTPTV) + ");");
+        else if (globalSTP instanceof CodeExprImpl)
+          code.pr("lf_set_stp_offset(" + ((CodeExprImpl) globalSTP).getCode().getBody() + ");");
+        else messageReporter.at(stpParam.get().eContainer()).error("Invalid STP offset");
       }
     }
 

test/Python/src/federated/Dataflow.lf

@@ -0,0 +1,68 @@
+target Python {
+  coordination: decentralized  # logging: debug
+}
+
+preamble {=
+  import time
+=}
+
+reactor Client(STP_offset = {= FOREVER =}) {
+  input server_message
+  output client_message
+
+  reaction(startup) {=
+    print("Client Startup!")
+  =}
+
+  reaction(server_message) -> client_message {=
+    val = server_message.value
+    time.sleep(0.1)
+    val += 1
+    print("client:", val)
+    if val==49:
+        print("client done")
+        request_stop()
+    if val<49:
+        client_message.set(val)
+  =} STP(10 s) {=
+    print("Client STP Violated!")
+    exit(1)
+  =}
+}
+
+reactor Server(STP_offset = {= FOREVER =}) {
+  output server_message
+  input client_message1
+  input client_message2
+
+  reaction(startup) -> server_message {=
+    print("Server Startup!")
+    server_message.set(0)
+  =}
+
+  reaction(client_message1, client_message2) -> server_message {=
+    val = max(client_message1.value, client_message2.value)
+    time.sleep(0.1)
+    val += 1
+    print("server:", val)
+    if val==48:
+        print("server done")
+        server_message.set(val)
+        request_stop()
+    if val<48:
+        server_message.set(val)
+  =} STP(10 s) {=
+    print("Server STP Violated!")
+    exit(1)
+  =}
+}
+
+federated reactor(STP_offset = {= FOREVER =}) {
+  client1 = new Client()
+  client2 = new Client()
+  server = new Server()
+  server.server_message -> client1.server_message
+  client1.client_message -> server.client_message1 after 0
+  server.server_message -> client2.server_message
+  client2.client_message -> server.client_message2 after 0
+}