Add a generic function for tx write out of order (timestamps not increasing)

Add an API to allow out-of-order writes to a radio interface, unlike the
current interface.

this function is called in UE in this commit, even if the tx might
still be in order.

Later, i plan to improve UE pulti-threading, that will lead to out of
order tx.  also, the gNB will benefit of this new function to replace
several specific pieces of code that reorder tx with over complex and
slow systems

executables/nr-ue.c

@@ -94,10 +94,6 @@
  *
  */
 
-
-#define RX_JOB_ID 0x1010
-#define TX_JOB_ID 100
-
 typedef enum {
   pss = 0,
   pbch = 1,
@@ -503,7 +499,6 @@ static void RU_write(nr_rxtx_thread_data_t *rxtxD) {
   if (mac->phy_config_request_sent &&
       openair0_cfg[0].duplex_mode == duplex_mode_TDD &&
       !get_softmodem_params()->continuous_tx) {
-
     int slots_frame = UE->frame_parms.slots_per_frame;
     int curr_slot = nr_ue_slot_select(&UE->nrUE_config, slot);
     if (curr_slot != NR_DOWNLINK_SLOT) {
@@ -520,18 +515,12 @@ static void RU_write(nr_rxtx_thread_data_t *rxtxD) {
     flags = TX_BURST_MIDDLE;
   }
 
-  if (flags || IS_SOFTMODEM_RFSIM)
-    AssertFatal(rxtxD->writeBlockSize ==
-                UE->rfdevice.trx_write_func(&UE->rfdevice,
-                                            proc->timestamp_tx,
-                                            txp,
-                                            rxtxD->writeBlockSize,
-                                            UE->frame_parms.nb_antennas_tx,
-                                            flags),"");
+  int tmp =
+      openair0_write_reorder(&UE->rfdevice, proc->timestamp_tx, txp, rxtxD->writeBlockSize, UE->frame_parms.nb_antennas_tx, flags);
+  AssertFatal(tmp == rxtxD->writeBlockSize, "");
 
   for (int i=0; i<UE->frame_parms.nb_antennas_tx; i++)
     memset(txp[i], 0, rxtxD->writeBlockSize);
-
 }
 
 void processSlotTX(void *arg)
@@ -666,14 +655,8 @@ void dummyWrite(PHY_VARS_NR_UE *UE,openair0_timestamp timestamp, int writeBlockS
   for (int i=0; i<UE->frame_parms.nb_antennas_tx; i++)
     dummy_tx[i]=dummy_tx_data[i];
 
-  AssertFatal( writeBlockSize ==
-               UE->rfdevice.trx_write_func(&UE->rfdevice,
-               timestamp,
-               dummy_tx,
-               writeBlockSize,
-               UE->frame_parms.nb_antennas_tx,
-               4),"");
-
+  int tmp = UE->rfdevice.trx_write_func(&UE->rfdevice, timestamp, dummy_tx, writeBlockSize, UE->frame_parms.nb_antennas_tx, 4);
+  AssertFatal(writeBlockSize == tmp, "");
 }
 
 void readFrame(PHY_VARS_NR_UE *UE,  openair0_timestamp *timestamp, bool toTrash) {
@@ -690,13 +673,13 @@ void readFrame(PHY_VARS_NR_UE *UE,  openair0_timestamp *timestamp, bool toTrash)
                    4*((x*UE->frame_parms.samples_per_subframe)+
                    UE->frame_parms.get_samples_slot_timestamp(slot,&UE->frame_parms,0));
       }
-        
-      AssertFatal( UE->frame_parms.get_samples_per_slot(slot,&UE->frame_parms) ==
-                   UE->rfdevice.trx_read_func(&UE->rfdevice,
-                   timestamp,
-                   rxp,
-                   UE->frame_parms.get_samples_per_slot(slot,&UE->frame_parms),
-                   UE->frame_parms.nb_antennas_rx), "");
+
+      int tmp = UE->rfdevice.trx_read_func(&UE->rfdevice,
+                                           timestamp,
+                                           rxp,
+                                           UE->frame_parms.get_samples_per_slot(slot, &UE->frame_parms),
+                                           UE->frame_parms.nb_antennas_rx);
+      AssertFatal(UE->frame_parms.get_samples_per_slot(slot, &UE->frame_parms) == tmp, "");
 
       if (IS_SOFTMODEM_RFSIM)
         dummyWrite(UE,*timestamp, UE->frame_parms.get_samples_per_slot(slot,&UE->frame_parms));
@@ -708,20 +691,20 @@ void readFrame(PHY_VARS_NR_UE *UE,  openair0_timestamp *timestamp, bool toTrash)
 
 }
 
-static void syncInFrame(PHY_VARS_NR_UE *UE, openair0_timestamp *timestamp, int rx_offset)
+static void syncInFrame(PHY_VARS_NR_UE *UE, openair0_timestamp *timestamp, openair0_timestamp rx_offset)
 {
-  LOG_I(PHY, "Resynchronizing RX by %d samples\n", rx_offset);
+  LOG_I(PHY, "Resynchronizing RX by %ld samples\n", rx_offset);
+
   if (IS_SOFTMODEM_IQPLAYER || IS_SOFTMODEM_IQRECORDER) {
     // Resynchonize by slot (will work with numerology 1 only)
     for (int size = rx_offset; size > 0; size -= UE->frame_parms.samples_per_subframe / 2) {
       int unitTransfer = size > UE->frame_parms.samples_per_subframe / 2 ? UE->frame_parms.samples_per_subframe / 2 : size;
-      AssertFatal(unitTransfer
-                      == UE->rfdevice.trx_read_func(&UE->rfdevice,
-                                                    timestamp,
-                                                    (void **)UE->common_vars.rxdata,
-                                                    unitTransfer,
-                                                    UE->frame_parms.nb_antennas_rx),
-                  "");
+      int tmp = UE->rfdevice.trx_read_func(&UE->rfdevice,
+                                           timestamp,
+                                           (void **)UE->common_vars.rxdata,
+                                           unitTransfer,
+                                           UE->frame_parms.nb_antennas_rx);
+      DevAssert(unitTransfer == tmp);
     }
   } else {
     *timestamp += UE->frame_parms.get_samples_per_slot(1, &UE->frame_parms);
@@ -731,13 +714,12 @@ static void syncInFrame(PHY_VARS_NR_UE *UE, openair0_timestamp *timestamp, int r
       // this happens here as the read size is samples_per_subframe which is very much larger than samp_per_slot
       if (IS_SOFTMODEM_RFSIM)
         dummyWrite(UE, *timestamp, unitTransfer);
-      AssertFatal(unitTransfer
-                      == UE->rfdevice.trx_read_func(&UE->rfdevice,
-                                                    timestamp,
-                                                    (void **)UE->common_vars.rxdata,
-                                                    unitTransfer,
-                                                    UE->frame_parms.nb_antennas_rx),
-                  "");
+      int res = UE->rfdevice.trx_read_func(&UE->rfdevice,
+                                           timestamp,
+                                           (void **)UE->common_vars.rxdata,
+                                           unitTransfer,
+                                           UE->frame_parms.nb_antennas_rx);
+      DevAssert(unitTransfer == res);
       *timestamp += unitTransfer; // this does not affect the read but needed for RFSIM write
     }
   }
@@ -767,10 +749,12 @@ void *UE_thread(void *arg)
   void *rxp[NB_ANTENNAS_RX];
   int start_rx_stream = 0;
   fapi_nr_config_request_t *cfg = &UE->nrUE_config;
-  AssertFatal(0 == openair0_device_load(&(UE->rfdevice), &openair0_cfg[0]), "Could not load the device\n");
+  int tmp = openair0_device_load(&(UE->rfdevice), &openair0_cfg[0]);
+  AssertFatal(tmp == 0, "Could not load the device\n");
   UE->rfdevice.host_type = RAU_HOST;
   UE->is_synchronized = 0;
-  AssertFatal(UE->rfdevice.trx_start_func(&UE->rfdevice) == 0, "Could not start the device\n");
+  int tmp2 = UE->rfdevice.trx_start_func(&UE->rfdevice);
+  AssertFatal(tmp2 == 0, "Could not start the device\n");
 
   notifiedFIFO_t nf;
   initNotifiedFIFO(&nf);
@@ -907,9 +891,8 @@ void *UE_thread(void *arg)
 
     const int readBlockSize = get_readBlockSize(slot_nr, &UE->frame_parms) - iq_shift_to_apply;
     openair0_timestamp rx_timestamp;
-    AssertFatal(readBlockSize
-                    == UE->rfdevice.trx_read_func(&UE->rfdevice, &rx_timestamp, rxp, readBlockSize, UE->frame_parms.nb_antennas_rx),
-                "");
+    int tmp = UE->rfdevice.trx_read_func(&UE->rfdevice, &rx_timestamp, rxp, readBlockSize, UE->frame_parms.nb_antennas_rx);
+    AssertFatal(readBlockSize == tmp, "");
 
     if(slot_nr == (nb_slot_frame - 1)) {
       // read in first symbol of next frame and adjust for timing drift
@@ -917,12 +900,13 @@ void *UE_thread(void *arg)
 
       if (first_symbols > 0) {
         openair0_timestamp ignore_timestamp;
-        AssertFatal(first_symbols ==
-                    UE->rfdevice.trx_read_func(&UE->rfdevice,
-                                               &ignore_timestamp,
-                                               (void **)UE->common_vars.rxdata,
-                                               first_symbols,
-                                               UE->frame_parms.nb_antennas_rx),"");
+        int tmp = UE->rfdevice.trx_read_func(&UE->rfdevice,
+                                             &ignore_timestamp,
+                                             (void **)UE->common_vars.rxdata,
+                                             first_symbols,
+                                             UE->frame_parms.nb_antennas_rx);
+        AssertFatal(first_symbols == tmp, "");
+
       } else
         LOG_E(PHY,"can't compensate: diff =%d\n", first_symbols);
     }
@@ -988,7 +972,8 @@ void init_NR_UE(int nb_inst, char *uecap_file, char *reconfig_file, char *rbconf
 
   for (int i = 0; i < nb_inst; i++) {
     NR_UE_MAC_INST_t *mac = get_mac_inst(i);
-    AssertFatal((mac->if_module = nr_ue_if_module_init(i)) != NULL, "can not initialize IF module\n");
+    mac->if_module = nr_ue_if_module_init(i);
+    AssertFatal(mac->if_module, "can not initialize IF module\n");
     if (!get_softmodem_params()->sa) {
       init_nsa_message(rrc_inst, reconfig_file, rbconfig_file);
       nr_rlc_activate_srb0(mac_inst->crnti, NULL, send_srb0_rrc);

openair1/PHY/impl_defs_top.h

@@ -272,7 +272,7 @@
 /* - between reception of pdsch and tarnsmission of its acknowlegment    */
 /* - between reception of un uplink grant and its related transmission   */
 // should be 2 as per NR standard, but current UE is not able to perform this value
-#define NR_UE_CAPABILITY_SLOT_RX_TO_TX           (3)
+#define NR_UE_CAPABILITY_SLOT_RX_TO_TX (3)
 
 #define DURATION_RX_TO_TX (NR_UE_CAPABILITY_SLOT_RX_TO_TX)
 

radio/COMMON/common_lib.c

@@ -43,6 +43,7 @@
 //#include "targets/RT/USER/lte-softmodem.h"
 #include "executables/softmodem-common.h"
 
+#define MAX_GAP 100ULL
 const char *const devtype_names[MAX_RF_DEV_TYPE] =
     {"", "USRP B200", "USRP X300", "USRP N300", "USRP X400", "BLADERF", "LMSSDR", "IRIS", "No HW", "UEDv2", "RFSIMULATOR"};
 
@@ -173,3 +174,98 @@ int openair0_transport_load(openair0_device *device,
 
   return rc;
 }
+
+static void writerEnqueue(re_order_t *ctx, openair0_timestamp timestamp, void **txp, int nsamps, int nbAnt, int flags)
+{
+  pthread_mutex_lock(&ctx->mutex_store);
+  LOG_D(HW, "Enqueue write for TS: %lu\n", timestamp);
+  int i;
+  for (i = 0; i < WRITE_QUEUE_SZ; i++)
+    if (!ctx->queue[i].active) {
+      ctx->queue[i].timestamp = timestamp;
+      ctx->queue[i].active = true;
+      ctx->queue[i].nsamps = nsamps;
+      ctx->queue[i].nbAnt = nbAnt;
+      ctx->queue[i].flags = flags;
+      AssertFatal(nbAnt <= NB_ANTENNAS_TX, "");
+      for (int j = 0; j < nbAnt; j++)
+        ctx->queue[i].txp[j] = txp[j];
+      break;
+    }
+  AssertFatal(i < WRITE_QUEUE_SZ, "Write queue full\n");
+  pthread_mutex_unlock(&ctx->mutex_store);
+}
+
+static void writerProcessWaitingQueue(openair0_device *device)
+{
+  bool found = false;
+  re_order_t *ctx = &device->reOrder;
+  do {
+    found = false;
+    pthread_mutex_lock(&ctx->mutex_store);
+    for (int i = 0; i < WRITE_QUEUE_SZ; i++) {
+      if (ctx->queue[i].active && llabs(ctx->queue[i].timestamp - ctx->nextTS) < MAX_GAP) {
+        openair0_timestamp timestamp = ctx->queue[i].timestamp;
+        LOG_D(HW, "Dequeue write for TS: %lu\n", timestamp);
+        int nsamps = ctx->queue[i].nsamps;
+        int nbAnt = ctx->queue[i].nbAnt;
+        int flags = ctx->queue[i].flags;
+        void *txp[NB_ANTENNAS_TX];
+        AssertFatal(nbAnt <= NB_ANTENNAS_TX, "");
+        for (int j = 0; j < nbAnt; j++)
+          txp[j] = ctx->queue[i].txp[j];
+        ctx->queue[i].active = false;
+        pthread_mutex_unlock(&ctx->mutex_store);
+        found = true;
+        if (flags || IS_SOFTMODEM_RFSIM) {
+          int wroteSamples = device->trx_write_func(device, timestamp, txp, nsamps, nbAnt, flags);
+          if (wroteSamples != nsamps)
+            LOG_E(HW, "Failed to write to rf\n");
+        }
+        ctx->nextTS += nsamps;
+        pthread_mutex_lock(&ctx->mutex_store);
+      }
+    }
+    pthread_mutex_unlock(&ctx->mutex_store);
+  } while (found);
+}
+
+// We assume the data behind *txp are permanently allocated
+// When we will go further, we can remove all RC.xxx.txdata buffers in xNB, in UE
+// but to make zerocopy and agnostic design, we need to make a proper ring buffer with mutex protection
+// mutex (or atomic flags) will be mandatory because this out order system root cause is there are several writer threads
+
+int openair0_write_reorder(openair0_device *device, openair0_timestamp timestamp, void **txp, int nsamps, int nbAnt, int flags)
+{
+  int wroteSamples = 0;
+  re_order_t *ctx = &device->reOrder;
+  LOG_D(HW, "received write order ts: %lu, nb samples %d, next ts %luflags %d\n", timestamp, nsamps, timestamp + nsamps, flags);
+  if (!ctx->initDone) {
+    ctx->nextTS = timestamp;
+    pthread_mutex_init(&ctx->mutex_write, NULL);
+    pthread_mutex_init(&ctx->mutex_store, NULL);
+    ctx->initDone = true;
+  }
+  if (pthread_mutex_trylock(&ctx->mutex_write) == 0) {
+    // We have the write exclusivity
+    if (llabs(timestamp - ctx->nextTS) < MAX_GAP) { // We are writing in sequence of the previous write
+      if (flags || IS_SOFTMODEM_RFSIM)
+        wroteSamples = device->trx_write_func(device, timestamp, txp, nsamps, nbAnt, flags);
+      else
+        wroteSamples = nsamps;
+      ctx->nextTS += nsamps;
+
+    } else {
+      writerEnqueue(ctx, timestamp, txp, nsamps, nbAnt, flags);
+    }
+    writerProcessWaitingQueue(device);
+    pthread_mutex_unlock(&ctx->mutex_write);
+    return wroteSamples ? wroteSamples : nsamps;
+  }
+  writerEnqueue(ctx, timestamp, txp, nsamps, nbAnt, flags);
+  if (pthread_mutex_trylock(&ctx->mutex_write) == 0) {
+    writerProcessWaitingQueue(device);
+    pthread_mutex_unlock(&ctx->mutex_write);
+  }
+  return nsamps;
+}

radio/COMMON/common_lib.h

@@ -384,6 +384,22 @@ typedef struct fhstate_s {
   int active;
 } fhstate_t;
 
+#define WRITE_QUEUE_SZ 20
+typedef struct {
+  bool initDone;
+  pthread_mutex_t mutex_write;
+  pthread_mutex_t mutex_store;
+  openair0_timestamp nextTS;
+  struct {
+    bool active;
+    openair0_timestamp timestamp;
+    void *txp[NB_ANTENNAS_TX];
+    int nsamps;
+    int nbAnt;
+    int flags;
+  } queue[WRITE_QUEUE_SZ];
+} re_order_t;
+
 /*!\brief structure holds the parameters to configure USRP devices */
 struct openair0_device_t {
   /*!tx write thread*/
@@ -586,6 +602,7 @@ struct openair0_device_t {
   /* \brief timing statistics for TX fronthaul (ethernet)
    */
   time_stats_t tx_fhaul;
+  re_order_t reOrder;
 };
 
 /* type of device init function, implemented in shared lib */
@@ -654,6 +671,7 @@ extern int read_recplayconfig(recplay_conf_t **recplay_conf, recplay_state_t **r
 /*! \brief store recorded iqs from memory to file. */
 extern void iqrecorder_end(openair0_device *device);
 
+int openair0_write_reorder(openair0_device *device, openair0_timestamp timestamp, void **txp, int nsamps, int nbAnt, int flags);
 
 #include <unistd.h>
 #ifndef gettid