Merge remote-tracking branch 'origin/fix-nr-rfsim-gNB-notx-slots2to19' into...

Merge remote-tracking branch 'origin/fix-nr-rfsim-gNB-notx-slots2to19' into integration-develop-nr-2019w46

cmake_targets/CMakeLists.txt

@@ -185,7 +185,7 @@ set(CMAKE_EXE_LINKER_FLAGS  "${CMAKE_EXE_LINKER_FLAGS} -Wl,-rpath -Wl,${CMAKE_CU
 # set a flag for changes in the source code
 # these changes are related to hardcoded path to include .h files
 set(CMAKE_C_FLAGS_DEBUG "${CMAKE_C_FLAGS} -g -DMALLOC_CHECK_=3")
-set(CMAKE_C_FLAGS_RELWITHDEBINFO "${CMAKE_C_FLAGS} -g -DMALLOC_CHECK_=3 -O2")
+set(CMAKE_C_FLAGS_RELWITHDEBINFO "${CMAKE_C_FLAGS} -g3 -DMALLOC_CHECK_=3 -O2")
 
 
 set(GIT_BRANCH        "UNKNOWN")

common/utils/threadPool/Makefile

-all: measurement_display thread-pool-test
+all:
+measurement_display thread-pool-test
 
-measurement_display: measurement_display.c thread-pool.h
-	gcc measurement_display.c -I ${OPENAIR_DIR}/ -I ${OPENAIR_DIR}/common/utils/  -I. ${OPENAIR_DIR}/common/utils/backtrace.c -lpthread -D TEST_THREAD_POOL -I../LOG -I../../utils/T -o measurement_display
+measurement_display:
+measurement_display.c thread-pool.h
+gcc measurement_display.c -I $ {OPENAIR_DIR}/ -I $ {OPENAIR_DIR}/common/utils/  -I. $ {OPENAIR_DIR}/common/utils/backtrace.c -lpthread -D TEST_THREAD_POOL -I../LOG -I../../utils/T -o
+measurement_display
 
-thread-pool-test: thread-pool.c thread-pool.h
-	gcc -g thread-pool.c -I ${OPENAIR_DIR}/ -I ${OPENAIR_DIR}/common/utils/  -I. ${OPENAIR_DIR}/common/utils/backtrace.c -I ${OPENAIR_DIR}/openair2/COMMON ${OPENAIR_DIR}/common/utils/LOG/log.c ${OPENAIR_DIR}/common/config/config_userapi.c ${OPENAIR_DIR}/common/config/config_load_configmodule.c ${OPENAIR_DIR}/common/config/config_cmdline.c -lpthread -ldl -D TEST_THREAD_POOL -I../LOG -I../../utils/T -o thread-pool-test
+thread-pool-test:
+thread-pool.c thread-pool.h
+gcc -g thread-pool.c -I $ {OPENAIR_DIR}/ -I $ {OPENAIR_DIR}/common/utils/  -I. $ {OPENAIR_DIR}/common/utils/backtrace.c -I $ {OPENAIR_DIR}/openair2/COMMON $ {OPENAIR_DIR}/common/utils/LOG/log.c $ {OPENAIR_DIR}/common/config/config_userapi.c
+$ {OPENAIR_DIR}/common/config/config_load_configmodule.c $ {OPENAIR_DIR}/common/config/config_cmdline.c -lpthread -ldl -D TEST_THREAD_POOL -I../LOG -I../../utils/T -o thread-pool-test
 

common/utils/threadPool/thread-pool.c

@@ -67,7 +67,7 @@ void *one_thread(void *arg) {
         delNotifiedFIFO_elt(elt);
       else
         pushNotifiedFIFO(elt->reponseFifo, elt);
-
+      myThread->runningOnKey=-1;
       mutexunlock(tp->incomingFifo.lockF);
     }
   } while (true);
