Draft: Introduce the next downstream tag (NDT) to optimize the communication in centralized federated execution

core/environment.c

@@ -106,14 +106,19 @@ static void environment_init_modes(environment_t* env, int num_modes, int num_st
  * @brief Initialize the federation-specific parts of the environment struct.
  */
 static void environment_init_federated(environment_t* env, int num_is_present_fields) {
+#ifdef FEDERATED_NDT_ENABLED
+    // FIXME: Create a queue saving tags instead of events. For now, ndt_q stores 
+    // dummy events.
+    env->ndt_q = pqueue_tag_init(10);
+#endif // FEDERATED_NDT_ENABLED
 #ifdef FEDERATED_DECENTRALIZED
     env->_lf_intended_tag_fields = (tag_t**) calloc(num_is_present_fields, sizeof(tag_t*));
     LF_ASSERT(env->_lf_intended_tag_fields, "Out of memory");
     env->_lf_intended_tag_fields_size = num_is_present_fields;
 #endif
 }
 
-void environment_init_tags( environment_t *env, instant_t start_time, interval_t duration) {
+void environment_init_tags(environment_t *env, instant_t start_time, interval_t duration) {
     env->current_tag = (tag_t){.time = start_time, .microstep = 0u};
 
     tag_t stop_tag = FOREVER_TAG_INITIALIZER;
@@ -165,6 +170,9 @@ void environment_free(environment_t* env) {
     pqueue_free(env->event_q);
     pqueue_free(env->recycle_q);
     pqueue_free(env->next_q);
+    #ifdef FEDERATED_NDT_ENABLED
+    pqueue_tag_free(env->ndt_q);
+    #endif // FEDERATED_NDT_ENABLED
 
     environment_free_threaded(env);
     environment_free_single_threaded(env);

core/federated/RTI/main.c

@@ -99,6 +99,8 @@ void usage(int argc, const char* argv[]) {
     lf_print("          clock sync attempt (default is 10). Applies to 'init' and 'on'.\n");
     lf_print("  -a, --auth Turn on HMAC authentication options.\n");
     lf_print("  -t, --tracing Turn on tracing.\n");
+    lf_print("  -v, --version The minimum required version of Lingua Franca.");
+    lf_print("      --ndt Turn on ndt optimization.\n");
 
     lf_print("Command given:");
     for (int i = 0; i < argc; i++) {
@@ -171,7 +173,10 @@ int process_clock_sync_args(int argc, const char* argv[]) {
 
 int process_args(int argc, const char* argv[]) {
     for (int i = 1; i < argc; i++) {
-        if (strcmp(argv[i], "-i") == 0 || strcmp(argv[i], "--id") == 0) {
+        if (strcmp(argv[i], "-v") == 0 || strcmp(argv[i], "--version") == 0) {
+            lf_print("%s", version_info);
+            return 0;
+        } else if (strcmp(argv[i], "-i") == 0 || strcmp(argv[i], "--id") == 0) {
             if (argc < i + 2) {
                 lf_print_error("--id needs a string argument.");
                 usage(argc, argv);
@@ -232,14 +237,16 @@ int process_args(int argc, const char* argv[]) {
             rti.authentication_enabled = true;
         } else if (strcmp(argv[i], "-t") == 0 || strcmp(argv[i], "--tracing") == 0) {
             rti.base.tracing_enabled = true;
+        } else if (strcmp(argv[i], "--ndt") == 0) {
+            rti.ndt_enabled = true;
         } else if (strcmp(argv[i], " ") == 0) {
             // Tolerate spaces
             continue;
         }  else {
            lf_print_error("Unrecognized command-line argument: %s", argv[i]);
            usage(argc, argv);
            return 0;
-       }
+        }
     }
     if (rti.base.number_of_scheduling_nodes == 0) {
         lf_print_error("--number_of_federates needs a valid positive integer argument.");

core/federated/RTI/rti_common.c

@@ -56,6 +56,8 @@ void initialize_scheduling_node(scheduling_node_t* e, uint16_t id) {
     e->downstream = NULL;
     e->num_downstream = 0;
     e->mode = REALTIME;
+    e->has_physical_action = false;
+    e->enable_ndt = false;
     invalidate_min_delays_upstream(e);
 }
 
@@ -298,6 +300,25 @@ static void _update_min_delays_upstream(scheduling_node_t* end, scheduling_node_
     }
 }
 
+bool check_physical_action_of_transitive_downstreams(scheduling_node_t* e, bool visited[]) {
+    if (visited[e->id] || e->state == NOT_CONNECTED) {
+        return false;
+    }
+
+    visited[e->id] = true;
+
+    for (int i = 0; i < e->num_downstream; i++) {
+        if (check_physical_action_of_transitive_downstreams(rti_common->scheduling_nodes[e->downstream[i]], visited)) {
+            return true;
+        }
+    }
+    if (e->has_physical_action) {
+        return true;
+    } else {
+        return false;
+    }
+}
+
 void update_min_delays_upstream(scheduling_node_t* node) {
     // Check whether cached result is valid.
     if (node->min_delays == NULL) {

core/federated/RTI/rti_common.h

@@ -65,6 +65,9 @@ typedef struct scheduling_node_t {
     int* downstream;                    // Array of downstream scheduling node ids.
     int num_downstream;                 // Size of the array of downstream scheduling nodes.
     execution_mode_t mode;              // FAST or REALTIME.
+    bool is_in_cycle;
+    bool has_physical_action;
+    bool enable_ndt;
     minimum_delay_t* min_delays;        // Array of minimum delays from upstream nodes, not including this node.
     size_t num_min_delays;              // Size of min_delays array.
     int flags;                          // Or of IS_IN_ZERO_DELAY_CYCLE, IS_IN_CYCLE
@@ -273,5 +276,9 @@ void invalidate_min_delays_upstream(scheduling_node_t* node);
  */
 void free_scheduling_nodes(scheduling_node_t** scheduling_nodes, uint16_t number_of_scheduling_nodes);
 
+bool check_cycle(scheduling_node_t* e, int target_id, bool visited[]);
+
+bool check_physical_action_of_transitive_downstreams(scheduling_node_t* e, bool visited[]);
+
 #endif // RTI_COMMON_H
 #endif // STANDALONE_RTI || LF_ENCLAVES