Multiple fixes for federated programs with TypeScript target

.github/workflows/py-tests.yml

@@ -33,6 +33,9 @@ jobs:
           fetch-depth: 0
       - name: Prepare build environment
         uses: ./.github/actions/prepare-build-env
+      - name: Install RTI
+        uses: ./.github/actions/install-rti
+        if: ${{ runner.os == 'macOS' || runner.os == 'Linux' }}
       - name: Setup Python
         uses: actions/setup-python@v4
         with:

.github/workflows/serialization-tests.yml

@@ -20,6 +20,9 @@ jobs:
           fetch-depth: 0
       - name: Prepare build environment
         uses: ./.github/actions/prepare-build-env
+      - name: Install RTI
+        uses: ./.github/actions/install-rti
+        if: ${{ runner.os == 'macOS' || runner.os == 'Linux' }}
       - name: Setup ROS2
         uses: ./.github/actions/setup-ros2
       - name: Install Protobuf Ubuntu

.github/workflows/ts-tests.yml

@@ -19,6 +19,7 @@ jobs:
         uses: actions/checkout@v3
         with:
           fetch-depth: 0
+          submodules: true
       - name: Prepare build environment
         uses: ./.github/actions/prepare-build-env
       - name: Setup Node.js environment
@@ -31,9 +32,12 @@ jobs:
         run: |
           brew install coreutils
         if: ${{ runner.os == 'macOS' }}
+      - name: Install RTI
+        uses: ./.github/actions/install-rti
+        if: ${{ runner.os == 'macOS' || runner.os == 'Linux' }}
       - name: Perform TypeScript tests
         run: |
-          ./gradlew targetTest -Ptarget=TypeScript -Druntime="git://github.com/lf-lang/reactor-ts.git#minor-reorg"
+          ./gradlew targetTest -Ptarget=TypeScript -Druntime="git://github.com/lf-lang/reactor-ts.git#master"
       - name: Collect code coverage
         run: ./gradlew jacocoTestReport
       - name: Report to CodeCov

core/src/main/java/org/lflang/TargetProperty.java

@@ -31,6 +31,7 @@
 import java.util.Arrays;
 import java.util.LinkedList;
 import java.util.List;
+import java.util.Objects;
 import java.util.Optional;
 import java.util.function.Predicate;
 import java.util.stream.Collectors;
@@ -283,10 +284,6 @@ public enum TargetProperty {
       (config, value, err) -> {
         config.files = ASTUtils.elementToListOfStrings(value);
       },
-      // FIXME: This merging of lists is potentially dangerous since
-      // the incoming list of files can belong to a .lf file that is
-      // located in a different location, and keeping just filename
-      // strings like this without absolute paths is incorrect.
       (config, value, err) -> {
         config.files.addAll(ASTUtils.elementToListOfStrings(value));
       }),
