Change the amount of threads added in the RU TP

common/utils/thread_pool/task_manager.c

@@ -335,7 +335,7 @@ label:
     int rc = pthread_mutex_unlock(&q->mtx);
     assert(rc == 0);
 
-    int val = *q->futex; // atomic_load_explicit(q->futex, memory_order_acquire);
+    int val = atomic_load_explicit(q->futex, memory_order_acquire);
     long r = syscall(SYS_futex, q->futex, FUTEX_WAIT_PRIVATE, val, NULL, 0);
     assert(r != -1);
     goto label;
@@ -469,7 +469,7 @@ void init_task_manager(task_manager_t* man, uint32_t num_threads)
 
   man->q_arr = calloc(num_threads, sizeof(not_q_t));
   assert(man->q_arr != NULL && "Memory exhausted");
-  man->futex = 0; 
+  atomic_store_explicit(&man->futex, 0, memory_order_seq_cst); 
 
  // man->waiting = false;
 
@@ -625,7 +625,7 @@ void trigger_all_task_manager(task_manager_t* man)
 {
   assert(man != NULL);
   //  FUTEX_WAKE_PRIVATE
-  man->futex = 0;
+  atomic_store_explicit(&man->futex, 0, memory_order_seq_cst);
   long r = syscall(SYS_futex, &man->futex, FUTEX_WAKE_PRIVATE, INT_MAX, NULL, NULL, 0);
   //printf("Number of threads woken %ld \n", r);
   if(r == -1){
@@ -657,21 +657,17 @@ void wait_task_status_completed(size_t len, task_status_t* arr)
   // We are believing Fedor
   const struct timespec ns = {0,1};
   int i = 0;
-  for(;;){
-    int cnt = 0;
-    for(int j = len -1; j > -1; --j){
-      if(atomic_load(&arr[j].completed) == 1){
-        cnt++; 
-      } else 
+  for(int j = len -1; j != -1 ; i++){
+    for(; j != -1; --j){
+      int const task_completed = 1;
+      if(atomic_load_explicit(&arr[j].completed, memory_order_acquire) != task_completed)
         break;
     }
-    if(i == 8){
+
+    if(i == 16){
       i = 0;
       nanosleep(&ns, NULL); 
     }
-    if(cnt == len)
-      break;
-    ++i;
   }
 }
 

common/utils/thread_pool/task_manager.h

@@ -5,7 +5,7 @@
 // Comment for deactivating ws tpool
 #define TASK_MANAGER
 #define TASK_MANAGER_CODING
-//#define TASK_MANAGER_RU
+#define TASK_MANAGER_RU
 
 #define TASK_MANAGER_UE
 #define TASK_MANAGER_UE_DECODING

executables/nr-ru.c

@@ -1876,10 +1876,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/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_status->completed) == 0);
-  atomic_store_explicit(&feptx->task_status->completed, 1, memory_order_seq_cst);
+  //assert(atomic_load(&feptx->task_status->completed) == 0);
+  atomic_store_explicit(&feptx->task_status->completed, 1, memory_order_release);
 #endif
 }
 
@@ -345,9 +345,10 @@ 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]; 
-  task_status_t task_status[sz];
-  memset(&task_status, 0, sz * sizeof(task_status_t));
+  assert(sz < 32);
+  feptx_cmd_t arr[32] = {0}; 
+  task_status_t task_status[32] = {0};
+  //memset(&task_status, 0, sz * sizeof(task_status_t));
 #endif
 
   for (int aid=0;aid<ru->nb_tx;aid++) {
@@ -440,8 +441,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_status->completed) == 0);
-  atomic_store(&feprx_cmd->task_status->completed, 1);
+  //assert(atomic_load(&feprx_cmd->task_status->completed) == 0);
+  atomic_store_explicit(&feprx_cmd->task_status->completed, 1, memory_order_release);
 #endif
 }
 
@@ -454,9 +455,10 @@ 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]; 
-  task_status_t task_status[sz];
-  memset(&task_status, 0, sizeof(task_status_t) * sz);
+  assert(sz < 32);
+  feprx_cmd_t arr[32] = {0}; 
+  task_status_t task_status[32] = {0};
+  //memset(&task_status, 0, sizeof(task_status_t) * sz);
 #endif
 
   for (int aid=0;aid<ru->nb_rx;aid++) {
@@ -505,8 +507,10 @@ 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_task_status_completed(nbfeprx, task_status);
+  if(nbfeptx > 0) {
+    trigger_all_task_manager(&ru->man);
+    wait_task_status_completed(nbfeprx, task_status);
+  }
 #else
   while (nbfeprx>0) {
     notifiedFIFO_elt_t *req=pullTpool(ru->respfeprx, ru->threadPool);