Stable index for _fed.inbound_net_listeners

Author: ChadliaJerad

core/federated/federate.c

@@ -1905,7 +1905,10 @@ void lf_terminate_execution(environment_t* env) {
   if (_fed.number_of_inbound_p2p_connections > 0 && _fed.inbound_net_listeners != NULL) {
     LF_PRINT_LOG("Waiting for %zu threads listening for incoming messages to exit.",
                  _fed.number_of_inbound_p2p_connections);
-    for (size_t i = 0; i < _fed.number_of_inbound_p2p_connections; i++) {
+    // inbound_net_listeners is indexed by federate ID (size NUMBER_OF_FEDERATES).
+    // Entries for federates that never connected remain zero-initialized; lf_thread_join
+    // on a zero handle returns an error which we ignore.
+    for (int i = 0; i < NUMBER_OF_FEDERATES; i++) {
       // Ignoring errors here.
       lf_thread_join(_fed.inbound_net_listeners[i], NULL);
     }
@@ -2279,8 +2282,13 @@ void lf_enqueue_port_absent_reactions(environment_t* env) {
 void* lf_handle_p2p_connections_from_federates(void* env_arg) {
   LF_ASSERT_NON_NULL(env_arg);
   size_t received_federates = 0;
-  // Allocate memory to store thread IDs.
-  _fed.inbound_net_listeners = (lf_thread_t*)calloc(_fed.number_of_inbound_p2p_connections, sizeof(lf_thread_t));
+  // Allocate memory to store thread IDs, indexed by federate ID.
+  // Using NUMBER_OF_FEDERATES (not number_of_inbound_p2p_connections) so that transient
+  // federates that reconnect can reuse their slot without overflowing the array.
+  _fed.inbound_net_listeners = (lf_thread_t*)calloc(NUMBER_OF_FEDERATES, sizeof(lf_thread_t));
+  // Track which federate IDs have connected at least once, to count unique connections.
+  bool seen_fed[NUMBER_OF_FEDERATES];
+  memset(seen_fed, 0, sizeof(seen_fed));
   while (!_lf_termination_executed) {
     // Case where all inbound connections are to persistent federates
     if (received_federates == _fed.number_of_inbound_p2p_connections && _fed.number_of_inbound_p2p_transients == 0) {
@@ -2369,7 +2377,14 @@ void* lf_handle_p2p_connections_from_federates(void* env_arg) {
     // Start a thread to listen for incoming messages from other federates.
     // The fed_id is a uint16_t, which we assume can be safely cast to and from void*.
     void* fed_id_arg = (void*)(uintptr_t)remote_fed_id;
-    int result = lf_thread_create(&_fed.inbound_net_listeners[received_federates], listen_to_federates, fed_id_arg);
+    // For a transient federate that is reconnecting, join its previous listener thread
+    // before overwriting the slot, to avoid leaking the thread handle.
+    if (remote_fed_is_transient && seen_fed[remote_fed_id]) {
+      lf_thread_join(_fed.inbound_net_listeners[remote_fed_id], NULL);
+    }
+    // Index by remote_fed_id so that reconnecting transient federates reuse the same slot
+    // rather than incrementing past the end of the array.
+    int result = lf_thread_create(&_fed.inbound_net_listeners[remote_fed_id], listen_to_federates, fed_id_arg);
     if (result != 0) {
       // Failed to create a listening thread.
       shutdown_net(_fed.net_for_inbound_p2p_connections[remote_fed_id], false);
@@ -2379,7 +2394,11 @@ void* lf_handle_p2p_connections_from_federates(void* env_arg) {
     }
     LF_MUTEX_UNLOCK(&lf_outbound_net_mutex);
 
-    received_federates++;
+    // Count each unique federate only once (not each reconnection).
+    if (!seen_fed[remote_fed_id]) {
+      seen_fed[remote_fed_id] = true;
+      received_federates++;
+    }
   }
 
   LF_PRINT_LOG("All %zu remote federates are connected.", _fed.number_of_inbound_p2p_connections);

include/core/federated/federate.h

@@ -63,9 +63,11 @@ typedef struct federate_instance_t {
   size_t number_of_inbound_p2p_transients;
 
   /**
-   * Array of thread IDs for threads that listen for incoming messages.
-   * This is NULL if there are none and otherwise has size given by
-   * number_of_inbound_p2p_connections.
+   * Array of thread IDs for threads that listen for incoming messages,
+   * indexed by federate ID. This is NULL if there are none and otherwise
+   * has size NUMBER_OF_FEDERATES. Using federate ID as the index allows
+   * transient federates that reconnect to reuse the same slot without
+   * overflowing the array.
    */
   lf_thread_t* inbound_net_listeners;