Data race in futex. Back to cond_var to test correctness

common/utils/thread_pool/task_manager.c

@@ -197,7 +197,7 @@ typedef struct {
   pthread_mutex_t mtx;
   pthread_cond_t cv;
   seq_ring_task_t r;
-  _Atomic int32_t* futex;
+ // _Atomic int32_t* futex;
   //_Atomic bool* waiting;
   _Atomic int done;
 } not_q_t;
@@ -209,7 +209,7 @@ typedef struct{
 
 
 static
-void init_not_q(not_q_t* q, _Atomic int32_t* futex /*, _Atomic bool* waiting */)
+void init_not_q(not_q_t* q /*, _Atomic int32_t* futex , _Atomic bool* waiting */)
 {
   assert(q != NULL);
 
@@ -229,7 +229,7 @@ void init_not_q(not_q_t* q, _Atomic int32_t* futex /*, _Atomic bool* waiting */)
   rc = pthread_cond_init(&q->cv, c_attr);
   assert(rc == 0);
 
-  q->futex = futex;
+  //q->futex = futex;
 }
 
 static
@@ -263,6 +263,8 @@ bool try_push_not_q(not_q_t* q, task_t t)
   int const rc = pthread_mutex_unlock(&q->mtx);
   assert(rc == 0);
 
+  pthread_cond_signal(&q->cv);
+
   return true;
 }
 
@@ -278,6 +280,8 @@ void push_not_q(not_q_t* q, task_t t)
 
   push_back_seq_ring_task(&q->r, t);
 
+  pthread_cond_signal(&q->cv);
+
   pthread_mutex_unlock(&q->mtx);
 }
 
@@ -305,6 +309,7 @@ ret_try_t try_pop_not_q(not_q_t* q)
     return ret;
   }
 
+  assert(sz > 0);
   ret.t = pop_seq_ring_task(&q->r);
 
   rc = pthread_mutex_unlock(&q->mtx);
@@ -321,7 +326,14 @@ bool pop_not_q(not_q_t* q, ret_try_t* out)
   assert(out != NULL);
   assert(q->done == 0 || q->done ==1);
 
-label:
+  int rc = pthread_mutex_lock(&q->mtx);
+  assert(rc == 0);
+  assert(q->done == 0 || q->done ==1);
+
+  while(size_seq_ring_task(&q->r) == 0 && q->done == 0)
+    pthread_cond_wait(&q->cv , &q->mtx);
+
+  /*
   // Let's be conservative and not use memory_order_relaxed
  // while (atomic_load_explicit(q->waiting, memory_order_seq_cst) == true){ //
       // Issue X86 PAUSE or ARM YIELD instruction to reduce contention between
@@ -340,11 +352,11 @@ label:
     assert(r != -1);
     goto label;
   }
-
-  assert(q->done == 0 || q->done ==1);
+*/
 
   //printf("Waking %ld id %ld \n", time_now_us(), pthread_self());
 
+  assert(q->done == 0 || q->done ==1);
   if(q->done == 1){
     //printf("Done, returning \n");
     int rc = pthread_mutex_unlock(&q->mtx);
@@ -354,7 +366,7 @@ label:
 
   out->t = pop_seq_ring_task(&q->r);
 
-  int rc = pthread_mutex_unlock(&q->mtx);
+  rc = pthread_mutex_unlock(&q->mtx);
   assert(rc == 0);
 
   return true;
@@ -369,13 +381,13 @@ void done_not_q(not_q_t* q)
   assert(rc == 0);
   
   q->done = 1;
-  long r = syscall(SYS_futex, q->futex, FUTEX_WAKE_PRIVATE, INT_MAX, NULL, NULL, 0);
-  assert(r != -1);
+  //long r = syscall(SYS_futex, q->futex, FUTEX_WAKE_PRIVATE, INT_MAX, NULL, NULL, 0);
+  //assert(r != -1);
   rc = pthread_mutex_unlock(&q->mtx);
   assert(rc == 0);
 
-//  rc = pthread_cond_signal(&q->cv);
-//  assert(rc == 0);
+  rc = pthread_cond_signal(&q->cv);
+  assert(rc == 0);
 
 //  q->futex++;
 }
