Use spsc_q for PUSCH queue

Use the spsc queue to pass jobs from TX thread (scheduler) to RX thread
(handling those jobs), reducing the amount spent searching for PUSCH
jobs, clarifying the design, and making it thread-safe.

Since the PUSCH contexts have to be stored somewhere, there is still a
single iteration over all contexts. Removing that is left for future
work. Thus, this commit (together with previous commits) reduces the
number of iterations over all PUSCH entries, only iterating over the
actually requested PUSCH jobs. On the other hand, because only the RX
thread iterates over PUSCH contexts, it should be thread-safe.

Remove flag "handled", which is always 0.
Signed-off-by: Robert Schmidt <robert.schmidt@openairinterface.org>

openair1/PHY/INIT/nr_init.c

@@ -442,6 +442,7 @@ void init_nr_transport(PHY_VARS_gNB *gNB)
 
   int max_nb_pucch = buffer_ul_slots ? MAX_MOBILES_PER_GNB * buffer_ul_slots : 1;
   gNB->pucch_queue = spsc_q_alloc(max_nb_pucch, sizeof(NR_gNB_PUCCH_job_t));
+  gNB->pusch_queue = spsc_q_alloc(gNB->max_nb_pusch, sizeof(NR_gNB_PUSCH_job_t));
 
   gNB->srs = (NR_gNB_SRS_t *)malloc16(gNB->max_nb_srs * sizeof(NR_gNB_SRS_t));
   for (int i = 0; i < gNB->max_nb_srs; i++)
@@ -463,6 +464,7 @@ void reset_nr_transport(PHY_VARS_gNB *gNB)
   const NR_DL_FRAME_PARMS *fp = &gNB->frame_parms;
 
   spsc_q_free(&gNB->pucch_queue);
+  spsc_q_free(&gNB->pusch_queue);
   free(gNB->srs);
 
   for (int i = 0; i < gNB->max_nb_pusch; i++)

openair1/PHY/NR_TRANSPORT/nr_ulsch.c

@@ -11,30 +11,6 @@
 #include "PHY/NR_TRANSPORT/nr_ulsch.h"
 #include "SCHED_NR/sched_nr.h"
 
