Network Flush Reactors embedded in FederatedOutputConnection (#330)

tanneberger · web-flow · commit c1dbc593ff31 · 2026-02-20T14:24:30.000+01:00
* reactor-uc: reworking init

* initial commit of rework

* small fix

* small fix

* add commets

* runtime path

* removing test-case
diff --git a/examples/posix/federated/receiver.c b/examples/posix/federated/receiver.c
@@ -78,7 +78,7 @@ typedef struct {
   Reactor super;
   LF_CHILD_REACTOR_INSTANCE(Receiver, receiver, 1);
   LF_FEDERATED_CONNECTION_BUNDLE_INSTANCE(Receiver, Sender);
-  LF_FEDERATE_BOOKKEEPING_INSTANCES(1);
+  LF_FEDERATE_BOOKKEEPING_INSTANCES(1, 0);
   LF_CHILD_INPUT_SOURCES(receiver, in, 1, 1, 0);
   LF_DEFINE_STARTUP_COORDINATOR(Federate);
   LF_DEFINE_CLOCK_SYNC(Federate);
diff --git a/examples/posix/federated/sender.c b/examples/posix/federated/sender.c
@@ -17,7 +17,7 @@ int serialize_msg_t(const void *user_struct, size_t user_struct_size, unsigned c
   const msg_t *msg = user_struct;
 
   memcpy(msg_buf, &msg->size, sizeof(msg->size));
-  memcpy(msg_buf + sizeof(msg->size), msg->msg, msg->size);
+  memcpy(msg_buf + sizeof(msg->size), msg->msg, user_struct_size);
 
   return sizeof(msg->size) + msg->size;
 }
@@ -90,7 +90,7 @@ typedef struct {
   Reactor super;
   LF_CHILD_REACTOR_INSTANCE(Sender, sender, 1);
   LF_FEDERATED_CONNECTION_BUNDLE_INSTANCE(Sender, Receiver);
-  LF_FEDERATE_BOOKKEEPING_INSTANCES(1);
+  LF_FEDERATE_BOOKKEEPING_INSTANCES(1, 1);
   LF_CHILD_OUTPUT_CONNECTIONS(sender, out, 1, 1, 1);
   LF_CHILD_OUTPUT_EFFECTS(sender, out, 1, 1, 0);
   LF_CHILD_OUTPUT_OBSERVERS(sender, out, 1, 1, 0);
diff --git a/examples/riot/coap_federated/receiver/main.c b/examples/riot/coap_federated/receiver/main.c
@@ -78,7 +78,7 @@ typedef struct {
   Reactor super;
   LF_CHILD_REACTOR_INSTANCE(Receiver, receiver, 1);
   LF_FEDERATED_CONNECTION_BUNDLE_INSTANCE(Receiver, Sender);
-  LF_FEDERATE_BOOKKEEPING_INSTANCES(1);
+  LF_FEDERATE_BOOKKEEPING_INSTANCES(1, 0);
   LF_CHILD_INPUT_SOURCES(receiver, in, 1, 1, 0);
   LF_DEFINE_STARTUP_COORDINATOR(Federate);
   LF_DEFINE_CLOCK_SYNC(Federate);
diff --git a/examples/riot/coap_federated/sender/main.c b/examples/riot/coap_federated/sender/main.c
@@ -85,7 +85,7 @@ typedef struct {
   Reactor super;
   LF_CHILD_REACTOR_INSTANCE(Sender, sender, 1);
   LF_FEDERATED_CONNECTION_BUNDLE_INSTANCE(Sender, Receiver);
-  LF_FEDERATE_BOOKKEEPING_INSTANCES(1);
+  LF_FEDERATE_BOOKKEEPING_INSTANCES(1, 1);
   LF_CHILD_OUTPUT_CONNECTIONS(sender, out, 1, 1, 1);
   LF_CHILD_OUTPUT_EFFECTS(sender, out, 1, 1, 0);
   LF_CHILD_OUTPUT_OBSERVERS(sender, out, 1, 1, 0);
diff --git a/examples/zephyr/basic_federated/common/receiver.h b/examples/zephyr/basic_federated/common/receiver.h
@@ -91,7 +91,7 @@ typedef struct {
   Reactor super;
   LF_CHILD_REACTOR_INSTANCE(Receiver, receiver, 1);
   LF_FEDERATED_CONNECTION_BUNDLE_INSTANCE(Receiver, Sender);
-  LF_FEDERATE_BOOKKEEPING_INSTANCES(1);
+  LF_FEDERATE_BOOKKEEPING_INSTANCES(1, 0);
   LF_CHILD_INPUT_SOURCES(receiver, in, 1, 1, 0);
   LF_DEFINE_STARTUP_COORDINATOR(Federate);
   LF_DEFINE_CLOCK_SYNC(Federate);
diff --git a/examples/zephyr/basic_federated/federated_sender/src/sender.c b/examples/zephyr/basic_federated/federated_sender/src/sender.c
@@ -148,7 +148,7 @@ typedef struct {
   LF_CHILD_OUTPUT_CONNECTIONS(sender, out, 1, 1, 2);
   LF_CHILD_OUTPUT_EFFECTS(sender, out, 1, 1, 0);
   LF_CHILD_OUTPUT_OBSERVERS(sender, out,1, 1, 0);
-  LF_FEDERATE_BOOKKEEPING_INSTANCES(2);
+  LF_FEDERATE_BOOKKEEPING_INSTANCES(2, 2);
   LF_DEFINE_STARTUP_COORDINATOR(Federate);
   LF_DEFINE_CLOCK_SYNC(Federate);
 } MainSender;
diff --git a/include/reactor-uc/federated.h b/include/reactor-uc/federated.h
@@ -8,6 +8,7 @@
 typedef struct FederatedConnectionBundle FederatedConnectionBundle;
 typedef struct FederatedOutputConnection FederatedOutputConnection;
 typedef struct FederatedInputConnection FederatedInputConnection;
+typedef struct FederatedFlushReactor FederatedFlushReactor;
 typedef struct NetworkChannel NetworkChannel;
 
 /**
@@ -63,6 +64,23 @@ void FederatedConnectionBundle_ctor(FederatedConnectionBundle* self, Reactor* pa
                                     FederatedInputConnection** inputs, deserialize_hook* deserialize_hooks,
                                     size_t inputs_size, FederatedOutputConnection** outputs,
                                     serialize_hook* serialize_hooks, size_t outputs_size, size_t index);
+/**
+ * @brief This reactor is part of the FederatedOutputConnection and has the purpose of flushing and sending
+ * the value transmitted by the last downstream port/reaction.
+ */
+struct FederatedFlushReactor {
+  Reactor super;
+  Port input_port;
+  Reaction flush_reaction;
+  Trigger* flush_triggers;
+  Reaction* reaction_array;
+};
+
+/**
+ * @brief Instantiates the FederatedFlushReactor object by initializing the reaction, input port, and reactor.
+ */
+void FederatedFlushReactor_ctor(FederatedFlushReactor* self, Reactor* parent, void* payload_buf, size_t payload_size,
+                                FederatedOutputConnection* connection);
 
 /**
  * @brief A single output connection from this federate to another federate.
@@ -74,16 +92,14 @@ void FederatedConnectionBundle_ctor(FederatedConnectionBundle* self, Reactor* pa
 struct FederatedOutputConnection {
   Connection super; // Inherits from Connection, it wastes some memory but makes for a nicer architecture.
   FederatedConnectionBundle* bundle; // A pointer to the super it is within
-  EventPayloadPool payload_pool;     // Output buffer
-  void* staged_payload_ptr;
   int conn_id;
+  FederatedFlushReactor flush_reactor;
 };
 
 void FederatedConnectionBundle_validate(FederatedConnectionBundle* bundle);
 
 void FederatedOutputConnection_ctor(FederatedOutputConnection* self, Reactor* parent, FederatedConnectionBundle* bundle,
-                                    int conn_id, void* payload_buf, bool* payload_used_buf, size_t payload_size,
-                                    size_t payload_buf_capacity);
+                                    int conn_id, void* payload_buf, size_t payload_size);
 
 /**
  * @brief A single input connection coming from another federate.
diff --git a/include/reactor-uc/macros_internal.h b/include/reactor-uc/macros_internal.h
@@ -118,8 +118,8 @@
 #define LF_FEDERATE_CONNECTION_BUNDLE_INSTANCE(ReactorName, OtherName)                                                 \
   ReactorName##_##OtherName##_Bundle ReactorName##_##OtherName_bundle
 
-#define LF_FEDERATE_BOOKKEEPING_INSTANCES(NumBundles)                                                                  \
-  LF_REACTOR_BOOKKEEPING_INSTANCES(0, 0, 1);                                                                           \
+#define LF_FEDERATE_BOOKKEEPING_INSTANCES(NumBundles, NumOutputs)                                                      \
+  LF_REACTOR_BOOKKEEPING_INSTANCES(0, 0, 1 + NumOutputs);                                                              \
   FederatedConnectionBundle* _bundles[NumBundles];
 
 #define LF_DEFINE_OUTPUT_STRUCT(ReactorName, PortName, SourceSize, BufferType)                                         \
@@ -472,21 +472,19 @@ typedef struct FederatedOutputConnection FederatedOutputConnection;
   typedef struct {                                                                                                     \
     FederatedOutputConnection super;                                                                                   \
     BufferType payload_buf[1];                                                                                         \
-    bool payload_used_buf[1];                                                                                          \
   } ReactorName##_##OutputName##_conn;
 
 #define LF_DEFINE_FEDERATED_OUTPUT_CONNECTION_STRUCT_ARRAY(ReactorName, OutputName, BufferType, ArrayLength)           \
   typedef struct {                                                                                                     \
     FederatedOutputConnection super;                                                                                   \
     BufferType payload_buf[1][(ArrayLength)];                                                                          \
-    bool payload_used_buf[1];                                                                                          \
   } ReactorName##_##OutputName##_conn;
 
 #define LF_DEFINE_FEDERATED_OUTPUT_CONNECTION_CTOR(ReactorName, OutputName, BufferType, DestinationConnId)             \
   void ReactorName##_##OutputName##_conn_ctor(ReactorName##_##OutputName##_conn* self, Reactor* parent,                \
                                               FederatedConnectionBundle* bundle) {                                     \
     FederatedOutputConnection_ctor(&self->super, parent, bundle, DestinationConnId, (void*)&self->payload_buf,         \
-                                   (bool*)&self->payload_used_buf, sizeof(self->payload_buf[0]), 1);                   \
+                                   sizeof(self->payload_buf[0]));                                                      \
   }
 
 #define LF_FEDERATED_OUTPUT_CONNECTION_INSTANCE(ReactorName, OutputName) ReactorName##_##OutputName##_conn OutputName
@@ -516,7 +514,11 @@ typedef struct FederatedOutputConnection FederatedOutputConnection;
 
 #define LF_INITIALIZE_FEDERATED_CONNECTION_BUNDLE(ReactorName, OtherName)                                              \
   ReactorName##_##OtherName##_Bundle_ctor(&self->ReactorName##_##OtherName##_bundle, &self->super, _bundle_idx);       \
-  self->_bundles[_bundle_idx++] = &self->ReactorName##_##OtherName##_bundle.super;
+  self->_bundles[_bundle_idx] = &self->ReactorName##_##OtherName##_bundle.super;                                       \
+  for (int j = 0; j < self->_bundles[_bundle_idx]->outputs_size; j++) {                                                \
+    self->_children[_child_idx++] = &self->_bundles[_bundle_idx]->outputs[j]->flush_reactor.super;                     \
+  }                                                                                                                    \
+  _bundle_idx++;
 
 #define LF_INITIALIZE_FEDERATED_OUTPUT_CONNECTION(ReactorName, OutputName, SerializeFunc)                              \
   ReactorName##_##OutputName##_conn_ctor(&self->OutputName, self->super.parent, &self->super);                         \
diff --git a/include/reactor-uc/trigger.h b/include/reactor-uc/trigger.h
@@ -14,16 +14,16 @@ typedef struct Trigger Trigger;
  * and output ports.
  */
 typedef enum {
-  TRIG_TIMER,
-  TRIG_ACTION,
-  TRIG_INPUT,
-  TRIG_OUTPUT,
-  TRIG_CONN,
-  TRIG_CONN_DELAYED,
-  TRIG_CONN_FEDERATED_INPUT,
-  TRIG_CONN_FEDERATED_OUTPUT,
-  TRIG_STARTUP,
-  TRIG_SHUTDOWN
+  TRIG_TIMER = 0,
+  TRIG_ACTION = 1,
+  TRIG_INPUT = 2,
+  TRIG_OUTPUT = 3,
+  TRIG_CONN = 4,
+  TRIG_CONN_DELAYED = 5,
+  TRIG_CONN_FEDERATED_INPUT = 6,
+  TRIG_CONN_FEDERATED_OUTPUT = 7,
+  TRIG_STARTUP = 8,
+  TRIG_SHUTDOWN = 9
 } TriggerType;
 
 /**
diff --git a/lfc/core/src/main/kotlin/org/lflang/generator/uc/UcConnectionGenerator.kt b/lfc/core/src/main/kotlin/org/lflang/generator/uc/UcConnectionGenerator.kt
@@ -259,6 +259,9 @@ class UcConnectionGenerator(
     allGroupedConnections.forEachIndexed { idx, el -> el.assignUid(idx) }
   }
 
+  fun getNumOutputs(federate: UcFederate) =
+      federatedConnectionBundles.sumOf { it.numOutputs(federate) }
+
   fun getNumFederatedConnectionBundles() = federatedConnectionBundles.size
 
   fun getNumConnectionsFromPort(instantiation: Instantiation?, port: Port): Int {
diff --git a/lfc/core/src/main/kotlin/org/lflang/generator/uc/UcFederateGenerator.kt b/lfc/core/src/main/kotlin/org/lflang/generator/uc/UcFederateGenerator.kt
@@ -40,6 +40,10 @@ class UcFederateGenerator(
     return connections.getMaxNumPendingEvents()
   }
 
+  fun getNumberOfOutputConnections(): Int {
+    return connections.getNumOutputs(currentFederate)
+  }
+
   private fun generateFederateStruct() =
       with(PrependOperator) {
         """
@@ -51,7 +55,7 @@ class UcFederateGenerator(
             |  // Startup and clock sync objects. 
         ${" |  "..startupCooordinator.generateFederateStructField()}
         ${" |  "..clockSync.generateFederateStructField()}
-            |  LF_FEDERATE_BOOKKEEPING_INSTANCES(${connections.getNumFederatedConnectionBundles()})
+            |  LF_FEDERATE_BOOKKEEPING_INSTANCES(${connections.getNumFederatedConnectionBundles()}, ${connections.getNumOutputs(currentFederate)})
             |} ${currentFederate.codeType};
             |
             """
diff --git a/lfc/core/src/main/kotlin/org/lflang/generator/uc/UcGenerator.kt b/lfc/core/src/main/kotlin/org/lflang/generator/uc/UcGenerator.kt
@@ -48,6 +48,7 @@ abstract class UcGenerator(
   // Contains the maximum number of pending events required by each reactor.
   // Is updated as reactors are analyzed and code-generated.
   val maxNumPendingEvents = mutableMapOf<Reactor, Int>()
+  val numberOfOutputConnections = mutableMapOf<Reactor, Int>()
 
   // Compute the total number of events and reactions within an instance (and its children)
   // Also returns whether there is any startup event within the instance.
diff --git a/lfc/core/src/main/kotlin/org/lflang/generator/uc/UcGeneratorFederated.kt b/lfc/core/src/main/kotlin/org/lflang/generator/uc/UcGeneratorFederated.kt
@@ -28,11 +28,12 @@ class UcGeneratorFederated(context: LFGeneratorContext, scopeProvider: LFGlobalS
   // computation for the non-federated case, but federated connections are also considered.
   fun totalNumEventsAndReactionsFederated(federate: UcFederate): Pair<Int, Int> {
     val eventsFromFederatedConnections = maxNumPendingEvents[mainDef.reactor]!!
+    val unaccountedNumberOfFlushReactions = numberOfOutputConnections[mainDef.reactor]!!
     val eventsAndReactionsInFederate =
         nonFederatedGenerator.totalNumEventsAndReactions(federate.inst.reactor)
     return Pair(
         eventsFromFederatedConnections + eventsAndReactionsInFederate.first,
-        eventsAndReactionsInFederate.second)
+        unaccountedNumberOfFlushReactions + eventsAndReactionsInFederate.second)
   }
 
   private fun doGenerateFederate(
@@ -178,6 +179,7 @@ class UcGeneratorFederated(context: LFGeneratorContext, scopeProvider: LFGlobalS
 
     // Record the number of events and reactions in this reactor
     maxNumPendingEvents[top] = generator.getMaxNumPendingEvents()
+    numberOfOutputConnections[top] = generator.getNumberOfOutputConnections()
 
     val headerFile = Paths.get("lf_federate.h")
     val sourceFile = Paths.get("lf_federate.c")
diff --git a/src/connection.c b/src/connection.c
@@ -7,7 +7,7 @@
 
 Port* Connection_get_final_upstream(Connection* self) {
 
-  if (!self->upstream || self->super.type != TRIG_CONN) {
+  if (!self->upstream || (self->super.type != TRIG_CONN && self->super.type != TRIG_CONN_FEDERATED_OUTPUT)) {
     return NULL;
   }
 
@@ -30,7 +30,7 @@ Port* Connection_get_final_upstream(Connection* self) {
 }
 
 void Connection_register_downstream(Connection* self, Port* port) {
-  LF_DEBUG(CONN, "Registering downstream %p with connection %p", port, self);
+  LF_DEBUG(CONN, "Registering downstream %s.%p with connection %s.%p", port, self);
   validate(self->downstreams_registered < self->downstreams_size);
 
   self->downstreams[self->downstreams_registered++] = port;
@@ -58,9 +58,9 @@ void LogicalConnection_trigger_downstreams(Connection* self, tag_t intended_tag,
       }
     }
 
-    for (size_t i = 0; i < down->conns_out_registered; i++) {
-      LF_DEBUG(CONN, "Found further downstream connection %p to recurse down", down->conns_out[i]);
-      down->conns_out[i]->trigger_downstreams(down->conns_out[i], intended_tag, value, value_size);
+    for (size_t j = 0; j < down->conns_out_registered; j++) {
+      LF_DEBUG(CONN, "Found further downstream connection %p to recurse down", down->conns_out[j]);
+      down->conns_out[j]->trigger_downstreams(down->conns_out[j], intended_tag, value, value_size);
     }
   }
 }
diff --git a/src/federated.c b/src/federated.c
diff --git a/src/schedulers/dynamic/scheduler.c b/src/schedulers/dynamic/scheduler.c
diff --git a/test/lf/src/FederatedConnection.lf b/test/lf/src/FederatedConnection.lf
diff --git a/test/platform/riot/coap_channel_federated_test/receiver/main.c b/test/platform/riot/coap_channel_federated_test/receiver/main.c
diff --git a/test/platform/riot/coap_channel_federated_test/sender/main.c b/test/platform/riot/coap_channel_federated_test/sender/main.c

Original file line number	Diff line number	Diff line change
`@@ -259,6 +259,9 @@ class UcConnectionGenerator(`
`259`	`259`	`allGroupedConnections.forEachIndexed { idx, el -> el.assignUid(idx) }`
`260`	`260`	`}`
`261`	`261`
	`262`	`+ fun getNumOutputs(federate: UcFederate) =`
	`263`	`+ federatedConnectionBundles.sumOf { it.numOutputs(federate) }`
	`264`	`+`
`262`	`265`	`fun getNumFederatedConnectionBundles() = federatedConnectionBundles.size`
`263`	`266`
`264`	`267`	`fun getNumConnectionsFromPort(instantiation: Instantiation?, port: Port): Int {`