Draft: Network Channel

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

@@ -520,9 +520,9 @@ public String generatePreamble(
         extern "C" {
         #endif""");
     includes.pr("#include \"core/federated/federate.h\"");
-    includes.pr("#include \"core/federated/network/net_common.h\"");
-    includes.pr("#include \"core/federated/network/net_util.h\"");
-    includes.pr("#include \"core/federated/network/socket_common.h\"");
+    includes.pr("#include \"network/api/net_driver.h\"");
+    includes.pr("#include \"network/api/net_common.h\"");
+    includes.pr("#include \"network/api/net_util.h\"");
     includes.pr("#include \"core/federated/clock-sync.h\"");
     includes.pr("#include \"core/threaded/reactor_threaded.h\"");
     includes.pr("#include \"core/utils/util.h\"");
@@ -543,9 +543,9 @@ protected String makePreamble(
     var code = new CodeBuilder();
 
     code.pr("#include \"core/federated/federate.h\"");
-    code.pr("#include \"core/federated/network/net_common.h\"");
-    code.pr("#include \"core/federated/network/net_util.h\"");
-    code.pr("#include \"core/federated/network/socket_common.h\"");
+    code.pr("#include \"network/api/net_driver.h\"");
+    code.pr("#include \"network/api/net_common.h\"");
+    code.pr("#include \"network/api/net_util.h\"");
     code.pr("#include \"core/federated/clock-sync.h\"");
     code.pr("#include \"core/threaded/reactor_threaded.h\"");
     code.pr("#include \"core/utils/util.h\"");
@@ -686,8 +686,9 @@ private String generateCodeToInitializeFederate(
         String.join(
             "\n",
             "// Initialize the socket mutexes",
-            "lf_mutex_init(&lf_outbound_socket_mutex);",
-            "lf_mutex_init(&socket_mutex);",
+            "lf_mutex_init(&lf_outbound_netchan_mutex);",
+            "lf_mutex_init(&lf_inbound_netchan_mutex);",
+            "lf_mutex_init(&shutdown_mutex);",
             "lf_cond_init(&lf_port_status_changed, &env->mutex);"));
 
     // Find the STA (A.K.A. the global STP offset) for this federate.
@@ -740,16 +741,16 @@ else if (globalSTP instanceof CodeExprImpl)
     code.pr(
         String.join(
             "\n",
-            "// Initialize the array of socket for incoming connections to -1.",
+            "// Initialize the array of network drivers for incoming connections to -1.",
             "for (int i = 0; i < NUMBER_OF_FEDERATES; i++) {",
-            "    _fed.sockets_for_inbound_p2p_connections[i] = -1;",
+            "    _fed.netchans_for_inbound_p2p_connections[i] = NULL;",
             "}"));
     code.pr(
         String.join(
             "\n",
-            "// Initialize the array of socket for outgoing connections to -1.",
+            "// Initialize the array of network drivers for outgoing connections to -1.",
             "for (int i = 0; i < NUMBER_OF_FEDERATES; i++) {",
-            "    _fed.sockets_for_outbound_p2p_connections[i] = -1;",
+            "    _fed.netchans_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.
@@ -765,7 +766,7 @@ else if (globalSTP instanceof CodeExprImpl)
     code.pr(
         String.join(
             "\n",
-            "// Connect to the RTI. This sets _fed.socket_TCP_RTI and _lf_rti_socket_UDP.",
+            "// Connect to the RTI. This sets _fed.netchan_to_RTI and _lf_rti_socket_UDP.",
             "lf_connect_to_rti("
                 + addDoubleQuotes(rtiConfig.getHost())
                 + ", "
@@ -775,14 +776,14 @@ else if (globalSTP instanceof CodeExprImpl)
     // Disable clock synchronization for the federate if it resides on the same host as the RTI,
     // unless that is overridden with the clock-sync-options target property.
     if (CExtensionUtils.clockSyncIsOn(federate, rtiConfig)) {
-      code.pr("synchronize_initial_physical_clock_with_rti(&_fed.socket_TCP_RTI);");
+      code.pr("synchronize_initial_physical_clock_with_rti(_fed.netchan_to_RTI);");
     }
 
     if (numberOfInboundConnections > 0) {
       code.pr(
           String.join(
               "\n",
-              "// Create a socket server to listen to other federates.",
+              "// Create a server to listen to other federates.",
               "// If a port is specified by the user, that will be used",
               "// as the only possibility for the server. If not, the port",
               "// will be selected by the OS (by specifying port 0).",

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

@@ -28,6 +28,7 @@
 import org.lflang.target.property.ClockSyncOptionsProperty;
 import org.lflang.target.property.ClockSyncOptionsProperty.ClockSyncOptions;
 import org.lflang.target.property.CmakeIncludeProperty;
+import org.lflang.target.property.CommunicationModeProperty;
 import org.lflang.target.property.CompileDefinitionsProperty;
 import org.lflang.target.property.CoordinationOptionsProperty;
 import org.lflang.target.property.CoordinationProperty;
@@ -201,6 +202,10 @@ public static void handleCompileDefinitions(
     if (federate.targetConfig.get(AuthProperty.INSTANCE)) {
       definitions.put("FEDERATED_AUTHENTICATED", "");
     }
+    if (federate.targetConfig.isSet(CommunicationModeProperty.INSTANCE)) {
+      definitions.put(
+          "COMM_TYPE", federate.targetConfig.get(CommunicationModeProperty.INSTANCE).toString());
+    }
     definitions.put("NUMBER_OF_FEDERATES", String.valueOf(federateNames.size()));
     definitions.put("EXECUTABLE_PREAMBLE", "");
     definitions.put("FEDERATE_ID", String.valueOf(federate.id));
@@ -325,7 +330,7 @@ public static void generateCMakeInclude(
 
   /**
    * Generate code that sends the neighbor structure message to the RTI. See {@code
-   * MSG_TYPE_NEIGHBOR_STRUCTURE} in {@code federated/net_common.h}.
+   * MSG_TYPE_NEIGHBOR_STRUCTURE} in {@code network/api/net_common.h}.
    *
    * @param federate The federate that is sending its neighbor structure
    */
@@ -342,7 +347,7 @@ public static String generateFederateNeighborStructure(FederateInstance federate
             "* information is needed for the RTI to perform the centralized coordination.",
             "* @see MSG_TYPE_NEIGHBOR_STRUCTURE in net_common.h",
             "*/",
-            "void lf_send_neighbor_structure_to_RTI(int rti_socket) {"));
+            "void lf_send_neighbor_structure_to_RTI(netchan_t rti_netchan) {"));
     code.indent();
     // Initialize the array of information about the federate's immediate upstream
     // and downstream relayed (through the RTI) logical connections, to send to the
@@ -444,8 +449,8 @@ public static String generateFederateNeighborStructure(FederateInstance federate
     code.pr(
         String.join(
             "\n",
-            "write_to_socket_fail_on_error(",
-            "    &rti_socket, ",
+            "write_to_netchan_fail_on_error(",
+            "    rti_netchan, ",
             "    buffer_size,",
             "    buffer_to_send,",
             "    NULL,",

core/src/main/java/org/lflang/generator/c/CCmakeGenerator.java

@@ -37,6 +37,7 @@
 import org.lflang.target.property.AuthProperty;
 import org.lflang.target.property.BuildTypeProperty;
 import org.lflang.target.property.CmakeIncludeProperty;
+import org.lflang.target.property.CommunicationModeProperty;
 import org.lflang.target.property.CompileDefinitionsProperty;
 import org.lflang.target.property.CompilerProperty;
 import org.lflang.target.property.PlatformProperty;
@@ -426,6 +427,10 @@ CodeBuilder generateCMakeCode(
       cMakeCode.pr("target_link_libraries( ${LF_MAIN_TARGET} PRIVATE OpenSSL::SSL)");
       cMakeCode.newLine();
     }
+    if (targetConfig.isSet(CommunicationModeProperty.INSTANCE)) {
+      cMakeCode.pr("set(COMM_TYPE " + targetConfig.get(CommunicationModeProperty.INSTANCE) + ")");
+      cMakeCode.newLine();
+    }
 
     if (!targetConfig.get(SingleThreadedProperty.INSTANCE)
         && platformOptions.platform() != Platform.ZEPHYR

core/src/main/java/org/lflang/generator/c/CGenerator.java

@@ -921,9 +921,13 @@ protected void copyTargetFiles() throws IOException {
         FileUtil.copyFromClassPath("/lib/c/reactor-c/" + directory, dest, true, false);
       }
       for (var directory :
-          List.of("logging", "platform", "low_level_platform", "trace", "version", "tag")) {
+          List.of(
+              "logging", "platform", "low_level_platform", "trace", "version", "tag", "network")) {
         var entry = "/lib/c/reactor-c/" + directory;
         if (arduino) {
+          if ("network".equals(directory)) {
+            continue; // Skip copying for the "network" directory
+          }
           if (FileConfig.class.getResource(entry + "/api") != null) {
             FileUtil.copyFromClassPath(
                 entry + "/api",

core/src/main/java/org/lflang/generator/c/CPreambleGenerator.java

@@ -65,7 +65,7 @@ public static String generateIncludeStatements(TargetConfig targetConfig, boolea
     code.pr("int lf_reactor_c_main(int argc, const char* argv[]);");
     if (targetConfig.isSet(FedSetupProperty.INSTANCE)) {
       code.pr("#include \"include/core/federated/federate.h\"");
-      code.pr("#include \"include/core/federated/network/net_common.h\"");
+      code.pr("#include \"network/api/net_common.h\"");
     }
     if (cppMode || arduinoBased(targetConfig)) {
       code.pr("}");

core/src/main/java/org/lflang/target/Target.java

@@ -556,6 +556,7 @@ public void initialize(TargetConfig config) {
               ClockSyncModeProperty.INSTANCE,
               ClockSyncOptionsProperty.INSTANCE,
               CmakeIncludeProperty.INSTANCE,
+              CommunicationModeProperty.INSTANCE,
               CompileDefinitionsProperty.INSTANCE,
               CompilerProperty.INSTANCE,
               CoordinationOptionsProperty.INSTANCE,

core/src/main/java/org/lflang/target/property/CommunicationModeProperty.java

@@ -0,0 +1,43 @@
+package org.lflang.target.property;
+
+import org.lflang.MessageReporter;
+import org.lflang.ast.ASTUtils;
+import org.lflang.lf.Element;
+import org.lflang.target.property.type.CommunicationModeType;
+import org.lflang.target.property.type.CommunicationModeType.CommunicationMode;
+
+public final class CommunicationModeProperty
+    extends TargetProperty<CommunicationMode, CommunicationModeType> {
+
+  /** Singleton target property instance. */
+  public static final CommunicationModeProperty INSTANCE = new CommunicationModeProperty();
+
+  private CommunicationModeProperty() {
+    super(new CommunicationModeType());
+  }
+
+  @Override
+  public Element toAstElement(CommunicationMode value) {
+    return ASTUtils.toElement(value.toString());
+  }
+
+  @Override
+  public CommunicationMode initialValue() {
+    return CommunicationMode.TCP;
+  }
+
+  @Override
+  public CommunicationMode fromAst(Element node, MessageReporter reporter) {
+    return fromString(ASTUtils.elementToSingleString(node), reporter);
+  }
+
+  @Override
+  protected CommunicationMode fromString(String string, MessageReporter reporter) {
+    return this.type.forName(string);
+  }
+
+  @Override
+  public String name() {
+    return "comm-type";
+  }
+}

core/src/main/java/org/lflang/target/property/type/CommunicationModeType.java

@@ -0,0 +1,44 @@
+package org.lflang.target.property.type;
+
+import java.util.Arrays;
+import java.util.List;
+import java.util.stream.Collectors;
+import org.lflang.target.property.type.CommunicationModeType.CommunicationMode;
+
+/** Enumeration of communication types */
+public class CommunicationModeType extends OptionsType<CommunicationMode> {
+
+  @Override
+  protected Class<CommunicationMode> enumClass() {
+    return CommunicationMode.class;
+  }
+
+  /** Enumeration of communication types. */
+  public enum CommunicationMode {
+    TCP("TCP");
+
+    // More communication modes will be added.
+
+    /** Alias used in toString method. */
+    private final String alias;
+
+    /** Private constructor for Cmake build types. */
+    CommunicationMode(String alias) {
+      this.alias = alias;
+    }
+
+    /** Return the alias. */
+    @Override
+    public String toString() {
+      return this.alias;
+    }
+
+    public static List<CommunicationMode> optionsList() {
+      return Arrays.stream(CommunicationMode.values()).collect(Collectors.toList());
+    }
+
+    public static CommunicationMode getDefault() {
+      return CommunicationMode.TCP;
+    }
+  }
+}