-static NR_gNB_ULSCH_t *find_nr_ulsch(PHY_VARS_gNB *gNB, uint16_t rnti, int pid)
-{
-  int16_t first_free_index = -1;
-  AssertFatal(gNB != NULL, "gNB is null\n");
-  NR_gNB_ULSCH_t *ulsch = NULL;
-
-  for (int i = 0; i < gNB->max_nb_pusch; i++) {
-    ulsch = &gNB->ulsch[i];
-    AssertFatal(ulsch, "gNB->ulsch[%d] is null\n", i);
-    if (!ulsch->active) {
-      if (first_free_index == -1)
-        first_free_index = i;
-    } else {
-      // if there is already an active ULSCH for this RNTI and HARQ_PID
-      if ((ulsch->harq_pid == pid) && (ulsch->rnti == rnti))
-        return ulsch;
-    }
-  }
-  if (first_free_index != -1)
-    ulsch = &gNB->ulsch[first_free_index];
-
-  return ulsch;
-}
-
 static void dump_pusch_pdu(int instance, int frame, int slot, nfapi_nr_pusch_pdu_t *pusch_pdu)
 {
   LOG_D(PHY,
@@ -117,49 +93,28 @@ static void dump_pusch_pdu(int instance, int frame, int slot, nfapi_nr_pusch_pdu
 void nr_fill_ulsch(PHY_VARS_gNB *gNB, int frame, int slot, nfapi_nr_pusch_pdu_t *ulsch_pdu)
 {
   dump_pusch_pdu(gNB->Mod_id, frame, slot, ulsch_pdu);
-  int harq_pid = ulsch_pdu->pusch_data.harq_process_id;
-  NR_gNB_ULSCH_t *ulsch = find_nr_ulsch(gNB, ulsch_pdu->rnti, harq_pid);
-  if (ulsch == NULL) {
-    LOG_E(NR_PHY, "No ulsch_id found for rnti %04x\n", ulsch_pdu->rnti);
-    return;
-  }
+  LOG_D(NR_PHY,
+        "%4d.%2d Programming ULSCH RNTI %04x HARQ PID %d new data indicator %d\n",
+        frame,
+        slot,
+        ulsch_pdu->rnti,
+        ulsch_pdu->pusch_data.harq_process_id,
+        ulsch_pdu->pusch_data.new_data_indicator);
 
-  ulsch->rnti = ulsch_pdu->rnti;
-  ulsch->harq_pid = harq_pid;
-  ulsch->handled = 0;
-  ulsch->active = true;
-  ulsch->beam_nb = 0;
+  NR_gNB_PUSCH_job_t pusch = {.frame = frame, .slot = slot, .pusch_pdu = *ulsch_pdu};
   if (gNB->common_vars.beam_id) {
     int fapi_beam_idx = ulsch_pdu->beamforming.prgs_list[0].dig_bf_interface_list[0].beam_idx;
     int bitmap = SL_to_bitmap(ulsch_pdu->start_symbol_index, ulsch_pdu->nr_of_symbols);
-    ulsch->beam_nb = beam_index_allocation(gNB->enable_analog_das,
+    pusch.beam_nb = beam_index_allocation(gNB->enable_analog_das,
                                            fapi_beam_idx,
                                            &gNB->common_vars,
                                            slot,
                                            gNB->frame_parms.symbols_per_slot,
                                            bitmap);
   }
-  ulsch->frame = frame;
-  ulsch->slot = slot;
-
-  NR_UL_gNB_HARQ_t *harq = ulsch->harq_process;
-  if (ulsch_pdu->pusch_data.new_data_indicator)
-    harq->harq_to_be_cleared = true;
-  LOG_D(NR_PHY,
-        "NEW ULSCH %d.%d RNTI %x HARQ PID %d new data indicator %d\n",
-        frame,
-        slot,
-        ulsch_pdu->rnti,
-        harq_pid,
-        ulsch_pdu->pusch_data.new_data_indicator);
-
-  if (ulsch_pdu->pusch_data.new_data_indicator)
-    harq->round = 0;
-  else
-    harq->round++;
-
-  memcpy(&ulsch->harq_process->ulsch_pdu, ulsch_pdu, sizeof(ulsch->harq_process->ulsch_pdu));
-  LOG_D(PHY, "Initializing nFAPI for ULSCH, harq_pid %d, layers %d\n", harq_pid, ulsch_pdu->nrOfLayers);
+  bool done = spsc_q_put(&gNB->pusch_queue, &pusch, sizeof(pusch));
+  if (!done)
+    LOG_W(NR_PHY, "PUSCH queue is full: dropping PUSCH UE %04x\n", ulsch_pdu->rnti);
 }
 
 void reset_active_ulsch(PHY_VARS_gNB *gNB, int frame)

openair1/PHY/defs_gNB.h

@@ -128,7 +128,7 @@ typedef struct {
 
 typedef struct {
   /// Nfapi ULSCH PDU
-  nfapi_nr_pusch_pdu_t ulsch_pdu;
+  nfapi_nr_pusch_pdu_t ulsch_pdu; // !!
   /// Index of current HARQ round for this DLSCH
   uint8_t round;
   bool new_rx;
@@ -183,10 +183,19 @@ typedef struct {
   int8_t last_iteration_cnt;
   /// Status Flag indicating for this ULSCH
   bool active;
-  /// Flag to indicate that the UL configuration has been handled. Used to remove a stale ULSCH when frame wraps around
-  uint8_t handled;
 } NR_gNB_ULSCH_t;
 
+typedef struct {
+  // identifier for concurrent beams
+  int beam_nb;
+  /// Frame where current PUSCH pdu was sent
+  uint32_t frame;
+  /// Slot where current PUSCH pdu was sent
+  uint32_t slot;
+  /// ULSCH PDU
+  nfapi_nr_pusch_pdu_t pusch_pdu;
+} NR_gNB_PUSCH_job_t;
+
 typedef struct {
   // identifier for concurrent beams
   int beam_nb;
@@ -379,6 +388,7 @@ typedef struct PHY_VARS_gNB_s {
   NR_gNB_PRS prs_vars;
   NR_gNB_PUSCH *pusch_vars;
   spsc_q_t pucch_queue;
+  spsc_q_t pusch_queue;
   NR_gNB_SRS_t *srs;
   NR_gNB_ULSCH_t *ulsch;
   NR_gNB_PHY_STATS_t phy_stats[MAX_MOBILES_PER_GNB];

openair1/SCHED_NR/phy_procedures_nr_gNB.c

@@ -450,7 +450,6 @@ static int nr_ulsch_procedures(PHY_VARS_gNB *gNB, int frame_rx, int slot_rx, int
             pusch_pdu->pusch_data.tb_size);
       nr_fill_indication(gNB, ulsch->frame, ulsch->slot, pusch, pusch_pdu, stats, NULL, 0, crc, pdu);
       gNBdumpScopeData(gNB, ulsch->slot, ulsch->frame, "ULSCH_NACK");
-      ulsch->handled = 1;
       LOG_D(PHY, "ULSCH %d in error\n",ULSCH_id);
       ulsch->last_iteration_cnt = ulsch->max_ldpc_iterations; // Setting to max_ldpc_iterations is sufficient given that this variable is only used for checking for failure
     }
@@ -535,7 +534,9 @@ static void fill_ul_rb_mask(PHY_VARS_gNB *gNB,
                             uint32_t rb_mask_ul[14][9],
                             nfapi_nr_max_num_of_symbol_per_slot_t *slot_conf,
                             const NR_gNB_PUCCH_job_t *pucch,
-                            int n_pucch)
+                            int n_pucch,
+                            const NR_gNB_PUSCH_job_t *pusch,
+                            int n_pusch)
 {
   for (int symbol = 0; symbol < 14; symbol++) {
     for (int m = 0; m < 9; m++) {
@@ -565,19 +566,15 @@ static void fill_ul_rb_mask(PHY_VARS_gNB *gNB,
     }
   }
 
-  for (int ULSCH_id = 0; ULSCH_id < gNB->max_nb_pusch; ULSCH_id++) {
-    NR_gNB_ULSCH_t *ulsch = &gNB->ulsch[ULSCH_id];
-    NR_UL_gNB_HARQ_t *ulsch_harq = ulsch->harq_process;
-    AssertFatal(ulsch_harq != NULL, "harq_pid %d is not allocated\n", ulsch->harq_pid);
-    if (!(ulsch->active && ulsch->frame == now.f && ulsch->slot == now.s && !ulsch->handled))
-      continue;
-    uint8_t symbol_start = ulsch_harq->ulsch_pdu.start_symbol_index;
-    uint8_t symbol_end = symbol_start + ulsch_harq->ulsch_pdu.nr_of_symbols;
+  for (int i = 0; i < n_pusch; i++) {
+    const nfapi_nr_pusch_pdu_t *pusch_pdu = &pusch[i].pusch_pdu;
+    uint8_t symbol_start = pusch_pdu->start_symbol_index;
+    uint8_t symbol_end = symbol_start + pusch_pdu->nr_of_symbols;
     for (int symbol = symbol_start; symbol < symbol_end; symbol++) {
       if (gNB->frame_parms.frame_type == FDD || (gNB->frame_parms.frame_type == TDD && slot_conf[symbol].slot_config.value == 1)) {
-        LOG_D(PHY, "symbol %d Filling rb_mask_ul rb_size %d\n", symbol, ulsch_harq->ulsch_pdu.rb_size);
-        for (int rb = 0; rb < ulsch_harq->ulsch_pdu.rb_size; rb++) {
-          int rb2 = rb + ulsch_harq->ulsch_pdu.rb_start + ulsch_harq->ulsch_pdu.bwp_start;
+        LOG_D(PHY, "symbol %d Filling rb_mask_ul rb_size %d\n", symbol, pusch_pdu->rb_size);
+        for (int rb = 0; rb < pusch_pdu->rb_size; rb++) {
+          int rb2 = rb + pusch_pdu->rb_start + pusch_pdu->bwp_start;
           rb_mask_ul[symbol][rb2 >> 5] |= 1U << (rb2 & 31);
         }
       }
@@ -1024,10 +1021,74 @@ static bool get_current_pucch(const void *data, void *user)
   return fsn_equal(t, *now);
 }
 
+static bool drop_old_pusch(const void *data, void *user)
+{
+  const NR_gNB_PUSCH_job_t *pusch = data;
+  const fsn_t *now = user;
+  const fsn_t t = {pusch->frame, pusch->slot, now->mu};
+  bool drop = fsn_in_the_past(t, *now);
+  if (drop)
+    LOG_E(NR_PHY, "%4d.%2d PUSCH job for UE %04x is in the past (%4d.%2d)\n", now->f, now->s, pusch->pusch_pdu.rnti, t.f, t.s);
+  return drop;
+}
+
+static bool get_current_pusch(const void *data, void *user)
+{
+  const NR_gNB_PUSCH_job_t *pusch = data;
+  const fsn_t *now = user;
+  const fsn_t t = {pusch->frame, pusch->slot, now->mu};
+  return fsn_equal(t, *now);
+}
+
+static int find_nr_ulsch_idx(PHY_VARS_gNB *gNB, uint16_t rnti, int pid)
+{
+  AssertFatal(gNB != NULL, "gNB is null\n");
+
+  int16_t first_free_index = -1;
+  for (int i = 0; i < gNB->max_nb_pusch; i++) {
+    const NR_gNB_ULSCH_t *ulsch = &gNB->ulsch[i];
+    AssertFatal(ulsch, "gNB->ulsch[%d] is null\n", i);
+    if (!ulsch->active) {
+      if (first_free_index == -1)
+        first_free_index = i;
+    } else {
+      // if there is already an active ULSCH for this RNTI and HARQ_PID
+      if ((ulsch->harq_pid == pid) && (ulsch->rnti == rnti))
+        return i;
+    }
+  }
+  return first_free_index;
+}
+static int handle_pusch_job_trigger(PHY_VARS_gNB *gNB, const NR_gNB_PUSCH_job_t *job)
+{
+  const nfapi_nr_pusch_pdu_t *pdu = &job->pusch_pdu;
+  int pid = pdu->pusch_data.harq_process_id;
+  int ULSCH_id = find_nr_ulsch_idx(gNB, pdu->rnti, pid);
+  if (ULSCH_id < 0) {
+    LOG_E(NR_PHY, "%4d.%2d cannot handle PUSCH job of RNTI %04x: no space\n", job->frame, job->slot, pdu->rnti);
+    return -1;
+  }
+  /* (re-)initialize this PUSCH context from job data */
+  NR_gNB_ULSCH_t *ulsch = &gNB->ulsch[ULSCH_id];
+  ulsch->active = true;
+  ulsch->frame = job->frame;
+  ulsch->slot = job->slot;
+  ulsch->rnti = pdu->rnti;
+  ulsch->harq_pid = pid;
+  ulsch->beam_nb = job->beam_nb;
+  ulsch->harq_process->ulsch_pdu = job->pusch_pdu;
+  if (pdu->pusch_data.new_data_indicator) {
+    ulsch->harq_process->harq_to_be_cleared = true;
+    ulsch->harq_process->round = 0;
+  } else {
+    ulsch->harq_process->round++;
+  }
+  return ULSCH_id;
+}
+
 int phy_procedures_gNB_uespec_RX(PHY_VARS_gNB *gNB, int frame_rx, int slot_rx, NR_UL_IND_t *UL_INFO)
 {
   /* those variables to log T_GNB_PHY_PUCCH_PUSCH_IQ only when we try to decode */
-  int pusch_decode_done = 0;
   int pusch_DTX = 0;
 
   const NR_DL_FRAME_PARMS *frame_parms = &gNB->frame_parms;
@@ -1040,6 +1101,10 @@ int phy_procedures_gNB_uespec_RX(PHY_VARS_gNB *gNB, int frame_rx, int slot_rx, N
   NR_gNB_PUCCH_job_t pucch[MAX_NUM_NR_UCI_PDUS];
   int n_pucch = spsc_q_get_while(&gNB->pucch_queue, get_current_pucch, &now, pucch, sizeof(*pucch), MAX_NUM_NR_UCI_PDUS);
 
+  spsc_q_drop_while(&gNB->pusch_queue, drop_old_pusch, &now);
+  NR_gNB_PUSCH_job_t pusch[MAX_UL_PDUS_PER_SLOT];
+  int n_pusch_jobs = spsc_q_get_while(&gNB->pusch_queue, get_current_pusch, &now, pusch, sizeof(*pusch), MAX_UL_PDUS_PER_SLOT);
+
   LOG_D(PHY,"phy_procedures_gNB_uespec_RX frame %d, slot %d\n",frame_rx,slot_rx);
   {
     // Mask of occupied RBs, per symbol and PRB
@@ -1047,7 +1112,7 @@ int phy_procedures_gNB_uespec_RX(PHY_VARS_gNB *gNB, int frame_rx, int slot_rx, N
     nfapi_nr_max_num_of_symbol_per_slot_t *slot_conf = NULL;
     if (frame_parms->frame_type == TDD)
       slot_conf = gNB->gNB_config.tdd_table.max_tdd_periodicity_list[slot_rx].max_num_of_symbol_per_slot_list;
-    fill_ul_rb_mask(gNB, now, rb_mask_ul, slot_conf, pucch, n_pucch);
+    fill_ul_rb_mask(gNB, now, rb_mask_ul, slot_conf, pucch, n_pucch, pusch, n_pusch_jobs);
 
     int first_symb = 0, num_symb = 0;
     if (frame_parms->frame_type == TDD)
@@ -1080,17 +1145,14 @@ int phy_procedures_gNB_uespec_RX(PHY_VARS_gNB *gNB, int frame_rx, int slot_rx, N
   UL_INFO->rx_ind.sfn = frame_rx;
   UL_INFO->rx_ind.slot = slot_rx;
   UL_INFO->rx_ind.pdu_list = UL_INFO->rx_pdu_list;
+  int ulsch_idx_to_decode[MAX_UL_PDUS_PER_SLOT];
   int num_pusch = 0;
-  int ulsch_idx_to_decode[gNB->max_nb_pusch];
-  for (int ULSCH_id = 0; ULSCH_id < gNB->max_nb_pusch; ULSCH_id++) {
-    ulsch_idx_to_decode[ULSCH_id] = -1;
-    NR_gNB_ULSCH_t *ulsch = &gNB->ulsch[ULSCH_id];
-    if (!(ulsch->active && ulsch->frame == frame_rx && ulsch->slot == slot_rx && !ulsch->handled))
+  for (int i = 0; i < n_pusch_jobs; ++i) {
+    int ULSCH_id = handle_pusch_job_trigger(gNB, &pusch[i]);
+    if (ULSCH_id < 0)
       continue;
-
-    pusch_decode_done = 1;
     NR_gNB_PUSCH *pusch_vars = &gNB->pusch_vars[ULSCH_id];
-
+    NR_gNB_ULSCH_t *ulsch = &gNB->ulsch[ULSCH_id];
     if (handle_pusch_decode_trigger(gNB, pusch_vars, ulsch, UL_INFO, &pusch_DTX))
       ulsch_idx_to_decode[num_pusch++] = ULSCH_id;
   }
@@ -1316,7 +1378,7 @@ int phy_procedures_gNB_uespec_RX(PHY_VARS_gNB *gNB, int frame_rx, int slot_rx, N
 
   stop_meas(&gNB->phy_proc_rx);
 
-  if (n_pucch > 0 || pusch_decode_done) {
+  if (n_pucch > 0 || num_pusch > 0) {
     T(T_GNB_PHY_PUCCH_PUSCH_IQ,
       T_INT(frame_rx),
       T_INT(slot_rx),