Commit ca782ef2 authored by Raymond Knopp's avatar Raymond Knopp

added threadPool to RU entity (FEP and FEP TX)

parent 2b426c0c
......@@ -58,6 +58,8 @@
unsigned short config_frames[4] = {2,9,11,13};
#endif
#define USE_TPOOL 1
/* these variables have to be defined before including ENB_APP/enb_paramdef.h and GNB_APP/gnb_paramdef.h */
static int DEFBANDS[] = {7};
static int DEFENBS[] = {0};
......@@ -276,7 +278,6 @@ int connect_rau(RU_t *ru) {
void fh_if5_south_out(RU_t *ru, int frame, int slot, uint64_t timestamp) {
if (ru == RC.ru[0]) VCD_SIGNAL_DUMPER_DUMP_VARIABLE_BY_NAME( VCD_SIGNAL_DUMPER_VARIABLES_TRX_TST, ru->proc.timestamp_tx&0xffffffff );
int offset = ru->nr_frame_parms->get_samples_slot_timestamp(slot,ru->nr_frame_parms,0);
return;
start_meas(&ru->tx_fhaul);
for (int aid=0;aid<ru->nb_tx;aid++)
ru->ifdevice.trx_write_func2(&ru->ifdevice,
......@@ -319,7 +320,6 @@ void fh_if5_south_in(RU_t *ru,
proc->frame_rx = ((proc->timestamp_rx-ru->ts_offset) / (fp->samples_per_subframe*10))&1023;
proc->tti_rx = fp->get_slot_from_timestamp(proc->timestamp_rx-ru->ts_offset,fp);
LOG_D(PHY,"IF5 %d.%d => RX %d.%d first_rx %d\n",*frame,*tti,proc->frame_rx,proc->tti_rx,proc->first_rx);
if (proc->first_rx == 0) {
if (proc->tti_rx != *tti) {
......@@ -338,6 +338,7 @@ void fh_if5_south_in(RU_t *ru,
}
stop_meas(&ru->rx_fhaul);
LOG_I(PHY,"IF5 %d.%d => RX %d.%d first_rx %d: time %f\n",*frame,*tti,proc->frame_rx,proc->tti_rx,proc->first_rx,ru->rx_fhaul.p_time/(cpu_freq_GHz*1000.0));
VCD_SIGNAL_DUMPER_DUMP_VARIABLE_BY_NAME( VCD_SIGNAL_DUMPER_VARIABLES_TRX_TS, proc->timestamp_rx&0xffffffff );
VCD_SIGNAL_DUMPER_DUMP_FUNCTION_BY_NAME( VCD_SIGNAL_DUMPER_FUNCTIONS_RECV_IF5, 0 );
......@@ -1179,7 +1180,6 @@ void *ru_stats_thread(void *param) {
if (ru->feptx_ofdm) {
print_meas(&ru->precoding_stats,"feptx_prec",NULL,NULL);
print_meas(&ru->txdataF_copy_stats,"txdataF_copy",NULL,NULL);
print_meas(&ru->ofdm_mod_stats,"feptx_ofdm",NULL,NULL);
if (ru->txfh_in_fep) {
print_meas(&ru->ofdm_mod_stats,"feptx_ofdm (with txfh)",NULL,NULL);
print_meas(&ru->ofdm_total_stats,"feptx_total (with txfh)",NULL,NULL);
......@@ -1560,9 +1560,10 @@ void init_RU_proc(RU_t *ru) {
if (opp_enabled == 1)
threadCreate( &ru->ru_stats_thread, ru_stats_thread, (void *)ru,"ru_stats", -1, OAI_PRIORITY_RT );
if (get_thread_worker_conf() == WORKER_ENABLE) {
#ifndef USE_TPOOL
if (ru->feprx) nr_init_feprx_thread(ru);
if (ru->feptx_ofdm) nr_init_feptx_thread(ru);
#endif
}
}
......@@ -1574,15 +1575,18 @@ void kill_NR_RU_proc(int inst) {
pthread_join(proc->pthread_FH, NULL);
if (get_nprocs() >= 2) {
#ifndef USE_TPOOL
if (ru->feprx) {
pthread_mutex_lock(&proc->mutex_fep);
proc->instance_cnt_fep = 0;
pthread_mutex_unlock(&proc->mutex_fep);
pthread_cond_signal(&proc->cond_fep);
LOG_D(PHY, "Joining pthread_fep\n");
pthread_join(proc->pthread_fep, NULL);
pthread_mutex_destroy(&proc->mutex_fep);
pthread_cond_destroy(&proc->cond_fep);
for (int aid=0;aid<ru->nb_rx;aid++) {
pthread_mutex_lock(&proc->mutex_fep[aid]);
proc->instance_cnt_fep[aid] = 0;
pthread_mutex_unlock(&proc->mutex_fep[aid]);
pthread_cond_signal(&proc->cond_fep[aid]);
LOG_D(PHY, "Joining pthread_fep %d\n",aid);
pthread_join(proc->pthread_fep[aid], NULL);
pthread_mutex_destroy(&proc->mutex_fep[aid]);
pthread_cond_destroy(&proc->cond_fep[aid]);
}
}
if (ru->feptx_ofdm) {
......@@ -1595,8 +1599,8 @@ void kill_NR_RU_proc(int inst) {
pthread_mutex_destroy(&proc->mutex_feptx);
pthread_cond_destroy(&proc->cond_feptx);
}
#endif
}
if (opp_enabled) {
LOG_D(PHY, "Joining ru_stats_thread\n");
pthread_join(ru->ru_stats_thread, NULL);
......@@ -1804,8 +1808,13 @@ void set_function_spec_param(RU_t *ru) {
ru->fh_north_out = fh_if4p5_north_out; // send_IF4p5 on reception
ru->fh_south_out = tx_rf; // send output to RF
ru->fh_north_asynch_in = fh_if4p5_north_asynch_in; // TX packets come asynchronously
ru->feprx = (get_thread_worker_conf() == WORKER_ENABLE) ? nr_fep_full_2thread : nr_fep_full; // RX DFTs
#ifdef USE_TPOOL
ru->feprx = (get_thread_worker_conf() == WORKER_ENABLE) ? nr_fep_tp : nr_fep_full; // this is frequency-shift + DFTs
ru->feptx_ofdm = (get_thread_worker_conf() == WORKER_ENABLE) ? nr_feptx_tp : nr_feptx_ofdm; // this is fep with idft only (no precoding in RRU)
#else
ru->feprx = (get_thread_worker_conf() == WORKER_ENABLE) ? nr_fep_full_2thread : nr_fep_full; // this is frequency-shift + DFTs
ru->feptx_ofdm = (get_thread_worker_conf() == WORKER_ENABLE) ? nr_feptx_ofdm_2thread : nr_feptx_ofdm; // this is fep with idft only (no precoding in RRU)
#endif
ru->feptx_prec = NULL;
ru->nr_start_if = nr_start_if; // need to start the if interface for if4p5
ru->ifdevice.host_type = RRU_HOST;
......@@ -1825,8 +1834,13 @@ void set_function_spec_param(RU_t *ru) {
malloc_IF4p5_buffer(ru);
} else if (ru->function == gNodeB_3GPP) {
ru->do_prach = 0; // no prach processing in RU
ru->feprx = (get_thread_worker_conf() == WORKER_ENABLE) ? nr_fep_full_2thread : nr_fep_full; // RX DFTs
#ifdef USE_TPOOL
ru->feprx = (get_thread_worker_conf() == WORKER_ENABLE) ? nr_fep_tp : nr_fep_full; // this is frequency-shift + DFTs
ru->feptx_ofdm = (get_thread_worker_conf() == WORKER_ENABLE) ? nr_feptx_tp: nr_feptx_ofdm; // this is fep with idft and precoding
#else
ru->feprx = (get_thread_worker_conf() == WORKER_ENABLE) ? nr_fep_full_2thread : nr_fep_full; // this is frequency-shift + DFTs
ru->feptx_ofdm = (get_thread_worker_conf() == WORKER_ENABLE) ? nr_feptx_ofdm_2thread : nr_feptx_ofdm; // this is fep with idft and precoding
#endif
ru->feptx_prec = (get_thread_worker_conf() == WORKER_ENABLE) ? NULL : nr_feptx_prec; // this is fep with idft and precoding
ru->fh_north_in = NULL; // no incoming fronthaul from north
ru->fh_north_out = NULL; // no outgoing fronthaul to north
......@@ -1856,12 +1870,18 @@ void set_function_spec_param(RU_t *ru) {
case REMOTE_IF5: // the remote unit is IF5 RRU
ru->do_prach = 0;
// ru->txfh_in_fep = 1;
ru->txfh_in_fep = 1;
#ifdef USE_TPOOL
ru->feprx = (get_thread_worker_conf() == WORKER_ENABLE) ? nr_fep_tp : nr_fep_full; // this is frequency-shift + DFTs
ru->feptx_prec = (get_thread_worker_conf() == WORKER_ENABLE) ? NULL : nr_feptx_prec; // need to do transmit Precoding + IDFTs
ru->feptx_ofdm = (get_thread_worker_conf() == WORKER_ENABLE) ? nr_feptx_tp: nr_feptx_ofdm; // need to do transmit Precoding + IDFTs
#else
ru->feprx = (get_thread_worker_conf() == WORKER_ENABLE) ? nr_fep_full_2thread : nr_fep_full; // this is frequency-shift + DFTs
ru->feptx_prec = (get_thread_worker_conf() == WORKER_ENABLE) ? NULL : nr_feptx_prec; // need to do transmit Precoding + IDFTs
ru->feptx_ofdm = (get_thread_worker_conf() == WORKER_ENABLE) ? nr_feptx_ofdm_2thread : nr_feptx_ofdm; // need to do transmit Precoding + IDFTs
#endif
ru->fh_south_in = fh_if5_south_in; // synchronous IF5 reception
ru->fh_south_out = (ru->txfh_in_fep) ? NULL : fh_if5_south_out; // synchronous IF5 transmission
ru->fh_south_out = (ru->txfh_in_fep>0) ? NULL : fh_if5_south_out; // synchronous IF5 transmission
ru->fh_south_asynch_in = NULL; // no asynchronous UL
ru->start_rf = ru->eth_params.transp_preference == ETH_UDP_IF5_ECPRI_MODE ? start_streaming : NULL;
ru->stop_rf = NULL;
......@@ -1998,6 +2018,25 @@ void init_NR_RU(char *rf_config_file) {
set_function_spec_param(ru);
LOG_I(PHY,"Starting ru_thread %d\n",ru_id);
init_RU_proc(ru);
int threadCnt = ru->nb_rx+ru->nb_tx;
if (threadCnt < 2) LOG_E(PHY,"Number of threads for gNB should be more than 1. Allocated only %d\n",threadCnt);
char pool[80];
sprintf(pool,"-1");
int s_offset = 0;
for (int icpu=1; icpu<threadCnt; icpu++) {
sprintf(pool+2+s_offset,",-1");
s_offset += 3;
}
#ifdef USE_TPOOL
ru->threadPool = (tpool_t*)malloc(sizeof(tpool_t));
initTpool(pool, ru->threadPool, cpumeas(CPUMEAS_GETSTATE));
// FEP RX result FIFO
ru->respfeprx = (notifiedFIFO_t*) malloc(sizeof(notifiedFIFO_t));
initNotifiedFIFO(ru->respfeprx);
// FEP TX result FIFO
ru->respfeptx = (notifiedFIFO_t*) malloc(sizeof(notifiedFIFO_t));
initNotifiedFIFO(ru->respfeptx);
#endif
} // for ru_id
// sleep(1);
......
......@@ -39,6 +39,7 @@
#include "time_meas.h"
#include "defs_common.h"
#include "nfapi_nr_interface_scf.h"
#include <common/utils/threadPool/thread-pool.h>
#define MAX_BANDS_PER_RRU 4
#define MAX_RRU_CONFIG_SIZE 1024
......@@ -188,6 +189,20 @@ typedef struct RU_feptx_t_s{
int index;
}RU_feptx_t;
typedef struct {
int aid;
struct RU_t_s *ru;
int start_symbol;
int end_symbol;
int slot;
} feprx_cmd_t;
typedef struct {
int aid;
struct RU_t_s *ru;
int slot;
} feptx_cmd_t;
typedef struct {
int frame;
int slot;
......@@ -246,7 +261,7 @@ typedef struct RU_proc_t_s {
/// \internal This variable is protected by \ref mutex_asynch_rxtx.
int instance_cnt_asynch_rxtx;
/// \internal This variable is protected by \ref mutex_fep
int instance_cnt_fep;
int instance_cnt_fep[8];
/// \internal This variable is protected by \ref mutex_feptx
int instance_cnt_feptx;
/// \internal This variable is protected by \ref mutex_ru_thread
......@@ -265,7 +280,7 @@ typedef struct RU_proc_t_s {
/// pthread struct for RU synch thread
pthread_t pthread_synch;
/// pthread struct for RU RX FEP worker thread
pthread_t pthread_fep;
pthread_t pthread_fep[8];
/// pthread struct for RU TX FEP worker thread
pthread_t pthread_feptx;
/// pthread struct for emulated RF
......@@ -318,7 +333,7 @@ typedef struct RU_proc_t_s {
/// condition variable for asynch RX/TX thread
pthread_cond_t cond_asynch_rxtx;
/// condition varible for RU RX FEP thread
pthread_cond_t cond_fep;
pthread_cond_t cond_fep[8];
/// condition varible for RU TX FEP thread
pthread_cond_t cond_feptx;
/// condition varible for emulated RF
......@@ -345,7 +360,7 @@ typedef struct RU_proc_t_s {
/// mutex for asynch RX/TX thread
pthread_mutex_t mutex_asynch_rxtx;
/// mutex for fep RX worker thread
pthread_mutex_t mutex_fep;
pthread_mutex_t mutex_fep[8];
/// mutex for fep TX worker thread
pthread_mutex_t mutex_feptx;
/// mutex for ru_thread
......@@ -433,6 +448,8 @@ typedef enum {
typedef struct RU_t_s {
/// ThreadPool for RU
tpool_t *threadPool;
/// index of this ru
uint32_t idx;
/// pointer to first RU
......@@ -647,6 +664,11 @@ typedef struct RU_t_s {
uint64_t if_frequency;
/// UL IF frequency offset to DL IF frequency in Hz
int if_freq_offset;
/// to signal end of feprx
notifiedFIFO_t *respfeprx;
/// to signal end of feptx
notifiedFIFO_t *respfeptx;
} RU_t;
......
......@@ -566,15 +566,15 @@ static void *fep_thread(void *param)
while (!oai_exit) {
if (wait_on_condition(&proc->mutex_fep,&proc->cond_fep,&proc->instance_cnt_fep,"fep thread")<0) break;
if (wait_on_condition(&proc->mutex_fep[0],&proc->cond_fep[0],&proc->instance_cnt_fep[0],"fep thread")<0) break;
if (oai_exit) break;
//stop_meas(&ru->ofdm_demod_wakeup_stats);
//VCD_SIGNAL_DUMPER_DUMP_FUNCTION_BY_NAME( VCD_SIGNAL_DUMPER_FUNCTIONS_PHY_PROCEDURES_RU_FEPRX1, 1 );
fep0(ru,0);
//VCD_SIGNAL_DUMPER_DUMP_FUNCTION_BY_NAME( VCD_SIGNAL_DUMPER_FUNCTIONS_PHY_PROCEDURES_RU_FEPRX1, 0 );
if (release_thread(&proc->mutex_fep,&proc->instance_cnt_fep,"fep thread")<0) break;
if (release_thread(&proc->mutex_fep[0],&proc->instance_cnt_fep[0],"fep thread")<0) break;
if (pthread_cond_signal(&proc->cond_fep) != 0) {
if (pthread_cond_signal(&proc->cond_fep[0]) != 0) {
printf("[eNB] ERROR pthread_cond_signal for fep thread exit\n");
exit_fun( "ERROR pthread_cond_signal" );
return NULL;
......@@ -607,26 +607,26 @@ void init_fep_thread(RU_t *ru,
{
RU_proc_t *proc = &ru->proc;
proc->instance_cnt_fep = -1;
proc->instance_cnt_fep[0] = -1;
pthread_mutex_init( &proc->mutex_fep, NULL);
pthread_cond_init( &proc->cond_fep, NULL);
pthread_mutex_init( &proc->mutex_fep[0], NULL);
pthread_cond_init( &proc->cond_fep[0], NULL);
threadCreate(&proc->pthread_fep, fep_thread, (void*)ru, "fep", -1, OAI_PRIORITY_RT);
threadCreate(&proc->pthread_fep[0], fep_thread, (void*)ru, "fep", -1, OAI_PRIORITY_RT);
}
extern void kill_fep_thread(RU_t *ru)
{
RU_proc_t *proc = &ru->proc;
pthread_mutex_lock( &proc->mutex_fep );
proc->instance_cnt_fep = 0;
pthread_cond_signal(&proc->cond_fep);
pthread_mutex_unlock( &proc->mutex_fep );
pthread_mutex_lock( &proc->mutex_fep[0] );
proc->instance_cnt_fep[0] = 0;
pthread_cond_signal(&proc->cond_fep[0]);
pthread_mutex_unlock( &proc->mutex_fep[0] );
LOG_D(PHY, "Joining pthread_fep\n");
pthread_join(proc->pthread_fep, NULL);
pthread_mutex_destroy( &proc->mutex_fep );
pthread_cond_destroy( &proc->cond_fep );
pthread_join(proc->pthread_fep[0], NULL);
pthread_mutex_destroy( &proc->mutex_fep[0] );
pthread_cond_destroy( &proc->cond_fep[0] );
}
......@@ -666,35 +666,35 @@ void ru_fep_full_2thread(RU_t *ru,
VCD_SIGNAL_DUMPER_DUMP_FUNCTION_BY_NAME( VCD_SIGNAL_DUMPER_FUNCTIONS_PHY_PROCEDURES_RU_FEPRX+ru->idx, 1 );
start_meas(&ru->ofdm_demod_stats);
if (pthread_mutex_timedlock(&proc->mutex_fep,&wait) != 0) {
printf("[RU] ERROR pthread_mutex_lock for fep thread (IC %d)\n", proc->instance_cnt_fep);
if (pthread_mutex_timedlock(&proc->mutex_fep[0],&wait) != 0) {
printf("[RU] ERROR pthread_mutex_lock for fep thread (IC %d)\n", proc->instance_cnt_fep[0]);
exit_fun( "error locking mutex_fep" );
return;
}
if (proc->instance_cnt_fep==0) {
if (proc->instance_cnt_fep[0]==0) {
printf("[RU] FEP thread busy\n");
exit_fun("FEP thread busy");
pthread_mutex_unlock( &proc->mutex_fep );
pthread_mutex_unlock( &proc->mutex_fep[0] );
return;
}
++proc->instance_cnt_fep;
++proc->instance_cnt_fep[0];
if (pthread_cond_signal(&proc->cond_fep) != 0) {
if (pthread_cond_signal(&proc->cond_fep[0]) != 0) {
printf("[RU] ERROR pthread_cond_signal for fep thread\n");
exit_fun( "ERROR pthread_cond_signal" );
return;
}
//start_meas(&ru->ofdm_demod_wakeup_stats);
pthread_mutex_unlock( &proc->mutex_fep );
pthread_mutex_unlock( &proc->mutex_fep[0] );
// call second slot in this symbol
fep0(ru,1);
start_meas(&ru->ofdm_demod_wait_stats);
wait_on_busy_condition(&proc->mutex_fep,&proc->cond_fep,&proc->instance_cnt_fep,"fep thread");
wait_on_busy_condition(&proc->mutex_fep[0],&proc->cond_fep[0],&proc->instance_cnt_fep[0],"fep thread");
stop_meas(&ru->ofdm_demod_wait_stats);
if(opp_enabled == 1 && ru->ofdm_demod_wakeup_stats.p_time>30*3000){
print_meas_now(&ru->ofdm_demod_wakeup_stats,"fep wakeup",stderr);
......
This diff is collapsed.
......@@ -52,6 +52,8 @@ void nr_feptx_prec_control(RU_t *ru,int frame,int tti_tx);
void nr_init_feprx_thread(RU_t *ru);
void nr_fep_full(RU_t *ru, int slot);
void nr_fep_full_2thread(RU_t *ru, int slot);
void nr_fep_tp(RU_t *ru, int slot);
void nr_feptx_tp(RU_t *ru, int frame_tx, int slot);
void feptx_prec(RU_t *ru,int frame_tx,int tti_tx);
int nr_phy_init_RU(RU_t *ru);
void nr_phy_free_RU(RU_t *ru);
......
......@@ -395,17 +395,26 @@ void *trx_eth_write_udp_cmd(void *arg) {
int trx_eth_write_udp(openair0_device *device, openair0_timestamp timestamp, void *buff, int aid, int nsamps, int flags) {
#ifdef USE_TPOOL
union udpTXReqUnion id = {.s={(uint64_t)timestamp,aid,nsamps,0}};
notifiedFIFO_elt_t *req=newNotifiedFIFO_elt(sizeof(udpTXelem_t), id.p, device->respudpTX, trx_eth_write_udp_cmd);
udpTXelem_t * udptxelem=(udpTXelem_t *) NotifiedFifoData(req);
#else
udpTXelem_t udptxelem0;
udpTXelem_t *udptxelem = &udptxelem0;
#endif
udptxelem->device = device;
udptxelem->timestamp = timestamp;
udptxelem->buff = buff;
udptxelem->aid = aid;
udptxelem->nsamps = nsamps;
udptxelem->flags = flags;
#ifdef USE_TPOOL
pushTpool(device->threadPool,req);
#else
trx_eth_write_udp_cmd(udptxelem);
#endif
}
extern int oai_exit;
......
......@@ -67,7 +67,7 @@ int trx_eth_start(openair0_device *device)
eth->num_fd = 1; //max(device->openair0_cfg->rx_num_channels,device->openair0_cfg->tx_num_channels);
printf("Creating %d UDP threads ...\n",device->openair0_cfg->rx_num_channels);
udp_read_t *u[device->openair0_cfg->rx_num_channels];
for (int i=0;i<device->openair0_cfg->rx_num_channels;i++) {
for (int i=0;i<(device->openair0_cfg->rx_num_channels);i++) {
u[i] = malloc(sizeof(udp_read_t));
u[i]->thread_id=i;
u[i]->device = device;
......
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment