Simplify NetworkChannel interface

examples/posix/federated/receiver.c

@@ -29,7 +29,7 @@ typedef struct {
   REACTION_INSTANCE(Receiver, r);
   PORT_INSTANCE(Receiver, in, 1);
   int cnt;
-  REACTOR_BOOKKEEPING_INSTANCES(1,1,0);
+  REACTOR_BOOKKEEPING_INSTANCES(1, 1, 0);
 } Receiver;
 
 DEFINE_REACTION_BODY(Receiver, r) {
@@ -56,12 +56,12 @@ typedef struct {
   FederatedConnectionBundle super;
   TcpIpChannel channel;
   FEDERATED_INPUT_CONNECTION_INSTANCE(Receiver, in);
-  FEDERATED_CONNECTION_BUNDLE_BOOKKEEPING_INSTANCES(1,0)
+  FEDERATED_CONNECTION_BUNDLE_BOOKKEEPING_INSTANCES(1, 0)
 } FEDERATED_CONNECTION_BUNDLE_NAME(Receiver, Sender);
 
 FEDERATED_CONNECTION_BUNDLE_CTOR_SIGNATURE(Receiver, Sender) {
   FEDERATED_CONNECTION_BUNDLE_CTOR_PREAMBLE();
-  TcpIpChannel_ctor(&self->channel, "127.0.0.1", PORT_NUM, AF_INET, false);
+  TcpIpChannel_ctor(&self->channel, parent->env, "127.0.0.1", PORT_NUM, AF_INET, false);
   FEDERATED_CONNECTION_BUNDLE_CALL_CTOR();
   INITIALIZE_FEDERATED_INPUT_CONNECTION(Receiver, in, deserialize_msg_t);
 }
@@ -71,14 +71,14 @@ typedef struct {
   CHILD_REACTOR_INSTANCE(Receiver, receiver, 1);
   FEDERATED_CONNECTION_BUNDLE_INSTANCE(Receiver, Sender);
   FEDERATE_BOOKKEEPING_INSTANCES(1);
-  CHILD_INPUT_SOURCES(receiver, in,1,1, 0);
+  CHILD_INPUT_SOURCES(receiver, in, 1, 1, 0);
 } MainRecv;
 
 REACTOR_CTOR_SIGNATURE(MainRecv) {
   FEDERATE_CTOR_PREAMBLE();
   REACTOR_CTOR(MainRecv);
   DEFINE_CHILD_INPUT_ARGS(receiver, in, 1, 1);
-  INITIALIZE_CHILD_REACTOR_WITH_PARAMETERS(Receiver, receiver,1, _receiver_in_args[i]);
+  INITIALIZE_CHILD_REACTOR_WITH_PARAMETERS(Receiver, receiver, 1, _receiver_in_args[i]);
   INITIALIZE_FEDERATED_CONNECTION_BUNDLE(Receiver, Sender);
   BUNDLE_REGISTER_DOWNSTREAM(Receiver, Sender, receiver, in);
 }

examples/posix/federated/sender.c

@@ -26,14 +26,14 @@ DEFINE_TIMER_CTOR(Sender, t, 1, 0)
 DEFINE_REACTION_STRUCT(Sender, r, 1)
 DEFINE_REACTION_CTOR(Sender, r, 0)
 DEFINE_OUTPUT_STRUCT(Sender, out, 1, msg_t)
-DEFINE_OUTPUT_CTOR(Sender, out, 1)  
+DEFINE_OUTPUT_CTOR(Sender, out, 1)
 
 typedef struct {
   Reactor super;
   TIMER_INSTANCE(Sender, t);
   REACTION_INSTANCE(Sender, r);
   PORT_INSTANCE(Sender, out, 1);
-  REACTOR_BOOKKEEPING_INSTANCES(1,2,0);
+  REACTOR_BOOKKEEPING_INSTANCES(1, 2, 0);
 } Sender;
 
 DEFINE_REACTION_BODY(Sender, r) {
@@ -65,15 +65,15 @@ typedef struct {
   FederatedConnectionBundle super;
   TcpIpChannel channel;
   FEDERATED_OUTPUT_CONNECTION_INSTANCE(Sender, out);
-  FEDERATED_CONNECTION_BUNDLE_BOOKKEEPING_INSTANCES(0,1);
+  FEDERATED_CONNECTION_BUNDLE_BOOKKEEPING_INSTANCES(0, 1);
 } FEDERATED_CONNECTION_BUNDLE_NAME(Sender, Receiver);
 
 FEDERATED_CONNECTION_BUNDLE_CTOR_SIGNATURE(Sender, Receiver) {
   FEDERATED_CONNECTION_BUNDLE_CTOR_PREAMBLE();
-  TcpIpChannel_ctor(&self->channel, "127.0.0.1", PORT_NUM, AF_INET, true);
+  TcpIpChannel_ctor(&self->channel, parent->env, "127.0.0.1", PORT_NUM, AF_INET, true);
 
   FEDERATED_CONNECTION_BUNDLE_CALL_CTOR();
-  
+
   INITIALIZE_FEDERATED_OUTPUT_CONNECTION(Sender, out, serialize_msg_t);
 }
 
@@ -82,17 +82,16 @@ typedef struct {
   Reactor super;
   CHILD_REACTOR_INSTANCE(Sender, sender, 1);
   FEDERATED_CONNECTION_BUNDLE_INSTANCE(Sender, Receiver);
-  TcpIpChannel channel;
   FEDERATE_BOOKKEEPING_INSTANCES(1);
-  CHILD_OUTPUT_CONNECTIONS(sender, out, 1,1, 1);
-  CHILD_OUTPUT_EFFECTS(sender, out, 1,1, 0);
-  CHILD_OUTPUT_OBSERVERS(sender, out, 1,1,0);
+  CHILD_OUTPUT_CONNECTIONS(sender, out, 1, 1, 1);
+  CHILD_OUTPUT_EFFECTS(sender, out, 1, 1, 0);
+  CHILD_OUTPUT_OBSERVERS(sender, out, 1, 1, 0);
 } MainSender;
 
 REACTOR_CTOR_SIGNATURE(MainSender) {
   FEDERATE_CTOR_PREAMBLE();
-  DEFINE_CHILD_OUTPUT_ARGS(sender, out,1,1);
-  INITIALIZE_CHILD_REACTOR_WITH_PARAMETERS(Sender, sender,1, _sender_out_args[i]);
+  DEFINE_CHILD_OUTPUT_ARGS(sender, out, 1, 1);
+  INITIALIZE_CHILD_REACTOR_WITH_PARAMETERS(Sender, sender, 1, _sender_out_args[i]);
   INITIALIZE_FEDERATED_CONNECTION_BUNDLE(Sender, Receiver);
   BUNDLE_REGISTER_UPSTREAM(Sender, Receiver, sender, out);
   REACTOR_CTOR(MainSender);

examples/zephyr/basic_federated/common/receiver.h

@@ -32,7 +32,7 @@ typedef struct {
   char msg[32];
 } msg_t;
 
-DEFINE_REACTION_STRUCT(Receiver, r,  0);
+DEFINE_REACTION_STRUCT(Receiver, r, 0);
 DEFINE_REACTION_CTOR(Receiver, r, 0)
 
 DEFINE_INPUT_STRUCT(Receiver, in, 1, 0, msg_t, 0)
@@ -42,7 +42,7 @@ typedef struct {
   Reactor super;
   REACTION_INSTANCE(Receiver, r);
   PORT_INSTANCE(Receiver, in, 1);
-  REACTOR_BOOKKEEPING_INSTANCES(1,1,0);
+  REACTOR_BOOKKEEPING_INSTANCES(1, 1, 0);
   int cnt;
 } Receiver;
 
@@ -59,25 +59,23 @@ REACTOR_CTOR_SIGNATURE_WITH_PARAMETERS(Receiver, InputExternalCtorArgs *in_exter
   REACTOR_CTOR(Receiver);
   INITIALIZE_REACTION(Receiver, r);
   INITIALIZE_INPUT(Receiver, in, 1, in_external);
-  
+
   // Register reaction as an effect of in
   PORT_REGISTER_EFFECT(self->in, self->r, 1);
 }
 
-
 DEFINE_FEDERATED_INPUT_CONNECTION(Receiver, in, msg_t, 5, MSEC(100), false);
 
 typedef struct {
   FederatedConnectionBundle super;
   TcpIpChannel channel;
   FEDERATED_INPUT_CONNECTION_INSTANCE(Receiver, in);
-  FEDERATED_CONNECTION_BUNDLE_BOOKKEEPING_INSTANCES(1,0)
+  FEDERATED_CONNECTION_BUNDLE_BOOKKEEPING_INSTANCES(1, 0)
 } FEDERATED_CONNECTION_BUNDLE_NAME(Receiver, Sender);
 
-
 FEDERATED_CONNECTION_BUNDLE_CTOR_SIGNATURE(Receiver, Sender) {
   FEDERATED_CONNECTION_BUNDLE_CTOR_PREAMBLE();
-  TcpIpChannel_ctor(&self->channel, IP_ADDR, PORT_NUM, AF_INET, false);
+  TcpIpChannel_ctor(&self->channel, parent->env, IP_ADDR, PORT_NUM, AF_INET, false);
   FEDERATED_CONNECTION_BUNDLE_CALL_CTOR();
   INITIALIZE_FEDERATED_INPUT_CONNECTION(Receiver, in, deserialize_payload_default);
 }
@@ -87,14 +85,14 @@ typedef struct {
   CHILD_REACTOR_INSTANCE(Receiver, receiver, 1);
   FEDERATED_CONNECTION_BUNDLE_INSTANCE(Receiver, Sender);
   FEDERATE_BOOKKEEPING_INSTANCES(1);
-  CHILD_INPUT_SOURCES(receiver, in, 1,1, 0);
+  CHILD_INPUT_SOURCES(receiver, in, 1, 1, 0);
 } MainRecv;
 
 REACTOR_CTOR_SIGNATURE(MainRecv) {
   FEDERATE_CTOR_PREAMBLE();
   REACTOR_CTOR(MainRecv);
-  DEFINE_CHILD_INPUT_ARGS(receiver, in,1,1);
-  INITIALIZE_CHILD_REACTOR_WITH_PARAMETERS(Receiver, receiver,1, _receiver_in_args[i]);
+  DEFINE_CHILD_INPUT_ARGS(receiver, in, 1, 1);
+  INITIALIZE_CHILD_REACTOR_WITH_PARAMETERS(Receiver, receiver, 1, _receiver_in_args[i]);
   INITIALIZE_FEDERATED_CONNECTION_BUNDLE(Receiver, Sender);
   BUNDLE_REGISTER_DOWNSTREAM(Receiver, Sender, receiver, in);
 }

examples/zephyr/basic_federated/federated_sender/src/sender.c

@@ -66,13 +66,12 @@ void setup_led() {
   gpio_pin_configure_dt(&led, GPIO_OUTPUT_ACTIVE);
 }
 
-
 typedef struct {
   Reactor super;
   REACTION_INSTANCE(Sender, r);
   ACTION_INSTANCE(Sender, act);
   PORT_INSTANCE(Sender, out, 1);
-  REACTOR_BOOKKEEPING_INSTANCES(1,2,0);
+  REACTOR_BOOKKEEPING_INSTANCES(1, 2, 0);
 } Sender;
 
 DEFINE_REACTION_BODY(Sender, r) {
@@ -104,31 +103,31 @@ typedef struct {
   FederatedConnectionBundle super;
   TcpIpChannel channel;
   FEDERATED_OUTPUT_CONNECTION_INSTANCE(Sender, out);
-  FEDERATED_CONNECTION_BUNDLE_BOOKKEEPING_INSTANCES(0,1);
+  FEDERATED_CONNECTION_BUNDLE_BOOKKEEPING_INSTANCES(0, 1);
 } FEDERATED_CONNECTION_BUNDLE_NAME(Sender, Receiver1);
 
 typedef struct {
   FederatedConnectionBundle super;
   TcpIpChannel channel;
   FEDERATED_OUTPUT_CONNECTION_INSTANCE(Sender, out);
-  FEDERATED_CONNECTION_BUNDLE_BOOKKEEPING_INSTANCES(0,1);
+  FEDERATED_CONNECTION_BUNDLE_BOOKKEEPING_INSTANCES(0, 1);
 } FEDERATED_CONNECTION_BUNDLE_NAME(Sender, Receiver2);
 
 FEDERATED_CONNECTION_BUNDLE_CTOR_SIGNATURE(Sender, Receiver1) {
   FEDERATED_CONNECTION_BUNDLE_CTOR_PREAMBLE();
-  TcpIpChannel_ctor(&self->channel, "192.168.1.100", PORT_CONN_1, AF_INET, true);
+  TcpIpChannel_ctor(&self->channel, parent->env, "192.168.1.100", PORT_CONN_1, AF_INET, true);
 
   FEDERATED_CONNECTION_BUNDLE_CALL_CTOR();
-  
+
   INITIALIZE_FEDERATED_OUTPUT_CONNECTION(Sender, out, serialize_payload_default);
 }
 
 FEDERATED_CONNECTION_BUNDLE_CTOR_SIGNATURE(Sender, Receiver2) {
   FEDERATED_CONNECTION_BUNDLE_CTOR_PREAMBLE();
-  TcpIpChannel_ctor(&self->channel, "192.168.1.100", PORT_CONN_2, AF_INET, true);
+  TcpIpChannel_ctor(&self->channel, parent->env, "192.168.1.100", PORT_CONN_2, AF_INET, true);
 
   FEDERATED_CONNECTION_BUNDLE_CALL_CTOR();
-  
+
   INITIALIZE_FEDERATED_OUTPUT_CONNECTION(Sender, out, serialize_payload_default);
 }
 
@@ -138,9 +137,9 @@ typedef struct {
   CHILD_REACTOR_INSTANCE(Sender, sender, 1);
   FEDERATED_CONNECTION_BUNDLE_INSTANCE(Sender, Receiver1);
   FEDERATED_CONNECTION_BUNDLE_INSTANCE(Sender, Receiver2);
-  CHILD_OUTPUT_CONNECTIONS(sender, out, 1,1, 2);
-  CHILD_OUTPUT_EFFECTS(sender, out, 1,1,0);
-  CHILD_OUTPUT_OBSERVERS(sender, out,1,1, 0);
+  CHILD_OUTPUT_CONNECTIONS(sender, out, 1, 1, 2);
+  CHILD_OUTPUT_EFFECTS(sender, out, 1, 1, 0);
+  CHILD_OUTPUT_OBSERVERS(sender, out, 1, 1, 0);
   FEDERATE_BOOKKEEPING_INSTANCES(2);
 } MainSender;
 

include/reactor-uc/error.h

@@ -21,12 +21,6 @@ typedef enum {
   LF_INVALID_VALUE,
   LF_OUT_OF_BOUNDS,
   LF_NO_MEM,
-  LF_COULD_NOT_CONNECT,
-  LF_NETWORK_SETUP_FAILED,
-  LF_CONNECTION_CLOSED,
-  LF_TIMEOUT,
-  LF_TRY_AGAIN,
-  LF_IN_PROGRESS
 } lf_ret_t;
 
 /**

include/reactor-uc/federated.h

@@ -31,7 +31,6 @@ struct FederatedConnectionBundle {
   serialize_hook *serialize_hooks;
   size_t outputs_size;
   bool server; // Does this federate work as server or client
-  void (*network_channel_state_changed)(FederatedConnectionBundle *self);
 };
 
 void FederatedConnectionBundle_ctor(FederatedConnectionBundle *self, Reactor *parent, NetworkChannel *net_channel,

include/reactor-uc/network_channel.h

@@ -51,26 +51,12 @@ struct NetworkChannel {
 
   /**
    * @brief Opens the connection to the corresponding NetworkChannel on another federate (non-blocking).
-   * For client-server channels this usually is implemented as the "bind" call on the server side.
-   * @return LF_OK if connection is opened, LF_INVALID_VALUE if the channel is configured incorrectly,
-   * LF_NETWORK_SETUP_FAILED if the connection open operation fails.
+   * The channel is not connected unless @p get_connection_state returns with NETWORK_CHANNEL_STATE_CONNECTED.
+   * @return LF_OK if channel opened without error, LF_ERR if the channel is configured incorrectly or the connection
+   * open operation fails.
    */
   lf_ret_t (*open_connection)(NetworkChannel *self);
 
-  /**
-   * @brief Try to connect to corresponding NetworkChannel on another federate (non-blocking).
-   * @return LF_OK if connection is established, LF_IN_PROGRESS if connection is in progress, LF_TRY_AGAIN if connection
-   * failed and should be retried, LF_ERR if connection failed and should not be retried.
-   */
-  lf_ret_t (*try_connect)(NetworkChannel *self);
-
-  /**
-   * @brief Try to reconnect to corresponding NetworkChannel after the connection broke of (non-blocking).
-   * @return LF_OK if connection is established, LF_IN_PROGRESS if connection is in progress, LF_TRY_AGAIN if connection
-   * failed and should be retried, LF_ERR if connection failed and should not be retried.
-   */
-  lf_ret_t (*try_reconnect)(NetworkChannel *self);
-
   /**
    * @brief Closes the connection to the corresponding NetworkChannel on another federate.
    */

include/reactor-uc/platform/posix/tcp_ip_channel.h

@@ -7,6 +7,7 @@
 #include "proto/message.pb.h"
 #include "reactor-uc/error.h"
 #include "reactor-uc/network_channel.h"
+#include "reactor-uc/environment.h"
 
 #define TCP_IP_CHANNEL_BUFFERSIZE 1024
 #define TCP_IP_CHANNEL_NUM_RETRIES 255
@@ -45,6 +46,7 @@ struct TcpIpChannel {
   void (*receive_callback)(FederatedConnectionBundle *conn, const FederateMessage *message);
 };
 
-void TcpIpChannel_ctor(TcpIpChannel *self, const char *host, unsigned short port, int protocol_family, bool server);
+void TcpIpChannel_ctor(TcpIpChannel *self, Environment *env, const char *host, unsigned short port, int protocol_family,
+                       bool server);
 
 #endif

include/reactor-uc/platform/riot/coap_udp_ip_channel.h

@@ -21,6 +21,7 @@ struct CoapUdpIpChannel {
 
   sock_udp_ep_t remote;
 
+  bool send_ack_received;
   FederateMessage output;
   uint8_t write_buffer[COAP_UDP_IP_CHANNEL_BUFFERSIZE];
 

lfc/core/src/main/kotlin/org/lflang/generator/uc/UcPreambleGenerator.kt

@@ -10,5 +10,5 @@ class UcPreambleGenerator(
     private val reactor: Reactor,
 ) {
     fun generateReactorPrivatePreamble() = reactor.preambles.joinToString(prefix= "// Private preambles\n", separator = "\n", postfix = "\n") { it.code.toText()}
-    fun generateReactorPublicPreamble() = resource.model.preambles.joinToString(prefix = "// Public preambles\n", separator = "\n", postfix = "\n") {it.code.toText()}
+    fun generateReactorPublicPreamble() = resource.model.preambles.joinToString(prefix = "// Public preambles ", separator = "\n", postfix = "\n") {it.code.toText()}
 }

lfc/core/src/main/kotlin/org/lflang/generator/uc/UcReactionGenerator.kt

@@ -48,15 +48,18 @@ class UcReactionGenerator(private val reactor: Reactor) {
             return res
         }
 
+
     private val Reaction.allContainedEffectsTriggersAndSources
         get() = run {
             val res = mutableMapOf<Instantiation, List<VarRef>>()
-            for (varRef in allContainedEffects.plus(allContainedSources).plus(allContainedTriggers)) {
-                val varRef = varRef as VarRef
-                if (varRef.container!! !in res.keys) {
-                    res[varRef.container] = mutableListOf()
-                }
-                res[varRef.container] = res[varRef.container]!!.plus(varRef)
+            for (effect in allContainedEffects) {
+                res.getOrPut(effect.container!!) { mutableListOf(effect) }.plus(effect)
+            }
+            for (trigger in allContainedTriggers) {
+                res.getOrPut((trigger as VarRef).container!!) { mutableListOf(trigger) }.plus(trigger)
+            }
+            for (source in allContainedSources) {
+                res.getOrPut((source as VarRef).container!!) { mutableListOf(source) }.plus(source)
             }
             res
         }
@@ -235,7 +238,7 @@ class UcReactionGenerator(private val reactor: Reactor) {
         """|
            |// Generated struct providing access to ports of child reactor `${inst.name}`
            |struct _${inst.reactor.codeType}_${inst.name} {
-        ${"|  "..triggers.joinToString(separator = "\n") { generateContainedTriggerInScope(it) }}
+        ${"|  "..triggers.joinToString { generateContainedTriggerInScope(it) }}
            |};
            |struct _${inst.reactor.codeType}_${inst.name} ${inst.name};
         ${"|"..triggers.joinToString(separator = "\n") {generateContainedTriggerFieldInit("${inst.name}", it)}}

lfc/core/src/main/kotlin/org/lflang/generator/uc/UcReactorGenerator.kt

@@ -37,7 +37,7 @@ class UcReactorGenerator(private val reactor: Reactor, fileConfig: UcFileConfig,
     private val timers = UcTimerGenerator(reactor)
     private val actions = UcActionGenerator(reactor)
     private val reactions = UcReactionGenerator(reactor)
-    private val preambles = UcPreambleGenerator(reactor.eResource(), reactor)
+    private val preambles = UcPreambleGenerator(fileConfig.resource, reactor)
     private val instances =
         UcInstanceGenerator(reactor, parameters, ports, connections, reactions, fileConfig, messageReporter)