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

Remove table of thread_id in the main UE thread

Browse Source
This commit is contained in:
laurent
2019-04-11 13:44:08 +02:00
parent 7b3615de64
commit 2e6d9d7d4a
2 changed files with 93 additions and 76 deletions
Download Patch File Download Diff File Expand all files Collapse all files
common/utils/threadPool/thread-pool.h
+29 -21
View File
@@ -77,16 +77,18 @@ static inline void delNotifiedFIFO_elt(notifiedFIFO_elt_t *elt) {
//LOG_W(UTIL,"delNotifiedFIFO on something not allocated by newNotifiedFIFO\n"); //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) { static inline void initNotifiedFIFO(notifiedFIFO_t *nf) {
mutexinit(nf->lockF); mutexinit(nf->lockF);
condinit (nf->notifF); condinit (nf->notifF);
nf->inF=NULL; initNotifiedFIFO_nothreadSafe(nf);
nf->outF=NULL;
// No delete function: the creator has only to free the memory // No delete function: the creator has only to free the memory
} }
static inline void pushNotifiedFIFO(notifiedFIFO_t *nf, notifiedFIFO_elt_t *msg) { static inline void pushNotifiedFIFO_nothreadSafe(notifiedFIFO_t *nf, notifiedFIFO_elt_t *msg) {
mutexlock(nf->lockF);
msg->next=NULL; msg->next=NULL;
if (nf->outF == NULL) if (nf->outF == NULL)
@@ -96,22 +98,39 @@ static inline void pushNotifiedFIFO(notifiedFIFO_t *nf, notifiedFIFO_elt_t *msg)
nf->inF->next = msg; nf->inF->next = msg;
nf->inF = 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); condbroadcast(nf->notifF);
mutexunlock(nf->lockF); mutexunlock(nf->lockF);
} }
static inline notifiedFIFO_elt_t *pullNotifiedFIFO(notifiedFIFO_t *nf) { static inline notifiedFIFO_elt_t *pullNotifiedFIFO_nothreadSafe(notifiedFIFO_t *nf) {
mutexlock(nf->lockF); if (nf->outF == NULL)
return NULL;
while(nf->outF == NULL)
condwait(nf->notifF, nf->lockF);
notifiedFIFO_elt_t *ret=nf->outF; 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; nf->outF=nf->outF->next;
if (nf->outF==NULL) if (nf->outF==NULL)
nf->inF=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); mutexunlock(nf->lockF);
return ret; return ret;
} }
@@ -122,18 +141,7 @@ static inline notifiedFIFO_elt_t *pollNotifiedFIFO(notifiedFIFO_t *nf) {
if (tmp != 0 ) if (tmp != 0 )
return NULL; return NULL;
notifiedFIFO_elt_t *ret=nf->outF; notifiedFIFO_elt_t *ret=pullNotifiedFIFO_nothreadSafe(nf);
if (ret!=NULL) {
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;
mutexunlock(nf->lockF); mutexunlock(nf->lockF);
return ret; return ret;
} }
executables/nr-ue.c
+64 -55
View File
@@ -156,7 +156,7 @@ PHY_VARS_NR_UE *init_nr_ue_vars(NR_DL_FRAME_PARMS *frame_parms,
*/ */
typedef struct syncData_s { typedef struct syncData_s {
UE_nr_rxtx_proc_t *proc; UE_nr_rxtx_proc_t proc;
PHY_VARS_NR_UE *UE; PHY_VARS_NR_UE *UE;
} syncData_t; } syncData_t;
@@ -239,7 +239,7 @@ static void UE_synch(void *arg) {
case pbch: case pbch:
LOG_I(PHY, "[UE thread Synch] Running Initial Synch (mode %d)\n",UE->mode); LOG_I(PHY, "[UE thread Synch] Running Initial Synch (mode %d)\n",UE->mode);
if (nr_initial_sync( syncD->proc, UE, UE->mode ) == 0) { if (nr_initial_sync( &syncD->proc, UE, UE->mode ) == 0) {
freq_offset = UE->common_vars.freq_offset; // frequency offset computed with pss in initial sync freq_offset = UE->common_vars.freq_offset; // frequency offset computed with pss in initial sync
hw_slot_offset = (UE->rx_offset<<1) / UE->frame_parms.samples_per_slot; hw_slot_offset = (UE->rx_offset<<1) / UE->frame_parms.samples_per_slot;
LOG_I(PHY,"Got synch: hw_slot_offset %d, carrier off %d Hz, rxgain %d (DL %u, UL %u), UE_scan_carrier %d\n", LOG_I(PHY,"Got synch: hw_slot_offset %d, carrier off %d Hz, rxgain %d (DL %u, UL %u), UE_scan_carrier %d\n",
@@ -548,40 +548,45 @@ void *UE_thread(void *arg) {
initNotifiedFIFO(&nf); initNotifiedFIFO(&nf);
int nbSlotProcessing=0; int nbSlotProcessing=0;
int thread_idx=0; int thread_idx=0;
notifiedFIFO_elt_t *processingMsg[RX_NB_TH]; notifiedFIFO_t freeBlocks;
initNotifiedFIFO_nothreadSafe(&freeBlocks);
for (int i=0; i<RX_NB_TH; i++) { for (int i=0; i<RX_NB_TH+1; i++) // RX_NB_TH working + 1 we are making to be pushed
processingMsg[i]= newNotifiedFIFO_elt(sizeof(processingData_t), 0,&nf,UE_processing); pushNotifiedFIFO_nothreadSafe(&freeBlocks,
processingData_t *tmp=(processingData_t *)NotifiedFifoData(processingMsg[i]); newNotifiedFIFO_elt(sizeof(processingData_t), 0,&nf,UE_processing));
tmp->UE=UE;
}
bool syncRunning=false; bool syncRunning=false;
notifiedFIFO_elt_t *syncMsg=newNotifiedFIFO_elt(sizeof(syncData_t),0,&nf,UE_synch);
syncData_t *syncD=(syncData_t *)NotifiedFifoData(syncMsg);
syncD->UE=UE;
syncD->proc=&((processingData_t *)NotifiedFifoData(processingMsg[0]))->proc;
const int nb_slot_frame = 10*UE->frame_parms.slots_per_subframe; const int nb_slot_frame = 10*UE->frame_parms.slots_per_subframe;
int absolute_slot, decoded_frame_rx=INT_MAX, trashed_frames=0; int absolute_slot=0, decoded_frame_rx=INT_MAX, trashed_frames=0;
while (!oai_exit) { while (!oai_exit) {
if (!syncD->UE->is_synchronized) { if (syncRunning) {
if (syncRunning) { notifiedFIFO_elt_t *res=tryPullTpool(&nf, Tpool);
notifiedFIFO_elt_t *res=tryPullTpool(&nf, Tpool);
if (res) { if (res) {
syncRunning=false; syncRunning=false;
} else { syncData_t *tmp=(syncData_t *)NotifiedFifoData(res);
trashFrame(UE, &timestamp); // shift the frame index with all the frames we trashed meanwhile we perform the synch search
trashed_frames++; decoded_frame_rx=(tmp->proc.decoded_frame_rx+trashed_frames) % MAX_FRAME_NUMBER;
} delNotifiedFIFO_elt(res);
} else { } else {
readFrame(UE, &timestamp); trashFrame(UE, &timestamp);
pushTpool(Tpool, syncMsg); trashed_frames++;
trashed_frames=0; continue;
syncRunning=true;
} }
}
AssertFatal( !syncRunning, "At this point synchronisation can't be running\n");
if (!UE->is_synchronized) {
readFrame(UE, &timestamp);
notifiedFIFO_elt_t *Msg=newNotifiedFIFO_elt(sizeof(syncData_t),0,&nf,UE_synch);
syncData_t *syncMsg=(syncData_t *)NotifiedFifoData(Msg);
syncMsg->UE=UE;
memset(&syncMsg->proc, 0, sizeof(syncMsg->proc));
pushTpool(Tpool, Msg);
trashed_frames=0;
syncRunning=true;
continue; continue;
} }
@@ -599,33 +604,32 @@ void *UE_thread(void *arg) {
UE->frame_parms.nb_antennas_rx),""); UE->frame_parms.nb_antennas_rx),"");
// we have the decoded frame index in the return of the synch process // we have the decoded frame index in the return of the synch process
// and we shifted above to the first slot of next frame // and we shifted above to the first slot of next frame
// the synch thread proc context is hard linked to regular processing thread context, thread id = 0 decoded_frame_rx++;
UE_nr_rxtx_proc_t *proc=&(((processingData_t *)NotifiedFifoData(processingMsg[0]))->proc); // we do ++ first in the regular processing, so it will be beging of frame;
// shift the frame index with all the frames we trashed meanwhile we perform the synch search
proc->decoded_frame_rx=(proc->decoded_frame_rx + trashed_frames) % MAX_FRAME_NUMBER;
decoded_frame_rx=proc->decoded_frame_rx;
// we do ++ first in the regular processing, so it will be 0;
absolute_slot=decoded_frame_rx*nb_slot_frame + nb_slot_frame -1; absolute_slot=decoded_frame_rx*nb_slot_frame + nb_slot_frame -1;
while (tryPullTpool(&nf, Tpool) != NULL) {
};
continue; continue;
} }
absolute_slot++; absolute_slot++;
// whatever means thread_idx
// Fix me: will be wrong when slot 1 is slow, as slot 2 finishes
// Slot 3 will overlap if RX_NB_TH is 2
// this is general failure in UE !!!
thread_idx = absolute_slot % RX_NB_TH; thread_idx = absolute_slot % RX_NB_TH;
int slot_nr = absolute_slot % nb_slot_frame; int slot_nr = absolute_slot % nb_slot_frame;
UE_nr_rxtx_proc_t *proc=&(((processingData_t *)NotifiedFifoData(processingMsg[thread_idx]))->proc); notifiedFIFO_elt_t *msgToPush;
AssertFatal((msgToPush=pullNotifiedFIFO_nothreadSafe(&freeBlocks)) != NULL,"chained list failure");
processingData_t *curMsg=(processingData_t *)NotifiedFifoData(msgToPush);
curMsg->UE=UE;
// update thread index for received subframe // update thread index for received subframe
proc->nr_tti_rx= slot_nr; curMsg->proc.nr_tti_rx= slot_nr;
UE->current_thread_id[slot_nr] = thread_idx; curMsg->UE->current_thread_id[slot_nr] = thread_idx;
proc->subframe_rx=table_sf_slot[slot_nr]; curMsg->proc.subframe_rx=table_sf_slot[slot_nr];
proc->nr_tti_tx = (absolute_slot + DURATION_RX_TO_TX) % nb_slot_frame; curMsg->proc.nr_tti_tx = (absolute_slot + DURATION_RX_TO_TX) % nb_slot_frame;
proc->subframe_tx=proc->nr_tti_rx; curMsg->proc.subframe_tx=curMsg->proc.nr_tti_rx;
proc->frame_rx = ( absolute_slot/nb_slot_frame ) % MAX_FRAME_NUMBER; curMsg->proc.frame_rx = ( absolute_slot/nb_slot_frame ) % MAX_FRAME_NUMBER;
proc->frame_tx = ( (absolute_slot + DURATION_RX_TO_TX) /nb_slot_frame ) % MAX_FRAME_NUMBER; curMsg->proc.frame_tx = ( (absolute_slot + DURATION_RX_TO_TX) /nb_slot_frame ) % MAX_FRAME_NUMBER;
proc->decoded_frame_rx=-1; curMsg->proc.decoded_frame_rx=-1;
LOG_D(PHY,"Process slot %d thread Idx %d \n", slot_nr, thread_idx); LOG_D(PHY,"Process slot %d thread Idx %d \n", slot_nr, thread_idx);
for (int i=0; i<UE->frame_parms.nb_antennas_rx; i++) for (int i=0; i<UE->frame_parms.nb_antennas_rx; i++)
@@ -634,7 +638,7 @@ void *UE_thread(void *arg) {
slot_nr*UE->frame_parms.samples_per_slot]; slot_nr*UE->frame_parms.samples_per_slot];
for (int i=0; i<UE->frame_parms.nb_antennas_tx; i++) for (int i=0; i<UE->frame_parms.nb_antennas_tx; i++)
txp[i] = (void *)&UE->common_vars.txdata[i][proc->nr_tti_tx*UE->frame_parms.samples_per_slot]; txp[i] = (void *)&UE->common_vars.txdata[i][curMsg->proc.nr_tti_tx*UE->frame_parms.samples_per_slot];
int readBlockSize, writeBlockSize; int readBlockSize, writeBlockSize;
@@ -683,31 +687,33 @@ void *UE_thread(void *arg) {
LOG_E(PHY,"can't compensate: diff =%d\n", first_symbols); LOG_E(PHY,"can't compensate: diff =%d\n", first_symbols);
} }
proc->timestamp_tx = timestamp+ curMsg->proc.timestamp_tx = timestamp+
(DURATION_RX_TO_TX*UE->frame_parms.samples_per_slot)- (DURATION_RX_TO_TX*UE->frame_parms.samples_per_slot)-
UE->frame_parms.ofdm_symbol_size-UE->frame_parms.nb_prefix_samples0; UE->frame_parms.ofdm_symbol_size-UE->frame_parms.nb_prefix_samples0;
notifiedFIFO_elt_t *res; notifiedFIFO_elt_t *res;
while (nbSlotProcessing >= RX_NB_TH ) { while (nbSlotProcessing >= RX_NB_TH) {
if ( (res=tryPullTpool(&nf, Tpool)) != NULL ) { if ( (res=tryPullTpool(&nf, Tpool)) != NULL ) {
nbSlotProcessing--; nbSlotProcessing--;
processingData_t *tmp=(processingData_t *)res->msgData; processingData_t *tmp=(processingData_t *)res->msgData;
if (tmp->proc.decoded_frame_rx != -1) if (tmp->proc.decoded_frame_rx != -1)
decoded_frame_rx=tmp->proc.decoded_frame_rx; decoded_frame_rx=tmp->proc.decoded_frame_rx;
pushNotifiedFIFO_nothreadSafe(&freeBlocks,res);
} }
usleep(200); usleep(200);
} }
if ( decoded_frame_rx != proc->frame_rx && if ( decoded_frame_rx != curMsg->proc.frame_rx &&
((decoded_frame_rx+1) % MAX_FRAME_NUMBER) != proc->frame_rx ) ((decoded_frame_rx+1) % MAX_FRAME_NUMBER) != curMsg->proc.frame_rx )
LOG_D(PHY,"Decoded frame index (%d) is not compatible with current context (%d), UE should go back to synch mode\n", LOG_D(PHY,"Decoded frame index (%d) is not compatible with current context (%d), UE should go back to synch mode\n",
decoded_frame_rx, proc->frame_rx); decoded_frame_rx, curMsg->proc.frame_rx );
nbSlotProcessing++; nbSlotProcessing++;
processingMsg[thread_idx]->key=slot_nr; msgToPush->key=slot_nr;
pushTpool(Tpool, processingMsg[thread_idx]); pushTpool(Tpool, msgToPush);
if (getenv("RFSIMULATOR")) { if (getenv("RFSIMULATOR")) {
// FixMe: Wait previous thread is done, because race conditions seems too bad // FixMe: Wait previous thread is done, because race conditions seems too bad
@@ -716,8 +722,11 @@ void *UE_thread(void *arg) {
res=pullTpool(&nf, Tpool); res=pullTpool(&nf, Tpool);
nbSlotProcessing--; nbSlotProcessing--;
processingData_t *tmp=(processingData_t *)res->msgData; processingData_t *tmp=(processingData_t *)res->msgData;
if (tmp->proc.decoded_frame_rx != -1) if (tmp->proc.decoded_frame_rx != -1)
decoded_frame_rx=tmp->proc.decoded_frame_rx; decoded_frame_rx=tmp->proc.decoded_frame_rx;
pushNotifiedFIFO_nothreadSafe(&freeBlocks,res);
} }
} // while !oai_exit } // while !oai_exit