Merge remote-tracking branch 'origin/fix-rfsim-multi-client-out-of-order' into integration_2025_w29 (!3455)

RF simulator: Fix concurrency issues during new client connections

- Fix concurrency issues happening when a new client connects, sometimes
  leading to a total softmodem deadlock.
- trace in 4G RAR processing to debug ci result

ci-scripts/yaml_files/5g_f1_rfsimulator/docker-compose.yaml

@@ -194,7 +194,6 @@ services:
                                     --uicc0.imsi 208990100001100
                                     --rfsimulator.options chanmod
                                     --rfsimulator.serveraddr server
-                                    --rfsimulator.hanging-workaround 1
                                     --telnetsrv --telnetsrv.shrmod ciUE --telnetsrv.listenaddr 192.168.71.181 --telnetsrv.listenport 8091
                                     --channelmod.modellist_rfsimu_1.[0].model_name rfsimu_channel_ue0
                                     --channelmod.modellist_rfsimu_1.[1].model_name rfsimu_channel_ue1

common/utils/telnetsrv/telnetsrv.c

@@ -692,7 +692,7 @@ void run_telnetsrv(void) {
         break;
       }
 
-      if (telnetparams.telnetdbg > 0)
+      //if (telnetparams.telnetdbg > 0)
       TELNET_LOG("Command received: readc %i filled %i \"%s\"\n", readc, filled, buf);
 
       if (buf[0] == '!') {

executables/nr-gnb.c

@@ -114,7 +114,7 @@ static void tx_func(processingData_L1tx_t *info)
   pushNotifiedFIFO(&gNB->resp_L1, res);
 
   int tx_slot_type = nr_slot_select(cfg, frame_tx, slot_tx);
-  if (tx_slot_type == NR_DOWNLINK_SLOT || tx_slot_type == NR_MIXED_SLOT || get_softmodem_params()->continuous_tx) {
+  if (tx_slot_type == NR_DOWNLINK_SLOT || tx_slot_type == NR_MIXED_SLOT || get_softmodem_params()->continuous_tx || IS_SOFTMODEM_RFSIM) {
     start_meas(&info->gNB->phy_proc_tx);
     phy_procedures_gNB_TX(info, frame_tx, slot_tx, 1);
 

executables/nr-ru.c

@@ -759,7 +759,7 @@ void tx_rf(RU_t *ru, int frame,int slot, uint64_t timestamp)
   radio_tx_burst_flag_t flags_burst = TX_BURST_INVALID;
   radio_tx_gpio_flag_t flags_gpio = 0;
 
-  if (cfg->cell_config.frame_duplex_type.value == TDD && !get_softmodem_params()->continuous_tx) {
+  if (cfg->cell_config.frame_duplex_type.value == TDD && !get_softmodem_params()->continuous_tx && !IS_SOFTMODEM_RFSIM) {
     int slot_type = nr_slot_select(cfg,frame,slot%fp->slots_per_frame);
     if(slot_type == NR_MIXED_SLOT) {
       int txsymb = 0;

openair1/PHY/LTE_UE_TRANSPORT/rar_tools_ue.c

@@ -105,7 +105,7 @@ int generate_ue_ulsch_params_from_rar(PHY_VARS_UE *ue,
   cqireq=rar[3]&1;
 
   if (rballoc>RIV_max) {
-    LOG_D(PHY,"rar_tools.c: ERROR: rb_alloc (%x) > RIV_max\n",rballoc);
+    LOG_W(PHY,"rar_tools.c: ERROR: rb_alloc (%x) > RIV_max\n",rballoc);
     return(-1);
   }
 

radio/rfsimulator/README.md

@@ -79,7 +79,6 @@ The RF simulator is using the configuration module, and its parameters are defin
 |`--rfsimulator.IQfile <file>`     | path to a file to store the IQ samples to (only with `saviq`)                  | `/tmp/rfsimulator.iqs` |
 |`--rfsimulator.prop_delay`        | simulated receive-path (gNB: UL, UE: DL) propagation delay in ms               | 0                      |
 |`--rfsimulator.wait_timeout`      | wait timeout when no UE is connected                                           | 1                      |
-|`--rfismulator.hanging-workaround`| Enable workaround to de-block potentially hanging server on new client connection.  | 0                 |
 
 Please refer to this document [`SIMULATION/TOOLS/DOC/channel_simulation.md`](../../openair1/SIMULATION/TOOLS/DOC/channel_simulation.md) for information about using the RFSimulator options to run the simulator with a channel model.
 

radio/rfsimulator/simulator.c

@@ -21,7 +21,6 @@
  *      contact@openairinterface.org
  */
 
-
 /*
  * Open issues and limitations
  * The read and write should be called in the same thread, that is not new USRP UHD design
@@ -80,12 +79,12 @@ typedef enum { SIMU_ROLE_SERVER = 1, SIMU_ROLE_CLIENT } simuRole;
 #define RFSIMU_SECTION "rfsimulator"
 #define RFSIMU_OPTIONS_PARAMNAME "options"
 
-
+#define RFSIM_CONFIG_HELP_OPTIONS                                                                  \
-#define RFSIM_CONFIG_HELP_OPTIONS     " list of comma separated options to enable rf simulator functionalities. Available options: \n"\
+  " list of comma separated options to enable rf simulator functionalities. Available options: \n" \
   "        chanmod:   enable channel modelisation\n"                                               \
   "        saviq:     enable saving written iqs to a file\n"
-#define CONFIG_HELP_HANG_WORKAROUND "Enable workaroud for server mode hanging on new client connection.\n"
 
+// clang-format off
 /*-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------*/
 /*                                            configuration parameters for the rfsimulator device */
 /*   optname                     helpstr                     paramflags           XXXptr                               defXXXval                          type         numelt  */
@@ -102,15 +101,15 @@ typedef enum { SIMU_ROLE_SERVER = 1, SIMU_ROLE_CLIENT } simuRole;
       {"offset", "<channel offset in samps>\n", simOpt, .u64ptr = &(rfsimulator->chan_offset), .defint64val = 0, TYPE_UINT64, 0}, \
     {"prop_delay",             "<propagation delay in ms>\n",         simOpt,  .dblptr=&(rfsimulator->prop_delay_ms),  .defdblval=0.0,                   TYPE_DOUBLE,    0 },\
     {"wait_timeout",           "<wait timeout if no UE connected>\n", simOpt,  .iptr=&(rfsimulator->wait_timeout),     .defintval=1,                     TYPE_INT,       0 },\
-    {"hanging-workaround",     CONFIG_HELP_HANG_WORKAROUND,           simOpt,  .iptr=&rfsimulator->hanging_workaround, .defintval=0,                     TYPE_INT,       0 },\
   };
-
+// clang-format on
 static void getset_currentchannels_type(char *buf, int debug, webdatadef_t *tdata, telnet_printfunc_t prnt);
 extern int get_currentchannels_type(char *buf, int debug, webdatadef_t *tdata, telnet_printfunc_t prnt); // in random_channel.c
 static int rfsimu_setchanmod_cmd(char *buff, int debug, telnet_printfunc_t prnt, void *arg);
 static int rfsimu_setdistance_cmd(char *buff, int debug, telnet_printfunc_t prnt, void *arg);
 static int rfsimu_getdistance_cmd(char *buff, int debug, telnet_printfunc_t prnt, void *arg);
 static int rfsimu_vtime_cmd(char *buff, int debug, telnet_printfunc_t prnt, void *arg);
+// clang-format off
 static telnetshell_cmddef_t rfsimu_cmdarray[] = {
     {"show models", "", (cmdfunc_t)rfsimu_setchanmod_cmd, {(webfunc_t)getset_currentchannels_type}, TELNETSRV_CMDFLAG_WEBSRVONLY | TELNETSRV_CMDFLAG_GETWEBTBLDATA, NULL},
     {"setmodel", "<model name> <model type>", (cmdfunc_t)rfsimu_setchanmod_cmd, {NULL}, TELNETSRV_CMDFLAG_PUSHINTPOOLQ | TELNETSRV_CMDFLAG_TELNETONLY, NULL},
@@ -119,14 +118,11 @@ static telnetshell_cmddef_t rfsimu_cmdarray[] = {
     {"vtime", "", (cmdfunc_t)rfsimu_vtime_cmd, {NULL}, TELNETSRV_CMDFLAG_PUSHINTPOOLQ | TELNETSRV_CMDFLAG_AUTOUPDATE},
     {"", "", NULL},
 };
+// clang-format on
 static telnetshell_cmddef_t *setmodel_cmddef = &(rfsimu_cmdarray[1]);
 
 static telnetshell_vardef_t rfsimu_vardef[] = {{"", 0, 0, NULL}};
-pthread_mutex_t Sockmutex;
-unsigned int nb_ue = 0;
-
 static uint64_t CirSize = minCirSize;
-
 typedef c16_t sample_t; // 2*16 bits complex number
 
 typedef struct buffer_s {
@@ -144,6 +140,8 @@ typedef struct buffer_s {
 
 typedef struct {
   int listen_sock, epollfd;
+  pthread_mutex_t Sockmutex;
+  unsigned int nb_cnx;
   openair0_timestamp nextRxTstamp;
   openair0_timestamp lastWroteTS;
   simuRole role;
@@ -165,13 +163,15 @@ typedef struct {
   poll_telnetcmdq_func_t poll_telnetcmdq;
   int wait_timeout;
   double prop_delay_ms;
-  int hanging_workaround;
 } rfsimulator_state_t;
 
+static bool flushInput(rfsimulator_state_t *t, int timeout, bool first_time);
+
 static buffer_t *allocCirBuf(rfsimulator_state_t *bridge, int sock)
 {
   uint64_t buff_index = bridge->next_buf++ % MAX_FD_RFSIMU;
   buffer_t *ptr = &bridge->buf[buff_index];
+  bridge->nb_cnx++;
   ptr->circularBuf = calloc(1, sampleToByte(CirSize, 1));
   if (ptr->circularBuf == NULL) {
     LOG_E(HW, "malloc(%lu) failed\n", sampleToByte(CirSize, 1));
@@ -214,7 +214,11 @@ static buffer_t *allocCirBuf(rfsimulator_state_t *bridge, int sock)
       init_done = true;
     }
     char modelname[30];
-    snprintf(modelname, sizeofArray(modelname), "rfsimu_channel_%s%d", (bridge->role == SIMU_ROLE_SERVER) ? "ue" : "enB", nb_ue);
+    snprintf(modelname,
+             sizeofArray(modelname),
+             "rfsimu_channel_%s%d",
+             (bridge->role == SIMU_ROLE_SERVER) ? "ue" : "enB",
+             bridge->nb_cnx);
     ptr->channel_model = find_channel_desc_fromname(modelname); // path_loss in dB
     if (!ptr->channel_model) {
       // Use legacy method to find channel model - this will use the same channel model for all clients
@@ -246,7 +250,7 @@ static void removeCirBuf(rfsimulator_state_t *bridge, buffer_t *buf)
   // free(bridge->buf[sock].channel_model);
   memset(buf, 0, sizeof(buffer_t));
   buf->conn_sock = -1;
-  nb_ue--;
+  bridge->nb_cnx--;
 }
 
 static void socketError(rfsimulator_state_t *bridge, buffer_t *buf)
@@ -286,8 +290,6 @@ static int setblocking(int sock, enum blocking_t active)
   return 0;
 }
 
-static bool flushInput(rfsimulator_state_t *t, int timeout, int nsamps);
-
 static void fullwrite(int fd, void *_buf, ssize_t count, rfsimulator_state_t *t) {
   if (t->saveIQfile != -1) {
     if (write(t->saveIQfile, _buf, count) != count)
@@ -450,25 +452,6 @@ static void getset_currentchannels_type(char *buf, int debug, webdatadef_t *tdat
     get_currentchannels_type("modify type", debug, tdata, prnt);
   }
 }
-/*getset_currentchannels_type */
-
-//static void print_cirBuf(struct complex16 *circularBuf,
-//                         uint64_t firstSample,
-//                         uint32_t cirSize,
-//                         int neg,
-//                         int pos,
-//                         int nbTx)
-//{
-//  for (int i = -neg; i < pos ; ++i) {
-//    for (int txAnt = 0; txAnt < nbTx; txAnt++) {
-//      const int idx = ((firstSample + i) * nbTx + txAnt + cirSize) % cirSize;
-//      if (i == 0)
-//        printf("->");
-//      printf("%08x%08x\n", circularBuf[idx].r, circularBuf[idx].i);
-//    }
-//  }
-//  printf("\n");
-//}
 
 static int rfsimu_setdistance_cmd(char *buff, int debug, telnet_printfunc_t prnt, void *arg)
 {
@@ -678,13 +661,34 @@ static int startClient(openair0_device *device)
   if (setblocking(sock, notBlocking) == -1) {
     return -1;
   }
-  return allocCirBuf(t, sock) == NULL ? -1 : 0;
+  buffer_t *b = allocCirBuf(t, sock);
+  if (!b)
+    return -1;
+  // read a 1 sample block to initialize the current time
+  bool have_to_wait;
+  do {
+    have_to_wait = true;
+    flushInput(t, 3, true);
+    if (b->lastReceivedTS)
+      have_to_wait = false;
+  } while (have_to_wait);
+  if (b->lastReceivedTS > 0)
+    b->lastReceivedTS--;
+  t->nextRxTstamp = b->lastReceivedTS;
+  LOG_D(HW, "Client got first timestamp: starting at %lu\n", t->nextRxTstamp);
+  if (b->channel_model)
+    b->channel_model->start_TS = t->nextRxTstamp;
+  return 0;
 }
 
-static int rfsimulator_write_internal(rfsimulator_state_t *t, openair0_timestamp timestamp, void **samplesVoid, int nsamps, int nbAnt, int flags, bool alreadyLocked) {
+static int rfsimulator_write_internal(rfsimulator_state_t *t,
-  if (!alreadyLocked)
+                                      openair0_timestamp timestamp,
-    pthread_mutex_lock(&Sockmutex);
+                                      void **samplesVoid,
-
+                                      int nsamps,
+                                      int nbAnt,
+                                      int flags)
+{
+  mutexlock(t->Sockmutex);
   LOG_D(HW, "Sending %d samples at time: %ld, nbAnt %d\n", nsamps, timestamp, nbAnt);
 
   for (int i = 0; i < MAX_FD_RFSIMU; i++) {
@@ -693,26 +697,19 @@ static int rfsimulator_write_internal(rfsimulator_state_t *t, openair0_timestamp
     if (b->conn_sock >= 0) {
       samplesBlockHeader_t header = {nsamps, nbAnt, timestamp};
       fullwrite(b->conn_sock, &header, sizeof(header), t);
-      sample_t tmpSamples[nsamps][nbAnt];
-
       if (nbAnt == 1) {
-        if (b->conn_sock >= 0) {
         fullwrite(b->conn_sock, samplesVoid[0], sampleToByte(nsamps, nbAnt), t);
-        }
       } else {
+        sample_t tmpSamples[nsamps][nbAnt];
         for (int a = 0; a < nbAnt; a++) {
           sample_t *in = (sample_t *)samplesVoid[a];
-
           for (int s = 0; s < nsamps; s++)
             tmpSamples[s][a] = in[s];
         }
-
-        if (b->conn_sock >= 0) {
         fullwrite(b->conn_sock, (void *)tmpSamples, sampleToByte(nsamps, nbAnt), t);
       }
     }
   }
-  }
 
   if (t->lastWroteTS != 0 && fabs((double)t->lastWroteTS - timestamp) > (double)CirSize)
     LOG_W(HW, "Discontinuous TX gap too large Tx:%lu, %lu\n", t->lastWroteTS, timestamp);
@@ -728,9 +725,7 @@ static int rfsimulator_write_internal(rfsimulator_state_t *t, openair0_timestamp
           timestamp - t->lastWroteTS);
 
   t->lastWroteTS = timestamp + nsamps;
-
+  mutexunlock(t->Sockmutex);
-  if (!alreadyLocked)
-    pthread_mutex_unlock(&Sockmutex);
 
   LOG_D(HW,
         "Sent %d samples at time: %ld->%ld, energy in first antenna: %d\n",
@@ -741,33 +736,22 @@ static int rfsimulator_write_internal(rfsimulator_state_t *t, openair0_timestamp
   return nsamps;
 }
 
-static int rfsimulator_write(openair0_device *device, openair0_timestamp timestamp, void **samplesVoid, int nsamps, int nbAnt, int flags) {
+static int rfsimulator_write(openair0_device *device,
+                             openair0_timestamp timestamp,
+                             void **samplesVoid,
+                             int nsamps,
+                             int nbAnt,
+                             int flags)
+{
   timestamp -= device->openair0_cfg->command_line_sample_advance;
-  return rfsimulator_write_internal(device->priv, timestamp, samplesVoid, nsamps, nbAnt, flags, false); // false = with lock
+  return rfsimulator_write_internal(device->priv, timestamp, samplesVoid, nsamps, nbAnt, flags);
-  // return rfsimulator_write_internal(device->priv, timestamp, samplesVoid, nsamps, nbAnt, flags, true);
 }
 
-static bool flushInput(rfsimulator_state_t *t, int timeout, int nsamps_for_initial) {
+static bool add_client(rfsimulator_state_t *t)
-  // Process all incoming events on sockets
+{
-  // store the data in lists
+  struct sockaddr_storage sa = {0};
-  struct epoll_event events[MAX_FD_RFSIMU] = {{0}};
+  socklen_t socklen = sizeof(sa);
-  int nfds = epoll_wait(t->epollfd, events, MAX_FD_RFSIMU, timeout);
+  int conn_sock = accept(t->listen_sock, (struct sockaddr *)&sa, &socklen);
-
-  if ( nfds==-1 ) {
-    if ( errno==EINTR || errno==EAGAIN ) {
-      return false;
-    } else {
-      LOG_W(HW, "epoll_wait() failed, errno(%d)\n", errno);
-      return false;
-    }
-  }
-
-  for (int nbEv = 0; nbEv < nfds; ++nbEv) {
-    buffer_t *b = events[nbEv].data.ptr;
-
-    if (events[nbEv].events & EPOLLIN && b == NULL) {
-      int conn_sock;
-      conn_sock = accept(t->listen_sock, NULL, NULL);
   if (conn_sock == -1) {
     LOG_E(HW, "accept() failed, errno(%d)\n", errno);
     return false;
@@ -775,109 +759,81 @@ static bool flushInput(rfsimulator_state_t *t, int timeout, int nsamps_for_initi
   if (setblocking(conn_sock, notBlocking)) {
     return false;
   }
+  mutexlock(t->Sockmutex);
   buffer_t *new_buf = allocCirBuf(t, conn_sock);
   if (new_buf == NULL) {
+    mutexunlock(t->Sockmutex);
     return false;
   }
-      LOG_I(HW, "A client connects, sending the current time\n");
+  new_buf->lastReceivedTS = t->lastWroteTS;
+  char ip[INET6_ADDRSTRLEN];
+  getnameinfo((struct sockaddr *)&sa, socklen, ip, sizeof(ip), NULL, 0, NI_NUMERICHOST);
+  uint16_t port = ((struct sockaddr_in *)&sa)->sin_port;
+  LOG_I(HW, "Client connects from %s:%d\n", ip, port);
   c16_t v = {0};
-      nb_ue++;
   void *samplesVoid[t->tx_num_channels];
-
   for (int i = 0; i < t->tx_num_channels; i++)
     samplesVoid[i] = (void *)&v;
-
+  samplesBlockHeader_t header = {1, t->tx_num_channels, t->lastWroteTS};
-      rfsimulator_write_internal(t, t->lastWroteTS > 1 ? t->lastWroteTS - 1 : 0, samplesVoid, 1, t->tx_num_channels, 1, false);
+  fullwrite(conn_sock, &header, sizeof(header), t);
+  fullwrite(conn_sock, samplesVoid, sampleToByte(1, t->tx_num_channels), t);
 
   if (new_buf->channel_model)
     new_buf->channel_model->start_TS = t->lastWroteTS;
-    } else {
+  mutexunlock(t->Sockmutex);
-      if ( events[nbEv].events & (EPOLLHUP | EPOLLERR | EPOLLRDHUP) ) {
+  return true;
-        socketError(t, b);
-        continue;
-      }
-      if ( b->circularBuf == NULL ) {
-        LOG_E(HW, "Received data on not connected socket %d\n", events[nbEv].data.fd);
-        continue;
 }
 
-      ssize_t blockSz;
+static void process_recv_header(rfsimulator_state_t *t, buffer_t *b, bool first_time)
-
+{
-      if ( b->headerMode)
-        blockSz=b->remainToTransfer;
-      else
-        blockSz= b->transferPtr + b->remainToTransfer <= b->circularBufEnd ?
-                 b->remainToTransfer :
-                 b->circularBufEnd - b->transferPtr ;
-
-      ssize_t sz = recv(b->conn_sock, b->transferPtr, blockSz, MSG_DONTWAIT);
-
-      if ( sz < 0 ) {
-        if ( errno != EAGAIN ) {
-          LOG_E(HW, "recv() failed, errno(%d)\n", errno);
-          //abort();
-        }
-      } else if ( sz == 0 )
-        continue;
-
-      LOG_D(HW, "Socket rcv %zd bytes\n", sz);
-      b->remainToTransfer -= sz;
-      b->transferPtr+=sz;
-
-      if (b->transferPtr==b->circularBufEnd )
-        b->transferPtr=(char *)b->circularBuf;
-
   // check the header and start block transfer
-      if ( b->headerMode==true && b->remainToTransfer==0) {
+  if (b->remainToTransfer != 0)
-        b->headerMode = false;
+    return;
 
-        if (t->nextRxTstamp == 0) { // First block in UE, resync with the gNB current TS
+  b->headerMode = false; // We got the header
-          t->nextRxTstamp=b->th.timestamp> nsamps_for_initial ?
+
-                           b->th.timestamp -  nsamps_for_initial :
+  if (first_time) {
-                           0;
+    b->lastReceivedTS = b->th.timestamp;
-          b->lastReceivedTS=b->th.timestamp> nsamps_for_initial ?
-                            b->th.timestamp :
-                            nsamps_for_initial;
-          LOG_D(HW, "UE got first timestamp: starting at %lu\n", t->nextRxTstamp);
     b->trashingPacket = true;
-          if (b->channel_model)
+  } else {
-            b->channel_model->start_TS = t->nextRxTstamp;
+    if (b->lastReceivedTS < b->th.timestamp) {
-        } else if (b->lastReceivedTS < b->th.timestamp) {
+      // We have a transmission hole to fill, like TDD
+      // we create no signal samples up to the beginning of this reception
       int nbAnt = b->th.nbAnt;
-
       if (b->th.timestamp - b->lastReceivedTS < CirSize) {
+        // case we wrap at circular buffer end
         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] = (c16_t){0};
-                b->circularBuf[(index*nbAnt+a)%CirSize].i = 0;
           }
         }
       } else {
+        // case no circular buffer wrap
         memset(b->circularBuf, 0, sampleToByte(CirSize, 1));
       }
-
       b->lastReceivedTS = b->th.timestamp;
-        } else if (b->lastReceivedTS > b->th.timestamp && b->th.size == 1) {
+    } else if (b->lastReceivedTS > b->th.timestamp) {
-          LOG_W(HW, "Received Rx/Tx synchro out of order\n");
-          b->trashingPacket=true;
-        } else if (b->lastReceivedTS == b->th.timestamp) {
-          // normal case
-        } else {
       LOG_W(HW, "Received data in past: current is %lu, new reception: %lu!\n", b->lastReceivedTS, b->th.timestamp);
       b->trashingPacket = true;
     }
-
+    // verify the time gap is not too large
-        pthread_mutex_lock(&Sockmutex);
+    mutexlock(t->Sockmutex);
-
     if (t->lastWroteTS != 0 && (fabs((double)t->lastWroteTS - b->lastReceivedTS) > (double)CirSize))
       LOG_W(HW, "UEsock(%d) Tx/Rx shift too large Tx:%lu, Rx:%lu\n", b->conn_sock, t->lastWroteTS, b->lastReceivedTS);
-
+    mutexunlock(t->Sockmutex);
-        pthread_mutex_unlock(&Sockmutex);
+  }
+  // move back the pointer to overwrite the header, we don't need it anymore
   b->transferPtr = (char *)&b->circularBuf[(b->lastReceivedTS * b->th.nbAnt) % CirSize];
+  // we now need to read the samples
   b->remainToTransfer = sampleToByte(b->th.size, b->th.nbAnt);
+  return;
 }
 
-      if ( b->headerMode==false ) {
+static void process_recv(rfsimulator_state_t *t, buffer_t *b, bool first_time)
+{
+  if (b->transferPtr == b->circularBufEnd)
+    b->transferPtr = (char *)b->circularBuf;
+
   if (!b->trashingPacket) {
     b->lastReceivedTS = b->th.timestamp + b->th.size - byteToSample(b->remainToTransfer, b->th.nbAnt);
     LOG_D(HW, "UEsock: %d Set b->lastReceivedTS %ld\n", b->conn_sock, b->lastReceivedTS);
@@ -891,16 +847,72 @@ static bool flushInput(rfsimulator_state_t *t, int timeout, int nsamps_for_initi
     b->trashingPacket = false;
   }
 }
+
+static bool flushInput(rfsimulator_state_t *t, int timeout, bool first_time)
+{
+  // Process all incoming events on sockets
+  // store the data in lists
+  struct epoll_event events[MAX_FD_RFSIMU] = {{0}};
+  int nfds = epoll_wait(t->epollfd, events, MAX_FD_RFSIMU, timeout);
+
+  if (nfds == -1) {
+    if (!(errno == EINTR || errno == EAGAIN))
+      LOG_W(HW, "epoll_wait() failed, errno(%d)\n", errno);
+    return false;
+  }
+
+  for (int nbEv = 0; nbEv < nfds; ++nbEv) {
+    buffer_t *b = events[nbEv].data.ptr;
+
+    if (events[nbEv].events & EPOLLIN && b == NULL) {
+      bool ret = add_client(t);
+      if (!ret)
+        return ret;
+      continue;
+    } else {
+      if (events[nbEv].events & (EPOLLHUP | EPOLLERR | EPOLLRDHUP)) {
+        socketError(t, b);
+        continue;
       }
     }
 
+    if (b->circularBuf == NULL) {
+      LOG_E(HW, "Received data on not connected socket %d\n", events[nbEv].data.fd);
+      continue;
+    }
+
+    ssize_t blockSz;
+    if (b->headerMode)
+      blockSz = b->remainToTransfer;
+    else
+      blockSz =
+          b->transferPtr + b->remainToTransfer <= b->circularBufEnd ? b->remainToTransfer : b->circularBufEnd - b->transferPtr;
+
+    ssize_t sz = recv(b->conn_sock, b->transferPtr, blockSz, MSG_DONTWAIT);
+    if (sz <= 0) {
+      if (sz < 0 && errno != EAGAIN)
+        LOG_E(HW, "recv() failed, errno(%d)\n", errno);
+      continue;
+    }
+    LOG_D(HW, "Socket rcv %zd bytes\n", sz);
+    b->remainToTransfer -= sz;
+    b->transferPtr += sz;
+    if (b->headerMode)
+      process_recv_header(t, b, first_time);
+    else
+      process_recv(t, b, first_time);
+  }
   return nfds > 0;
 }
 
 static int rfsimulator_read(openair0_device *device, openair0_timestamp *ptimestamp, void **samplesVoid, int nsamps, int nbAnt)
 {
   rfsimulator_state_t *t = device->priv;
-  LOG_D(HW, "Enter rfsimulator_read, expect %d samples, will release at TS: %ld, nbAnt %d\n", nsamps, t->nextRxTstamp+nsamps, nbAnt);
+  LOG_D(HW,
+        "Enter rfsimulator_read, expect %d samples, will release at TS: %ld, nbAnt %d\n",
+        nsamps,
+        t->nextRxTstamp + nsamps,
+        nbAnt);
 
   // deliver data from received data
   // check if a UE is connected
@@ -915,7 +927,7 @@ static int rfsimulator_read(openair0_device *device, openair0_timestamp *ptimest
     if (t->nextRxTstamp == 0)
       LOG_I(HW, "No connected device, generating void samples...\n");
 
-    if (!flushInput(t, t->wait_timeout,  nsamps)) {
+    if (!flushInput(t, t->wait_timeout, false)) {
       for (int x = 0; x < nbAnt; x++)
         memset(samplesVoid[x], 0, sampleToByte(nsamps, 1));
 
@@ -929,16 +941,12 @@ static int rfsimulator_read(openair0_device *device, openair0_timestamp *ptimest
     }
   } else {
     bool have_to_wait;
-    int loops = 0;
     do {
       have_to_wait = false;
-
       buffer_t *b = NULL;
       for (int sock = 0; sock < MAX_FD_RFSIMU; sock++) {
         b = &t->buf[sock];
-
+        if (b->circularBuf && (t->nextRxTstamp + nsamps) > b->lastReceivedTS) {
-        if ( b->circularBuf )
-          if ( t->nextRxTstamp+nsamps > b->lastReceivedTS ) {
             have_to_wait = true;
             break;
         }
@@ -949,18 +957,11 @@ static int rfsimulator_read(openair0_device *device, openair0_timestamp *ptimest
               "Waiting on socket, current last ts: %ld, expected at least : %ld\n",
               b->lastReceivedTS,
               t->nextRxTstamp + nsamps);
-        flushInput(t, 3, nsamps);
+        flushInput(t, 3, false);
-      }
-      if (t->hanging_workaround && loops++ > 10 && t->role == SIMU_ROLE_SERVER && nb_ue > 1) {
-        // Just start producing samples. The clients will catch up.
-        have_to_wait = false;
-        LOG_W(HW,
-              "No longer waiting for clients to catch up, starting to produce samples\n");
       }
     } while (have_to_wait);
   }
 
-
   struct timespec start_time;
   int ret = clock_gettime(CLOCK_REALTIME, &start_time);
   AssertFatal(ret == 0, "clock_gettime() failed: errno %d, %s\n", errno, strerror(errno));
@@ -1000,12 +1001,11 @@ static int rfsimulator_read(openair0_device *device, openair0_timestamp *ptimest
                      t->nextRxTstamp,
                      CirSize,
                      apply_noise_per_channel);
-        }
+        } else { // no channel modeling
-        else { // no channel modeling
           int nbAnt_tx = ptr->th.nbAnt; // number of Tx antennas
           int firstIndex = (CirSize + t->nextRxTstamp - t->chan_offset) % CirSize;
           sample_t *out = (sample_t *)samplesVoid[a];
-          if ((nbAnt_tx == 1) && ((nb_ue == 1) || (t->role == SIMU_ROLE_CLIENT))) { // optimized for 1 Tx and 1 UE
+          if (nbAnt_tx == 1 && t->nb_cnx == 1) { // optimized for 1 Tx and 1 UE
             sample_t *firstSample = (sample_t *)&(ptr->circularBuf[firstIndex]);
             if (firstIndex + nsamps > CirSize) {
               int tailSz = CirSize - firstIndex;
@@ -1137,7 +1137,7 @@ int device_init(openair0_device *device, openair0_config_t *openair0_cfg) {
     rfsimulator->prop_delay_ms = rfsimulator->chan_offset * 1000 / rfsimulator->sample_rate;
     LOG_I(HW, "propagation delay %f ms, %lu samples\n", rfsimulator->prop_delay_ms, rfsimulator->chan_offset);
   }
-  pthread_mutex_init(&Sockmutex, NULL);
+  mutexinit(rfsimulator->Sockmutex);
   LOG_I(HW,
         "Running as %s\n",
         rfsimulator->role == SIMU_ROLE_SERVER ? "server waiting opposite rfsimulators to connect"