@@ -1001,15 +998,34 @@ public static TargetDecl extractTargetDecl(Target target, TargetConfig config) {
    *
    * @param config The configuration object to update.
    * @param properties AST node that holds all the target properties.
+   * @param relativePath The path from the main resource to the resource from which the new
+   *     properties originate.
    */
-  public static void update(TargetConfig config, List<KeyValuePair> properties, ErrorReporter err) {
+  public static void update(
+      TargetConfig config, List<KeyValuePair> properties, Path relativePath, ErrorReporter err) {
     properties.forEach(
         property -> {
           TargetProperty p = forName(property.getName());
           if (p != null) {
             // Mark the specified target property as set by the user
             config.setByUser.add(p);
-            p.updater.parseIntoTargetConfig(config, property.getValue(), err);
+            var value = property.getValue();
+            if (property.getName().equals("files")) {
+              var array = LfFactory.eINSTANCE.createArray();
+              ASTUtils.elementToListOfStrings(property.getValue()).stream()
+                  .map(relativePath::resolve) // assume all paths are relative
+                  .map(Objects::toString)
+                  .map(
+                      s -> {
+                        var element = LfFactory.eINSTANCE.createElement();
+                        element.setLiteral(s);
+                        return element;
+                      })
+                  .forEach(array.getElements()::add);
+              value = LfFactory.eINSTANCE.createElement();
+              value.setArray(array);
+            }
+            p.updater.parseIntoTargetConfig(config, value, err);
           }
         });
   }

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

@@ -120,7 +120,7 @@ public String generatePreamble(
             fast: false,
             federateID: %d,
             federationID: "Unidentified Federation",
-            keepAlive: false,
+            keepAlive: true,
             minOutputDelay: %s,
             networkMessageActions: [%s],
             rtiHost: "%s",

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

@@ -2,6 +2,7 @@
 
 import static org.lflang.ast.ASTUtils.convertToEmptyListIfNull;
 
+import java.nio.file.Path;
 import org.eclipse.emf.ecore.resource.Resource;
 import org.lflang.ErrorReporter;
 import org.lflang.TargetConfig;
@@ -57,11 +58,20 @@ private void mergeImportedConfig(
       if (targetProperties != null) {
         // Merge properties
         TargetProperty.update(
-            this, convertToEmptyListIfNull(targetProperties.getPairs()), errorReporter);
+            this,
+            convertToEmptyListIfNull(targetProperties.getPairs()),
+            getRelativePath(mainResource, federateResource),
+            errorReporter);
       }
     }
   }
 
+  private Path getRelativePath(Resource source, Resource target) {
+    return Path.of(source.getURI().toFileString())
+        .getParent()
+        .relativize(Path.of(target.getURI().toFileString()).getParent());
+  }
+
   /** Method for the removal of things that should not appear in the target config of a federate. */
   private void clearPropertiesToIgnore() {
     this.setByUser.remove(TargetProperty.CLOCK_SYNC);

core/src/main/java/org/lflang/federated/launcher/FedLauncherGenerator.java

@@ -205,7 +205,8 @@ public void doGenerate(List<FederateInstance> federates, RtiConfig rtiConfig) {
                 "do",
                 "    wait $pid",
                 "done",
-                "echo \"All done.\"")
+                "echo \"All done.\"",
+                "EXITED_SUCCESSFULLY=true")
             + "\n");
 
     // Create bin directory for the script.
