TASK_MANAGER_LTE NOT tested

common/utils/thread_pool/task_manager.h

@@ -7,10 +7,9 @@
 #define TASK_MANAGER_CODING
 #define TASK_MANAGER_RU 
 
-//#define TASK_MANAGER_UE
 #define TASK_MANAGER_UE_DECODING
 #define TASK_MANAGER_SIM
-//#define TASK_MANAGER_LTE
+#define TASK_MANAGER_LTE
 
 #include "task.h"
 

executables/lte-softmodem.c

@@ -73,7 +73,7 @@ unsigned short config_frames[4] = {2,9,11,13};
 #include "common/utils/LOG/vcd_signal_dumper.h"
 #include "UTIL/OPT/opt.h"
 #include "enb_config.h"
-
+#include "common/utils/thread_pool/task_manager.h"
 
 #include "create_tasks.h"
 
@@ -545,19 +545,39 @@ int main ( int argc, char **argv )
     printf("Initializing eNB threads single_thread_flag:%d wait_for_sync:%d\n", get_softmodem_params()->single_thread_flag,get_softmodem_params()->wait_for_sync);
     init_eNB(get_softmodem_params()->single_thread_flag,get_softmodem_params()->wait_for_sync);
   }
+ 
   for (int x=0; x < RC.nb_L1_inst; x++)
     for (int CC_id=0; CC_id<RC.nb_L1_CC[x]; CC_id++) {
       L1_rxtx_proc_t *L1proc= &RC.eNB[x][CC_id]->proc.L1_proc;
       L1_rxtx_proc_t *L1proctx= &RC.eNB[x][CC_id]->proc.L1_proc_tx;
-      L1proc->threadPool = (tpool_t *)malloc(sizeof(tpool_t));
       L1proc->respDecode=(notifiedFIFO_t*) malloc(sizeof(notifiedFIFO_t));
-      if ( strlen(get_softmodem_params()->threadPoolConfig) > 0 )
-       initTpool(get_softmodem_params()->threadPoolConfig, L1proc->threadPool, true);
-      else
-        initTpool("n", L1proc->threadPool, true);
+      if ( strlen(get_softmodem_params()->threadPoolConfig) > 0 ){
+#ifdef TASK_MANAGER_LTE
+      L1proc->man = calloc(1, sizeof(task_manager_t));
+      assert(L1proc->man != NULL && "Memory exhausted");
+      int const num_threads = parse_num_threads(get_softmodem_params()->threadPoolConfig);
+      init_task_manager(L1proc->man, num_threads);
+#else
+      initTpool(get_softmodem_params()->threadPoolConfig, L1proc->threadPool, true);
+#endif
+      }else {
+#ifdef TASK_MANAGER_LTE
+      L1proc->man = calloc(1, sizeof(task_manager_t));
+      assert(L1proc->man != NULL && "Memory exhausted");
+      int const num_threads = parse_num_threads("n");
+      init_task_manager(L1proc->man, num_threads);
+#else
+      initTpool("n", L1proc->threadPool, true);
+#endif
+      }
       initNotifiedFIFO(L1proc->respDecode);
+#ifdef TASK_MANAGER_LTE
+      L1proctx->man = L1proc->man;
+#else
       L1proctx->threadPool = L1proc->threadPool;
-  }
+#endif
+
+    }
 
   printf("wait_eNBs()\n");
   wait_eNBs();

openair1/PHY/LTE_TRANSPORT/dlsch_coding.c

@@ -43,6 +43,7 @@
 #include "executables/lte-softmodem.h"
 #include <syscall.h>
 #include <common/utils/threadPool/thread-pool.h>
+#include "common/utils/thread_pool/task_manager.h"
 
 //#define DEBUG_DLSCH_CODING
 //#define DEBUG_DLSCH_FREE 1
@@ -280,6 +281,10 @@ static void TPencode(void * arg) {
 			  rdata->r,
 			  hadlsch->nb_rb);
     stop_meas(rdata->rm_stats);