@@ -469,13 +481,13 @@ 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");
-  atomic_store_explicit(&man->futex, 0, memory_order_seq_cst); 
+  //atomic_store_explicit(&man->futex, 0, memory_order_seq_cst); 
 
  // man->waiting = false;
 
   not_q_t* q_arr = (not_q_t*)man->q_arr;
   for(uint32_t i = 0; i < num_threads; ++i){
-    init_not_q(&q_arr[i], &man->futex /*, &man->waiting */);   
+    init_not_q(&q_arr[i] /*,&man->futex, &man->waiting */);   
   }
 
   man->t_arr = calloc(num_threads, sizeof(pthread_t));
@@ -506,6 +518,7 @@ void init_task_manager(task_manager_t* man, uint32_t num_threads)
 
   man->index = 0;
 
+  /*
   pthread_mutexattr_t attr = {0};
 #ifdef _DEBUG
   int const rc_mtx = pthread_mutexattr_settype(&attr, PTHREAD_MUTEX_ERRORCHECK);
@@ -517,6 +530,7 @@ void init_task_manager(task_manager_t* man, uint32_t num_threads)
   pthread_condattr_t* c_attr = NULL; 
   rc = pthread_cond_init(&man->wait_cv, c_attr);
   assert(rc == 0);
+*/
 
   //pin_thread_to_core(3);
 }
@@ -542,12 +556,13 @@ void free_task_manager(task_manager_t* man, void (*clean)(task_t*))
   free(man->q_arr);
 
   free(man->t_arr);
-
+/*
   int rc = pthread_mutex_destroy(&man->wait_mtx);
   assert(rc == 0);
 
   rc = pthread_cond_destroy(&man->wait_cv);
   assert(rc == 0);
+  */
 }
 
 void async_task_manager(task_manager_t* man, task_t t)
@@ -621,6 +636,7 @@ void trigger_and_wait_all_task_manager(task_manager_t* man)
 
 */
 
