gNB working

executables/nr-gnb.c

@@ -284,8 +284,13 @@ void rx_func(void *param)
     } else {
       //notifiedFIFO_elt_t * res = pullTpool(&gNB->L1_tx_filled, &gNB->threadPool);
       notifiedFIFO_elt_t* res = pullNotifiedFIFO(&gNB->L1_tx_filled);
-      if (res == NULL)
-        return; // Tpool has been stopped
+      if (res == NULL){
+#ifdef TASK_MANAGER_RU
+         if(info->elm != NULL)
+            pushNotifiedFIFO(info->elm->reponseFifo, info->elm);
+#endif
+        return; // Queue aborted 
+      }
        processingData_L1tx_t* syncMsg = (processingData_L1tx_t *)NotifiedFifoData(res);
       syncMsg->gNB = gNB;
       syncMsg->timestamp_tx = info->timestamp_tx;
@@ -301,8 +306,13 @@ void rx_func(void *param)
     }
   } else if (get_softmodem_params()->continuous_tx) {
     notifiedFIFO_elt_t *res = pullNotifiedFIFO(&gNB->L1_tx_free);
-    if (res == NULL)
-      return; // Tpool has been stopped
+    if (res == NULL){
+#ifdef TASK_MANAGER_RU
+         if(info->elm != NULL)
+            pushNotifiedFIFO(info->elm->reponseFifo, info->elm);
+#endif
+      return;  // Queue aborted 
+    }
     processingData_L1tx_t *syncMsg = (processingData_L1tx_t *)NotifiedFifoData(res);
     syncMsg->gNB = gNB;
     syncMsg->timestamp_tx = info->timestamp_tx;
@@ -311,6 +321,11 @@ void rx_func(void *param)
     res->key = slot_tx;
     pushNotifiedFIFO(&gNB->L1_tx_out, res);
   }