+
+#ifdef TASK_MANAGER_LTE
+    completed_task_ans(rdata->ans);
+#endif
 }
 
 int dlsch_encoding(PHY_VARS_eNB *eNB,
@@ -345,13 +350,23 @@ int dlsch_encoding(PHY_VARS_eNB *eNB,
       return(-1);
   }
 
+#ifdef TASK_MANAGER_LTE
+  turboEncode_t arr[hadlsch->C];
+  task_ans_t ans[hadlsch->C];
+  memset(ans, 0, hadlsch->C * sizeof(task_ans_t));
+#endif
   notifiedFIFO_t respEncode;
   initNotifiedFIFO(&respEncode);
+
   for (int r=0, r_offset=0; r<hadlsch->C; r++) {
-    
+#ifdef TASK_MANAGER_LTE 
+  turboEncode_t* rdata = &arr[r];
+  rdata->ans = &ans[r];
+#else 
     union turboReqUnion id= {.s={dlsch->rnti,frame,subframe,r,0}};
     notifiedFIFO_elt_t *req = newNotifiedFIFO_elt(sizeof(turboEncode_t), id.p, &respEncode, TPencode);
     turboEncode_t * rdata=(turboEncode_t *) NotifiedFifoData(req);
+#endif
     rdata->input=hadlsch->c[r];
     rdata->Kr_bytes= ( r<hadlsch->Cminus ? hadlsch->Kminus : hadlsch->Kplus) >>3;
     rdata->filler=(r==0) ? hadlsch->F : 0;
@@ -364,8 +379,12 @@ int dlsch_encoding(PHY_VARS_eNB *eNB,
     rdata->round=hadlsch->round;
     rdata->r_offset=r_offset;
     rdata->G=G;
-
-    pushTpool(proc->threadPool, req);
+#ifdef TASK_MANAGER_LTE
+  task_t t = {.func =  TPencode, .args = rdata};
+  async_task_manager(proc->man, t);
+#else
+  pushTpool(proc->threadPool, req);
+#endif
     nbEncode++;
 
     int Qm=hadlsch->Qm;
@@ -378,6 +397,12 @@ int dlsch_encoding(PHY_VARS_eNB *eNB,
     else
       r_offset += Nl*Qm * ((GpmodC==0?0:1) + (Gp/C));
   }
+ 
+#ifdef TASK_MANAGER_LTE
+  if(nbEncode > 0){
+   join_task_ans(ans, hadlsch->C);
+  }
+#else
   // Wait all other threads finish to process
   while (nbEncode) {
     notifiedFIFO_elt_t *res = pullTpool(&respEncode, proc->threadPool);
@@ -386,6 +411,8 @@ int dlsch_encoding(PHY_VARS_eNB *eNB,
     delNotifiedFIFO_elt(res);
     nbEncode--;
   }
+#endif
+
   VCD_SIGNAL_DUMPER_DUMP_FUNCTION_BY_NAME(VCD_SIGNAL_DUMPER_FUNCTIONS_ENB_DLSCH_ENCODING, VCD_FUNCTION_OUT);
   return(0);
 }
@@ -451,13 +478,23 @@ int dlsch_encoding_fembms_pmch(PHY_VARS_eNB *eNB,
       return(-1);
   }
   int nbEncode = 0;
+#ifdef TASK_MANAGER_LTE
+  turboEncode_t arr[hadlsch->C];
+  task_ans_t ans[hadlsch->C];
+  memset(ans, 0, hadlsch->C * sizeof(task_ans_t));
+#endif
   notifiedFIFO_t respEncode;
   initNotifiedFIFO(&respEncode);
+
   for (int r=0, r_offset=0; r<hadlsch->C; r++) {
-    
+#ifdef TASK_MANAGER_LTE    
+  turboEncode_t* rdata = &arr[r];
+  rdata->ans = &ans[r];
+#else 
     union turboReqUnion id= {.s={dlsch->rnti,frame,subframe,r,0}};
     notifiedFIFO_elt_t *req = newNotifiedFIFO_elt(sizeof(turboEncode_t), id.p, &respEncode, TPencode);
     turboEncode_t * rdata=(turboEncode_t *) NotifiedFifoData(req);
+#endif
     rdata->input=hadlsch->c[r];
     rdata->Kr_bytes= ( r<hadlsch->Cminus ? hadlsch->Kminus : hadlsch->Kplus) >>3;
     rdata->filler=(r==0) ? hadlsch->F : 0;
@@ -471,7 +508,12 @@ int dlsch_encoding_fembms_pmch(PHY_VARS_eNB *eNB,
     rdata->r_offset=r_offset;
     rdata->G=G;
 
+#ifdef TASK_MANAGER_LTE
+  task_t t = {.func = TPencode, .args = rdata};
+  async_task_manager(proc->man, t);
+#else
     pushTpool(proc->threadPool, req);
+#endif
     nbEncode++;
 
     int Qm=hadlsch->Qm;
@@ -484,6 +526,12 @@ int dlsch_encoding_fembms_pmch(PHY_VARS_eNB *eNB,
     else
       r_offset += Nl*Qm * ((GpmodC==0?0:1) + (Gp/C));
   }
+
+#ifdef TASK_MANAGER_LTE
+  if(nbEncode > 0){
+    join_task_ans(ans, hadlsch->C); 
+  }
+#else
   // Wait all other threads finish to process
   while (nbEncode) {
     notifiedFIFO_elt_t *res = pullTpool(&respEncode, proc->threadPool);
@@ -492,6 +540,7 @@ int dlsch_encoding_fembms_pmch(PHY_VARS_eNB *eNB,
     delNotifiedFIFO_elt(res);
     nbEncode--;
   }
+#endif
   return(0);
 }
 

openair1/PHY/LTE_TRANSPORT/transport_proto.h

@@ -36,6 +36,8 @@
 #include "nfapi_interface.h"
 #include "transport_common_proto.h"
 
+#include "common/utils/thread_pool/task_manager.h"
+
 // Functions below implement 36-211 and 36-212
 
 /** @addtogroup _PHY_TRANSPORT_
@@ -525,7 +527,11 @@ unsigned int  ulsch_decoding(PHY_VARS_eNB *phy_vars_eNB,
                              uint8_t UE_id,
                              uint8_t control_only_flag,
                              uint8_t Nbundled,
-                             uint8_t llr8_flag);
+                             uint8_t llr8_flag
+#ifdef TASK_MANAGER_LTE
+                             ,thread_info_tm_t* t_info
+#endif			     
+			     );
 
 void generate_phich_top(PHY_VARS_eNB *phy_vars_eNB,
                         L1_rxtx_proc_t *proc,

openair1/PHY/LTE_TRANSPORT/ulsch_decoding.c

@@ -41,6 +41,7 @@
 #include "RRC/LTE/rrc_extern.h"
 #include "PHY_INTERFACE/phy_interface.h"
 #include "transport_proto.h"
+#include "common/utils/thread_pool/task_manager.h"
 
 extern int oai_exit;
 
@@ -239,8 +240,11 @@ void processULSegment(void * arg) {
                                    1,
                                    r,
                                    &E)==-1) {
-      LOG_E(PHY,"ulsch_decoding.c: Problem in rate matching\n");
-      return;
+#ifdef TASK_MANAGER_LTE
+     completed_task_ans(rdata->ans);
+#endif
+     LOG_E(PHY,"ulsch_decoding.c: Problem in rate matching\n");
+     return;
    }
    stop_meas(&eNB->ulsch_rate_unmatching_stats);
    int max_Ncb = 3*ulsch_harq->RTC[r]*32 ;
@@ -284,6 +288,10 @@ void processULSegment(void * arg) {
                                             &eNB->ulsch_tc_intl1_stats,
                                             &eNB->ulsch_tc_intl2_stats,
                                             &ulsch_harq->abort_decode);
+
+#ifdef TASK_MANAGER_LTE
+     completed_task_ans(rdata->ans);
+#endif 
 }
 
 /*!
@@ -295,7 +303,11 @@ void processULSegment(void * arg) {
   @returns 0 on success
 */
 
+#ifdef TASK_MANAGER_LTE
+static int ulsch_decoding_data(PHY_VARS_eNB *eNB, L1_rxtx_proc_t *proc, int UE_id, int harq_pid, int llr8_flag, thread_info_tm_t* t_info)
+#else
 static int ulsch_decoding_data(PHY_VARS_eNB *eNB, L1_rxtx_proc_t *proc, int UE_id, int harq_pid, int llr8_flag)
+#endif
 {
   unsigned int r_offset=0;
   int offset = 0;
@@ -330,14 +342,20 @@ static int ulsch_decoding_data(PHY_VARS_eNB *eNB, L1_rxtx_proc_t *proc, int UE_i
       E = ulsch_harq->Qm * (Gp/ulsch_harq->C);
     else
       E = ulsch_harq->Qm * ((GpmodC==0?0:1) + (Gp/ulsch_harq->C));
-    
+
+#ifdef TASK_MANAGER_LTE
+    turboDecode_t* rdata = &((turboDecode_t*)t_info->buf)[t_info->len];
+     assert(t_info->len < 64);
+     rdata->ans = &t_info->ans[t_info->len];
+     t_info->len += 1;
+#else
     union turboReqUnion id= {.s={ulsch->rnti,proc->frame_rx,proc->subframe_rx,0,0}};
     notifiedFIFO_elt_t *req=newNotifiedFIFO_elt(sizeof(turboDecode_t),
                                                 id.p,
                                                 proc->respDecode,
                                                 processULSegment);
     turboDecode_t * rdata=(turboDecode_t *) NotifiedFifoData(req);
-
+#endif
     rdata->eNB=eNB;
     rdata->frame=proc->frame_rx;
     rdata->subframe=proc->subframe_rx;
@@ -355,12 +373,18 @@ static int ulsch_decoding_data(PHY_VARS_eNB *eNB, L1_rxtx_proc_t *proc, int UE_i
     rdata->function=td;
     int Fbytes=(r==0) ? rdata->Fbits>>3 : 0;
     int sz=Kr_bytes - Fbytes - ((ulsch_harq->C>1)?3:0);
+#ifdef TASK_MANAGER_LTE
+    task_t t = { .args = rdata, .func = &processULSegment };
+    async_task_manager(proc->man, t);
+#else
     pushTpool(proc->threadPool,req);
+#endif
     proc->nbDecode++;
     LOG_D(PHY,"Added a block to decode, in pipe: %d\n",proc->nbDecode);
     r_offset+=E;
     offset+=sz;	    
   }
+
   return(ret);
 }
 
@@ -395,7 +419,11 @@ unsigned int  ulsch_decoding(PHY_VARS_eNB *eNB,
                              uint8_t UE_id,
                              uint8_t control_only_flag,
                              uint8_t Nbundled,
-                             uint8_t llr8_flag)
+                             uint8_t llr8_flag
+#ifdef TASK_MANAGER_LTE
+                              ,thread_info_tm_t* t_info
+#endif
+			     )
 {
   int16_t *ulsch_llr = eNB->pusch_vars[UE_id]->llr;
   LTE_DL_FRAME_PARMS *frame_parms = &eNB->frame_parms;
@@ -1082,7 +1110,11 @@ unsigned int  ulsch_decoding(PHY_VARS_eNB *eNB,
 
   LOG_D(PHY,"frame %d subframe %d O_ACK:%d o_ACK[]=%d:%d:%d:%d\n",frame,subframe,ulsch_harq->O_ACK,ulsch_harq->o_ACK[0],ulsch_harq->o_ACK[1],ulsch_harq->o_ACK[2],ulsch_harq->o_ACK[3]);
   // Do ULSCH Decoding for data portion
+#ifdef TASK_MANAGER_LTE
+  ret = ulsch_decoding_data(eNB, proc, UE_id, harq_pid, llr8_flag, t_info);
+#else
   ret = ulsch_decoding_data(eNB, proc, UE_id, harq_pid, llr8_flag);
+#endif
   return(ret);
 }
 

openair1/PHY/defs_eNB.h

@@ -66,7 +66,7 @@
 #include "PHY/LTE_TRANSPORT/transport_eNB.h"
 #include "openair2/PHY_INTERFACE/IF_Module.h"
 #include "common/openairinterface5g_limits.h"
-
+#include "common/utils/thread_pool/task_manager.h"
 
 #define PBCH_A 24
 #define MAX_NUM_RU_PER_eNB 64
@@ -252,7 +252,11 @@ typedef struct {
   pthread_cond_t cond_RUs;
   /// mutex for RXn-TXnp4 processing thread
   pthread_mutex_t mutex_RUs;
+#ifdef TASK_MANAGER_LTE
+  task_manager_t* man; // non-owning ptr
+#else
   tpool_t *threadPool;
+#endif
   int nbDecode;
   notifiedFIFO_t *respDecode;
   pthread_mutex_t mutex_emulateRF;
@@ -793,6 +797,9 @@ typedef struct TurboDecode_s {
     int offset;
     int maxIterations;
     int decodeIterations;
+#ifdef TASK_MANAGER_LTE
+ task_ans_t* ans;
+#endif
 } turboDecode_t;
 
 #define TURBO_SIMD_SOFTBITS   96+12+3+3*6144
@@ -809,6 +816,9 @@ typedef struct turboEncode_s {
   time_stats_t *rm_stats;
   time_stats_t *te_stats;
   time_stats_t *i_stats;
+#ifdef TASK_MANAGER_LTE
+ task_ans_t* ans;
+#endif
 } turboEncode_t;
 
 

openair1/SCHED/phy_procedures_lte_eNb.c

@@ -41,6 +41,7 @@
 #include <common/utils/system.h>
 #include "common/utils/LOG/vcd_signal_dumper.h"
 #include <nfapi/oai_integration/nfapi_pnf.h>
+#include "common/utils/thread_pool/task_manager.h"
 
 #include "assertions.h"
 
@@ -1232,10 +1233,13 @@ uci_procedures(PHY_VARS_eNB *eNB,
   } // end loop for (int i = 0; i < NUMBER_OF_UCI_MAX; i++) {
 }
 
+#ifdef TASK_MANAGER_LTE
+void postDecode(L1_rxtx_proc_t *proc, turboDecode_t* rdata){
+#else
 void postDecode(L1_rxtx_proc_t *proc, notifiedFIFO_elt_t *req)
 {
   turboDecode_t * rdata=(turboDecode_t *) NotifiedFifoData(req);
-
+#endif
   LTE_eNB_ULSCH_t *ulsch = rdata->eNB->ulsch[rdata->UEid];
   LTE_UL_eNB_HARQ_t *ulsch_harq = rdata->ulsch_harq;
   PHY_VARS_eNB *eNB=rdata->eNB;
@@ -1330,6 +1334,12 @@ void pusch_procedures(PHY_VARS_eNB *eNB,L1_rxtx_proc_t *proc) {
   const int frame    = proc->frame_rx;
   uint32_t harq_pid0 = subframe2harq_pid(&eNB->frame_parms,frame,subframe);
 
+#ifdef TASK_MANAGER_LTE
+  turboDecode_t arr[64] = {0};
+  task_ans_t ans[64] = {0};
+  thread_info_tm_t t_info = { .ans = ans, .buf = (uint8_t*)arr };
+#endif
+
   for (i = 0; i < NUMBER_OF_ULSCH_MAX; i++) {
     ulsch = eNB->ulsch[i];
     if (!ulsch) continue; 
@@ -1387,7 +1397,11 @@ void pusch_procedures(PHY_VARS_eNB *eNB,L1_rxtx_proc_t *proc) {
                      i,
                      0, // control_only_flag
                      ulsch_harq->V_UL_DAI,
-                     ulsch_harq->nb_rb > 20 ? 1 : 0);
+                     ulsch_harq->nb_rb > 20 ? 1 : 0
+#ifdef TASK_MANAGER_LTE
+		     ,&t_info 
+#endif		     
+		     );
     }
     else if ((ulsch) &&
              (ulsch->rnti>0) &&
@@ -1404,6 +1418,15 @@ void pusch_procedures(PHY_VARS_eNB *eNB,L1_rxtx_proc_t *proc) {
   }   //   for (i=0; i<NUMBER_OF_ULSCH_MAX; i++)
 
   const bool decode = proc->nbDecode;
+#ifdef TASK_MANAGER_LTE
+  assert(t_info.len == proc->nbDecode);
+  if (proc->nbDecode > 0) {
+     join_task_ans(t_info.ans, t_info.len);  
+     for(size_t i = 0; i < t_info.len; ++i){
+       postDecode(proc, &arr[i]);
+     }
+  }
+#else
   while (proc->nbDecode > 0) {
     notifiedFIFO_elt_t *req=pullTpool(proc->respDecode, proc->threadPool);
     if (req == NULL)
@@ -1413,6 +1436,7 @@ void pusch_procedures(PHY_VARS_eNB *eNB,L1_rxtx_proc_t *proc) {
     merge_meas(&eNB->ulsch_turbo_decoding_stats, &ts);
     delNotifiedFIFO_elt(req);
   }
+#endif
   if (decode)
     stop_meas(&eNB->ulsch_decoding_stats);
 }

openair1/SIMULATION/LTE_PHY/dlsim.c

@@ -56,6 +56,7 @@
 #include "dummy_functions.c"
 #include "executables/thread-common.h"
 #include "common/ran_context.h"
+#include "common/utils/thread_pool/task_manager.h"
 void feptx_ofdm(RU_t *ru, int frame, int subframe);
 void feptx_prec(RU_t *ru, int frame, int subframe);
 
@@ -589,6 +590,9 @@ int main(int argc, char **argv) {
   nfapi_tx_request_t TX_req;
   Sched_Rsp_t sched_resp;
   int pa=dB0;
+#ifdef TASK_MANAGER_LTE
+  task_manager_t man = {0};
+#endif
   cpu_freq_GHz = get_cpu_freq_GHz();
   printf("Detected cpu_freq %f GHz\n",cpu_freq_GHz);
   memset((void *)&sched_resp,0,sizeof(sched_resp));
@@ -1254,10 +1258,16 @@ int main(int argc, char **argv) {
   }
 
   L1_rxtx_proc_t *proc_eNB = &eNB->proc.L1_proc;
-  proc_eNB->threadPool = (tpool_t *)malloc(sizeof(tpool_t));
   proc_eNB->respDecode=(notifiedFIFO_t*) malloc(sizeof(notifiedFIFO_t));
-  initTpool("n", proc_eNB->threadPool, true);
   initNotifiedFIFO(proc_eNB->respDecode);
+#ifdef TASK_MANAGER_LTE
+  int const n_threads = parse_num_threads("n");
+  init_task_manager(&man, n_threads);
+  proc_eNB->man = &man;
+#else
+  proc_eNB->threadPool = (tpool_t *)malloc(sizeof(tpool_t));
+  initTpool("n", proc_eNB->threadPool, true);
+#endif
 
   proc_eNB->frame_tx=0;
 

openair1/SIMULATION/LTE_PHY/ulsim.c

@@ -55,6 +55,7 @@
 #include "PHY/LTE_ESTIMATION/lte_estimation.h"
 #include "openair1/PHY/LTE_TRANSPORT/dlsch_tbs_full.h"
 #include "PHY/phy_extern.h"
+#include "common/utils/thread_pool/task_manager.h"
 
 const char *__asan_default_options()
 {
@@ -787,9 +788,17 @@ int main(int argc, char **argv) {
   proc_rxtx_ue->frame_rx = (subframe<4)?(proc_rxtx->frame_tx-1):(proc_rxtx->frame_tx);
   proc_rxtx_ue->subframe_tx = proc_rxtx->subframe_rx;
   proc_rxtx_ue->subframe_rx = (proc_rxtx->subframe_tx+6)%10;
+ 
+#ifdef TASK_MANAGER_LTE
+  int const n_threads = parse_num_threads("n");
+  proc_rxtx->man = calloc(1, sizeof(task_manager_t));
+  assert(proc_rxtx->man != NULL && "Memory exhausted");
+  init_task_manager(proc_rxtx->man, n_threads);
+#else
   proc_rxtx->threadPool = (tpool_t *)malloc(sizeof(tpool_t));
-  proc_rxtx->respDecode=(notifiedFIFO_t*) malloc(sizeof(notifiedFIFO_t));
   initTpool("n", proc_rxtx->threadPool, true);
+#endif
+  proc_rxtx->respDecode=(notifiedFIFO_t*) malloc(sizeof(notifiedFIFO_t));
   initNotifiedFIFO(proc_rxtx->respDecode);
 
   printf("Init UL hopping UE\n");