Merge pull request #2032 from lf-lang/decentralized-small-delay-bugfix

Fix bug in C decentralized execution with small after delays

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

@@ -549,6 +549,7 @@ protected String makePreamble(
     int numOfNetworkActions = federate.networkMessageActions.size();
     code.pr(
         """
+        interval_t _lf_action_delay_table[%1$s];
         lf_action_base_t* _lf_action_table[%1$s];
         size_t _lf_action_table_size = %1$s;
         lf_action_base_t* _lf_zero_delay_action_table[%2$s];
@@ -572,7 +573,7 @@ protected String makePreamble(
         """
             .formatted(numOfPortAbsentReactions));
 
-    int numOfSTAAOffsets = federate.stpOffsets.size();
+    int numOfSTAAOffsets = federate.staaOffsets.size();
     code.pr(
         CExtensionUtils.surroundWithIfFederatedDecentralized(
             """

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

@@ -61,6 +61,13 @@ public static String initializeTriggersForNetworkActions(
         var reactor = main.lookupReactorInstance(federate.networkReceiverInstantiations.get(i));
         var actionInstance = reactor.lookupActionInstance(action);
         var trigger = CUtil.actionRef(actionInstance, null);
+        var delay = federate.networkMessageActionDelays.get(i);
+        code.pr(
+            "_lf_action_delay_table["
+                + actionTableCount
+                + "] = "
+                + getNetworkDelayLiteral(delay)
+                + "; \\");
         code.pr(
             "_lf_action_table["
                 + actionTableCount++
@@ -91,24 +98,24 @@ public static String initializeTriggersForNetworkActions(
    */
   public static String stpStructs(FederateInstance federate) {
     CodeBuilder code = new CodeBuilder();
-    federate.stpOffsets.sort((d1, d2) -> (int) (d1.time - d2.time));
-    if (!federate.stpOffsets.isEmpty()) {
+    federate.staaOffsets.sort((d1, d2) -> (int) (d1.time - d2.time));
+    if (!federate.staaOffsets.isEmpty()) {
       // Create a static array of trigger_t pointers.
       // networkMessageActions is a list of Actions, but we
       // need a list of trigger struct names for ActionInstances.
       // There should be exactly one ActionInstance in the
       // main reactor for each Action.
-      for (int i = 0; i < federate.stpOffsets.size(); ++i) {
+      for (int i = 0; i < federate.staaOffsets.size(); ++i) {
         // Find the corresponding ActionInstance.
         List<Action> networkActions =
-            federate.stpToNetworkActionMap.get(federate.stpOffsets.get(i));
+            federate.stpToNetworkActionMap.get(federate.staaOffsets.get(i));
 
         code.pr("staa_lst[" + i + "] = (staa_t*) malloc(sizeof(staa_t));");
         code.pr(
             "staa_lst["
                 + i
                 + "]->STAA = "
-                + CTypes.getInstance().getTargetTimeExpr(federate.stpOffsets.get(i))
+                + CTypes.getInstance().getTargetTimeExpr(federate.staaOffsets.get(i))
                 + ";");
         code.pr("staa_lst[" + i + "]->numActions = " + networkActions.size() + ";");
         code.pr(

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

@@ -273,18 +273,19 @@ private static void addNetworkReceiverReactor(
 
     // Keep track of this action in the destination federate.
     connection.dstFederate.networkMessageActions.add(networkAction);
+    connection.dstFederate.networkMessageActionDelays.add(connection.getDefinition().getDelay());
     if (connection.getDefinition().getDelay() == null)
       connection.dstFederate.zeroDelayNetworkMessageActions.add(networkAction);
 
     TimeValue maxSTP = findMaxSTP(connection, coordination);
 
     if (!connection.dstFederate.currentSTPOffsets.contains(maxSTP.time)) {
       connection.dstFederate.currentSTPOffsets.add(maxSTP.time);
-      connection.dstFederate.stpOffsets.add(maxSTP);
+      connection.dstFederate.staaOffsets.add(maxSTP);
       connection.dstFederate.stpToNetworkActionMap.put(maxSTP, new ArrayList<>());
     } else {
       // TODO: Find more efficient way to reuse timevalues
-      for (var offset : connection.dstFederate.stpOffsets) {
+      for (var offset : connection.dstFederate.staaOffsets) {
         if (maxSTP.time == offset.time) {
           maxSTP = offset;
           break;

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

@@ -170,6 +170,13 @@ public Instantiation getInstantiation() {
    */
   public List<Action> networkMessageActions = new ArrayList<>();
 
+  /**
+   * List of after delay values of the corresponding entries of {@code networkMessageActions}. These
+   * will be {@code null} in the case of zero-delay connections and {@code 0} in the case of
+   * microstep-delay connections.
+   */
+  public List<Expression> networkMessageActionDelays = new ArrayList<>();
+
   /**
    * List of networkMessage actions corresponding to zero-delay connections. This should be a subset
    * of the networkMessageActions.
@@ -258,7 +265,7 @@ public Instantiation getInstantiation() {
   private Set<Reaction> excludeReactions = null;
 
   /** Keep a unique list of enabled serializers */
-  public List<TimeValue> stpOffsets = new ArrayList<>();
+  public List<TimeValue> staaOffsets = new ArrayList<>();
 
   /** The STP offsets that have been recorded in {@code stpOffsets thus far. */
   public Set<Long> currentSTPOffsets = new HashSet<>();

test/C/src/federated/SmallDelayDecentralized.lf

@@ -0,0 +1,62 @@
+target C {
+  timeout: 1 sec,
+  coordination: decentralized
+}
+
+preamble {=
+  #include "platform.h"
+=}
+
+reactor Count {
+  state count: int = 1
+  output out: int
+  logical action loop
+
+  reaction(startup) -> loop {=
+    lf_schedule(loop, 0);
+  =}
+
+  reaction(loop) -> out {=
+    if (self->count < 6) {
+      lf_sleep(MSEC(50));
+      lf_set(out, self->count++);
+      lf_schedule(loop, 0);
+    }
+  =}
+}
+
+reactor Print {
+  input in: int
+  state c: int = 1
+  state checks: int = 0
+
+  logical action loop
+
+  reaction(startup) -> loop {=
+    lf_schedule(loop, 0);
+  =}
+
+  reaction(in) {=
+    interval_t elapsed_time = lf_time_logical_elapsed();
+    lf_print("At time %lld, received %d", elapsed_time, in->value);
+    if (in->value != self->c) {
+      lf_print_error_and_exit("Expected to receive %d.", self->c);
+    }
+    self->c++;
+  =} STP(1 sec) {=
+    lf_print_error_and_exit("STP violation. This should not happen because the STP offset is large.");
+  =}
+
+  reaction(loop) {=
+    lf_print("checking self.checks, which is now %d...", self->checks);
+    if (self->checks++ <= 3) {
+      lf_schedule(loop, 0);
+    }
+  =}
+}
+
+federated reactor {
+  c = new Count()
+  p = new Print()
+  c.out -> p.in after 0
+}