+#ifdef TASK_MANAGER_RU
+  // I can also write bad code
+  if(info->elm != NULL)
+    pushNotifiedFIFO(info->elm->reponseFifo, info->elm);
+#endif
 
 #if 0
   LOG_D(PHY, "rxtx:%lld nfapi:%lld phy:%lld tx:%lld rx:%lld prach:%lld ofdm:%lld ",
@@ -523,19 +538,22 @@ void init_gNB_Tpool(int inst) {
   PHY_VARS_gNB *gNB;
   gNB = RC.gNB[inst];
   gNB_L1_proc_t *proc = &gNB->proc;
-#ifdef TASK_MANAGER
-  //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
-  int n_threads = num_threads(get_softmodem_params()->threadPoolConfig);
-  init_task_manager(&gNB->man, n_threads ); //log_cores/2);
-#endif
+
   // PUSCH symbols per thread need to be calculated by how many threads we have
   gNB->num_pusch_symbols_per_thread = 1;
+
+#if defined(TASK_MANAGER) && defined(TASK_MANAGER_CODING) && defined(TASK_MANAGER_DEMODULATION) && defined(TASK_MANAGER_RU)
+  int n_threads = num_threads(get_softmodem_params()->threadPoolConfig);
+  init_task_manager(&gNB->man, n_threads); //log_cores/2);
+#elif !defined(TASK_MANAGER) ||  !defined(TASK_MANAGER_CODING) || !defined(TASK_MANAGER_DEMODULATION) || !defined(TASK_MANAGER_RU)
+  int n_threads = num_threads(get_softmodem_params()->threadPoolConfig);
+  init_task_manager(&gNB->man, n_threads ); //log_cores/2);
   // ULSCH decoding threadpool
   initTpool(get_softmodem_params()->threadPoolConfig, &gNB->threadPool, cpumeas(CPUMEAS_GETSTATE));
-
+#else
+  // ULSCH decoding threadpool
+  initTpool(get_softmodem_params()->threadPoolConfig, &gNB->threadPool, cpumeas(CPUMEAS_GETSTATE));
+#endif
    // ULSCH decoder result FIFO
   initNotifiedFIFO(&gNB->respPuschSymb);
   initNotifiedFIFO(&gNB->respDecode);

executables/nr-ru.c

@@ -1191,9 +1191,6 @@ void *ru_thread( void *param ) {
   pthread_mutex_unlock(&RC.ru_mutex);
   wait_sync("ru_thread");
 
-  processingData_L1_t *syncMsg;
-  notifiedFIFO_elt_t *res;
-
   if(!emulate_rf) {
     // Start RF device if any
     if (ru->start_rf) {
@@ -1341,9 +1338,10 @@ void *ru_thread( void *param ) {
       }
     } // end if (slot_type == NR_UPLINK_SLOT || slot_type == NR_MIXED_SLOT) {
 
+
+    notifiedFIFO_elt_t *res = NULL;
     // At this point, all information for subframe has been received on FH interface
     if (!get_softmodem_params()->reorder_thread_disable) {
-      //res = pullTpool(&gNB->resp_L1, &gNB->threadPool);
       res = pullNotifiedFIFO(&gNB->resp_L1);
       if (res == NULL)
         break; // Tpool has been stopped
@@ -1351,6 +1349,8 @@ void *ru_thread( void *param ) {
       res = newNotifiedFIFO_elt(sizeof(processingData_L1_t), 0, &gNB->resp_L1, NULL);
     }
 
+    processingData_L1_t *syncMsg = NULL;
+
     syncMsg = (processingData_L1_t *)NotifiedFifoData(res);
     syncMsg->gNB = gNB;
     syncMsg->frame_rx = proc->frame_rx;
@@ -1361,14 +1361,11 @@ void *ru_thread( void *param ) {
     res->key = proc->tti_rx;
 #ifdef TASK_MANAGER // No TASK_MANAGER_RU
     if (!get_softmodem_params()->reorder_thread_disable) {
-      assert(res->processingFunc != NULL);
+      assert(res->processingFunc == rx_func);
       assert(res->reponseFifo != NULL);
-      //assert(gNB->threadPool.incomingFifo == gNB->resp_L1);
-
-      pushNotifiedFIFO(&gNB->threadPool.incomingFifo, res);
-
-      //res->processingFunc(NotifiedFifoData(res));
-      //pushNotifiedFIFO(res->reponseFifo, res);
+      syncMsg->elm = res; 
+      task_t t = {.func = rx_func, .args = syncMsg};
+      async_task_manager(&gNB->man, t);
     } 
 #else
     if (!get_softmodem_params()->reorder_thread_disable) 

openair1/PHY/NR_TRANSPORT/nr_dlsch_coding.c

@@ -277,8 +277,8 @@ static void ldpc8blocks(void *p)
   }
 
 #ifdef TASK_MANAGER_CODING
-  assert(atomic_load(impp->task_done) == 0);
-  atomic_store_explicit(impp->task_done, 1, memory_order_seq_cst);
+  //assert(atomic_load(impp->task_done) == 0);
+  atomic_store_explicit(&impp->task_done->completed, 1, memory_order_release);
 #endif
 }
 
@@ -424,14 +424,39 @@ int nr_dlsch_encoding(PHY_VARS_gNB *gNB,
     notifiedFIFO_t nf;
     initNotifiedFIFO(&nf);
     int nbJobs = 0;
+
+#ifdef TASK_MANAGER_CODING
+      size_t const sz = (impp.n_segments/8+((impp.n_segments&7)==0 ? 0 : 1));
+      encoder_implemparams_t arr[sz];
+      task_status_t task_status[sz];
+      memset(task_status, 0, sz * sizeof(task_status_t));
+ #endif
     for (int j = 0; j < (impp.n_segments / 8 + ((impp.n_segments & 7) == 0 ? 0 : 1)); j++) {
+#ifdef TASK_MANAGER_CODING
+      assert(nbJobs < sz);
+      encoder_implemparams_t* perJobImpp = &arr[nbJobs];
+#else
       notifiedFIFO_elt_t *req = newNotifiedFIFO_elt(sizeof(impp), j, &nf, ldpc8blocks);
       encoder_implemparams_t *perJobImpp = (encoder_implemparams_t *)NotifiedFifoData(req);
+#endif
       *perJobImpp = impp;
       perJobImpp->macro_num = j;
+#ifdef TASK_MANAGER_CODING
+        perJobImpp->task_done = &task_status[nbJobs];
+        task_t t = {.args = perJobImpp, .func = ldpc8blocks};
+        //assert(atomic_load(&perJobImpp->task_status->completed) == 0); 
+        async_task_manager(&gNB->man, t); 
+#else
       pushTpool(&gNB->threadPool, req);
+#endif
+
       nbJobs++;
     }
+#ifdef TASK_MANAGER_CODING
+      if(nbJobs > 0) {
+        wait_task_status_completed(nbJobs, task_status);
+      }
+#else
     while (nbJobs) {
       notifiedFIFO_elt_t *req = pullTpool(&nf, &gNB->threadPool);
       if (req == NULL)
@@ -439,6 +464,7 @@ int nr_dlsch_encoding(PHY_VARS_gNB *gNB,
       delNotifiedFIFO_elt(req);
       nbJobs--;
     }
+#endif
   }
   VCD_SIGNAL_DUMPER_DUMP_FUNCTION_BY_NAME(VCD_SIGNAL_DUMPER_FUNCTIONS_gNB_DLSCH_ENCODING, VCD_FUNCTION_OUT);
   return 0;

openair1/PHY/defs_gNB.h

@@ -728,7 +728,6 @@ typedef struct PHY_VARS_gNB_s {
   notifiedFIFO_t L1_tx_filled;
   notifiedFIFO_t L1_tx_out;
   notifiedFIFO_t resp_RU_tx;
-  tpool_t threadPool;
   int nbSymb;
   int num_pusch_symbols_per_thread;
   pthread_t L1_rx_thread;
@@ -739,8 +738,14 @@ 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) && defined(TASK_MANAGER_RU) && defined(TASK_MANAGER_SIM)
+  task_manager_t man;
+#elif !defined(TASK_MANAGER) ||  !defined(TASK_MANAGER_CODING) || !defined(TASK_MANAGER_DEMODULATION) || !defined(TASK_MANAGER_RU) || !defined(TASK_MANAGER_SIM) 
   task_manager_t man;
+  tpool_t threadPool;
+#else
+  tpool_t threadPool;
 #endif
 } PHY_VARS_gNB;
 
@@ -818,6 +823,9 @@ typedef struct processingData_L1 {
   int slot_tx;
   openair0_timestamp timestamp_tx;
   PHY_VARS_gNB *gNB;
+#ifdef TASK_MANAGER_RU
+  notifiedFIFO_elt_t* elm; 
+#endif
 } processingData_L1_t;
 
 typedef enum {