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

this function is called in UE in this MR, even if the tx should 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,
@@ -502,7 +498,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) {
@@ -519,18 +514,17 @@ 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),"");
+  AssertFatal(rxtxD->writeBlockSize
+                  == openair0_write_in_order(&UE->rfdevice,
+                                             proc->timestamp_tx,
+                                             txp,
+                                             rxtxD->writeBlockSize,
+                                             UE->frame_parms.nb_antennas_tx,
+                                             flags),
+              "");
 
   for (int i=0; i<UE->frame_parms.nb_antennas_tx; i++)
     memset(txp[i], 0, rxtxD->writeBlockSize);
-
 }
 
 void processSlotTX(void *arg) {
@@ -868,7 +862,6 @@ void *UE_thread(void *arg)
       continue;
     }
 
-
     absolute_slot++;
 
     int slot_nr = absolute_slot % nb_slot_frame;

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,93 @@ 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 *tx are permanently allocated
+int openair0_write_in_order(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, flags %d\n", 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_in_order(openair0_device *device, openair0_timestamp timestamp, void **txp, int nsamps, int nbAnt, int flags);
 
 #include <unistd.h>
 #ifndef gettid