task manager RU working

common/utils/thread_pool/task_manager.h

@@ -4,8 +4,8 @@
 // Comment for deactivating ws tpool
 #define TASK_MANAGER
 #define TASK_MANAGER_DEMODULATION
-//#define TASK_MANAGER_CODING
-//#define TASK_MANAGER_RU
+#define TASK_MANAGER_CODING
+#define TASK_MANAGER_RU
 
 //#define TASK_MANAGER_UE
 //#define TASK_MANAGER_UE_DECODING

executables/nr-ru.c

@@ -1359,12 +1359,16 @@ void *ru_thread( void *param ) {
     syncMsg->slot_tx = proc->tti_tx;
     syncMsg->timestamp_tx = proc->timestamp_tx;
     res->key = proc->tti_rx;
-#ifdef TASK_MANAGER_RU
+#ifdef TASK_MANAGER // No TASK_MANAGER_RU
     if (!get_softmodem_params()->reorder_thread_disable) {
       assert(res->processingFunc != NULL);
       assert(res->reponseFifo != NULL);
-      res->processingFunc(NotifiedFifoData(res));
-      pushNotifiedFIFO(res->reponseFifo, res);
+      //assert(gNB->threadPool.incomingFifo == gNB->resp_L1);
+
+      pushNotifiedFIFO(&gNB->threadPool.incomingFifo, res);
+
+      //res->processingFunc(NotifiedFifoData(res));
+      //pushNotifiedFIFO(res->reponseFifo, res);
     } 
 #else
     if (!get_softmodem_params()->reorder_thread_disable) 
@@ -1876,10 +1880,7 @@ void init_NR_RU(configmodule_interface_t *cfg, char *rf_config_file)
       }
       LOG_I(PHY,"RU thread-pool core string %s\n",pool);
 #ifdef TASK_MANAGER_RU
-      int const log_cores = get_nprocs_conf();
-      assert(log_cores > 0);
-      // Assuming: Physical cores = Logical cores / 2
-      init_task_manager(&ru->man, log_cores/2); 
+      init_task_manager(&ru->man, ru->num_tpcores); 
 #else
       ru->threadPool = (tpool_t*)malloc(sizeof(tpool_t));
       initTpool(pool, ru->threadPool, cpumeas(CPUMEAS_GETSTATE));

openair1/PHY/CODING/nrLDPC_defs.h

@@ -67,7 +67,7 @@ typedef struct {
   /// LDPC-code outputs
   uint8_t *d[MAX_NUM_NR_DLSCH_SEGMENTS_PER_LAYER*NR_MAX_NB_LAYERS];
 #ifdef TASK_MANAGER_CODING
-  _Atomic int* task_done; 
+  task_status_t* task_done; 
 #endif
 } encoder_implemparams_t;
 

openair1/PHY/defs_RU.h

@@ -179,7 +179,7 @@ typedef struct {
  int endSymbol;
  int slot; 
 #ifdef TASK_MANAGER_RU
- _Atomic int* task_done; 
+ task_status_t* task_done; 
 #endif
 } feprx_cmd_t;
 
@@ -190,7 +190,7 @@ typedef struct {
  int startSymbol;
  int numSymbols;
 #ifdef TASK_MANAGER_RU
- _Atomic int* task_done; 
+ task_status_t* task_done; 
 #endif
 } feptx_cmd_t;
 

openair1/SCHED_NR/nr_ru_procedures.c

@@ -327,8 +327,8 @@ void nr_feptx(void *arg) {
   nr_feptx0(ru,slot,startSymbol,numSymbols,aa);
 
 #ifdef TASK_MANAGER_RU
-  assert(atomic_load(feptx->task_done) == 0);
-  atomic_store(feptx->task_done, 1);
+  //assert(atomic_load(&feptx->task_done->completed) == 0);
+  atomic_store_explicit(&feptx->task_done->completed, 1, memory_order_release);
 #endif
 }
 
@@ -346,8 +346,8 @@ void nr_feptx_tp(RU_t *ru, int frame_tx, int slot) {
 #ifdef TASK_MANAGER_RU
   size_t const sz = ru->nb_tx + (ru->half_slot_parallelization>0)*ru->nb_tx;
   feptx_cmd_t arr[sz]; 
-  _Atomic int tasks_done[sz];
-  memset(&tasks_done, 0, sizeof(_Atomic int) * sz);
+  task_status_t tasks_done[sz];
+  memset(&tasks_done, 0, sizeof(task_status_t) * sz);
 #endif
 
   for (int aid=0;aid<ru->nb_tx;aid++) {
@@ -395,9 +395,8 @@ void nr_feptx_tp(RU_t *ru, int frame_tx, int slot) {
   }
 
 #ifdef TASK_MANAGER_RU
-  stop_spining_task_manager(&ru->man);
-  trigger_all_task_manager(&ru->man);
-  wait_spin_all_atomics_one( nbfeptx, tasks_done);
+  if(nbfeptx > 0)
+    wait_task_status_completed(nbfeptx, tasks_done);
 #else
   while (nbfeptx>0) {
     notifiedFIFO_elt_t *req=pullTpool(ru->respfeptx, ru->threadPool);
@@ -437,8 +436,8 @@ void nr_fep(void* arg) {
   VCD_SIGNAL_DUMPER_DUMP_FUNCTION_BY_NAME(VCD_SIGNAL_DUMPER_FUNCTIONS_PHY_PROCEDURES_RU_FEPRX+aid, 0);
 
 #ifdef TASK_MANAGER_RU
-  assert(atomic_load(feprx_cmd->task_done) == 0);
-  atomic_store(feprx_cmd->task_done, 1);
+ // assert(atomic_load(feprx_cmd->task_done->completed) == 0);
+  atomic_store_explicit(&feprx_cmd->task_done->completed, 1, memory_order_release);
 #endif
 }
 
@@ -452,8 +451,8 @@ void nr_fep_tp(RU_t *ru, int slot)
 #ifdef TASK_MANAGER_RU
   size_t const sz = ru->nb_rx + (ru->half_slot_parallelization>0)*ru->nb_rx;
   feprx_cmd_t arr[sz]; 
-  _Atomic int tasks_done[sz];
-  memset(&tasks_done, 0, sizeof(_Atomic int) * sz);
+  task_status_t tasks_done[sz];
+  memset(&tasks_done, 0, sizeof(task_status_t) * sz);
 #endif
 
   for (int aid=0;aid<ru->nb_rx;aid++) {
@@ -501,9 +500,8 @@ void nr_fep_tp(RU_t *ru, int slot)
        }
   }
 #ifdef TASK_MANAGER_RU
-  stop_spining_task_manager(&ru->man);
-  trigger_all_task_manager(&ru->man);
-  wait_spin_all_atomics_one(nbfeprx, tasks_done);
+  if(nbfeprx > 0)
+    wait_task_status_completed(nbfeprx, tasks_done);
 #else
   while (nbfeprx>0) {
     notifiedFIFO_elt_t *req=pullTpool(ru->respfeprx, ru->threadPool);