du in 2 threads, and fix UL in FS6 split

common/utils/ocp_itti/intertask_interface.h

@@ -285,7 +285,7 @@ typedef struct {
   TASK_DEF(TASK_PDCP_ENB, TASK_PRIORITY_MED, 200, NULL, NULL)   \
   TASK_DEF(TASK_DATA_FORWARDING, TASK_PRIORITY_MED, 200, NULL, NULL)   \
   TASK_DEF(TASK_END_MARKER, TASK_PRIORITY_MED, 200, NULL, NULL)   \
-  TASK_DEF(TASK_RRC_ENB,  TASK_PRIORITY_MED,  200, NULL,NULL)\
+  TASK_DEF(TASK_RRC_ENB,  TASK_PRIORITY_MED,  200, rrc_enb_process_itti_msg,NULL)\
   TASK_DEF(TASK_RAL_ENB,  TASK_PRIORITY_MED,  200, NULL, NULL)  \
   TASK_DEF(TASK_S1AP,     TASK_PRIORITY_MED,  200, NULL, NULL)  \
   TASK_DEF(TASK_X2AP,     TASK_PRIORITY_MED,  200, NULL, NULL)  \

executables/main-fs6.c

@@ -1384,39 +1384,43 @@ void phy_procedures_eNB_TX_tosplit(uint8_t *bufferZone, PHY_VARS_eNB *eNB, L1_rx
   return;
 }
 
-void DL_du_fs6(void *arg) {
-  RU_t *ru = *(RU_t **)arg;
+void *DL_du_fs6(void *arg) {
+  RU_t *ru=(RU_t *)arg;
   static uint64_t lastTS;
   L1_rxtx_proc_t L1_proc= {0};
 
-  for (int i=0; i<ru->num_eNB; i++) {
-    initBufferZone(bufferZone);
-    initStaticTime(begingWait);
-    initRefTimes(fullLoop);
-    pickStaticTime(begingWait);
-    int nb_blocks=receiveSubFrame(&sockFS6, bufferZone, sizeof(bufferZone), CTsentCUv0 );
-    updateTimesReset(begingWait, &fullLoop, 1000, false, "DU wait CU");
-
-    if (nb_blocks > 0) {
-      if ( lastTS+ru->eNB_list[i]->frame_parms.samples_per_tti != hUDP(bufferZone)->timestamp) {
-        LOG_E(HW,"Missed a subframe: expecting: %lu, received %lu\n",
-              lastTS+ru->eNB_list[i]->frame_parms.samples_per_tti,
-              hUDP(bufferZone)->timestamp);
-      }
+  while (1) {
+    for (int i=0; i<ru->num_eNB; i++) {
+      initBufferZone(bufferZone);
+      initStaticTime(begingWait);
+      initRefTimes(fullLoop);
+      pickStaticTime(begingWait);
+      int nb_blocks=receiveSubFrame(&sockFS6, bufferZone, sizeof(bufferZone), CTsentCUv0 );
+      updateTimesReset(begingWait, &fullLoop, 1000, false, "DU wait CU");
+
+      if (nb_blocks > 0) {
+        if ( lastTS+ru->eNB_list[i]->frame_parms.samples_per_tti != hUDP(bufferZone)->timestamp) {
+          LOG_E(HW,"Missed a subframe: expecting: %lu, received %lu\n",
+                lastTS+ru->eNB_list[i]->frame_parms.samples_per_tti,
+                hUDP(bufferZone)->timestamp);
+        }
+
+        lastTS=hUDP(bufferZone)->timestamp;
+        setAllfromTS(hUDP(bufferZone)->timestamp - sf_ahead*ru->eNB_list[i]->frame_parms.samples_per_tti, &L1_proc);
+        phy_procedures_eNB_TX_fromsplit( bufferZone, nb_blocks, ru->eNB_list[i], &L1_proc, 1);
+      } else
+        LOG_E(PHY,"DL not received for subframe\n");
+    }
 
-      lastTS=hUDP(bufferZone)->timestamp;
-      setAllfromTS(hUDP(bufferZone)->timestamp - sf_ahead*ru->eNB_list[i]->frame_parms.samples_per_tti, &L1_proc);
-      phy_procedures_eNB_TX_fromsplit( bufferZone, nb_blocks, ru->eNB_list[i], &L1_proc, 1);
-    } else
-      LOG_E(PHY,"DL not received for subframe\n");
+    feptx_prec(ru, &L1_proc);
+    feptx_ofdm(ru, &L1_proc);
+    tx_rf(ru, &L1_proc);
   }
 
-  feptx_prec(ru, &L1_proc);
-  feptx_ofdm(ru, &L1_proc);
-  tx_rf(ru, &L1_proc);
+  return NULL;
 }
 
-void UL_du_fs6(RU_t *ru, L1_rxtx_proc_t *proc, int frame, int subframe) {
+void UL_du_fs6(RU_t *ru, L1_rxtx_proc_t *proc) {
   initStaticTime(begingWait);
   initRefTimes(fullLoop);
   pickStaticTime(begingWait);
@@ -1560,7 +1564,7 @@ void *du_fs6(void *arg) {
   fill_rf_config(ru,ru->rf_config_file);
   init_frame_parms(&ru->frame_parms,1);
   phy_init_RU(ru);
-  openair0_device_load(&ru->rfdevice,&ru->openair0_cfg);
+  init_rf(ru);
   wait_sync("ru_thread");
   char *remoteIP;
 
@@ -1570,33 +1574,26 @@ void *du_fs6(void *arg) {
     remoteIP=CU_IP;
 
   AssertFatal(createUDPsock(NULL, DU_PORT, remoteIP, CU_PORT, &sockFS6), "");
-  tpool_t pool;
-  tpool_t *Tpool = &pool;
-  char params[]="-1,-1";
-  initTpool(params, Tpool, false);
 
   if (ru->start_rf) {
     if (ru->start_rf(ru) != 0)
       LOG_E(HW,"Could not start the RF device\n");
-    else LOG_I(PHY,"RU %d rf device ready\n",ru->idx);
-  } else LOG_I(PHY,"RU %d no rf device\n",ru->idx);
+    else
+      LOG_I(PHY,"RU %d rf device ready\n",ru->idx);
+  } else
+    LOG_I(PHY,"RU %d no rf device\n",ru->idx);
 
   initStaticTime(begingWait);
   initRefTimes(waitRxAndProcessingUL);
   initRefTimes(fullLoop);
-  L1_rxtx_proc_t L1proc;
-  notifiedFIFO_t nf;
-  initNotifiedFIFO(&nf);
+  pthread_t t;
+  threadCreate(&t, DL_du_fs6, (void *)ru, "MainDuTx", -1, OAI_PRIORITY_RT_MAX);
 
   while(1) {
-    L1_rxtx_proc_t *proc = &L1proc;
+    L1_rxtx_proc_t L1proc;
     updateTimesReset(begingWait, &fullLoop, 1000,  true,"DU for full SubFrame (must be less 1ms)");
     pickStaticTime(begingWait);
-    notifiedFIFO_elt_t *Msg=newNotifiedFIFO_elt(sizeof(ru),0,&nf, DL_du_fs6);
-    *(RU_t **)NotifiedFifoData(Msg)=ru;
-    pushTpool(Tpool, Msg);
-    UL_du_fs6(ru, proc, proc->frame_rx,proc->subframe_rx);
-    notifiedFIFO_elt_t *res=pullTpool(&nf, Tpool);
+    UL_du_fs6(ru, &L1proc);
     updateTimesReset(begingWait, &waitRxAndProcessingUL, 1000,  true,"DU Time in wait Rx + Ul processing");
   }
 

executables/main-ocp.c

@@ -11,10 +11,11 @@
  * The merger of OpenAir central code to this branch
  * should check if these 3 files are modified and analyze if code code has to be copied in here
  */
+#define  _GNU_SOURCE
+#include <pthread.h>
 
 #include <common/utils/LOG/log.h>
 #include <common/utils/system.h>
-
 static int DEFBANDS[] = {7};
 static int DEFENBS[] = {0};
 #include <common/config/config_userapi.h>
@@ -44,7 +45,6 @@ void reset_opp_meas(void) {
 // Fixme: there are many mistakes in the datamodel and in redondant variables
 // TDD is also mode complex
 void setAllfromTS(uint64_t TS, L1_rxtx_proc_t *proc) {
-
   for (int i=0; i < RC.nb_inst; i++) {
     for (int j=0; j<RC.nb_CC[i]; j++) {
       LTE_DL_FRAME_PARMS *fp=&RC.eNB[i][j]->frame_parms;
@@ -96,7 +96,6 @@ void init_eNB_proc(int inst) {
 }
 
 void init_RU_proc(RU_t *ru) {
-
   pthread_t t;
   char *fs6=getenv("fs6");
 
@@ -104,7 +103,7 @@ void init_RU_proc(RU_t *ru) {
     if ( strncasecmp(fs6,"cu", 2) == 0 )
       threadCreate(&t, cu_fs6, (void *)ru, "MainCu", -1, OAI_PRIORITY_RT_MAX);
     else if ( strncasecmp(fs6,"du", 2) == 0 ) {
-      threadCreate(&t, du_fs6, (void *)ru, "MainDu", -1, OAI_PRIORITY_RT_MAX);
+      threadCreate(&t, du_fs6, (void *)ru, "MainDuRx", -1, OAI_PRIORITY_RT_MAX);
     } else
       AssertFatal(false, "environement variable fs6 is not cu or du");
   } else
@@ -415,7 +414,7 @@ void init_precoding_weights(PHY_VARS_eNB *eNB) {
   }
 }
 
-void prach_procedures_ocp(PHY_VARS_eNB *eNB, L1_rxtx_proc_t * proc, int br_flag) {
+void prach_procedures_ocp(PHY_VARS_eNB *eNB, L1_rxtx_proc_t *proc, int br_flag) {
   uint16_t max_preamble[4],max_preamble_energy[4],max_preamble_delay[4],avg_preamble_energy[4];
   RU_t *ru;
   int aa=0;
@@ -436,7 +435,7 @@ void prach_procedures_ocp(PHY_VARS_eNB *eNB, L1_rxtx_proc_t * proc, int br_flag)
 
   // run PRACH detection for CE-level 0 only for now when br_flag is set
   rx_prach(eNB,
-	   proc,
+           proc,
            eNB->RU_list[0],
            &max_preamble[0],
            &max_preamble_energy[0],
@@ -536,7 +535,7 @@ void prach_procedures_ocp(PHY_VARS_eNB *eNB, L1_rxtx_proc_t * proc, int br_flag)
   }
 } // else br_flag
 
-void prach_eNB(PHY_VARS_eNB *eNB, L1_rxtx_proc_t * proc, int frame,int subframe) {
+void prach_eNB(PHY_VARS_eNB *eNB, L1_rxtx_proc_t *proc, int frame,int subframe) {
   // check if we have to detect PRACH first
   if (is_prach_subframe(&eNB->frame_parms, frame,subframe)>0) {
     prach_procedures_ocp(eNB, proc, 0);
@@ -576,7 +575,6 @@ static inline int rxtx(PHY_VARS_eNB *eNB,L1_rxtx_proc_t *proc, char *thread_name
 }
 
 void eNB_top(PHY_VARS_eNB *eNB, L1_rxtx_proc_t *proc, int dummy1, int dummy2,  char *string,RU_t *ru) {
-
   if (!oai_exit) {
     if (rxtx(eNB,proc,string) < 0)
       LOG_E(PHY,"eNB %d CC_id %d failed during execution\n",eNB->Mod_id,eNB->CC_id);
@@ -590,7 +588,7 @@ void rx_rf(RU_t *ru, L1_rxtx_proc_t *proc) {
   int i;
   openair0_timestamp ts=0, timestamp_rx;
   static openair0_timestamp old_ts=0;
-  
+
   for (i=0; i<ru->nb_rx; i++)
     //receive in the next slot
     rxp[i] = (void *)&ru->common.rxdata[i][((proc->subframe_rx+1)%10)*fp->samples_per_tti];
@@ -615,15 +613,17 @@ void rx_rf(RU_t *ru, L1_rxtx_proc_t *proc) {
   }
 
   if (timestamp_rx - old_ts != fp->samples_per_tti) {
-      LOG_E(HW,"impossible shift in rx stream, rx: %ld, previous rx distance: %ld, should be %d\n", timestamp_rx, proc->timestamp_rx - old_ts, fp->samples_per_tti);
-      //ru->ts_offset += (proc->timestamp_rx - old_ts - fp->samples_per_tti);
-      //proc->timestamp_rx = ts-ru->ts_offset;
+    LOG_E(HW,"impossible shift in rx stream, rx: %ld, previous rx distance: %ld, should be %d\n", timestamp_rx, proc->timestamp_rx - old_ts, fp->samples_per_tti);
+    //ru->ts_offset += (proc->timestamp_rx - old_ts - fp->samples_per_tti);
+    //proc->timestamp_rx = ts-ru->ts_offset;
   }
+
   old_ts=timestamp_rx;
   setAllfromTS(timestamp_rx, proc);
 }
 
-void tx_rf(RU_t *ru, L1_rxtx_proc_t *proc) {
+int tx_rf(RU_t *ru, L1_rxtx_proc_t *proc) {
+  int ret=0;
   LTE_DL_FRAME_PARMS *fp = &ru->frame_parms;
   void *txp[ru->nb_tx];
   int i;
@@ -671,29 +671,31 @@ void tx_rf(RU_t *ru, L1_rxtx_proc_t *proc) {
 
     /* add fail safe for late command end */
     // prepare tx buffer pointers
-    ru->rfdevice.trx_write_func(&ru->rfdevice,
-                                proc->timestamp_tx+ru->ts_offset-ru->openair0_cfg.tx_sample_advance-sf_extension,
-                                txp,
-                                siglen+sf_extension,
-                                ru->nb_tx,
-                                flags);
+    ret=ru->rfdevice.trx_write_func(&ru->rfdevice,
+                                    proc->timestamp_tx+ru->ts_offset-ru->openair0_cfg.tx_sample_advance-sf_extension,
+                                    txp,
+                                    siglen+sf_extension,
+                                    ru->nb_tx,
+                                    flags);
     LOG_D(PHY,"[TXPATH] RU %d tx_rf, writing to TS %llu, frame %d, subframe %d\n",ru->idx,
           (long long unsigned int)proc->timestamp_tx,proc->frame_tx,proc->subframe_tx);
   }
+
+  return ret;
 }
 
 static void *ru_thread( void *param ) {
-   setbuf(stdout, NULL);
-   setbuf(stderr, NULL);
+  setbuf(stdout, NULL);
+  setbuf(stderr, NULL);
   RU_t               *ru      = (RU_t *)param;
   L1_rxtx_proc_t L1proc;
-  L1_rxtx_proc_t * proc=&L1proc;
+  L1_rxtx_proc_t *proc=&L1proc;
 
   if (ru->if_south == LOCAL_RF) { // configure RF parameters only
     fill_rf_config(ru,ru->rf_config_file);
     init_frame_parms(&ru->frame_parms,1);
     phy_init_RU(ru);
-    openair0_device_load(&ru->rfdevice,&ru->openair0_cfg);
+    init_rf(ru);
   }
 
   AssertFatal(setup_RU_buffers(ru)==0, "Exiting, cannot initialize RU Buffers\n");
@@ -714,7 +716,7 @@ static void *ru_thread( void *param ) {
 
     // do RX front-end processing (frequency-shift, dft) if needed
     if (ru->feprx)
-      ru->feprx(ru, proc); 
+      ru->feprx(ru, proc);
 
     // At this point, all information for subframe has been received on FH interface
     // If this proc is to provide synchronization, do so
@@ -750,8 +752,18 @@ static void *ru_thread( void *param ) {
   return NULL;
 }
 
+int init_rf(RU_t *ru) {
+  char name[256];
+  pthread_getname_np(pthread_self(),name, 255);
+  pthread_setname_np(pthread_self(),"UHD for OAI");
+  int ret=openair0_device_load(&ru->rfdevice,&ru->openair0_cfg);
+  pthread_setname_np(pthread_self(),name);
+  return ret;
+}
+
 int start_rf(RU_t *ru) {
-  return(ru->rfdevice.trx_start_func(&ru->rfdevice));
+  int ret=ru->rfdevice.trx_start_func(&ru->rfdevice);
+  return ret;
 }
 
 int stop_rf(RU_t *ru) {

executables/split_headers.h

@@ -11,7 +11,8 @@
 #define DU_PORT "8787"
 
 #define MTU 65536
-#define UDP_TIMEOUT 1000L // in micro  second (struct timeval, NOT struct timespec)
+#define UDP_TIMEOUT 900000L // in micro  second (struct timeval, NOT struct timespec)
+// linux may timeout for a much longer time (up to 10ms)
 #define MAX_BLOCKS 16
 #define blockAlign 32 //bytes
 
@@ -236,8 +237,9 @@ void sendFs6Ul(PHY_VARS_eNB *eNB, int UE_id, int harq_pid, int segmentID, int16_
 void *cu_fs6(void *arg);
 void *du_fs6(void *arg);
 void fill_rf_config(RU_t *ru, char *rf_config_file);
+int init_rf(RU_t *ru);
 void rx_rf(RU_t *ru, L1_rxtx_proc_t *proc);
-void tx_rf(RU_t *ru, L1_rxtx_proc_t *proc);
+int tx_rf(RU_t *ru, L1_rxtx_proc_t *proc);
 void common_signal_procedures (PHY_VARS_eNB *eNB,int frame, int subframe);
 void pmch_procedures(PHY_VARS_eNB *eNB,L1_rxtx_proc_t *proc);
 bool dlsch_procedures(PHY_VARS_eNB *eNB,

openair2/RRC/LTE/rrc_eNB.c

@@ -9124,9 +9124,10 @@ rrc_enb_task(
   itti_mark_task_ready(TASK_RRC_ENB);
   LOG_I(RRC,"Entering main loop of RRC message task\n");
 
-  while (1) {
-    (void) rrc_enb_process_itti_msg(NULL);
-  }
+  //while (1) {
+    //(void) rrc_enb_process_itti_msg(NULL);
+  //}
+  return NULL;
 }
 
 /*------------------------------------------------------------------------------*/

targets/COMMON/create_tasks.c

@@ -61,7 +61,7 @@ int create_tasks(uint32_t enb_nb) {
   rc = itti_create_task (TASK_RRC_ENB, rrc_enb_task, NULL);
   AssertFatal(rc >= 0, "Create task for RRC eNB failed\n");
 
-  if (EPC_MODE_ENABLED) {
+  if (EPC_MODE_ENABLED && !fs6Du ) {
     rc = itti_create_task(TASK_SCTP, sctp_eNB_task, NULL);
     AssertFatal(rc >= 0, "Create task for SCTP failed\n");
   }