From 0f207da28a582be26d873c462be9d4db6f77d6ff Mon Sep 17 00:00:00 2001 From: Raymond Knopp Date: Sun, 8 May 2022 18:36:42 +0200 Subject: [PATCH 1/3] initial changes for multi-threaded TX for IF5 --- executables/nr-ru.c | 33 +++--- openair1/PHY/defs_RU.h | 2 + openair1/SCHED_NR/nr_ru_procedures.c | 25 ++++- .../LAYER2/NR_MAC_gNB/gNB_scheduler_dlsch.c | 17 +-- targets/ARCH/COMMON/common_lib.h | 2 +- targets/ARCH/ETHERNET/USERSPACE/LIB/eth_udp.c | 100 +++++++++--------- .../ETHERNET/USERSPACE/LIB/ethernet_lib.h | 2 +- 7 files changed, 92 insertions(+), 89 deletions(-) diff --git a/executables/nr-ru.c b/executables/nr-ru.c index d14b3f1a88..662cbdb9e1 100644 --- a/executables/nr-ru.c +++ b/executables/nr-ru.c @@ -275,15 +275,15 @@ int connect_rau(RU_t *ru) { // southbound IF5 fronthaul for 16-bit OAI format 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 ); - void *buffs[ru->nb_tx]; int offset = ru->nr_frame_parms->get_samples_slot_timestamp(slot,ru->nr_frame_parms,0); - for (int aid=0;aidnb_tx;aid++) buffs[aid]=&ru->common.txdata[aid][offset]; start_meas(&ru->tx_fhaul); - ru->ifdevice.trx_write_func2(&ru->ifdevice, - timestamp, - buffs, - ru->nr_frame_parms->get_samples_per_slot(slot,ru->nr_frame_parms), - 0); + for (int aid=0;aidnb_tx;aid++) + ru->ifdevice.trx_write_func2(&ru->ifdevice, + timestamp, + (void*)&ru->common.txdata[aid][offset], + aid, + ru->nr_frame_parms->get_samples_per_slot(slot,ru->nr_frame_parms), + 0); stop_meas(&ru->tx_fhaul); } @@ -488,7 +488,7 @@ void fh_if5_north_asynch_in(RU_t *ru,int *frame,int *slot) { RU_proc_t *proc = &ru->proc; int tti_tx,frame_tx; openair0_timestamp timestamp_tx; - recv_IF5(ru, ×tamp_tx, *slot, IF5_RRH_GW_DL,1); + AssertFatal(1==0,"Shouldn't get here\n"); // printf("Received subframe %d (TS %llu) from RCC\n",tti_tx,timestamp_tx); frame_tx = (timestamp_tx / (fp->samples_per_subframe*10))&1023; uint32_t idx_sf = timestamp_tx / fp->samples_per_subframe; @@ -565,11 +565,9 @@ void fh_if4p5_north_asynch_in(RU_t *ru,int *frame,int *slot) { } void fh_if5_north_out(RU_t *ru) { - RU_proc_t *proc=&ru->proc; - uint8_t seqno=0; /// **** send_IF5 of rxdata to BBU **** /// VCD_SIGNAL_DUMPER_DUMP_FUNCTION_BY_NAME( VCD_SIGNAL_DUMPER_FUNCTIONS_SEND_IF5, 1 ); - send_IF5(ru, proc->timestamp_rx, proc->tti_rx, &seqno, IF5_RRH_GW_UL); + AssertFatal(1==0,"Shouldn't get here\n"); VCD_SIGNAL_DUMPER_DUMP_FUNCTION_BY_NAME( VCD_SIGNAL_DUMPER_FUNCTIONS_SEND_IF5, 0 ); } @@ -1181,9 +1179,15 @@ void *ru_stats_thread(void *param) { 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); - print_meas(&ru->ofdm_total_stats,"feptx_total",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); + } + else { + print_meas(&ru->ofdm_mod_stats,"feptx_ofdm",NULL,NULL); + print_meas(&ru->ofdm_total_stats,"feptx_total",NULL,NULL); + } } - print_meas(&ru->rx_fhaul,"rx_fhaul",NULL,NULL); print_meas(&ru->tx_fhaul,"tx_fhaul",NULL,NULL); @@ -1851,11 +1855,12 @@ 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->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 ru->fh_south_in = fh_if5_south_in; // synchronous IF5 reception - ru->fh_south_out = fh_if5_south_out; // synchronous IF5 transmission + ru->fh_south_out = (ru->txfh_in_fep) ? 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; diff --git a/openair1/PHY/defs_RU.h b/openair1/PHY/defs_RU.h index 0320697ecf..6078e2f857 100644 --- a/openair1/PHY/defs_RU.h +++ b/openair1/PHY/defs_RU.h @@ -507,6 +507,8 @@ typedef struct RU_t_s { int sf_ahead; /// TX processing advance in slots (for NR) int sl_ahead; + /// flag to indicated TX FH is embedded in TX FEP + int txfh_in_fep; /// FAPI confiuration nfapi_nr_config_request_scf_t config; /// Frame parameters diff --git a/openair1/SCHED_NR/nr_ru_procedures.c b/openair1/SCHED_NR/nr_ru_procedures.c index b90f14a129..ffe847b6ce 100644 --- a/openair1/SCHED_NR/nr_ru_procedures.c +++ b/openair1/SCHED_NR/nr_ru_procedures.c @@ -56,7 +56,7 @@ void nr_feptx0(RU_t *ru,int tti_tx,int first_symbol, int num_symbols, int aa) { NR_DL_FRAME_PARMS *fp = ru->nr_frame_parms; - unsigned int slot_offset,slot_offsetF; + unsigned int slot_offset,slot_offsetF,initial_slot_offset,num_samples;; int slot = tti_tx; @@ -71,17 +71,19 @@ void nr_feptx0(RU_t *ru,int tti_tx,int first_symbol, int num_symbols, int aa) { slot_offset += (idx_sym%(0x7<numerology_index)) ? fp->nb_prefix_samples : fp->nb_prefix_samples0; slot_offset += fp->ofdm_symbol_size*first_symbol; + initial_slot_offset = slot_offset; LOG_D(PHY,"SFN/SF:RU:TX:%d/%d Generating slot %d (first_symbol %d num_symbols %d)\n",ru->proc.frame_tx, ru->proc.tti_tx,slot,first_symbol,num_symbols); - if (fp->Ncp == 1) + if (fp->Ncp == 1) { PHY_ofdm_mod(&ru->common.txdataF_BF[aa][slot_offsetF], (int*)&ru->common.txdata[aa][slot_offset], fp->ofdm_symbol_size, num_symbols, fp->nb_prefix_samples, CYCLIC_PREFIX); - else { + num_samples=num_symbols*(fp->nb_prefix_samples+fp->ofdm_symbol_size); + } else { if (fp->numerology_index != 0) { if (!(slot%(fp->slots_per_subframe/2))&&(first_symbol==0)) { // case where first symbol in slot has longer prefix @@ -98,6 +100,8 @@ void nr_feptx0(RU_t *ru,int tti_tx,int first_symbol, int num_symbols, int aa) { num_symbols-1, fp->nb_prefix_samples, CYCLIC_PREFIX); + num_samples=(num_symbols-1)*(fp->nb_prefix_samples+fp->ofdm_symbol_size)+ + fp->nb_prefix_samples0+fp->ofdm_symbol_size; } else { // all symbols in slot have shorter prefix PHY_ofdm_mod(&ru->common.txdataF_BF[aa][slot_offsetF], @@ -106,6 +110,7 @@ void nr_feptx0(RU_t *ru,int tti_tx,int first_symbol, int num_symbols, int aa) { num_symbols, fp->nb_prefix_samples, CYCLIC_PREFIX); + num_samples=num_symbols*(fp->nb_prefix_samples+fp->ofdm_symbol_size); } } // numerology_index!=0 else { //numerology_index == 0 @@ -130,10 +135,20 @@ void nr_feptx0(RU_t *ru,int tti_tx,int first_symbol, int num_symbols, int aa) { slot_offset += fp->nb_prefix_samples0+fp->ofdm_symbol_size; slot_offsetF += fp->ofdm_symbol_size; } - } - } + } // for(idx_symbol.. + num_samples=slot_offset-initial_slot_offset; + } // numerology 0 } + if (ru->txfh_in_fep) { + ru->ifdevice.trx_write_func2(&ru->ifdevice, + ru->proc.timestamp_tx, + (void*)&ru->common.txdata[aa][initial_slot_offset], + aa, + num_samples, + 0); + + } //VCD_SIGNAL_DUMPER_DUMP_FUNCTION_BY_NAME(VCD_SIGNAL_DUMPER_FUNCTIONS_PHY_PROCEDURES_RU_FEPTX_OFDM+(first_symbol!=0?1:0), 0); } diff --git a/openair2/LAYER2/NR_MAC_gNB/gNB_scheduler_dlsch.c b/openair2/LAYER2/NR_MAC_gNB/gNB_scheduler_dlsch.c index c751b33ac0..1c0c23bb0d 100644 --- a/openair2/LAYER2/NR_MAC_gNB/gNB_scheduler_dlsch.c +++ b/openair2/LAYER2/NR_MAC_gNB/gNB_scheduler_dlsch.c @@ -608,25 +608,10 @@ bool allocate_dl_retransmission(module_id_t module_id, /* the retransmission will use a different time domain allocation, check * that we have enough resources */ NR_pdsch_semi_static_t temp_ps = *ps; -<<<<<<< HEAD + const uint16_t slbitmap = SL_to_bitmap(ps->startSymbolIndex, ps->nrOfSymbols); nr_set_pdsch_semi_static(sib1,scc, cg, sched_ctrl->active_bwp, bwpd,tda, ps->nrOfLayers, sched_ctrl, &temp_ps); while (rbStart < bwpSize && !(rballoc_mask[rbStart]&SL_to_bitmap(temp_ps.startSymbolIndex, temp_ps.nrOfSymbols))) -======= - - nr_set_pdsch_semi_static(sib1, - scc, - cg, - sched_ctrl->active_bwp, - bwpd, - tda, - ps->nrOfLayers, - sched_ctrl, - &temp_ps); - - const uint16_t slbitmap = SL_to_bitmap(temp_ps.startSymbolIndex, temp_ps.nrOfSymbols); - while (rbStart < bwpSize && (rballoc_mask[rbStart] & slbitmap) != slbitmap) ->>>>>>> origin/develop rbStart++; while (rbStart + rbSize < bwpSize && (rballoc_mask[rbStart + rbSize] & slbitmap) == slbitmap) diff --git a/targets/ARCH/COMMON/common_lib.h b/targets/ARCH/COMMON/common_lib.h index e30a78ef75..f3dcc35e53 100644 --- a/targets/ARCH/COMMON/common_lib.h +++ b/targets/ARCH/COMMON/common_lib.h @@ -423,7 +423,7 @@ struct openair0_device_t { @param antenna_id index of the antenna if the device has multiple anteannas @param flags flags must be set to TRUE if timestamp parameter needs to be applied */ - int (*trx_write_func2)(openair0_device *device, openair0_timestamp timestamp, void **buff, int nsamps, int flags); + int (*trx_write_func2)(openair0_device *device, openair0_timestamp timestamp, void *buff, int aid, int nsamps, int flags); /*! \brief Receive samples from hardware. * Read \ref nsamps samples from each channel to buffers. buff[0] is the array for diff --git a/targets/ARCH/ETHERNET/USERSPACE/LIB/eth_udp.c b/targets/ARCH/ETHERNET/USERSPACE/LIB/eth_udp.c index dcb23b5970..f043b6ded4 100644 --- a/targets/ARCH/ETHERNET/USERSPACE/LIB/eth_udp.c +++ b/targets/ARCH/ETHERNET/USERSPACE/LIB/eth_udp.c @@ -289,101 +289,98 @@ int trx_eth_write_udp_IF4p5(openair0_device *device, openair0_timestamp timestam return (bytes_sent); } -int trx_eth_write_udp(openair0_device *device, openair0_timestamp timestamp, void **buff, int nsamps, int flags) { +int trx_eth_write_udp(openair0_device *device, openair0_timestamp timestamp, void *buff, int aid, int nsamps, int flags) { int bytes_sent=0; eth_state_t *eth = (eth_state_t*)device->priv; int sendto_flag =0; fhstate_t *fhstate = &device->fhstate; - AssertFatal((nsamps&255) == 0,"transport size should be a multiple of 256 samples, have %d\n",nsamps); //sendto_flag|=flags; eth->tx_nsamps=nsamps; - int ntx = device->openair0_cfg->tx_num_channels; - void *buff2[ntx]; + void *buff2; #if defined(__x86_64) || defined(__i386__) #ifdef __AVX2__ int nsamps2 = 256>>3; - __m256i buff_tx[ntx][nsamps2+1]; - for (int aid=0;aid>2; - __m128i buff_tx[ntx][nsamps2+2]; - for (int aid=0;aid>2; - int16x8_t buff_tx[ntx][nsamps2+2]; - for (int aid=0;aid>6; - *(uint8_t *)(buff2[aid] + 3) = 0;// 1024&0xff; - // ECPRI PC_ID (2 bytes) - *(uint16_t *)(buff2[aid] + 4) = aid; + /* construct application header */ + // ECPRI Protocol revision + reserved bits (1 byte) + *(uint8_t *)buff2 = ECPRIREV; + // ECPRI Message type (1 byte) + *(uint8_t *)(buff2 + 1) = 64; + // ECPRI PC_ID (2 bytes) + *(uint16_t *)(buff2 + 4) = aid; - } openair0_timestamp TS = timestamp + fhstate->TS0; TS = 6*device->sampling_rate_ratio_d*(TS/device->sampling_rate_ratio_n); TS -= device->txrx_offset; int TSinc = 6*256*device->sampling_rate_ratio_d/device->sampling_rate_ratio_n; + int len=256; for (int offset=0;offset>8; + *(uint8_t *)(buff2 + 3) = len&0xff; + //LOG_I(PHY,"TS %llu aa %d : offset %d, len %d\n",(unsigned long long)TS,aid,offset,len); // bring TX data into 12 MSBs #if defined(__x86_64__) || defined(__i386__) #ifdef __AVX2__ - __m256i *buff256 = (__m256i *)&((uint32_t*)buff[aid])[offset]; - for (int j=0; j<32; j+=8) { - buff_tx[aid][1+j] = _mm256_slli_epi16(buff256[j],4); - buff_tx[aid][2+j] = _mm256_slli_epi16(buff256[j+1],4); - buff_tx[aid][3+j] = _mm256_slli_epi16(buff256[j+2],4); - buff_tx[aid][4+j] = _mm256_slli_epi16(buff256[j+3],4); - buff_tx[aid][5+j] = _mm256_slli_epi16(buff256[j+4],4); - buff_tx[aid][6+j] = _mm256_slli_epi16(buff256[j+5],4); - buff_tx[aid][7+j] = _mm256_slli_epi16(buff256[j+6],4); - buff_tx[aid][8+j] = _mm256_slli_epi16(buff256[j+7],4); - } + __m256i *buff256 = (__m256i *)&((uint32_t*)buff)[offset]; + for (int j=0; j<32; j+=8) { + buff_tx[1+j] = _mm256_slli_epi16(buff256[j],4); + buff_tx[2+j] = _mm256_slli_epi16(buff256[j+1],4); + buff_tx[3+j] = _mm256_slli_epi16(buff256[j+2],4); + buff_tx[4+j] = _mm256_slli_epi16(buff256[j+3],4); + buff_tx[5+j] = _mm256_slli_epi16(buff256[j+4],4); + buff_tx[6+j] = _mm256_slli_epi16(buff256[j+5],4); + buff_tx[7+j] = _mm256_slli_epi16(buff256[j+6],4); + buff_tx[8+j] = _mm256_slli_epi16(buff256[j+7],4); + } #else - __m128i *buff128 = (__m128i *)&buff[aid][offset]; - for (int j=0; j<64; j++) buff_tx[aid][2+j] = _mm_slli_epi16(buff128[j],4); + __m128i *buff128 = (__m128i *)&buff[offset]; + for (int j=0; j<64; j++) buff_tx[2+j] = _mm_slli_epi16(buff128[j],4); #endif #elif defined(__arm__) - int16x8_t *buff128 = (__int16x8_t*)&buff[aid][offset]; - for (int j=0; j<64; j++) buff_tx[aid][2+j] = vshlq_n_s16(((int16x8_t *)buff128)[j],4); + int16x8_t *buff128 = (__int16x8_t*)&buff[offset]; + for (int j=0; j<64; j++) buff_tx[2+j] = vshlq_n_s16(((int16x8_t *)buff128)[j],4); #endif - /* Send packet */ - bytes_sent = sendto(eth->sockfdd[aid%eth->num_fd], - buff2[aid], - UDP_PACKET_SIZE_BYTES(256), - sendto_flag, - (struct sockaddr*)ð->dest_addrd, - eth->addr_len); + bytes_sent = sendto(eth->sockfdd[aid%eth->num_fd], + buff2, + UDP_PACKET_SIZE_BYTES(len>>2), + sendto_flag, + (struct sockaddr*)ð->dest_addrd, + eth->addr_len); - if ( bytes_sent == -1) { + if ( bytes_sent == -1) { eth->num_tx_errors++; perror("ETHERNET WRITE: "); exit(-1); - } else { + } else { eth->tx_actual_nsamps=bytes_sent>>2; eth->tx_count++; - } } - } + } return (bytes_sent-APP_HEADER_SIZE_BYTES)>>2; } @@ -426,12 +423,11 @@ void *udp_read_thread(void *arg) { } sleep(1); } - + return(0); } int trx_eth_read_udp(openair0_device *device, openair0_timestamp *timestamp, uint32_t **buff, int nsamps) { - eth_state_t *eth = (eth_state_t*)device->priv; fhstate_t *fhstate = &device->fhstate; openair0_timestamp prev_read_TS= fhstate->TS_read, min_TS; // block until FH is ready diff --git a/targets/ARCH/ETHERNET/USERSPACE/LIB/ethernet_lib.h b/targets/ARCH/ETHERNET/USERSPACE/LIB/ethernet_lib.h index 72a3cfef63..ccf67b4ca0 100644 --- a/targets/ARCH/ETHERNET/USERSPACE/LIB/ethernet_lib.h +++ b/targets/ARCH/ETHERNET/USERSPACE/LIB/ethernet_lib.h @@ -241,7 +241,7 @@ int ethernet_tune(openair0_device *device, unsigned int option, int value); * @ingroup _oai */ int eth_socket_init_udp(openair0_device *device); -int trx_eth_write_udp(openair0_device *device, openair0_timestamp timestamp, void **buff, int nsamps, int flags); +int trx_eth_write_udp(openair0_device *device, openair0_timestamp timestamp, void *buf, int aid, int nsamps, int flags); int trx_eth_read_udp(openair0_device *device, openair0_timestamp *timestamp, uint32_t **buff, int nsamps); From 2b426c0cc90528147b60fa0cb4d5396bec0f6116 Mon Sep 17 00:00:00 2001 From: Raymond Knopp Date: Sat, 28 May 2022 07:47:41 +0200 Subject: [PATCH 2/3] added threadPool for IF5 TX path --- executables/nr-ru.c | 7 +- targets/ARCH/COMMON/common_lib.h | 8 +++ targets/ARCH/ETHERNET/USERSPACE/LIB/eth_udp.c | 69 ++++++++++++++----- .../ETHERNET/USERSPACE/LIB/ethernet_lib.c | 17 ++++- .../ETHERNET/USERSPACE/LIB/ethernet_lib.h | 22 ++++++ 5 files changed, 104 insertions(+), 19 deletions(-) diff --git a/executables/nr-ru.c b/executables/nr-ru.c index 662cbdb9e1..a9c7c65b42 100644 --- a/executables/nr-ru.c +++ b/executables/nr-ru.c @@ -276,6 +276,7 @@ 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;aidnb_tx;aid++) ru->ifdevice.trx_write_func2(&ru->ifdevice, @@ -1276,9 +1277,9 @@ void *ru_thread( void *param ) { memcpy((void *)&ru->config,(void *)&RC.gNB[0]->gNB_config,sizeof(ru->config)); if(emulate_rf) { - fill_rf_config(ru,ru->rf_config_file); nr_init_frame_parms(&ru->config, fp); nr_dump_frame_parms(fp); + fill_rf_config(ru,ru->rf_config_file); nr_phy_init_RU(ru); if (setup_RU_buffers(ru)!=0) { @@ -1870,6 +1871,7 @@ void set_function_spec_param(RU_t *ru) { ru->ifdevice.eth_params = &ru->eth_params; ru->ifdevice.configure_rru = configure_ru; + printf("starting transport : rx_num_antennas %d, tx_num_antennas %d\n",ru->openair0_cfg.rx_num_channels,ru->openair0_cfg.tx_num_channels); ret = openair0_transport_load(&ru->ifdevice,&ru->openair0_cfg,&ru->eth_params); printf("openair0_transport_init returns %d for ru_id %u\n", ret, ru->idx); @@ -1990,7 +1992,8 @@ void init_NR_RU(char *rf_config_file) { } } } - + ru->openair0_cfg.rx_num_channels = ru->nb_rx; + ru->openair0_cfg.tx_num_channels = ru->nb_tx; //LOG_I(PHY,"Initializing RRU descriptor %d : (%s,%s,%d)\n",ru_id,ru_if_types[ru->if_south],NB_timing[ru->if_timing],ru->function); set_function_spec_param(ru); LOG_I(PHY,"Starting ru_thread %d\n",ru_id); diff --git a/targets/ARCH/COMMON/common_lib.h b/targets/ARCH/COMMON/common_lib.h index f3dcc35e53..d8a9d11c8a 100644 --- a/targets/ARCH/COMMON/common_lib.h +++ b/targets/ARCH/COMMON/common_lib.h @@ -37,6 +37,7 @@ #include #include #include "record_player.h" +#include /* default name of shared library implementing the radio front end */ #define OAI_RF_LIBNAME "oai_device" @@ -325,6 +326,7 @@ typedef struct fhstate_s { openair0_timestamp olddeltaTS[8]; openair0_timestamp oldTS[8]; openair0_timestamp TS_read; + int first_read; uint32_t *buff[8]; uint32_t buff_size; int r[8]; @@ -368,6 +370,12 @@ struct openair0_device_t { /*!brief pointer to FH state, used in ECPRI split 8*/ fhstate_t fhstate; + /*!brief threadpool for UDP write*/ + tpool_t *threadPool; + + /*!brief message response for notification fifo*/ + notifiedFIFO_t *respudpTX; + /*!brief Used in ECPRI split 8 to indicate numerator of sampling rate ratio*/ int sampling_rate_ratio_n; diff --git a/targets/ARCH/ETHERNET/USERSPACE/LIB/eth_udp.c b/targets/ARCH/ETHERNET/USERSPACE/LIB/eth_udp.c index f043b6ded4..79b0c84ac3 100644 --- a/targets/ARCH/ETHERNET/USERSPACE/LIB/eth_udp.c +++ b/targets/ARCH/ETHERNET/USERSPACE/LIB/eth_udp.c @@ -152,11 +152,11 @@ int eth_socket_init_udp(openair0_device *device) { printf("ETHERNET: Cannot set SO_REUSEPORT option on socket (user %d)",i); exit(0); } -#ifdef SO_ATTACH_REUSEPORT_EBPF +#ifdef SO_ATTACH_REUSEPORT_EBPF*/ struct sock_filter code[]={ - { BPF_LD | BPF_W | BPF_ABS, 0, 0, SKF_AD_OFF + SKF_AD_CPU }, - /* return A */ - { BPF_RET | BPF_A, 0, 0, 0 }, + { BPF_LD | BPF_W | BPF_ABS, 0, 0, SKF_AD_OFF + SKF_AD_CPU }, // A = #cpu + { BPF_ALU | BPF_MOD | BPF_K, 0, 0, eth->num_fd}, // A = A % group_size + { BPF_RET | BPF_A, 0, 0, 0 }, // return A }; struct sock_fprog bpf = { .len = sizeof(code)/sizeof(struct sock_filter), @@ -166,6 +166,7 @@ int eth_socket_init_udp(openair0_device *device) { printf("ETHERNET: Cannot set SO_ATTACH_REUSEPORT_EBPF option on socket (user %d)",i); exit(0); } + else printf("ETHERNET: set SO_ATTACH_REUSEPORT_EBPF option on socket (user %d)\n",i); #endif } @@ -205,7 +206,7 @@ int trx_eth_read_udp_IF4p5(openair0_device *device, openair0_timestamp *timestam while(bytes_received == -1) { again: - bytes_received = recvfrom(eth->sockfdd[cc%eth->num_fd], + bytes_received = recvfrom(eth->sockfdd[0/*cc%eth->num_fd*/], buff[0], packet_size, 0, @@ -236,7 +237,6 @@ int trx_eth_read_udp_IF4p5(openair0_device *device, openair0_timestamp *timestam eth->rx_count++; } } - eth->rx_nsamps = nsamps; return(bytes_received); } @@ -271,7 +271,7 @@ int trx_eth_write_udp_IF4p5(openair0_device *device, openair0_timestamp timestam eth->tx_nsamps = nblocks; - bytes_sent = sendto(eth->sockfdd[cc%eth->num_fd], + bytes_sent = sendto(eth->sockfdd[0/*cc%eth->num_fd*/], buff[0], packet_size, 0, @@ -289,7 +289,17 @@ int trx_eth_write_udp_IF4p5(openair0_device *device, openair0_timestamp timestam return (bytes_sent); } -int trx_eth_write_udp(openair0_device *device, openair0_timestamp timestamp, void *buff, int aid, int nsamps, int flags) { + + +void *trx_eth_write_udp_cmd(void *arg) { + + udpTXelem_t *udpTXelem=(udpTXelem_t*)arg; + + openair0_device *device=udpTXelem->device; + openair0_timestamp timestamp = udpTXelem->timestamp; + void *buff = udpTXelem->buff; + int aid = udpTXelem->buff; + int nsamps = udpTXelem->nsamps; int bytes_sent=0; eth_state_t *eth = (eth_state_t*)device->priv; @@ -365,7 +375,7 @@ int trx_eth_write_udp(openair0_device *device, openair0_timestamp timestamp, voi #endif /* Send packet */ - bytes_sent = sendto(eth->sockfdd[aid%eth->num_fd], + bytes_sent = sendto(eth->sockfdd[0/*aid%eth->num_fd*/], buff2, UDP_PACKET_SIZE_BYTES(len>>2), sendto_flag, @@ -381,9 +391,22 @@ int trx_eth_write_udp(openair0_device *device, openair0_timestamp timestamp, voi eth->tx_count++; } } - return (bytes_sent-APP_HEADER_SIZE_BYTES)>>2; } +int trx_eth_write_udp(openair0_device *device, openair0_timestamp timestamp, void *buff, int aid, int nsamps, int flags) { + + 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); + + udptxelem->device = device; + udptxelem->timestamp = timestamp; + udptxelem->buff = buff; + udptxelem->aid = aid; + udptxelem->nsamps = nsamps; + udptxelem->flags = flags; + pushTpool(device->threadPool,req); +} extern int oai_exit; void *udp_read_thread(void *arg) { @@ -394,11 +417,11 @@ void *udp_read_thread(void *arg) { openair0_device *device=u->device; fhstate_t *fhstate = &device->fhstate; char buffer[UDP_PACKET_SIZE_BYTES(256)]; - + int first_read=0; while (oai_exit == 0) { LOG_I(PHY,"UDP read thread %d, waiting for start sampling_rate_d %d, sampling_rate_n %d\n",u->thread_id,device->sampling_rate_ratio_n,device->sampling_rate_ratio_d); while (fhstate->active > 0) { - size_t count = recvfrom(((eth_state_t*)device->priv)->sockfdd[u->thread_id], + size_t count = recvfrom(((eth_state_t*)device->priv)->sockfdd[0], buffer,sizeof(buffer),0, (struct sockaddr *)&((eth_state_t*)device->priv)->dest_addrd, (socklen_t *)&((eth_state_t*)device->priv)->addr_len); @@ -409,7 +432,8 @@ void *udp_read_thread(void *arg) { if ((int)count <= 0) continue; AssertFatal(aid < 8,"Cannot handle more than 8 antennas, got aid %d\n",aid); fhstate->r[aid]=1; - if (aid==0 && fhstate->TS[0] == 0) fhstate->TS0 = TS; + if (aid==0 && first_read == 0) fhstate->TS0 = TS; + first_read = 1; /* store the timestamp value from packet's header */ fhstate->TS[aid] = TS; int64_t offset = TS - fhstate->TS0; @@ -420,6 +444,7 @@ void *udp_read_thread(void *arg) { memcpy((void*)(device->openair0_cfg->rxbase[aid]+offset), (void*)&buffer[APP_HEADER_SIZE_BYTES], count-APP_HEADER_SIZE_BYTES); + LOG_D(PHY,"thread_id %d, aid %d, TS %llu, TS0 %llu, offset %d\n",u->thread_id,aid,(unsigned long long)TS,fhstate->TS0,offset); } sleep(1); } @@ -438,14 +463,26 @@ int trx_eth_read_udp(openair0_device *device, openair0_timestamp *timestamp, uin min_TS = fhstate->TS[0]; for (int i=1;iopenair0_cfg->rx_num_channels;i++) min_TS = min(min_TS,fhstate->TS[i]); // poll/sleep until we accumulated enough samples on each antenna port - while (min_TS < fhstate->TS0+prev_read_TS + nsamps) { + int count=0; + while (fhstate->first_read == 1 && min_TS < (fhstate->TS0+prev_read_TS + nsamps)) { usleep(100); min_TS = fhstate->TS[0]; for (int i=1;iopenair0_cfg->rx_num_channels;i++) min_TS = min(min_TS,fhstate->TS[i]); + count++; } - *timestamp = fhstate->TS_read; - fhstate->TS_read = prev_read_TS + nsamps; + if (fhstate->first_read == 0) { + *timestamp = min_TS - fhstate->TS0; + fhstate->TS_read = *timestamp+nsamps; + LOG_D(PHY,"first read : TS_read %llu, TS %llu state (%d,%d,%d,%d,%d,%d,%d,%d)\n",fhstate->TS_read,*timestamp, + fhstate->r[0],fhstate->r[1],fhstate->r[2],fhstate->r[3],fhstate->r[4],fhstate->r[5],fhstate->r[6],fhstate->r[7]); + } + else { + *timestamp = fhstate->TS_read; + fhstate->TS_read = prev_read_TS + nsamps; + LOG_D(PHY,"TS_read %llu, min_TS %llu, prev_read_TS %llu, nsamps %d, fhstate->TS0+prev_TS+nsamps %llu, wait count %d x 100us\n",fhstate->TS_read,min_TS,prev_read_TS,nsamps,fhstate->TS0+prev_read_TS+nsamps,count); + } + fhstate->first_read = 1; return (nsamps); } diff --git a/targets/ARCH/ETHERNET/USERSPACE/LIB/ethernet_lib.c b/targets/ARCH/ETHERNET/USERSPACE/LIB/ethernet_lib.c index b5a21c5d16..dc80c8dcb4 100644 --- a/targets/ARCH/ETHERNET/USERSPACE/LIB/ethernet_lib.c +++ b/targets/ARCH/ETHERNET/USERSPACE/LIB/ethernet_lib.c @@ -64,7 +64,7 @@ int trx_eth_start(openair0_device *device) AssertFatal(device->thirdparty_init != NULL, "device->thirdparty_init is null\n"); AssertFatal(device->thirdparty_init(device) == 0, "third-party init failed\n"); device->openair0_cfg->samples_per_packet = 256; - eth->num_fd = max(device->openair0_cfg->rx_num_channels,device->openair0_cfg->tx_num_channels); + 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;iopenair0_cfg->rx_num_channels;i++) { @@ -112,6 +112,20 @@ int trx_eth_start(openair0_device *device) } else AssertFatal(1==0,"num_rb_dl %d not ok with ECPRI\n",device->openair0_cfg->num_rb_dl); } + int threadCnt = device->openair0_cfg->tx_num_channels; + 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; icputhreadPool = (tpool_t*)malloc(sizeof(tpool_t)); + initTpool(pool, device->threadPool, cpumeas(CPUMEAS_GETSTATE)); + // ULSCH decoder result FIFO + device->respudpTX = (notifiedFIFO_t*) malloc(sizeof(notifiedFIFO_t)); + initNotifiedFIFO(device->respudpTX); } /* initialize socket */ if (eth->flags == ETH_RAW_MODE) { @@ -502,6 +516,7 @@ int transport_init(openair0_device *device, device->trx_ctlsend_func = trx_eth_ctlsend_udp; device->trx_ctlrecv_func = trx_eth_ctlrecv_udp; memset((void*)&device->fhstate,0,sizeof(device->fhstate)); + printf("Setting %d RX channels to waiting\n",openair0_cfg->rx_num_channels); for (int i=openair0_cfg->rx_num_channels;i<8;i++) device->fhstate.r[i] = 1; } else if (eth->flags == ETH_RAW_IF4p5_MODE) { device->trx_write_func = trx_eth_write_raw_IF4p5; diff --git a/targets/ARCH/ETHERNET/USERSPACE/LIB/ethernet_lib.h b/targets/ARCH/ETHERNET/USERSPACE/LIB/ethernet_lib.h index ccf67b4ca0..60c10abd8a 100644 --- a/targets/ARCH/ETHERNET/USERSPACE/LIB/ethernet_lib.h +++ b/targets/ARCH/ETHERNET/USERSPACE/LIB/ethernet_lib.h @@ -41,6 +41,7 @@ #include #include #include +#include #define MAX_INST 4 #define DEFAULT_IF "lo" @@ -209,6 +210,27 @@ typedef struct { short q; } iqoai_t ; +typedef struct udpTXelem_s { + openair0_device *device; + openair0_timestamp timestamp; + void *buff; + int aid; + int nsamps; + int flags; +} udpTXelem_t; + +struct udpTXReqId { + uint64_t TS; + int aid; + int length; + uint16_t spare; +} __attribute__((packed)); + +union udpTXReqUnion { + struct udpTXReqId s; + uint64_t p; +}; + void dump_packet(char *title, unsigned char* pkt, int bytes, unsigned int tx_rx_flag); unsigned short calc_csum (unsigned short *buf, int nwords); void dump_dev(openair0_device *device); From ca782ef2cba24e5b9dc068e49e9d60f7be00c21e Mon Sep 17 00:00:00 2001 From: Raymond Knopp Date: Mon, 30 May 2022 07:21:39 +0200 Subject: [PATCH 3/3] added threadPool to RU entity (FEP and FEP TX) --- executables/nr-ru.c | 73 ++++-- openair1/PHY/defs_RU.h | 30 ++- openair1/SCHED/ru_procedures.c | 44 ++-- openair1/SCHED_NR/nr_ru_procedures.c | 246 +++++++++++++----- openair1/SCHED_NR/sched_nr.h | 2 + targets/ARCH/ETHERNET/USERSPACE/LIB/eth_udp.c | 11 +- .../ETHERNET/USERSPACE/LIB/ethernet_lib.c | 2 +- 7 files changed, 304 insertions(+), 104 deletions(-) diff --git a/executables/nr-ru.c b/executables/nr-ru.c index a9c7c65b42..5e562eff31 100644 --- a/executables/nr-ru.c +++ b/executables/nr-ru.c @@ -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;aidnb_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;aidnb_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; icputhreadPool = (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); diff --git a/openair1/PHY/defs_RU.h b/openair1/PHY/defs_RU.h index 6078e2f857..20505ca6f4 100644 --- a/openair1/PHY/defs_RU.h +++ b/openair1/PHY/defs_RU.h @@ -39,6 +39,7 @@ #include "time_meas.h" #include "defs_common.h" #include "nfapi_nr_interface_scf.h" +#include #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; diff --git a/openair1/SCHED/ru_procedures.c b/openair1/SCHED/ru_procedures.c index dc84279a96..73e174cef2 100644 --- a/openair1/SCHED/ru_procedures.c +++ b/openair1/SCHED/ru_procedures.c @@ -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); diff --git a/openair1/SCHED_NR/nr_ru_procedures.c b/openair1/SCHED_NR/nr_ru_procedures.c index ffe847b6ce..22d4b97a74 100644 --- a/openair1/SCHED_NR/nr_ru_procedures.c +++ b/openair1/SCHED_NR/nr_ru_procedures.c @@ -527,35 +527,25 @@ void nr_feptx_prec(RU_t *ru,int frame_tx,int tti_tx) { stop_meas(&ru->precoding_stats); } -void nr_fep0(RU_t *ru, int first_half) { +void nr_fep0(RU_t *ru, int aid) { - uint8_t start_symbol, end_symbol, l, aa; + int l; RU_proc_t *proc = &ru->proc; NR_DL_FRAME_PARMS *fp = ru->nr_frame_parms; - if(first_half == 1) { - start_symbol = 0; - end_symbol = NR_SYMBOLS_PER_SLOT/2; - } else { - start_symbol = NR_SYMBOLS_PER_SLOT/2; - end_symbol = NR_SYMBOLS_PER_SLOT; - } - - LOG_D(PHY,"In fep0 for slot = %d, first_half = %d, start_symbol = %d, end_symbol = %d\n", proc->tti_rx, first_half, start_symbol, end_symbol); + LOG_D(PHY,"In fep0 for slot = %d, aid = %d\n", proc->tti_rx, aid); // printf("fep0: slot %d\n",slot); VCD_SIGNAL_DUMPER_DUMP_FUNCTION_BY_NAME(VCD_SIGNAL_DUMPER_FUNCTIONS_PHY_PROCEDURES_RU_FEPRX+proc->tti_rx, 1); int offset = (proc->tti_rx&3) * fp->symbols_per_slot * fp->ofdm_symbol_size; - for (l = start_symbol; l < end_symbol; l++) { - for (aa = 0; aa < fp->nb_antennas_rx; aa++) { - nr_slot_fep_ul(fp, - ru->common.rxdata[aa], - &ru->common.rxdataF[aa][offset], - l, - proc->tti_rx, - ru->N_TA_offset); - } + for (l = 0; l < NR_SYMBOLS_PER_SLOT; l++) { + nr_slot_fep_ul(fp, + ru->common.rxdata[aid], + &ru->common.rxdataF[aid][offset], + l, + proc->tti_rx, + ru->N_TA_offset); } VCD_SIGNAL_DUMPER_DUMP_FUNCTION_BY_NAME(VCD_SIGNAL_DUMPER_FUNCTIONS_PHY_PROCEDURES_RU_FEPRX+proc->tti_rx, 0); } @@ -563,31 +553,32 @@ void nr_fep0(RU_t *ru, int first_half) { static void *nr_feprx_thread(void *param) { - RU_t *ru = (RU_t *)param; + feprx_cmd_t *feprx_cmd = (feprx_cmd_t*)param; + RU_t *ru = feprx_cmd->ru; RU_proc_t *proc = &ru->proc; - + int aid=feprx_cmd->aid; while (!oai_exit) { - if (wait_on_condition(&proc->mutex_fep,&proc->cond_fep,&proc->instance_cnt_fep,"NR feprx thread")<0) break; + if (wait_on_condition(&proc->mutex_fep[aid-1],&proc->cond_fep[aid-1],&proc->instance_cnt_fep[aid-1],"NR feprx thread")<0) break; if (oai_exit) break; VCD_SIGNAL_DUMPER_DUMP_FUNCTION_BY_NAME( VCD_SIGNAL_DUMPER_FUNCTIONS_PHY_PROCEDURES_RU_FEPRX1, 1 ); - nr_fep0(ru, 0); + nr_fep0(ru, aid); 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,"nr_feprx thread")<0) break; + if (release_thread(&proc->mutex_fep[aid-1],&proc->instance_cnt_fep[aid-1],"nr_feprx thread")<0) break; - if (pthread_cond_signal(&proc->cond_fep) != 0) { + if (pthread_cond_signal(&proc->cond_fep[aid-1]) != 0) { printf("[gNB] ERROR pthread_cond_signal for nr_feprx thread exit\n"); exit_fun( "ERROR pthread_cond_signal" ); return NULL; } } - + free(param); return NULL; } @@ -595,13 +586,17 @@ static void *nr_feprx_thread(void *param) { void nr_init_feprx_thread(RU_t *ru) { RU_proc_t *proc = &ru->proc; - - proc->instance_cnt_fep = -1; - - pthread_mutex_init( &proc->mutex_fep, NULL); - pthread_cond_init( &proc->cond_fep, NULL); - - threadCreate(&proc->pthread_fep, nr_feprx_thread, (void*)ru, "feprx", -1, OAI_PRIORITY_RT); + AssertFatal(ru->nb_rx<8,"FEP is limited to 8 antennas maximum\n"); + feprx_cmd_t *feprx_cmd; + for (int aid=0;aid<(ru->nb_rx-1);aid++) { + proc->instance_cnt_fep[aid]= -1; + pthread_mutex_init( &proc->mutex_fep[aid], NULL); + pthread_cond_init( &proc->cond_fep[aid], NULL); + feprx_cmd = malloc(sizeof(feprx_cmd_t)); + feprx_cmd->aid=aid+1; + feprx_cmd->ru = ru; + threadCreate(&proc->pthread_fep[aid], nr_feprx_thread, (void*)feprx_cmd, "feprx", -1, OAI_PRIORITY_RT); + } } @@ -619,41 +614,42 @@ void nr_fep_full_2thread(RU_t *ru, int slot) { if (ru->idx == 0) VCD_SIGNAL_DUMPER_DUMP_FUNCTION_BY_NAME( VCD_SIGNAL_DUMPER_FUNCTIONS_PHY_PROCEDURES_RU_FEPRX, 1 ); wait.tv_sec=0; - wait.tv_nsec=5000000L; + wait.tv_nsec=7000000L; VCD_SIGNAL_DUMPER_DUMP_FUNCTION_BY_NAME( VCD_SIGNAL_DUMPER_FUNCTIONS_PHY_PROCEDURES_RU_FEPRX, 1 ); start_meas(&ru->ofdm_demod_stats); + for (int aid=0;aidnb_rx-1;aid++) { + if (pthread_mutex_timedlock(&proc->mutex_fep[aid],&wait) != 0) { + printf("[RU] ERROR timeout on pthread_mutex_lock for nr_feprx thread aid %d (IC %d)\n", aid,proc->instance_cnt_fep[aid]); + exit_fun( "error locking mutex_fep" ); + return; + } - if (pthread_mutex_timedlock(&proc->mutex_fep,&wait) != 0) { - printf("[RU] ERROR pthread_mutex_lock for nr_feprx thread (IC %d)\n", proc->instance_cnt_fep); - exit_fun( "error locking mutex_fep" ); - return; - } - - if (proc->instance_cnt_fep==0) { - printf("[RU] FEP thread busy\n"); - exit_fun("FEP thread busy"); - pthread_mutex_unlock( &proc->mutex_fep ); - return; - } + if (proc->instance_cnt_fep[aid]==0) { + printf("[RU] FEP thread busy\n"); + exit_fun("FEP thread busy"); + pthread_mutex_unlock( &proc->mutex_fep[aid] ); + return; + } - ++proc->instance_cnt_fep; + ++proc->instance_cnt_fep[aid]; - if (pthread_cond_signal(&proc->cond_fep) != 0) { - printf("[RU] ERROR pthread_cond_signal for nr_feprx thread\n"); - exit_fun( "ERROR pthread_cond_signal" ); - return; - } - //start_meas(&ru->ofdm_demod_wakeup_stats); + if (pthread_cond_signal(&proc->cond_fep[aid]) != 0) { + printf("[RU] ERROR pthread_cond_signal for nr_feprx thread\n"); + exit_fun( "ERROR pthread_cond_signal" ); + return; + } + //start_meas(&ru->ofdm_demod_wakeup_stats); - pthread_mutex_unlock( &proc->mutex_fep ); - - // call second slot in this symbol - nr_fep0(ru, 1); + pthread_mutex_unlock( &proc->mutex_fep[aid] ); + } + // call first antenna in this thread + nr_fep0(ru, 0); start_meas(&ru->ofdm_demod_wait_stats); - wait_on_busy_condition(&proc->mutex_fep,&proc->cond_fep,&proc->instance_cnt_fep,"nr_feprx thread"); + for (int aid=0;aidnb_rx-1;aid++) + wait_on_busy_condition(&proc->mutex_fep[aid],&proc->cond_fep[aid],&proc->instance_cnt_fep[aid],"nr_feprx 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); @@ -700,3 +696,135 @@ void nr_fep_full(RU_t *ru, int slot) { } + + +void nr_feptx(void *arg) { + + feptx_cmd_t *feptx = (feptx_cmd_t *)arg; + + RU_t *ru = feptx->ru; + int slot = feptx->slot; + int aa = feptx->aid; + NR_DL_FRAME_PARMS *fp = ru->nr_frame_parms; + + int txdataF_offset = (slot*fp->samples_per_slot_wCP); + ////////////precoding//////////// + VCD_SIGNAL_DUMPER_DUMP_FUNCTION_BY_NAME(VCD_SIGNAL_DUMPER_FUNCTIONS_PHY_PROCEDURES_RU_FEPTX_PREC+feptx->aid , 1); + + if (aa==0) start_meas(&ru->precoding_stats); + + if (ru->do_precoding == 1) { + for(int i=0; inb_log_antennas; ++i) { + memcpy((void*) &ru->common.beam_id[i][slot*fp->symbols_per_slot], + (void*) &ru->gNB_list[0]->common_vars.beam_id[i][slot*fp->symbols_per_slot], + (fp->symbols_per_slot)*sizeof(uint8_t)); + } + } + + if (ru->nb_tx == 1 && ru->nb_log_antennas == 1) { + memcpy((void*)&ru->common.txdataF_BF[0][0], + (void*)&ru->gNB_list[0]->common_vars.txdataF[0][txdataF_offset], + (fp->samples_per_slot_wCP)*sizeof(int32_t)); + } + else if (ru->do_precoding == 0) { + int gNB_tx = ru->gNB_list[0]->frame_parms.nb_antennas_tx; + memcpy((void*)&ru->common.txdataF_BF[aa][0], + (void*)&ru->gNB_list[0]->common_vars.txdataF[aa%gNB_tx][txdataF_offset], + (fp->samples_per_slot_wCP)*sizeof(int32_t)); + } + else { + int32_t ***bw = ru->beam_weights[0]; + for(int i=0; isymbols_per_slot; ++i){ + nr_beam_precoding(ru->gNB_list[0]->common_vars.txdataF, + ru->common.txdataF_BF, + fp, + bw, + slot, + i, + aa, + ru->nb_log_antennas, + txdataF_offset);//here + } + } + if (aa==0) stop_meas(&ru->precoding_stats); + VCD_SIGNAL_DUMPER_DUMP_FUNCTION_BY_NAME(VCD_SIGNAL_DUMPER_FUNCTIONS_PHY_PROCEDURES_RU_FEPTX_PREC+feptx->aid , 0); + + ////////////FEPTX//////////// + if (aa==0) start_meas(&ru->ofdm_mod_stats); + nr_feptx0(ru,slot,0,fp->symbols_per_slot,aa); + if (aa==0) stop_meas(&ru->ofdm_mod_stats); +} +void nr_feptx_tp(RU_t *ru, int frame_tx, int slot) { + + nfapi_nr_config_request_scf_t *cfg = &ru->gNB_list[0]->gNB_config; + int nbfeptx=0; + if (nr_slot_select(cfg,frame_tx,slot) == NR_UPLINK_SLOT) return; + for (int aa=0; aanb_tx; aa++) memset(ru->common.txdataF[aa],0,ru->nr_frame_parms->samples_per_slot_wCP*sizeof(int32_t)); + + if (ru->idx == 0) VCD_SIGNAL_DUMPER_DUMP_FUNCTION_BY_NAME( VCD_SIGNAL_DUMPER_FUNCTIONS_PHY_PROCEDURES_RU_FEPTX_OFDM, 1 ); + start_meas(&ru->ofdm_total_stats); + for (int aid=0;aidnb_tx;aid++) { + notifiedFIFO_elt_t *req=newNotifiedFIFO_elt(sizeof(feptx_cmd_t), 2000 + aid,ru->respfeptx,nr_feptx); + feptx_cmd_t *feptx_cmd=(feptx_cmd_t*)NotifiedFifoData(req); + feptx_cmd->aid = aid; + feptx_cmd->ru = ru; + feptx_cmd->slot = slot; + pushTpool(ru->threadPool,req); + nbfeptx++; + } + while (nbfeptx>0) { + notifiedFIFO_elt_t *req=pullTpool(ru->respfeptx, ru->threadPool); + delNotifiedFIFO_elt(req); + nbfeptx--; + } + stop_meas(&ru->ofdm_total_stats); + if (ru->idx == 0) VCD_SIGNAL_DUMPER_DUMP_FUNCTION_BY_NAME( VCD_SIGNAL_DUMPER_FUNCTIONS_PHY_PROCEDURES_RU_FEPTX_OFDM, 0 ); +} + +void nr_fep(void* arg) { + + feprx_cmd_t *feprx_cmd = (feprx_cmd_t *)arg; + + RU_t *ru = feprx_cmd->ru; + int aid = feprx_cmd->aid; + int tti_rx = feprx_cmd->slot; + + NR_DL_FRAME_PARMS *fp = ru->nr_frame_parms; + + LOG_D(PHY,"In nr_fep for aid %d, slot = %d\n", aid, tti_rx); + + VCD_SIGNAL_DUMPER_DUMP_FUNCTION_BY_NAME(VCD_SIGNAL_DUMPER_FUNCTIONS_PHY_PROCEDURES_RU_FEPRX+aid, 1); + + int offset = (tti_rx&3) * fp->symbols_per_slot * fp->ofdm_symbol_size; + for (int l = 0; l < fp->symbols_per_slot; l++) + nr_slot_fep_ul(fp, + ru->common.rxdata[aid], + &ru->common.rxdataF[aid][offset], + l, + tti_rx, + ru->N_TA_offset); + VCD_SIGNAL_DUMPER_DUMP_FUNCTION_BY_NAME(VCD_SIGNAL_DUMPER_FUNCTIONS_PHY_PROCEDURES_RU_FEPRX+aid, 0); +} +void nr_fep_tp(RU_t *ru, int slot) { + + int nbfeprx=0; + if (ru->idx == 0) VCD_SIGNAL_DUMPER_DUMP_FUNCTION_BY_NAME( VCD_SIGNAL_DUMPER_FUNCTIONS_PHY_PROCEDURES_RU_FEPRX, 1 ); + start_meas(&ru->ofdm_demod_stats); + for (int aid=0;aidnb_rx;aid++) { + notifiedFIFO_elt_t *req=newNotifiedFIFO_elt(sizeof(feprx_cmd_t), 1000 + aid,ru->respfeprx,nr_fep); + feprx_cmd_t *feprx_cmd=(feprx_cmd_t*)NotifiedFifoData(req); + feprx_cmd->aid = aid; + feprx_cmd->ru = ru; + feprx_cmd->slot = ru->proc.tti_rx; + pushTpool(ru->threadPool,req); + nbfeprx++; + } + while (nbfeprx>0) { + notifiedFIFO_elt_t *req=pullTpool(ru->respfeprx, ru->threadPool); + delNotifiedFIFO_elt(req); + nbfeprx--; + } + stop_meas(&ru->ofdm_demod_stats); + if (ru->idx == 0) VCD_SIGNAL_DUMPER_DUMP_FUNCTION_BY_NAME( VCD_SIGNAL_DUMPER_FUNCTIONS_PHY_PROCEDURES_RU_FEPRX, 0 ); +} + diff --git a/openair1/SCHED_NR/sched_nr.h b/openair1/SCHED_NR/sched_nr.h index 2124397c47..cfd911b36b 100644 --- a/openair1/SCHED_NR/sched_nr.h +++ b/openair1/SCHED_NR/sched_nr.h @@ -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); diff --git a/targets/ARCH/ETHERNET/USERSPACE/LIB/eth_udp.c b/targets/ARCH/ETHERNET/USERSPACE/LIB/eth_udp.c index 79b0c84ac3..f91537ba31 100644 --- a/targets/ARCH/ETHERNET/USERSPACE/LIB/eth_udp.c +++ b/targets/ARCH/ETHERNET/USERSPACE/LIB/eth_udp.c @@ -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; diff --git a/targets/ARCH/ETHERNET/USERSPACE/LIB/ethernet_lib.c b/targets/ARCH/ETHERNET/USERSPACE/LIB/ethernet_lib.c index dc80c8dcb4..fbe9aed34b 100644 --- a/targets/ARCH/ETHERNET/USERSPACE/LIB/ethernet_lib.c +++ b/targets/ARCH/ETHERNET/USERSPACE/LIB/ethernet_lib.c @@ -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;iopenair0_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;