@@ -95,7 +95,7 @@ void initTpool(char *params,tpool_t *pool, bool performanceMeas) {
   pool->nbThreads=0;
   pool->restrictRNTI=false;
   curptr=strtok_r(params,",",&saveptr);
-
+  struct one_thread * ptr;
   while ( curptr!=NULL ) {
     int c=toupper(curptr[0]);
 
@@ -109,8 +109,9 @@ void initTpool(char *params,tpool_t *pool, bool performanceMeas) {
         break;
 
       default:
+	ptr=pool->allthreads;
         pool->allthreads=(struct one_thread *)malloc(sizeof(struct one_thread));
-        pool->allthreads->next=pool->allthreads;
+        pool->allthreads->next=ptr;
         printf("create a thread for core %d\n", atoi(curptr));
         pool->allthreads->coreID=atoi(curptr);
         pool->allthreads->id=pool->nbThreads;

common/utils/threadPool/thread-pool.h

@@ -150,8 +150,9 @@ static inline  notifiedFIFO_elt_t *pollNotifiedFIFO(notifiedFIFO_t *nf) {
 // This function aborts all messages matching the key
 // If the queue is used in thread pools, it doesn't cancels already running processing
 // because the message has already been picked
-static inline void abortNotifiedFIFO(notifiedFIFO_t *nf, uint64_t key) {
+static inline int abortNotifiedFIFO(notifiedFIFO_t *nf, uint64_t key) {
   mutexlock(nf->lockF);
+  int nbDeleted=0;
   notifiedFIFO_elt_t **start=&nf->outF;
 
   while(*start!=NULL) {
@@ -159,13 +160,16 @@ static inline void abortNotifiedFIFO(notifiedFIFO_t *nf, uint64_t key) {
       notifiedFIFO_elt_t *request=*start;
       *start=(*start)->next;
       delNotifiedFIFO_elt(request);
-    }
-
-    if (*start != NULL)
+      nbDeleted++;
+    } else
       start=&(*start)->next;
   }
 
+  if (nf->outF == NULL)
+    nf->inF=NULL;
+
   mutexunlock(nf->lockF);
+  return nbDeleted;
 }
 
 struct one_thread {
@@ -195,7 +199,20 @@ typedef struct thread_pool {
 static inline void pushTpool(tpool_t *t, notifiedFIFO_elt_t *msg) {
   if (t->measurePerf) msg->creationTime=rdtsc();
 
-  pushNotifiedFIFO(&t->incomingFifo, msg);
+  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);
+  }
 }
 
 static inline notifiedFIFO_elt_t *pullTpool(notifiedFIFO_t *responseFifo, tpool_t *t) {
@@ -225,7 +242,8 @@ static inline notifiedFIFO_elt_t *tryPullTpool(notifiedFIFO_t *responseFifo, tpo
   return msg;
 }
 
-static inline void abortTpool(tpool_t *t, uint64_t key) {
+static inline int abortTpool(tpool_t *t, uint64_t key) {
+  int nbRemoved=0;
   notifiedFIFO_t *nf=&t->incomingFifo;
   mutexlock(nf->lockF);
   notifiedFIFO_elt_t **start=&nf->outF;
@@ -235,22 +253,27 @@ static inline void abortTpool(tpool_t *t, uint64_t key) {
       notifiedFIFO_elt_t *request=*start;
       *start=(*start)->next;
       delNotifiedFIFO_elt(request);
-    }
-
-    if (*start != NULL)
+      nbRemoved++;
+    } else
       start=&(*start)->next;
   }
 
+  if (t->incomingFifo.outF==NULL)
+    t->incomingFifo.inF=NULL;
+
   struct one_thread *ptr=t->allthreads;
 
   while(ptr!=NULL) {
-    if (ptr->runningOnKey==key)
+    if (ptr->runningOnKey==key) {
       ptr->abortFlag=true;
+      nbRemoved++;
+    }
 
     ptr=ptr->next;
   }
 
   mutexunlock(nf->lockF);
+  return nbRemoved;
 }
 void initTpool(char *params,tpool_t *pool, bool performanceMeas);
 

common/utils/threadPool/thread-pool.md

@@ -3,69 +3,90 @@
 The thread pool is a working server, made of a set of worker threads that can be mapped on CPU cores.
 
 Each worker loop on pick from the same input queue jobs to do.
+
 When a job is done, the worker sends a return if a return is defined.
 
-A selective abort allows to cancel parallel jobs (usage: a client pushed jobs, but from a response of one job, the other linked jobs becomes useless).
+    A selective abort allows to cancel parallel jobs (usage: a client pushed jobs, but from a response of one job, the other linked jobs becomes useless).
 
-All the thread pool functions are thread safe, nevertheless the working functions are implemented by the thread pool client, so the client has to tackle the parallel execution of his functions called "processingFunc" hereafter.
+      All the thread pool functions are thread safe, nevertheless the working functions are implemented by the thread pool client,
+      so the client has to tackle the parallel execution of his functions called "processingFunc" hereafter.
 
 ## license
-Author: Laurent Thomas, Open cells project 
-The owner share this piece code to Openairsoftware alliance as per OSA license terms
+    Author:
+      Laurent Thomas, Open cells project
+      The owner share this piece code to Openairsoftware alliance as per OSA license terms
 
 # jobs
 
-A job is a message (notifiedFIFO_elt_t):
-next: internal FIFO chain, do not set it
-key:  a long int that the client can use to identify a message or a group of messages
-responseFifo: if the client defines a response FIFO, the message will be posted back after processing
-processingFunc: any funtion (type void processingFunc(void *)) that the worker will launch
-msgData: the data passed to processingFunc. It can be added automatically, or you can set it to a buffer you are managing
-malloced: a boolean that enable internal free in these cases: no return Fifo or Abort feature
-
-The job messages can be created with newNotifiedFIFO_elt() and delNotifiedFIFO_elt() or managed by the client.
+      A job is a message (notifiedFIFO_elt_t):
+    next:
+        internal FIFO chain, do not set it
+    key:
+          a long int that the client can use to identify a message or a group of messages
+    responseFifo:
+          if the client defines a response FIFO, the message will be posted back after processing
+    processingFunc:
+          any funtion (type void processingFunc(void *)) that the worker will launch
+    msgData:
+            the data passed to processingFunc. It can be added automatically, or you can set it to a buffer you are managing
+    malloced:
+    a boolean that enable internal free in these cases:
+            no return Fifo or Abort feature
+
+                      The job messages can be created with newNotifiedFIFO_elt() and delNotifiedFIFO_elt() or managed by the client.
 
 # Queues of jobs
 
-Queues are type of: notifiedFIFO_t that must be initialized by init_notifiedFIFO()
-No delete function is required, the creator has only to free the data of type notifiedFIFO_t
+              Queues are type of:
+                      notifiedFIFO_t that must be initialized by init_notifiedFIFO()
+                      No delete function is required, the creator has only to free the data of type notifiedFIFO_t
 
-push_notifiedFIFO() add a job in the queue
-pull_notifiedFIFO() is blocking, poll_notifiedFIFO() is non blocking
+                      push_notifiedFIFO() add a job in the queue
+                      pull_notifiedFIFO() is blocking, poll_notifiedFIFO() is non blocking
 
-abort_notifiedFIFO() allows the customer to delete all waiting jobs that match with the key (see key in jobs definition)
+                      abort_notifiedFIFO() allows the customer to delete all waiting jobs that match with the key (see key in jobs definition)
 
 # Thread pools
 
 ## initialization
-The clients can create one or more thread pools with init_tpool()
-the params string structure: describes a list of cores, separated by "," that run a worker thread
+                      The clients can create one or more thread pools with init_tpool()
+              the params string structure:
+                      describes a list of cores, separated by "," that run a worker thread
 
-If the core exists on the CPU, the thread pool initialization sets the affinity between this thread and the related code (use negative values is allowed, so the thread will never be mapped on a specific core).
+                      If the core exists on the CPU, the thread pool initialization sets the affinity between this thread and
+                      the related code (use negative values is allowed, so the thread will never be mapped on a specific core).
 
-The threads are all Linux real time scheduler, their name is set automatically is "Tpool_<core id>"
+                      The threads are all Linux real time scheduler, their name is set automatically is "Tpool_<core id>"
 
 ## adding jobs
-The client create their jobs messages as a notifiedFIFO_elt_t, then they push it with pushTpool() (that internally calls push_notifiedFIFO())
+                      The client create their jobs messages as a notifiedFIFO_elt_t, then they push it with pushTpool() (that internally calls push_notifiedFIFO())
 
-If they need a return, they have to create response queues with init_notifiedFIFO() and set this FIFO pointer in the notifiedFIFO_elt_t before pushing the job.
+                      If they need a return, they have to create response queues with init_notifiedFIFO() and set this FIFO pointer in the notifiedFIFO_elt_t before pushing the job.
 
 ## abort
 
-A abort service abortTpool() allows to abort all jobs that match a key (see jobs "key"). When the abort returns, it garanties no job (matching the key) response will be posted on response queues.
+                      A abort service abortTpool() allows to abort all jobs that match a key (see jobs "key"). When the abort returns, it garanties no job (matching the key) response will be posted on response queues.
 
-Nevertheless, jobs already performed before the return of abortTpool() are pushed in the response Fifo queue.
+                      Nevertheless, jobs already performed before the return of abortTpool() are pushed in the response Fifo queue.
 
 ## Performance measurements
 
-A performance measurement is integrated: the pool will automacillay fill timestamps:
+                  A performance measurement is integrated:
+                  the pool will automacillay fill timestamps:
+
+                  * creationTime:
+                          time the request is push to the pool;
 
-* creationTime: time the request is push to the pool;
-* startProcessingTime: time a worker start to run on the job
-* endProcessingTime: time the worker finished the job
-* returnTime: time the client reads the result
+* startProcessingTime:
+time a worker start to run on the job
+* endProcessingTime:
+time the worker finished the job
+* returnTime:
+time the client reads the result
 
-if you set the environement variable: thread-pool-measurements to a valid file name
+if you set the environement variable:
+thread-pool-measurements to a valid file name
 These measurements will be wrote to this Linux pipe.
 
-A tool to read the linux fifo and display it in ascii is provided: see the local directory Makefile for this tool and to compile the thread pool unitary tests.
+A tool to read the linux fifo and display it in ascii is provided:
+see the local directory Makefile for this tool and to compile the thread pool unitary tests.

executables/nr-ru.c

@@ -81,6 +81,7 @@
 #include "common/utils/LOG/vcd_signal_dumper.h"
 
 #include "enb_config.h"
+#include <executables/nr-softmodem.h>
 
 #ifdef SMBV
 #include "PHY/TOOLS/smbv.h"
@@ -730,13 +731,15 @@ void tx_rf(RU_t *ru,int frame,int slot, uint64_t timestamp) {
   //nr_subframe_t SF_type     = nr_slot_select(cfg,slot%fp->slots_per_frame);
 
   if ((slot == 0) ||
-      (slot == 1)) {
+      (slot == 1) || IS_SOFTMODEM_RFSIM ) {
     int siglen=fp->samples_per_slot;
     int flags;
     if (slot==0)
       flags = 2;
     else if (slot==1)
       flags=3;
+    else
+      flags=4;
 
     /*
         if (SF_type == SF_S) {

executables/nr-softmodem.h

@@ -49,10 +49,12 @@
 
 #define SOFTMODEM_NOS1_BIT            (1<<0)
 #define SOFTMODEM_NOKRNMOD_BIT        (1<<1)
+#define SOFTMODEM_RFSIM_BIT           (1<<10)
 
 #define IS_SOFTMODEM_NOS1            ( get_softmodem_optmask() & SOFTMODEM_NOS1_BIT)
 #define IS_SOFTMODEM_NOKRNMOD        ( get_softmodem_optmask() & SOFTMODEM_NOKRNMOD_BIT)
 
+#define IS_SOFTMODEM_RFSIM           ( get_softmodem_optmask() & SOFTMODEM_RFSIM_BIT)
 extern uint64_t get_softmodem_optmask(void);
 extern void get_common_options(void);
 

executables/nr-ue.c

@@ -538,6 +538,9 @@ void trashFrame(PHY_VARS_NR_UE *UE, openair0_timestamp *timestamp) {
                                dummy_rx,
                                UE->frame_parms.samples_per_subframe,
                                UE->frame_parms.nb_antennas_rx);
+    if (IS_SOFTMODEM_RFSIM ) {
+	 usleep(1000); // slow down, as would do actuall rf to let cpu for the synchro thread
+    }
   }
 
   for (int i=0; i<UE->frame_parms.nb_antennas_tx; i++)

targets/ARCH/COMMON/common_lib.c

@@ -126,7 +126,7 @@ int load_lib(openair0_device *device,
 	  libname=OAI_RFSIM_LIBNAME;
 	  shlib_fdesc[0].fname="device_init";
   } else if (flag == RAU_LOCAL_RADIO_HEAD) {
-	  if (getenv("RFSIMULATOR") != NULL)
+	  if (IS_SOFTMODEM_RFSIM)
 		  libname="rfsimulator";
 	  else
           libname=OAI_RF_LIBNAME;

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,11 +103,11 @@ 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*10;
+  int sendbuff=1000*1000*100;
   AssertFatal ( setsockopt(sock, SOL_SOCKET, SO_SNDBUF, &sendbuff, sizeof(sendbuff)) == 0, "");
   struct epoll_event ev= {0};
   ev.events = EPOLLIN | EPOLLRDHUP;
@@ -183,7 +184,7 @@ void setblocking(int sock, enum blocking_t active) {
   AssertFatal(fcntl(sock, F_SETFL, opts) >= 0, "");
 }
 
-static bool flushInput(rfsimulator_state_t *t, int timeout);
+static bool flushInput(rfsimulator_state_t *t, int timeout, int nsamps);
 
 void fullwrite(int fd, void *_buf, ssize_t count, rfsimulator_state_t *t) {
   if (t->saveIQfile != -1) {
@@ -207,7 +208,8 @@ void fullwrite(int fd, void *_buf, ssize_t count, rfsimulator_state_t *t) {
       if(errno==EAGAIN) {
         // The opposite side is saturated
         // we read incoming sockets meawhile waiting
-        flushInput(t, 5);
+        //flushInput(t, 5);
+        usleep(500);
         continue;
       } else
         return;
@@ -218,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;
@@ -316,19 +316,18 @@ sin_addr:
   return 0;
 }
 
-int rfsimulator_write(openair0_device *device, openair0_timestamp timestamp, void **samplesVoid, int nsamps, int nbAnt, int flags) {
-  rfsimulator_state_t *t = device->priv;
+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; i<FD_SETSIZE; i++) {
     buffer_t *b=&t->buf[i];
 
     if (b->conn_sock >= 0 ) {
-      if ( abs((double)b->lastWroteTS-timestamp) > (double)CirSize )
-        LOG_E(HW,"Tx/Rx shift too large Tx:%lu, Rx:%lu\n", b->lastWroteTS, b->lastReceivedTS);
-
       samplesBlockHeader_t header= {t->typeStamp, nsamps, nbAnt, timestamp};
-      fullwrite(b->conn_sock, &header, sizeof(header), t);
+      fullwrite(b->conn_sock,&header, sizeof(header), t);
       sample_t tmpSamples[nsamps][nbAnt];
 
       for(int a=0; a<nbAnt; a++) {
@@ -340,21 +339,30 @@ int rfsimulator_write(openair0_device *device, openair0_timestamp timestamp, voi
 
       if (b->conn_sock >= 0 ) {
         fullwrite(b->conn_sock, (void *)tmpSamples, sampleToByte(nsamps,nbAnt), t);
-        b->lastWroteTS=timestamp+nsamps;
       }
     }
   }
 
+  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) );
-  // Let's verify we don't have incoming data
-  // This is mandatory when the opposite side doesn't transmit
-  flushInput(t, 0);
-  pthread_mutex_unlock(&Sockmutex);
   return nsamps;
 }
 
-static bool flushInput(rfsimulator_state_t *t, int timeout) {
+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, 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
   struct epoll_event events[FD_SETSIZE]= {0};
@@ -375,7 +383,16 @@ static bool flushInput(rfsimulator_state_t *t, int timeout) {
       AssertFatal( (conn_sock = accept(t->listen_sock,NULL,NULL)) != -1, "");
       setblocking(conn_sock, notBlocking);
       allocCirBuf(t, conn_sock);
-      LOG_I(HW,"A ue connected\n");
+      LOG_I(HW,"A ue connected, sending the current time\n");
+      struct complex16 v= {0};
+      void *samplesVoid[t->tx_num_channels];
+
+      for ( int i=0; i < t->tx_num_channels; i++)
+        samplesVoid[i]=(void *)&v;
+
+      rfsimulator_write_internal(t, t->lastWroteTS > 1 ? t->lastWroteTS-1 : 0,
+                                 samplesVoid, 1,
+                                 t->tx_num_channels, 1, false);
     } else {
       if ( events[nbEv].events & (EPOLLHUP | EPOLLERR | EPOLLRDHUP) ) {
         socketError(t,fd);
@@ -421,40 +438,60 @@ static bool flushInput(rfsimulator_state_t *t, int timeout) {
                      (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;
             }
           }
-
-          LOG_W(HW,"gap of: %ld in reception\n", b->th.timestamp-b->lastReceivedTS );
-        }
-
-        b->lastReceivedTS=b->th.timestamp;
-        AssertFatal(b->lastWroteTS == 0 || ( abs((double)b->lastWroteTS-b->lastReceivedTS) < (double)CirSize),
-                    "Tx/Rx shift too large Tx:%lu, Rx:%lu\n", b->lastWroteTS, b->lastReceivedTS);
+          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);	}
+
+        pthread_mutex_lock(&Sockmutex);
+
+        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];
         b->remainToTransfer=sampleToByte(b->th.size, b->th.nbAnt);
       }
 
       if ( b->headerMode==false ) {
-        LOG_D(HW,"Set b->lastReceivedTS %ld\n", b->lastReceivedTS);
-        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 )
-          t->nextTimestamp=b->lastReceivedTS-b->th.size;
-
-        if ( b->remainToTransfer==0) {
-          LOG_D(HW,"Completed block reception: %ld\n", b->lastReceivedTS);
-          b->headerMode=true;
-          b->transferPtr=(char *)&b->th;
-          b->remainToTransfer=sizeof(samplesBlockHeader_t);
-          b->th.magic=-1;
+        LOG_D(HW,"UEsock: %d Set b->lastReceivedTS %ld\n", fd, b->lastReceivedTS);
+	if ( ! b->trashingPacket ) {
+	  b->lastReceivedTS=b->th.timestamp+b->th.size-byteToSample(b->remainToTransfer,b->th.nbAnt);
+	}
+
+	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;
         }
       }
     }
@@ -468,7 +505,6 @@ int rfsimulator_read(openair0_device *device, openair0_timestamp *ptimestamp, vo
     LOG_W(HW, "rfsimulator: only 1 antenna tested\n");
   }
 
-  pthread_mutex_lock(&Sockmutex);
   rfsimulator_state_t *t = device->priv;
   LOG_D(HW, "Enter rfsimulator_read, expect %d samples, will release at TS: %ld\n", nsamps, t->nextTimestamp+nsamps);
   // deliver data from received data
@@ -484,45 +520,58 @@ int rfsimulator_read(openair0_device *device, openair0_timestamp *ptimestamp, vo
     if ( t->nextTimestamp == 0)
       LOG_W(HW,"No connected device, generating void samples...\n");
 
-    if (!flushInput(t, 10)) {
+    if (!flushInput(t, 10,  nsamps)) {
       for (int x=0; x < nbAnt; x++)
         memset(samplesVoid[x],0,sampleToByte(nsamps,1));
 
       t->nextTimestamp+=nsamps;
-      LOG_D(HW,"Generated void samples for Rx: %ld\n", t->nextTimestamp);
+
+      if ( ((t->nextTimestamp/nsamps)%100) == 0)
+        LOG_W(HW,"No UE, Generated void samples for Rx: %ld\n", t->nextTimestamp);
+
       *ptimestamp = t->nextTimestamp-nsamps;
-      pthread_mutex_unlock(&Sockmutex);
       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 {
       have_to_wait=false;
 
-      for (int sock=0; sock<FD_SETSIZE; sock++) {
+      for ( int sock=0; sock<FD_SETSIZE; sock++) {
         buffer_t *b=&t->buf[sock];
 
-        if ( b->circularBuf) {
-          LOG_D(HW,"sock: %d, lastWroteTS: %lu, lastRecvTS: %lu, TS must be avail: %lu\n",
-                sock, b->lastWroteTS,
-                b->lastReceivedTS,
-                t->nextTimestamp+nsamps);
-
-          if (  b->lastReceivedTS > b->lastWroteTS ) {
-            // The caller modem (NB, UE, ...) must send Tx in advance, so we fill TX if Rx is in advance
-            // This occurs for example when UE is in sync mode: it doesn't transmit
-            // with USRP, it seems ok: if "tx stream" is off, we may consider it actually cuts the Tx power
-            struct complex16 v = {0};
-            void *samplesVoid[b->th.nbAnt];
-
-            for ( int i=0; i < b->th.nbAnt; i++)
-              samplesVoid[i]=(void *)&v;
-
-            rfsimulator_write(device, b->lastReceivedTS, samplesVoid, 1, b->th.nbAnt, 0);
-          }
-        }
-
         if ( b->circularBuf )
           if ( t->nextTimestamp+nsamps > b->lastReceivedTS ) {
             have_to_wait=true;
@@ -535,7 +584,7 @@ int rfsimulator_read(openair0_device *device, openair0_timestamp *ptimestamp, vo
           ptr->lastReceivedTS,
           t->nextTimestamp+nsamps);
         */
-        flushInput(t, 3);
+        flushInput(t, 3, nsamps);
     } while (have_to_wait);
   }
 
@@ -582,7 +631,6 @@ int rfsimulator_read(openair0_device *device, openair0_timestamp *ptimestamp, vo
         nsamps,
         *ptimestamp, t->nextTimestamp,
         signal_energy(samplesVoid[0], nsamps));
-  pthread_mutex_unlock(&Sockmutex);
   return nsamps;
 }
 int rfsimulator_request(openair0_device *device, void *msg, ssize_t msg_len) {
@@ -613,7 +661,6 @@ __attribute__((__visibility__("default")))
 int device_init(openair0_device *device, openair0_config_t *openair0_cfg) {
   // to change the log level, use this on command line
   // --log_config.hw_log_level debug
-  // (for phy layer, replace "hw" by "phy")
   rfsimulator_state_t *rfsimulator = (rfsimulator_state_t *)calloc(sizeof(rfsimulator_state_t),1);
   rfsimulator_readconfig(rfsimulator);
   pthread_mutex_init(&Sockmutex, NULL);
@@ -629,7 +676,6 @@ int device_init(openair0_device *device, openair0_config_t *openair0_cfg) {
   device->trx_set_gains_func   = rfsimulator_set_gains;
   device->trx_write_func       = rfsimulator_write;
   device->trx_read_func      = rfsimulator_read;
-  device->uhd_set_thread_priority = NULL;
   /* let's pretend to be a b2x0 */
   device->type = USRP_B200_DEV;
   device->openair0_cfg=&openair0_cfg[0];

targets/PROJECTS/GENERIC-LTE-EPC/CONF/gnb.band78.tm1.106PRB.usrpx300.conf

@@ -262,7 +262,7 @@ RUs = (
 THREAD_STRUCT = (
   {
     #three config for level of parallelism "PARALLEL_SINGLE_THREAD", "PARALLEL_RU_L1_SPLIT", or "PARALLEL_RU_L1_TRX_SPLIT"
-    parallel_config    = "PARALLEL_RU_L1_TRX_SPLIT";
+    parallel_config    = "PARALLEL_SINGLE_THREAD";
     #two option for worker "WORKER_DISABLE" or "WORKER_ENABLE"
     worker_config      = "WORKER_DISABLE";
   }