Merge remote-tracking branch 'origin/fix-tpool-deadlock' into integration_2025_w02 (!3181)

Use UL actor for processSlotTx

Use UL actor instead of Thread pool for processSlotTX.

Closes #887

doc/nr-ue-design.md

@@ -13,7 +13,7 @@ The `UE_thread` function in `nr-ue.c` is the main top level thread that interact
 
 The UE exits when at any point in operation it gets out of synchronization. When the command line option `--non-stop` is used, the UE goes to 'Initial Synchronization' mode when it loses synchronization with gNB. However, this feature is not fully implemented and it is a work in progress at the time of writing this documentation. This will be the default behavior (not a command line option) when the feature is fully implemented.
 
-UE uses actors which are threads dedicated to particular activity. Sync Actor handles initial sync. DL Actors handle DLSCH PHY procedures. UL procedures are processed directly on the threadpool
+UE uses actors which are threads dedicated to particular activity. Sync Actor handles initial sync. DL Actors handle DLSCH PHY procedures. UL procedures are are run on the UL Actor
 
 ![design](nr-ue-threads.svg)
 

executables/nr-ue.c

@@ -103,10 +103,9 @@
 static void *NRUE_phy_stub_standalone_pnf_task(void *arg);
 
 static void start_process_slot_tx(void* arg) {
-  task_t task;
-  task.args = arg;
-  task.func = processSlotTX;
-  pushTpool(&(get_nrUE_params()->Tpool), task);
+  notifiedFIFO_elt_t *newTx = arg;
+  nr_rxtx_thread_data_t *curMsgTx = NotifiedFifoData(newTx);
+  pushNotifiedFIFO(&curMsgTx->UE->ul_actor.fifo, newTx);
 }
 
 static size_t dump_L1_UE_meas_stats(PHY_VARS_NR_UE *ue, char *output, size_t max_len)
@@ -563,13 +562,7 @@ void processSlotTX(void *arg)
       phy_procedures_nrUE_TX(UE, proc, &phy_data);
     }
   }
-
-  int slots_per_frame = (UE->sl_mode == 2) ? UE->SL_UE_PHY_PARAMS.sl_frame_params.slots_per_frame
-                                           : UE->frame_parms.slots_per_frame;
-  int next_slot_and_frame = proc->nr_slot_tx + 1  + proc->nr_slot_tx_offset + proc->frame_tx * slots_per_frame;
-  dynamic_barrier_join(&UE->process_slot_tx_barriers[next_slot_and_frame % NUM_PROCESS_SLOT_TX_BARRIERS]);
   RU_write(rxtxD, sl_tx_action);
-  free(rxtxD);
   TracyCZoneEnd(ctx);
 }
 
@@ -609,7 +602,7 @@ static int handle_sync_req_from_mac(PHY_VARS_NR_UE *UE)
     for (int i = 0; i < NUM_DL_ACTORS; i++) {
       flush_actor(UE->dl_actors + i);
     }
-    /*TODO: Flush UL jobs */
+    flush_actor(&UE->ul_actor);
 
     clean_UE_harq(UE);
     UE->is_synchronized = 0;
@@ -1082,22 +1075,22 @@ void *UE_thread(void *arg)
 
     // Start TX slot processing here. It runs in parallel with RX slot processing
     // in current code, DURATION_RX_TO_TX constant is the limit to get UL data to encode from a RX slot
-    nr_rxtx_thread_data_t *curMsgTx = calloc(1, sizeof(*curMsgTx));
+    notifiedFIFO_elt_t *newTx = newNotifiedFIFO_elt(sizeof(nr_rxtx_thread_data_t), 0, 0, processSlotTX);
+    nr_rxtx_thread_data_t *curMsgTx = NotifiedFifoData(newTx);
+    memset(curMsgTx, 0, sizeof(*curMsgTx));
     curMsgTx->proc = curMsg.proc;
     curMsgTx->writeBlockSize = writeBlockSize;
     curMsgTx->proc.timestamp_tx = writeTimestamp;
     curMsgTx->UE = UE;
-    curMsgTx->stream_status = stream_status;
     curMsgTx->proc.nr_slot_tx_offset = nr_slot_tx_offset;
 
-    int sync_to_previous_thread = stream_status == STREAM_STATUS_SYNCED ? 1 : 0;
     int slot = curMsgTx->proc.nr_slot_tx;
     int slot_and_frame = slot + curMsgTx->proc.frame_tx * UE->frame_parms.slots_per_frame;
 
     dynamic_barrier_update(&UE->process_slot_tx_barriers[slot_and_frame % NUM_PROCESS_SLOT_TX_BARRIERS],
-                           tx_wait_for_dlsch[slot] + sync_to_previous_thread,
+                           tx_wait_for_dlsch[slot],
                            start_process_slot_tx,
-                           curMsgTx);
+                           newTx);
     stream_status = STREAM_STATUS_SYNCED;
     tx_wait_for_dlsch[slot] = 0;
     // apply new duration next run to avoid thread dead lock

executables/nr-uesoftmodem.c

@@ -497,6 +497,7 @@ int main(int argc, char **argv)
         for (int i = 0; i < NUM_DL_ACTORS; i++) {
           init_actor(&UE[CC_id]->dl_actors[i], "DL_", -1);
         }
+        init_actor(&UE[CC_id]->ul_actor, "UL_", -1);
         init_nr_ue_vars(UE[CC_id], inst);
 
         if (UE[CC_id]->sl_mode) {

openair1/PHY/defs_nr_UE.h

@@ -548,9 +548,8 @@ typedef struct PHY_VARS_NR_UE_s {
   sl_nr_ue_phy_params_t SL_UE_PHY_PARAMS;
   Actor_t sync_actor;
   Actor_t dl_actors[NUM_DL_ACTORS];
-  
+  Actor_t ul_actor;
   ntn_config_message_t* ntn_config_message;
-
 } PHY_VARS_NR_UE;
 
 typedef struct {
@@ -632,7 +631,6 @@ typedef struct nr_rxtx_thread_data_s {
   nr_phy_data_t phy_data;
   int tx_wait_for_dlsch;
   int rx_offset;
-  enum stream_status_e stream_status;
 } nr_rxtx_thread_data_t;
 
 typedef struct LDPCDecode_ue_s {