+/*
 void trigger_all_task_manager(task_manager_t* man)
 {
   assert(man != NULL);
@@ -633,6 +649,7 @@ void trigger_all_task_manager(task_manager_t* man)
   }
   assert(r != -1);
 }
+*/
 
 /*
 void wait_all_task_manager(task_manager_t* man)
@@ -660,7 +677,7 @@ void wait_task_status_completed(size_t len, task_status_t* arr)
   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)
+      if(atomic_load_explicit(&arr[j].completed, memory_order_seq_cst) != task_completed) //  memory_order_acquire
         break;
     }
 

common/utils/thread_pool/task_manager.h

@@ -4,6 +4,7 @@
 
 // Comment for deactivating ws tpool
 #define TASK_MANAGER
+#define TASK_MANAGER_DEMODULATION
 #define TASK_MANAGER_CODING
 #define TASK_MANAGER_RU
 
@@ -66,10 +67,10 @@ typedef struct{
 
   _Atomic(uint64_t) num_task;
 
-  pthread_cond_t  wait_cv; 
-  pthread_mutex_t wait_mtx;
+//  pthread_cond_t  wait_cv; 
+//  pthread_mutex_t wait_mtx;
 
-  _Atomic(int32_t) futex;
+//  _Atomic(int32_t) futex;
 
   //_Atomic(bool) waiting;
 } task_manager_t;
@@ -84,7 +85,7 @@ void async_task_manager(task_manager_t* man, task_t t);
 // Note that if the thread was working, this call is superflous
 // This method proved a bit faster as it is only one syscall
 // instead of a syscall when async_task_manager is called 
-void trigger_all_task_manager(task_manager_t* man);
+// void trigger_all_task_manager(task_manager_t* man);
 
 //void trigger_and_spin_task_manager(task_manager_t* man);
 

executables/nr-gnb.c

@@ -481,12 +481,15 @@ void init_gNB_Tpool(int inst) {
   PHY_VARS_gNB *gNB;
   gNB = RC.gNB[inst];
   gNB_L1_proc_t *proc = &gNB->proc;
-#ifdef TASK_MANAGER
+#if defined(TASK_MANAGER) && defined(TASK_MANAGER_CODING) && defined(TASK_MANAGER_DEMODULATION) 
   int const log_cores = get_nprocs_conf();
   assert(log_cores > 0);
   printf("[MIR]: log cores %d \n", log_cores);
   // Assuming: 2 x Physical cores = Logical cores
-  init_task_manager(&gNB->man, log_cores/2);
+  init_task_manager(&gNB->man, 4);
+#elif !defined(TASK_MANAGER) || !defined(TASK_MANAGER_CODING) || !defined(TASK_MANAGER_DEMODULATION) 
+  init_task_manager(&gNB->man, 4);
+  initTpool(get_softmodem_params()->threadPoolConfig, &gNB->threadPool, cpumeas(CPUMEAS_GETSTATE));
 #else
   initTpool(get_softmodem_params()->threadPoolConfig, &gNB->threadPool, cpumeas(CPUMEAS_GETSTATE));
 #endif

openair1/PHY/LTE_TRANSPORT/dlsch_coding.c

@@ -403,7 +403,7 @@ int dlsch_encoding(PHY_VARS_eNB *eNB,
   }
  
 #ifdef TASK_MANAGER_LTE
-  trigger_all_task_manager(proc->man);
+  //trigger_all_task_manager(proc->man);
   wait_task_status_completed(hadlsch->C, task_status);
 #else
   // Wait all other threads finish to process
@@ -533,7 +533,7 @@ int dlsch_encoding_fembms_pmch(PHY_VARS_eNB *eNB,
   }
 
 #ifdef TASK_MANAGER_LTE
-  trigger_all_task_manager(proc->man);
+  //trigger_all_task_manager(proc->man);
   wait_task_status_completed(hadlsch->C, task_status);
 #else
   // Wait all other threads finish to process

openair1/PHY/LTE_TRANSPORT/ulsch_decoding.c

@@ -392,7 +392,7 @@ static int ulsch_decoding_data(PHY_VARS_eNB *eNB, L1_rxtx_proc_t *proc, int UE_i
   }
 
 #ifdef TASK_MANAGER_LTE
-  trigger_all_task_manager(proc->man);
+  //trigger_all_task_manager(proc->man);
 #endif
   return(ret);
 }

openair1/PHY/NR_TRANSPORT/nr_dlsch_coding.c

@@ -453,7 +453,7 @@ int nr_dlsch_encoding(PHY_VARS_gNB *gNB,
 
 #ifdef TASK_MANAGER_CODING
     if(nbJobs > 0) {
-      trigger_all_task_manager(&gNB->man);
+      //trigger_all_task_manager(&gNB->man);
       wait_task_status_completed(nbJobs, task_status); 
       nbJobs = 0; 
     }

openair1/PHY/NR_TRANSPORT/nr_ulsch_decoding.c

@@ -189,7 +189,10 @@ static void nr_processULSegment(void *arg)
 
     LOG_E(PHY, "ulsch_decoding.c: Problem in rate_matching\n");
     rdata->decodeIterations = max_ldpc_iterations + 1;
-
+#ifdef TASK_MANAGER
+  assert(rdata->task_status != NULL && atomic_load(&rdata->task_status->completed) == 0);
+  atomic_store_explicit(&rdata->task_status->completed, 1, memory_order_release); //  memory_order order );
+#endif
     return;
   }
 
@@ -233,8 +236,7 @@ static void nr_processULSegment(void *arg)
 
 #ifdef TASK_MANAGER
   assert(rdata->task_status != NULL && atomic_load(&rdata->task_status->completed) == 0);
-  atomic_store_explicit(&rdata->task_status->completed, 1, memory_order_seq_cst); //  memory_order order );
-  // atomic_fetch_sub_explicit(rdata->tasks_remaining, 1, memory_order_seq_cst);
+  atomic_store_explicit(&rdata->task_status->completed, 1, memory_order_release); //  memory_order order );
 #endif
 }
 
@@ -499,10 +501,6 @@ int nr_ulsch_decoding(PHY_VARS_gNB *phy_vars_gNB,
     r_offset += E;
     offset += ((harq_process->K >> 3) - (harq_process->F >> 3) - ((harq_process->C > 1) ? 3 : 0));
   }
-#ifdef TASK_MANAGER
-  //stop_spining_task_manager(&phy_vars_gNB->man);
-  trigger_all_task_manager(&phy_vars_gNB->man);
-#endif
 
   return harq_process->C;
 }

openair1/PHY/NR_TRANSPORT/nr_ulsch_demodulation.c

@@ -1441,7 +1441,7 @@ static void nr_pusch_symbol_processing(void *arg)
       llr16[i] = llr_ptr[i] * rdata->s[i];
   }
 
-#ifdef TASK_MANAGER
+#ifdef TASK_MANAGER_DEMODULATION
   assert(rdata->task_status != NULL);
   atomic_store_explicit(&rdata->task_status->completed, 1, memory_order_seq_cst); //  memory_order order );
 #endif
@@ -1655,7 +1655,7 @@ int nr_rx_pusch_tp(PHY_VARS_gNB *gNB,
   start_meas(&gNB->rx_pusch_symbol_processing_stats);
   int numSymbols = gNB->num_pusch_symbols_per_thread;
 
-#ifdef TASK_MANAGER
+#ifdef TASK_MANAGER_DEMODULATION
   int const loop_iter = rel15_ul->nr_of_symbols/numSymbols;
   puschSymbolProc_t arr[loop_iter];
   task_status_t task_status[loop_iter];
@@ -1679,7 +1679,7 @@ int nr_rx_pusch_tp(PHY_VARS_gNB *gNB,
       total_res+=pusch_vars->ul_valid_re_per_slot[symbol+s];
     }
     if (total_res > 0) {
-#ifdef TASK_MANAGER 
+#ifdef TASK_MANAGER_DEMODULATION 
       puschSymbolProc_t *rdata = &arr[sz_arr]; 
       rdata->task_status = &task_status[sz_arr];
       ++sz_arr;
@@ -1706,7 +1706,7 @@ int nr_rx_pusch_tp(PHY_VARS_gNB *gNB,
         // Obvious memory leak when TASK_MANAGER not defined
         nr_pusch_symbol_processing(rdata);
       } else {
-#ifdef TASK_MANAGER
+#ifdef TASK_MANAGER_DEMODULATION
         task_t t = { .args = rdata, .func = &nr_pusch_symbol_processing };
         async_task_manager(&gNB->man , t);
 #else
@@ -1719,9 +1719,9 @@ int nr_rx_pusch_tp(PHY_VARS_gNB *gNB,
     }
   } // symbol loop
 
-#ifdef TASK_MANAGER
+#ifdef TASK_MANAGER_DEMODULATION
   if(gNB->nbSymb > 0){
-    trigger_all_task_manager(&gNB->man);
+    //trigger_all_task_manager(&gNB->man);
     wait_task_status_completed(sz_arr, task_status);
     gNB->nbSymb = 0; 
   }

openair1/PHY/NR_UE_TRANSPORT/nr_dlsch_decoding.c

@@ -479,7 +479,7 @@ uint32_t nr_dlsch_decoding(PHY_VARS_NR_UE *phy_vars_ue,
   int num_seg_ok = 0;
   int nbDecode = harq_process->C;
 #ifdef TASK_MANAGER_UE_DECODING
-  trigger_all_task_manager(&get_nrUE_params()->man);
+  //trigger_all_task_manager(&get_nrUE_params()->man);
   wait_task_status_completed(nbDecode, task_status); 
   for(size_t i = 0; i < harq_process->C; ++i){
     nr_ue_postDecode(phy_vars_ue, &arr[i], nbDecode == 1, b_size, b, &num_seg_ok, proc);

openair1/PHY/defs_gNB.h

@@ -738,8 +738,11 @@ typedef struct PHY_VARS_gNB_s {
   void *scopeData;
   /// structure for analyzing high-level RT measurements
   rt_L1_profiling_t rt_L1_profiling; 
-#ifdef TASK_MANAGER
+#if defined(TASK_MANAGER) && defined(TASK_MANAGER_CODING) && defined(TASK_MANAGER_DEMODULATION)
+  task_manager_t man;
+#elif !defined(TASK_MANAGER) || !defined(TASK_MANAGER_CODING) || !defined(TASK_MANAGER_DEMODULATION)
   task_manager_t man;
+  tpool_t threadPool;
 #else
   tpool_t threadPool;
 #endif
@@ -757,7 +760,7 @@ typedef struct puschSymbolProc_s {
   int16_t **llr_layers;
   int16_t *s;
   uint32_t nvar;
-#ifdef TASK_MANAGER
+#ifdef TASK_MANAGER_DEMODULATION 
  task_status_t* task_status;
 #endif
 } puschSymbolProc_t;

openair1/SCHED/phy_procedures_lte_eNb.c

@@ -1422,7 +1422,7 @@ void pusch_procedures(PHY_VARS_eNB *eNB,L1_rxtx_proc_t *proc) {
 #ifdef TASK_MANAGER_LTE
   if (proc->nbDecode) {
     // Not needed, but won't hurt performance
-    trigger_all_task_manager(proc->man);
+    //trigger_all_task_manager(proc->man);
     wait_task_status_completed(t_info.len, t_info.task_status);
     for(int i = 0; i < t_info.len; ++i){
       postDecode(proc, &arr[i]); 

openair1/SCHED_NR/nr_ru_procedures.c

@@ -398,7 +398,7 @@ void nr_feptx_tp(RU_t *ru, int frame_tx, int slot) {
 #ifdef TASK_MANAGER_RU
   if(nbfeptx > 0) {
     //stop_spining_task_manager(&ru->man);
-    trigger_all_task_manager(&ru->man);
+    //trigger_all_task_manager(&ru->man);
     wait_task_status_completed(nbfeptx, task_status);
     nbfeptx = 0; 
   }
@@ -508,7 +508,7 @@ void nr_fep_tp(RU_t *ru, int slot)
 #ifdef TASK_MANAGER_RU
   //stop_spining_task_manager(&ru->man);
   if(nbfeprx > 0) {
-    trigger_all_task_manager(&ru->man);
+    //trigger_all_task_manager(&ru->man);
     wait_task_status_completed(nbfeprx, task_status);
   }
 #else

openair1/SCHED_NR/phy_procedures_nr_gNB.c

@@ -1002,10 +1002,12 @@ int phy_procedures_gNB_uespec_RX(PHY_VARS_gNB *gNB, int frame_rx, int slot_rx)
   
 #ifdef TASK_MANAGER
   if (totalDecode > 0) {
+	  assert(totalDecode == t_info.len);
     wait_task_status_completed(t_info.len, t_info.task_status);
     for(int i = 0; i < t_info.len; ++i){
       nr_postDecode(gNB, &arr[i]); 
     } 
+    totalDecode = 0;
     //printf("Decoding time %ld \n", time_now_ns() - t0);
   }
 #else