From 08ea6180670b359ae96feb6a9a58fb86136710bc Mon Sep 17 00:00:00 2001 From: laurent Date: Tue, 19 Nov 2019 13:27:28 +0100 Subject: [PATCH] fix important bug in CU, merge thread pool improvement and rf simulator improvements made initialy in 5G branch --- common/utils/threadPool/thread-pool.c | 30 ++---- common/utils/threadPool/thread-pool.h | 75 ++++++++------ executables/main-fs6.c | 17 ++- executables/main-ocp.c | 2 +- targets/ARCH/rfsimulator/simulator.c | 142 ++++++++++++++++---------- 5 files changed, 156 insertions(+), 110 deletions(-) diff --git a/common/utils/threadPool/thread-pool.c b/common/utils/threadPool/thread-pool.c index 1b7aa960ba..dd3d16d717 100644 --- a/common/utils/threadPool/thread-pool.c +++ b/common/utils/threadPool/thread-pool.c @@ -49,23 +49,6 @@ void *one_thread(void *arg) { struct one_thread *myThread=(struct one_thread *) arg; struct thread_pool *tp=myThread->pool; - // configure the thread core assignment - // TBD: reserve the core for us exclusively - if ( myThread->coreID >= 0 && myThread->coreID < get_nprocs_conf()) { - cpu_set_t cpuset; - CPU_ZERO(&cpuset); - CPU_SET(myThread->coreID, &cpuset); - pthread_setaffinity_np(pthread_self(), sizeof(cpu_set_t), &cpuset); - } - - //Configure the thread scheduler policy for Linux - struct sched_param sparam= {0}; - sparam.sched_priority = sched_get_priority_max(SCHED_RR); - pthread_setschedparam(pthread_self(), SCHED_RR, &sparam); - // set the thread name for debugging - sprintf(myThread->name,"Tpool_%d",myThread->coreID); - pthread_setname_np(pthread_self(), myThread->name ); - // Infinite loop to process requests do { notifiedFIFO_elt_t *elt=pullNotifiedFifoRemember(&tp->incomingFifo, myThread); @@ -106,10 +89,6 @@ void initTpool(char *params,tpool_t *pool, bool performanceMeas) { } else pool->traceFd=-1; - //Configure the thread scheduler policy for Linux - struct sched_param sparam= {0}; - sparam.sched_priority = sched_get_priority_max(SCHED_RR)-1; - pthread_setschedparam(pthread_self(), SCHED_RR, &sparam); pool->activated=true; initNotifiedFIFO(&pool->incomingFifo); char *saveptr, * curptr; @@ -137,7 +116,11 @@ void initTpool(char *params,tpool_t *pool, bool performanceMeas) { pool->allthreads->coreID=atoi(curptr); pool->allthreads->id=pool->nbThreads; pool->allthreads->pool=pool; - pthread_create(&pool->allthreads->threadID, NULL, one_thread, (void *)pool->allthreads); + //Configure the thread scheduler policy for Linux + // set the thread name for debugging + sprintf(pool->allthreads->name,"Tpool_%d",pool->allthreads->coreID); + threadCreate(&pool->allthreads->threadID, one_thread, (void *)pool->allthreads, + pool->allthreads->name, pool->allthreads->coreID, OAI_PRIORITY_RT); pool->nbThreads++; } @@ -152,6 +135,9 @@ void initTpool(char *params,tpool_t *pool, bool performanceMeas) { #ifdef TEST_THREAD_POOL +void exit_function(const char *file, const char *function, const int line, const char *s) { +} + struct testData { int id; char txt[30]; diff --git a/common/utils/threadPool/thread-pool.h b/common/utils/threadPool/thread-pool.h index d68918a163..d358db5f00 100644 --- a/common/utils/threadPool/thread-pool.h +++ b/common/utils/threadPool/thread-pool.h @@ -11,6 +11,7 @@ #include #include #include +#include #ifdef DEBUG #define THREADINIT PTHREAD_ERRORCHECK_MUTEX_INITIALIZER_NP @@ -77,41 +78,60 @@ static inline void delNotifiedFIFO_elt(notifiedFIFO_elt_t *elt) { //LOG_W(UTIL,"delNotifiedFIFO on something not allocated by newNotifiedFIFO\n"); } +static inline void initNotifiedFIFO_nothreadSafe(notifiedFIFO_t *nf) { + nf->inF=NULL; + nf->outF=NULL; +} static inline void initNotifiedFIFO(notifiedFIFO_t *nf) { mutexinit(nf->lockF); condinit (nf->notifF); - nf->inF=NULL; - nf->outF=NULL; + initNotifiedFIFO_nothreadSafe(nf); // No delete function: the creator has only to free the memory } -static inline void pushNotifiedFIFO(notifiedFIFO_t *nf, notifiedFIFO_elt_t *msg) { - mutexlock(nf->lockF); +static inline void pushNotifiedFIFO_nothreadSafe(notifiedFIFO_t *nf, notifiedFIFO_elt_t *msg) { msg->next=NULL; if (nf->outF == NULL) nf->outF = msg; - if (nf->inF) + if (nf->inF != NULL) nf->inF->next = msg; nf->inF = msg; +} + +static inline void pushNotifiedFIFO(notifiedFIFO_t *nf, notifiedFIFO_elt_t *msg) { + mutexlock(nf->lockF); + pushNotifiedFIFO_nothreadSafe(nf,msg); condbroadcast(nf->notifF); mutexunlock(nf->lockF); } -static inline notifiedFIFO_elt_t *pullNotifiedFIFO(notifiedFIFO_t *nf) { - mutexlock(nf->lockF); - - while(!nf->outF) - condwait(nf->notifF, nf->lockF); +static inline notifiedFIFO_elt_t *pullNotifiedFIFO_nothreadSafe(notifiedFIFO_t *nf) { + if (nf->outF == NULL) + return NULL; notifiedFIFO_elt_t *ret=nf->outF; + + if (nf->outF==nf->outF->next) + LOG_E(TMR,"Circular list in thread pool: push several times the same buffer is forbidden\n"); + nf->outF=nf->outF->next; if (nf->outF==NULL) nf->inF=NULL; + return ret; +} + +static inline notifiedFIFO_elt_t *pullNotifiedFIFO(notifiedFIFO_t *nf) { + mutexlock(nf->lockF); + notifiedFIFO_elt_t *ret; + + while((ret=pullNotifiedFIFO_nothreadSafe(nf)) == NULL) + condwait(nf->notifF, nf->lockF); + mutexunlock(nf->lockF); return ret; } @@ -122,14 +142,7 @@ static inline notifiedFIFO_elt_t *pollNotifiedFIFO(notifiedFIFO_t *nf) { if (tmp != 0 ) return NULL; - notifiedFIFO_elt_t *ret=nf->outF; - - if (ret!=NULL) - nf->outF=nf->outF->next; - - if (nf->outF==NULL) - nf->inF=NULL; - + notifiedFIFO_elt_t *ret=pullNotifiedFIFO_nothreadSafe(nf); mutexunlock(nf->lockF); return ret; } @@ -151,6 +164,7 @@ static inline int abortNotifiedFIFO(notifiedFIFO_t *nf, uint64_t key) { } else start=&(*start)->next; } + if (nf->outF == NULL) nf->inF=NULL; @@ -184,16 +198,20 @@ typedef struct thread_pool { static inline void pushTpool(tpool_t *t, notifiedFIFO_elt_t *msg) { if (t->measurePerf) msg->creationTime=rdtsc(); + if ( t->activated) pushNotifiedFIFO(&t->incomingFifo, msg); else { - if (t->measurePerf) - msg->startProcessingTime=rdtsc(); - msg->processingFunc(NotifiedFifoData(msg)); - if (t->measurePerf) - msg->endProcessingTime=rdtsc(); - if (msg->reponseFifo) - pushNotifiedFIFO(msg->reponseFifo, msg); + if (t->measurePerf) + msg->startProcessingTime=rdtsc(); + + msg->processingFunc(NotifiedFifoData(msg)); + + if (t->measurePerf) + msg->endProcessingTime=rdtsc(); + + if (msg->reponseFifo) + pushNotifiedFIFO(msg->reponseFifo, msg); } } @@ -202,10 +220,8 @@ static inline notifiedFIFO_elt_t *pullTpool(notifiedFIFO_t *responseFifo, tpool_ if (t->measurePerf) msg->returnTime=rdtsc(); - - if (t->traceFd) + if (t->traceFd >= 0) if(write(t->traceFd, msg, sizeof(*msg))); - return msg; } @@ -217,10 +233,8 @@ static inline notifiedFIFO_elt_t *tryPullTpool(notifiedFIFO_t *responseFifo, tpo if (t->measurePerf) msg->returnTime=rdtsc(); - if (t->traceFd) if(write(t->traceFd, msg, sizeof(*msg))); - return msg; } @@ -239,6 +253,7 @@ static inline int abortTpool(tpool_t *t, uint64_t key) { } else start=&(*start)->next; } + if (t->incomingFifo.outF==NULL) t->incomingFifo.inF=NULL; diff --git a/executables/main-fs6.c b/executables/main-fs6.c index 471ad0f1eb..638db0ac11 100644 --- a/executables/main-fs6.c +++ b/executables/main-fs6.c @@ -1324,6 +1324,9 @@ void *DL_du_fs6(void *arg) { feptx_prec(ru, &L1_proc); feptx_ofdm(ru, &L1_proc); tx_rf(ru, &L1_proc); + + if ( IS_SOFTMODEM_RFSIM ) + return NULL; } return NULL; @@ -1443,13 +1446,22 @@ void *cu_fs6(void *arg) { remoteIP=DU_IP; AssertFatal(createUDPsock(NULL, CU_PORT, remoteIP, DU_PORT, &sockFS6), ""); + + L1_rxtx_proc_t L1proc= {0}; + if ( strlen(get_softmodem_params()->threadPoolConfig) > 0 ) + initTpool(get_softmodem_params()->threadPoolConfig, &L1proc.threadPool, true); + else + initTpool("n", &L1proc.threadPool, true); + + initNotifiedFIFO(&L1proc.respEncode); + initNotifiedFIFO(&L1proc.respDecode); + uint64_t timeStamp=0; initStaticTime(begingWait); initStaticTime(begingWait2); initRefTimes(waitDUAndProcessingUL); initRefTimes(makeSendDL); initRefTimes(fullLoop); - L1_rxtx_proc_t L1proc= {0}; while(1) { timeStamp+=ru->frame_parms.samples_per_tti; @@ -1496,6 +1508,7 @@ void *du_fs6(void *arg) { initRefTimes(waitRxAndProcessingUL); initRefTimes(fullLoop); pthread_t t; + if ( !IS_SOFTMODEM_RFSIM ) threadCreate(&t, DL_du_fs6, (void *)ru, "MainDuTx", -1, OAI_PRIORITY_RT_MAX); while(1) { @@ -1503,6 +1516,8 @@ void *du_fs6(void *arg) { updateTimesReset(begingWait, &fullLoop, 1000, true,"DU for full SubFrame (must be less 1ms)"); pickStaticTime(begingWait); UL_du_fs6(ru, &L1proc); + if ( IS_SOFTMODEM_RFSIM ) + DL_du_fs6((void *)ru); updateTimesReset(begingWait, &waitRxAndProcessingUL, 1000, true,"DU Time in wait Rx + Ul processing"); } diff --git a/executables/main-ocp.c b/executables/main-ocp.c index 8810115197..819eab0ee6 100644 --- a/executables/main-ocp.c +++ b/executables/main-ocp.c @@ -688,7 +688,7 @@ static void *ru_thread( void *param ) { setbuf(stdout, NULL); setbuf(stderr, NULL); RU_t *ru = (RU_t *)param; - L1_rxtx_proc_t L1proc; + L1_rxtx_proc_t L1proc={0}; L1_rxtx_proc_t *proc=&L1proc; if ( strlen(get_softmodem_params()->threadPoolConfig) > 0 ) diff --git a/targets/ARCH/rfsimulator/simulator.c b/targets/ARCH/rfsimulator/simulator.c index b85cdb507a..95eeb38aa0 100644 --- a/targets/ARCH/rfsimulator/simulator.c +++ b/targets/ARCH/rfsimulator/simulator.c @@ -70,8 +70,8 @@ typedef struct complex16 sample_t; // 2*16 bits complex number typedef struct buffer_s { int conn_sock; openair0_timestamp lastReceivedTS; - openair0_timestamp lastWroteTS; bool headerMode; + bool trashingPacket; samplesBlockHeader_t th; char *transferPtr; uint64_t remainToTransfer; @@ -83,6 +83,7 @@ typedef struct buffer_s { typedef struct { int listen_sock, epollfd; openair0_timestamp nextTimestamp; + openair0_timestamp lastWroteTS; uint64_t typeStamp; char *ip; uint16_t port; @@ -102,8 +103,8 @@ void allocCirBuf(rfsimulator_state_t *bridge, int sock) { ptr->circularBufEnd=((char *)ptr->circularBuf)+sampleToByte(CirSize,1); ptr->conn_sock=sock; ptr->lastReceivedTS=0; - ptr->lastWroteTS=0; ptr->headerMode=true; + ptr->trashingPacket=false; ptr->transferPtr=(char *)&ptr->th; ptr->remainToTransfer=sizeof(samplesBlockHeader_t); int sendbuff=1000*1000*100; @@ -219,8 +220,6 @@ void fullwrite(int fd, void *_buf, ssize_t count, rfsimulator_state_t *t) { } } - - void rfsimulator_readconfig(rfsimulator_state_t *rfsimulator) { char *saveF=NULL; char *modelname=NULL; @@ -317,19 +316,16 @@ sin_addr: return 0; } -static int rfsimulator_write_internal(rfsimulator_state_t *t, openair0_timestamp timestamp, void **samplesVoid, int nsamps, int nbAnt, int flags) { +static int rfsimulator_write_internal(rfsimulator_state_t *t, openair0_timestamp timestamp, void **samplesVoid, int nsamps, int nbAnt, int flags, bool alreadyLocked) { + if (!alreadyLocked) + pthread_mutex_lock(&Sockmutex); + LOG_D(HW,"sending %d samples at time: %ld\n", nsamps, timestamp); for (int i=0; ibuf[i]; if (b->conn_sock >= 0 ) { - pthread_mutex_lock(&Sockmutex); - - if ( b->lastWroteTS != 0 && abs((double)b->lastWroteTS-timestamp) > (double)CirSize) - LOG_E(HW,"Discontinuous TX gap too large Tx:%lu, %lu\n", b->lastWroteTS, timestamp); - - AssertFatal(b->lastWroteTS <= timestamp+1, " Not supported to send Tx out of order (same in USRP) %lu, %lu\n", b->lastWroteTS, timestamp); samplesBlockHeader_t header= {t->typeStamp, nsamps, nbAnt, timestamp}; fullwrite(b->conn_sock,&header, sizeof(header), t); sample_t tmpSamples[nsamps][nbAnt]; @@ -343,24 +339,29 @@ static int rfsimulator_write_internal(rfsimulator_state_t *t, openair0_timestamp if (b->conn_sock >= 0 ) { fullwrite(b->conn_sock, (void *)tmpSamples, sampleToByte(nsamps,nbAnt), t); - b->lastWroteTS=timestamp+nsamps; } - - pthread_mutex_unlock(&Sockmutex); } } + if ( t->lastWroteTS != 0 && abs((double)t->lastWroteTS-timestamp) > (double)CirSize) + LOG_E(HW,"Discontinuous TX gap too large Tx:%lu, %lu\n", t->lastWroteTS, timestamp); + + AssertFatal(t->lastWroteTS <= timestamp+1, " Not supported to send Tx out of order (same in USRP) %lu, %lu\n", + t->lastWroteTS, timestamp); + t->lastWroteTS=timestamp+nsamps; + + if (!alreadyLocked) + pthread_mutex_unlock(&Sockmutex); + LOG_D(HW,"sent %d samples at time: %ld->%ld, energy in first antenna: %d\n", nsamps, timestamp, timestamp+nsamps, signal_energy(samplesVoid[0], nsamps) ); return nsamps; } int rfsimulator_write(openair0_device *device, openair0_timestamp timestamp, void **samplesVoid, int nsamps, int nbAnt, int flags) { - return rfsimulator_write_internal(device->priv, timestamp, samplesVoid, nsamps, nbAnt, flags); + return rfsimulator_write_internal(device->priv, timestamp, samplesVoid, nsamps, nbAnt, flags, false); } - - static bool flushInput(rfsimulator_state_t *t, int timeout, int nsamps_for_initial) { // Process all incoming events on sockets // store the data in lists @@ -389,22 +390,9 @@ static bool flushInput(rfsimulator_state_t *t, int timeout, int nsamps_for_initi for ( int i=0; i < t->tx_num_channels; i++) samplesVoid[i]=(void *)&v; - pthread_mutex_lock(&Sockmutex); - openair0_timestamp timestamp=1; - - for (int i=0; ibuf[i]; - - if (t->buf[i].circularBuf != NULL && t->buf[i].lastWroteTS > 1) { - timestamp=b->lastWroteTS; - break; - } - } - - pthread_mutex_unlock(&Sockmutex); - rfsimulator_write_internal(t, timestamp-1, + rfsimulator_write_internal(t, t->lastWroteTS > 1 ? t->lastWroteTS-1 : 0, samplesVoid, 1, - t->tx_num_channels, 1); + t->tx_num_channels, 1, false); } else { if ( events[nbEv].events & (EPOLLHUP | EPOLLERR | EPOLLRDHUP) ) { socketError(t,fd); @@ -450,25 +438,41 @@ static bool flushInput(rfsimulator_state_t *t, int timeout, int nsamps_for_initi (t->typeStamp == ENB_MAGICDL_FDD && b->th.magic==UE_MAGICDL_FDD), "Socket Error in protocol"); b->headerMode=false; - if ( b->lastReceivedTS != b->th.timestamp) { + if ( t->nextTimestamp == 0 ) { // First block in UE, resync with the eNB current TS + t->nextTimestamp=b->th.timestamp> nsamps_for_initial ? + b->th.timestamp - nsamps_for_initial : + 0; + b->lastReceivedTS=b->th.timestamp> nsamps_for_initial ? + b->th.timestamp : + nsamps_for_initial; + LOG_W(HW,"UE got first timestamp: starting at %lu\n", t->nextTimestamp); + b->trashingPacket=true; + } else if ( b->lastReceivedTS < b->th.timestamp) { int nbAnt= b->th.nbAnt; - + for (uint64_t index=b->lastReceivedTS; index < b->th.timestamp; index++ ) { for (int a=0; a < nbAnt; a++) { b->circularBuf[(index*nbAnt+a)%CirSize].r=0; b->circularBuf[(index*nbAnt+a)%CirSize].i=0; } } - - if ( abs(b->th.timestamp-b->lastReceivedTS) > 50 ) + if (b->lastReceivedTS != 0 && b->th.timestamp-b->lastReceivedTS > 50 ) LOG_W(HW,"UEsock: %d gap of: %ld in reception\n", fd, b->th.timestamp-b->lastReceivedTS ); - } + b->lastReceivedTS=b->th.timestamp; + + } else if ( b->lastReceivedTS > b->th.timestamp && b->th.size == 1 ) { + LOG_W(HW,"Received Rx/Tx synchro out of order\n"); + b->trashingPacket=true; + } else if ( b->lastReceivedTS == b->th.timestamp ) { + // normal case + } else { + abort(); + AssertFatal(false, "received data in past: current is %lu, new reception: %lu!\n", b->lastReceivedTS, b->th.timestamp); } - b->lastReceivedTS=b->th.timestamp; pthread_mutex_lock(&Sockmutex); - if (b->lastWroteTS != 0 && ( abs((double)b->lastWroteTS-b->lastReceivedTS) > (double)CirSize)) - LOG_E(HW,"UEsock: %d Tx/Rx shift too large Tx:%lu, Rx:%lu\n", fd, b->lastWroteTS, b->lastReceivedTS); + if (t->lastWroteTS != 0 && ( abs((double)t->lastWroteTS-b->lastReceivedTS) > (double)CirSize)) + LOG_E(HW,"UEsock: %d Tx/Rx shift too large Tx:%lu, Rx:%lu\n", fd, t->lastWroteTS, b->lastReceivedTS); pthread_mutex_unlock(&Sockmutex); b->transferPtr=(char *)&b->circularBuf[b->lastReceivedTS%CirSize]; @@ -477,22 +481,17 @@ static bool flushInput(rfsimulator_state_t *t, int timeout, int nsamps_for_initi if ( b->headerMode==false ) { LOG_D(HW,"UEsock: %d Set b->lastReceivedTS %ld\n", fd, b->lastReceivedTS); - b->lastReceivedTS=b->th.timestamp+b->th.size-byteToSample(b->remainToTransfer,b->th.nbAnt); + if ( ! b->trashingPacket ) { + b->lastReceivedTS=b->th.timestamp+b->th.size-byteToSample(b->remainToTransfer,b->th.nbAnt); + } - // First block in UE, resync with the eNB current TS - if ( t->nextTimestamp == 0 && b->th.size < b->lastReceivedTS) { - t->nextTimestamp=b->lastReceivedTS > nsamps_for_initial ? - b->lastReceivedTS - nsamps_for_initial : - 0; - LOG_W(HW,"UE got first timestamp: starting at %lu\n", t->nextTimestamp); - } - - if ( b->remainToTransfer==0) { - LOG_D(HW,"UEsock: %d Completed block reception: %ld\n", fd, b->lastReceivedTS); - b->headerMode=true; - b->transferPtr=(char *)&b->th; - b->remainToTransfer=sizeof(samplesBlockHeader_t); - b->th.magic=-1; + if ( b->remainToTransfer==0) { + LOG_D(HW,"UEsock: %d Completed block reception: %ld\n", fd, b->lastReceivedTS); + b->headerMode=true; + b->transferPtr=(char *)&b->th; + b->remainToTransfer=sizeof(samplesBlockHeader_t); + b->th.magic=-1; + b->trashingPacket=false; } } } @@ -528,12 +527,43 @@ int rfsimulator_read(openair0_device *device, openair0_timestamp *ptimestamp, vo t->nextTimestamp+=nsamps; if ( ((t->nextTimestamp/nsamps)%100) == 0) - LOG_W(HW,"Generated void samples for Rx: %ld\n", t->nextTimestamp); + LOG_W(HW,"No UE, Generated void samples for Rx: %ld\n", t->nextTimestamp); *ptimestamp = t->nextTimestamp-nsamps; return nsamps; } } else { + pthread_mutex_lock(&Sockmutex); + + if ( t->nextTimestamp > 0 && t->lastWroteTS < t->nextTimestamp) { + pthread_mutex_unlock(&Sockmutex); + usleep(10000); + pthread_mutex_lock(&Sockmutex); + if ( t->lastWroteTS < t->nextTimestamp ) { + // Assuming Tx is not done fully in another thread + // We can never write is the past from the received time + // So, the node perform receive but will never write these symbols + // let's tell this to the opposite node + // We send timestamp for nb samples required + // assuming this should have been done earlier if a Tx would exist + pthread_mutex_unlock(&Sockmutex); + struct complex16 v= {0}; + void *samplesVoid[t->tx_num_channels]; + + for ( int i=0; i < t->tx_num_channels; i++) + samplesVoid[i]=(void *)&v; + LOG_I(HW, "No samples Tx occured, so we send 1 sample to notify it: Tx:%lu, Rx:%lu\n", + t->lastWroteTS, t->nextTimestamp); + rfsimulator_write_internal(t, t->nextTimestamp, + samplesVoid, 1, + t->tx_num_channels, 1, true); + } else { + pthread_mutex_unlock(&Sockmutex); + LOG_W(HW, "trx_write came from another thread\n"); + } + } else + pthread_mutex_unlock(&Sockmutex); + bool have_to_wait; do {