@@ -281,24 +282,19 @@ private String getSetupCode() {
         "# Set a trap to kill all background jobs on error or control-C",
         "# Use two distinct traps so we can see which signal causes this.",
         "cleanup() {",
-        "    printf \"Killing federate %s.\\n\" ${pids[*]}",
-        "    # The || true clause means this is not an error if kill fails.",
-        "    kill ${pids[@]} || true",
-        "    printf \"#### Killing RTI %s.\\n\" ${RTI}",
-        "    kill ${RTI} || true",
-        "    exit 1",
-        "}",
-        "cleanup_err() {",
-        "    echo \"#### Received ERR signal on line $1. Invoking cleanup().\"",
-        "    cleanup",
-        "}",
-        "cleanup_sigint() {",
-        "    echo \"#### Received SIGINT signal on line $1. Invoking cleanup().\"",
-        "    cleanup",
+        "    if [ \"$EXITED_SUCCESSFULLY\" = true ] ; then",
+        "        exit 0",
+        "    else",
+        "        printf \"Killing federate %s.\\n\" ${pids[*]}",
+        "        # The || true clause means this is not an error if kill fails.",
+        "        kill ${pids[@]} || true",
+        "        printf \"#### Killing RTI %s.\\n\" ${RTI}",
+        "        kill ${RTI} || true",
+        "        exit 1",
+        "    fi",
         "}",
         "",
-        "trap 'cleanup_err $LINENO' ERR",
-        "trap 'cleanup_sigint $LINENO' SIGINT",
+        "trap 'cleanup; exit' EXIT",
         "",
         "# Create a random 48-byte text ID for this federation.",
         "# The likelihood of two federations having the same ID is 1/16,777,216 (1/2^24).",
@@ -354,7 +350,7 @@ private String getLaunchCode(String rtiLaunchCode) {
         "    echo \"RTI could not be found.\"",
         "    echo \"The source code can be obtained from"
             + " https://github.com/lf-lang/reactor-c/tree/main/core/federated/RTI\"",
-        "    exit",
+        "    exit 1",
         "fi                ",
         "# The RTI is started first to allow proper boot-up",
         "# before federates will try to connect.",
@@ -388,7 +384,7 @@ private String getRemoteLaunchCode(
         "    then",
         "        echo \"RTI could not be found.\"",
         "        echo \"The source code can be found in org.lflang/src/lib/core/federated/RTI\"",
-        "        exit",
+        "        exit 1",
         "    fi",
         "    " + rtiLaunchString + " 2>&1 | tee -a " + logFileName + "' &",
         "# Store the PID of the channel to RTI",

core/src/main/kotlin/org/lflang/generator/ts/TSConstructorGenerator.kt

@@ -59,7 +59,6 @@ class TSConstructorGenerator(
                     |        }
                     |        federateConfig.federationID = __federationID;
                     |        federateConfig.fast = __fast;
-                    |        federateConfig.keepAlive = __keepAlive;
                     |        super(federateConfig, success, fail);
                     """.trimMargin()
             } else {

core/src/testFixtures/java/org/lflang/tests/LFTest.java

@@ -258,6 +258,10 @@ private Thread recordStream(StringBuffer builder, InputStream inputStream) {
               char[] buf = new char[1024];
               while ((len = reader.read(buf)) > 0) {
                 builder.append(buf, 0, len);
+                if (Runtime.getRuntime().freeMemory()
+                    < Runtime.getRuntime().totalMemory() * 3 / 4) {
+                  builder.delete(0, builder.length() / 2);
+                }
               }
             } catch (IOException e) {
               throw new RuntimeIOException(e);

core/src/testFixtures/java/org/lflang/tests/TestBase.java

@@ -520,7 +520,7 @@ private void execute(LFTest test) throws TestError {
       if (!p.waitFor(MAX_EXECUTION_TIME_SECONDS, TimeUnit.SECONDS)) {
         stdout.interrupt();
         stderr.interrupt();
-        p.destroyForcibly();
+        p.destroy();
         throw new TestError(Result.TEST_TIMEOUT);
       } else {
         if (stdoutException.get() != null || stderrException.get() != null) {

test/Python/src/docker/FilesPropertyContainerized.lf

@@ -7,15 +7,15 @@ preamble {=
     try:
         import hello
     except:
-        lf_request_stop()
+        request_stop()
 =}
 
 main reactor {
     preamble {=
         try:
             import hello
         except:
-            lf_request_stop()
+            request_stop()
     =}
     state passed = False
     timer t(1 msec)

test/Python/src/docker/federated/DistributedStopDecentralizedContainerized.lf

@@ -1,5 +1,5 @@
 /**
- * Test for lf_request_stop() in federated execution with decentralized
+ * Test for request_stop() in federated execution with decentralized
  * coordination.
  *
  * @author Soroush Bateni

test/Python/src/federated/CycleDetection.lf

@@ -36,7 +36,7 @@ reactor UserInput {
             self.sys.stderr.write("Did not receive the expected balance. Expected: 200. Got: {}.\n".format(balance.value))
             self.sys.exit(1)
         print("Balance: {}".format(balance.value))
-        lf_request_stop()
+        request_stop()
     =}
 
     reaction(shutdown) {= print("Test passed!") =}

test/Python/src/federated/DistributedCountDecentralizedLate.lf

@@ -20,8 +20,8 @@ reactor Print {
     # STP(in, 30 msec);
     state success = 0
     state success_stp_violation = 0
-    # Force a timer to be invoke periodically
-    timer t(0, 10 usec)
+    # Force a timer to be invoked periodically
+    timer t(0, 1 msec)
     # to ensure logical time will advance in the absence of incoming messages.
     state c = 0
 

test/Python/src/federated/DistributedCountDecentralizedLateHierarchy.lf

@@ -14,7 +14,48 @@ target Python {
 }
 
 import Count from "../lib/Count.lf"
-import ImportantActuator, Print from "DistributedCountDecentralizedLateDownstream.lf"
+import Print from "DistributedCountDecentralizedLateDownstream.lf"
+
+reactor ImportantActuator {
+    input inp
+    # Count messages that arrive without STP violation.
+    state success = 0
+    state success_stp_violation = 0
+    # Force a timer to be invoked periodically
+    timer t(0, 10 usec)
+    # to ensure logical time will advance in the absence of incoming messages.
+    state c = 0
+
+    reaction(inp) {=
+        current_tag = lf.tag()
+        print(f"ImportantActuator: At tag ({lf.tag().time}, {lf.tag().microstep}) received {inp.value}. "
+              f"Intended tag is ({inp.intended_tag.time - lf.time.start()}, {inp.intended_tag.microstep}).")
+        if lf.tag() == Tag((SEC(1) *  self.c) + + lf.time.start(), 0):
+            self.success += 1 # Message was on-time
+        else:
+            self.sys.stderr.write("Normal reaction was invoked, but current tag doesn't match expected tag.")
+            self.sys.exit(1)
+        self.c += 1
+    =} STP(0) {=
+        current_tag = lf.tag()
+        print(f"ImportantActuator: At tag ({lf.time.logical_elapsed()}, {lf.tag().microstep}), message has violated the STP offset "
+              f"by ({lf.tag().time - inp.intended_tag.time}, {lf.tag().microstep - inp.intended_tag.microstep}).")
+        self.success_stp_violation += 1
+        self.c += 1
+    =}
+
+    reaction(t) {=
+        # Do nothing.
+    =}
+
+    reaction(shutdown) {=
+        if (self.success +  self.success_stp_violation) != 5:
+            self.sys.stderr.write("Failed to detect STP violation in messages.")
+            self.sys.exit(1)
+        else:
+            print(f"Successfully detected STP violations ({self.success_stp_violation} violations, {self.success} on-time).")
+    =}
+}
 
 reactor Receiver {
     input inp

test/Python/src/federated/DistributedSendClass.lf

@@ -9,7 +9,7 @@ preamble {=
 reactor A {
     input o
 
-    reaction(o) {= lf_request_stop() =}
+    reaction(o) {= request_stop() =}
 }
 
 reactor B {

test/Python/src/federated/DistributedStop.lf

@@ -1,6 +1,5 @@
 /**
- * Test for lf_request_stop() in federated execution with centralized
- * coordination.
+ * Test for request_stop() in federated execution with centralized coordination.
  *
  * @author Soroush Bateni
  */
@@ -18,23 +17,23 @@ reactor Sender {
         tag = lf.tag()
         print("Sending 42 at ({}, {}).".format(
                      lf.time.logical_elapsed(),
-                     lf.tag().microstep))
+                     tag.microstep))
         out.set(42)
-        if lf.tag().microstep == 0:
+        if tag.microstep == 0:
             # Instead of having a separate reaction
             # for 'act' like Stop.lf, we trigger the
-            # same reaction to test lf_request_stop() being
+            # same reaction to test request_stop() being
             # called multiple times
             act.schedule(0)
-        if tag.time == USEC(1):
-            # Call lf_request_stop() both at (1 usec, 0) and
+        if lf.time.logical_elapsed() == USEC(1):
+            # Call request_stop() both at (1 usec, 0) and
             # (1 usec, 1)
             print("Requesting stop at ({}, {}).".format(
                      lf.time.logical_elapsed(),
                      lf.tag().microstep))
-            lf_request_stop()
+            request_stop()
 
-        _1usec1 = Tag(time=USEC(1) + get_start_time(), microstep=1)
+        _1usec1 = Tag(time=USEC(1) + lf.time.start(), microstep=1)
         if lf.tag_compare(lf.tag(), _1usec1) == 0:
             # The reaction was invoked at (1 usec, 1) as expected
             self.reaction_invoked_correctly = True
@@ -77,12 +76,12 @@ reactor Receiver(
             print("Requesting stop at ({}, {}).".format(
                      lf.time.logical_elapsed(),
                      lf.tag().microstep))
-            lf_request_stop()
+            request_stop()
             # The receiver should receive a message at tag
             # (1 usec, 1) and trigger this reaction
             self.reaction_invoked_correctly = True
 
-        _1usec1 = Tag(time=USEC(1) + get_start_time(), microstep=1)
+        _1usec1 = Tag(time=USEC(1) + lf.time.start(), microstep=1)
         if lf.tag_compare(lf.tag(), _1usec1) > 0:
             self.reaction_invoked_correctly = False
     =}

test/Python/src/federated/DistributedStopDecentralized.lf

@@ -1,5 +1,5 @@
 /**
- * Test for lf_request_stop() in federated execution with decentralized
+ * Test for request_stop() in federated execution with decentralized
  * coordination.
  *
  * @author Soroush Bateni