maki/openairinterface5g
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge remote-tracking branch 'origin/develop' into if5_ECPRI_rework

Browse Source
This commit is contained in:
Raymond Knopp
2022-07-30 19:36:27 +02:00
parent 6236fe8e77 5091682551
commit 3760ffa4af
507 changed files with 151617 additions and 14431 deletions
Download Patch File Download Diff File Expand all files Collapse all files
targets/ARCH/ETHERNET/USERSPACE/LIB/eth_udp.c
+78 -59
View File
@@ -47,6 +47,7 @@
#include "common_lib.h"
#include "ethernet_lib.h"
#include "common/utils/threadPool/thread-pool.h"
//#define DEBUG 1
@@ -291,16 +292,13 @@ int trx_eth_write_udp_IF4p5(openair0_device *device, openair0_timestamp timestam
void *trx_eth_write_udp_cmd(void *arg) {
udpTXelem_t *udpTXelem=(udpTXelem_t*)arg;
void *trx_eth_write_udp_cmd(udpTXelem_t *udpTXelem) {
openair0_device *device=udpTXelem->device;
openair0_timestamp timestamp = udpTXelem->timestamp;
void *buff = udpTXelem->buff;
int aid = udpTXelem->buff;
void **buff = udpTXelem->buff;
int nsamps = udpTXelem->nsamps;
int nant = udpTXelem->nant;
int bytes_sent=0;
eth_state_t *eth = (eth_state_t*)device->priv;
int sendto_flag =0;
@@ -309,6 +307,9 @@ void *trx_eth_write_udp_cmd(void *arg) {
//sendto_flag|=flags;
eth->tx_nsamps=nsamps;
struct timespec txstart;
clock_gettime(CLOCK_MONOTONIC, &txstart);
LOG_D(PHY,"Starting TX FH for TS %llu @ %d\n",(unsigned long long)timestamp,txstart.tv_nsec);
void *buff2;
#if defined(__x86_64) || defined(__i386__)
#ifdef __AVX2__
@@ -318,7 +319,7 @@ void *trx_eth_write_udp_cmd(void *arg) {
#else
int nsamps2 = 256>>2;
__m128i buff_tx[nsamps2+2];
buff=(void*)&buff_tx[2] - APP_HEADER_SIZE_BYTES;
buff2=(void*)&buff_tx[2] - APP_HEADER_SIZE_BYTES;
#endif
#elif defined(__arm__) || defined(__aarch64__)
int nsamps2 = 256>>2;
@@ -333,96 +334,114 @@ void *trx_eth_write_udp_cmd(void *arg) {
*(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;
LOG_D(PHY,"TS %llu (%llu),txrx_offset %d,d %d, n %d, buff[0] %p buff[1] %p\n",
(unsigned long long)TS,timestamp,device->txrx_offset,device->sampling_rate_ratio_d,device->sampling_rate_ratio_n,
buff[0],buff[1]);
for (int offset=0;offset<nsamps;offset+=256,TS+=TSinc) {
// OAI modified SEQ_ID (4 bytes)
*(uint64_t *)(buff2 + 6) = TS;
if ((offset + 256) <= nsamps) len=1024;
else len = (nsamps-offset)<<2;
// ECPRI Payload Size (2 bytes)
*(uint8_t *)(buff2 + 2) = len>>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
for (int aid = 0; aid<nant; aid++) {
LOG_D(PHY,"TS %llu (TS0 %llu) aa %d : offset %d, len %d\n",(unsigned long long)TS,(unsigned long long)fhstate->TS0,aid,offset,len);
// ECPRI PC_ID (2 bytes)
*(uint16_t *)(buff2 + 4) = aid;
// bring TX data into 12 MSBs
#if defined(__x86_64__) || defined(__i386__)
#ifdef __AVX2__
__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);
}
__m256i *buff256 = (__m256i *)&(((int32_t*)buff[aid])[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[offset];
for (int j=0; j<64; j++) buff_tx[2+j] = _mm_slli_epi16(buff128[j],4);
__m128i *buff128 = (__m128i *)&buff[aid][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[offset];
for (int j=0; j<64; j++) buff_tx[2+j] = vshlq_n_s16(((int16x8_t *)buff128)[j],4);
int16x8_t *buff128 = (__int16x8_t*)&buff[aid][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[0/*aid%eth->num_fd*/],
buff2,
UDP_PACKET_SIZE_BYTES(len>>2),
sendto_flag,
(struct sockaddr*)&eth->dest_addrd,
eth->addr_len);
if ( bytes_sent == -1) {
eth->num_tx_errors++;
/* Send packet */
bytes_sent = sendto(eth->sockfdd[0],
buff2,
UDP_PACKET_SIZE_BYTES(len>>2),
sendto_flag,
(struct sockaddr*)&eth->dest_addrd,
eth->addr_len);
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++;
}
}
}
} // aid
} // offset
free(buff);
}
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);
int trx_eth_write_udp(openair0_device *device, openair0_timestamp timestamp, void **buff, int fd_ind, int nsamps, int flags, int nant) {
union udpTXReqUnion id = {.s={(uint64_t)timestamp,nsamps,0}};
notifiedFIFO_elt_t *req=newNotifiedFIFO_elt(sizeof(udpTXelem_t), id.p, device->utx[fd_ind]->resp,NULL);
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->buff = calloc(nant,sizeof(void*));
memcpy(udptxelem->buff,buff,nant*sizeof(void*));
udptxelem->fd_ind = fd_ind;
udptxelem->nsamps = nsamps;
udptxelem->flags = flags;
#ifdef USE_TPOOL
pushTpool(device->threadPool,req);
#else
trx_eth_write_udp_cmd(udptxelem);
#endif
udptxelem->nant = nant;
pushNotifiedFIFO(device->utx[fd_ind]->resp, req);
LOG_D(PHY,"Pushed to TX FH FIFO, TS %llu, nsamps %d, nant %d buffs[0] %p buffs[1] %p\n",
timestamp,nsamps,nant,udptxelem->buff[0],udptxelem->buff[1]);
}
extern int oai_exit;
void *udp_write_thread(void *arg) {
udp_ctx_t *utx = (udp_ctx_t *)arg;
utx->resp = (notifiedFIFO_elt_t*) malloc(sizeof(notifiedFIFO_elt_t));
initNotifiedFIFO(utx->resp);
LOG_D(PHY,"UDP write thread started on core %d\n",sched_getcpu());
reset_meas(&utx->device->tx_fhaul);
while (oai_exit == 0) {
notifiedFIFO_elt_t *res = pullNotifiedFIFO(utx->resp);
udpTXelem_t *udptxelem = (udpTXelem_t *)NotifiedFifoData(res);
LOG_D(PHY,"Pulled from TX FH FIFO, TS %llu, nsamps %d, nant %d\n",udptxelem->timestamp,udptxelem->nsamps,udptxelem->nant);
start_meas(&utx->device->tx_fhaul);
trx_eth_write_udp_cmd(udptxelem);
stop_meas(&utx->device->tx_fhaul);
// send data to RU
delNotifiedFIFO_elt(res);
}
free(utx->resp);
}
void *udp_read_thread(void *arg) {
openair0_timestamp TS;
int aid;
udp_read_t *u = (udp_read_t *)arg;
udp_ctx_t *u = (udp_ctx_t *)arg;
openair0_device *device=u->device;
fhstate_t *fhstate = &device->fhstate;
char buffer[UDP_PACKET_SIZE_BYTES(256)];
@@ -453,7 +472,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 (%d), aid %d, TS %llu, TS0 %llu, offset %d\n",u->thread_id,sched_getcpu(),aid,(unsigned long long)TS,fhstate->TS0,offset);
LOG_D(PHY,"UDP read thread_id %d (%d), aid %d, TS %llu, TS0 %llu, offset %d\n",u->thread_id,sched_getcpu(),aid,(unsigned long long)TS,fhstate->TS0,offset);
}
sleep(1);
}
targets/ARCH/ETHERNET/USERSPACE/LIB/ethernet_lib.c
+11 -5
View File
@@ -65,13 +65,19 @@ int trx_eth_start(openair0_device *device)
AssertFatal(device->thirdparty_init(device) == 0, "third-party init failed\n");
device->openair0_cfg->samples_per_packet = 256;
eth->num_fd = 1; //max(device->openair0_cfg->rx_num_channels,device->openair0_cfg->tx_num_channels);
udp_read_t *u[2];
udp_ctx_t *u[1+eth->num_fd];
device->utx = (udp_ctx_t**)malloc(sizeof(device->utx));
for (int i=0;i<eth->num_fd;i++) {
u[i] = malloc(sizeof(udp_read_t));
u[i] = malloc(sizeof(udp_ctx_t));
u[i]->thread_id=i;
u[i]->device = device;
printf("UDP Read Thread %d on core %d\n",i,device->openair0_cfg->fh_cores[i]);
threadCreate(&u[i]->pthread,udp_read_thread,u[i],"udp read thread",device->openair0_cfg->fh_cores[i],OAI_PRIORITY_RT_MAX);
printf("UDP Read Thread %d on core %d\n",i,device->openair0_cfg->rxfh_cores[i]);
threadCreate(&u[i]->pthread,udp_read_thread,u[i],"udp read thread",device->openair0_cfg->rxfh_cores[i],OAI_PRIORITY_RT_MAX);
device->utx[i] = malloc(sizeof(udp_ctx_t));
device->utx[i]->thread_id=i;
device->utx[i]->device = device;
printf("UDP Write Thread %d on core %d\n",i,device->openair0_cfg->txfh_cores[i]);
threadCreate(&device->utx[i]->pthread,udp_write_thread,device->utx[i],"udp write thread",device->openair0_cfg->txfh_cores[i],OAI_PRIORITY_RT_MAX);
}
device->sampling_rate_ratio_n=1;
device->sampling_rate_ratio_d=1;
@@ -125,7 +131,7 @@ int trx_eth_start(openair0_device *device)
device->threadPool = (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));
device->respudpTX = (notifiedFIFO_elt_t*) malloc(sizeof(notifiedFIFO_elt_t));
initNotifiedFIFO(device->respudpTX);
#endif
}
targets/ARCH/ETHERNET/USERSPACE/LIB/ethernet_lib.h
+5 -3
View File
@@ -213,8 +213,9 @@ typedef struct {
typedef struct udpTXelem_s {
openair0_device *device;
openair0_timestamp timestamp;
void *buff;
int aid;
void **buff;
int fd_ind;
int nant;
int nsamps;
int flags;
} udpTXelem_t;
@@ -241,6 +242,7 @@ void inline dump_txcounters(openair0_device *device);
void dump_iqs(char * buff, int iq_cnt);
void *udp_read_thread(void *arg);
void *udp_write_thread(void *arg);
/*! \fn int ethernet_tune (openair0_device *device, unsigned int option, int value);
* \brief this function allows you to configure certain ethernet parameters in socket or device level
@@ -263,7 +265,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 *buf, int aid, int nsamps, int flags);
int trx_eth_write_udp(openair0_device *device, openair0_timestamp timestamp, void **buf, int fd_ind, int nsamps, int flags,int nant);
int trx_eth_read_udp(openair0_device *device, openair0_timestamp *timestamp, uint32_t **buff, int nsamps);