lte,enb,rlc: atomically compute DL buffer state and update the scheduler in the eNB

lib/include/srsran/rlc/rlc.h

@@ -59,7 +59,6 @@ public:
   // MAC interface
   bool     has_data_locked(const uint32_t lcid);
   uint32_t get_buffer_state(const uint32_t lcid);
-  void     get_buffer_state(uint32_t lcid, uint32_t& tx_queue, uint32_t& prio_tx_queue);
   uint32_t get_total_mch_buffer_state(uint32_t lcid);
   uint32_t read_pdu(uint32_t lcid, uint8_t* payload, uint32_t nof_bytes);
   uint32_t read_pdu_mch(uint32_t lcid, uint8_t* payload, uint32_t nof_bytes);
@@ -89,6 +88,8 @@ public:
 private:
   void reset_metrics();
 
+  void get_buffer_state(uint32_t lcid, uint32_t& tx_queue, uint32_t& prio_tx_queue);
+
   srslog::basic_logger&      logger;
   byte_buffer_pool*          pool   = nullptr;
   srsue::pdcp_interface_rlc* pdcp   = nullptr;

lib/include/srsran/rlc/rlc_am_lte.h

@@ -401,6 +401,8 @@ private:
     int  required_buffer_size(const rlc_amd_retx_t& retx);
     void retransmit_pdu(uint32_t sn);
 
+    void get_buffer_state_nolock(uint32_t& new_tx, uint32_t& prio_tx);
+
     // Helpers
     bool poll_required();
     bool do_status();

lib/include/srsran/rlc/rlc_tm.h

@@ -58,7 +58,7 @@ public:
   uint32_t read_pdu(uint8_t* payload, uint32_t nof_bytes) override;
   void     write_pdu(uint8_t* payload, uint32_t nof_bytes) override;
 
-  void set_bsr_callback(bsr_callback_t callback) override {}
+  void set_bsr_callback(bsr_callback_t callback) override;
 
 private:
   byte_buffer_pool*          pool = nullptr;
@@ -67,6 +67,9 @@ private:
   srsue::pdcp_interface_rlc* pdcp = nullptr;
   srsue::rrc_interface_rlc*  rrc  = nullptr;
 
+  std::mutex     bsr_callback_mutex;
+  bsr_callback_t bsr_callback;
+
   std::atomic<bool> tx_enabled = {true};
 
   std::mutex           metrics_mutex;

lib/include/srsran/rlc/rlc_um_base.h

@@ -66,7 +66,9 @@ public:
   rlc_bearer_metrics_t get_metrics();
   void                 reset_metrics();
 
-  void set_bsr_callback(bsr_callback_t callback) {}
+  void set_bsr_callback(bsr_callback_t callback);
+
+  uint32_t get_lcid() const { return lcid; }
 
 protected:
   // Transmitter sub-class base
@@ -88,11 +90,14 @@ protected:
     bool             has_data();
     virtual uint32_t get_buffer_state() = 0;
 
+    void set_bsr_callback(bsr_callback_t callback);
+
   protected:
     byte_buffer_pool*     pool = nullptr;
     srslog::basic_logger& logger;
     std::string           rb_name;
     rlc_um_base*          parent = nullptr;
+    bsr_callback_t        bsr_callback;
 
     rlc_config_t cfg = {};
 

lib/src/rlc/rlc.cc

@@ -57,8 +57,8 @@ void rlc::init(srsue::pdcp_interface_rlc* pdcp_,
                uint32_t                   lcid_,
                bsr_callback_t             bsr_callback_)
 {
-  init(pdcp_, rrc_, timers_, lcid_);
   bsr_callback = bsr_callback_;
+  init(pdcp_, rrc_, timers_, lcid_);
 }
 
 void rlc::reset_metrics()
@@ -598,16 +598,13 @@ void rlc::update_bsr(uint32_t lcid)
   if (bsr_callback) {
     uint32_t tx_queue = 0, prio_tx_queue = 0;
     get_buffer_state(lcid, tx_queue, prio_tx_queue);
-    bsr_callback(lcid, tx_queue, prio_tx_queue);
   }
 }
 
 void rlc::update_bsr_mch(uint32_t lcid)
 {
   if (bsr_callback) {
-    uint32_t tx_queue   = get_total_mch_buffer_state(lcid);
-    uint32_t retx_queue = 0; // todo: separate tx_queue and retx_queue
-    bsr_callback(lcid, tx_queue, retx_queue);
+    uint32_t tx_queue = get_total_mch_buffer_state(lcid);
   }
 }
 

lib/src/rlc/rlc_am_lte.cc

@@ -452,13 +452,17 @@ uint32_t rlc_am_lte::rlc_am_lte_tx::get_buffer_state()
 }
 
 void rlc_am_lte::rlc_am_lte_tx::get_buffer_state(uint32_t& n_bytes_newtx, uint32_t& n_bytes_prio)
+{
+  std::lock_guard<std::mutex> lock(mutex);
+  get_buffer_state_nolock(n_bytes_newtx, n_bytes_prio);
+}
+
+void rlc_am_lte::rlc_am_lte_tx::get_buffer_state_nolock(uint32_t& n_bytes_newtx, uint32_t& n_bytes_prio)
 {
   n_bytes_newtx   = 0;
   n_bytes_prio    = 0;
   uint32_t n_sdus = 0;
 
-  std::lock_guard<std::mutex> lock(mutex);
-
   logger.debug("%s Buffer state - do_status=%s, status_prohibit_running=%s (%d/%d)",
                RB_NAME,
                do_status() ? "yes" : "no",
@@ -513,6 +517,10 @@ void rlc_am_lte::rlc_am_lte_tx::get_buffer_state(uint32_t& n_bytes_newtx, uint32
     n_bytes_newtx += 2; // Two bytes for fixed header with SN length = 10
     logger.debug("%s Total buffer state - %d SDUs (%d B)", RB_NAME, n_sdus, n_bytes_newtx);
   }
+
+  if (bsr_callback) {
+    bsr_callback(parent->lcid, n_bytes_newtx, n_bytes_prio);
+  }
 }
 
 int rlc_am_lte::rlc_am_lte_tx::write_sdu(unique_byte_buffer_t sdu)
@@ -627,12 +635,9 @@ void rlc_am_lte::rlc_am_lte_tx::timer_expired(uint32_t timeout_id)
     logger.debug("%s Status prohibit timer expired after %dms", RB_NAME, status_prohibit_timer.duration());
   }
 
-  lock.unlock();
-
   if (bsr_callback) {
     uint32_t new_tx_queue = 0, prio_tx_queue = 0;
-    get_buffer_state(new_tx_queue, prio_tx_queue);
-    bsr_callback(parent->lcid, new_tx_queue, prio_tx_queue);
+    get_buffer_state_nolock(new_tx_queue, prio_tx_queue);
   }
 }
 

lib/src/rlc/rlc_tm.cc

@@ -121,13 +121,20 @@ bool rlc_tm::has_data()
 
 uint32_t rlc_tm::get_buffer_state()
 {
-  return ul_queue.size_bytes();
+  uint32_t newtx_queue   = 0;
+  uint32_t prio_tx_queue = 0;
+  get_buffer_state(newtx_queue, prio_tx_queue);
+  return newtx_queue + prio_tx_queue;
 }
 
 void rlc_tm::get_buffer_state(uint32_t& newtx_queue, uint32_t& prio_tx_queue)
 {
-  newtx_queue   = get_buffer_state();
+  std::lock_guard<std::mutex> lock(bsr_callback_mutex);
+  newtx_queue   = ul_queue.size_bytes();
   prio_tx_queue = 0;
+  if (bsr_callback) {
+    bsr_callback(lcid, newtx_queue, prio_tx_queue);
+  }
 }
 
 rlc_bearer_metrics_t rlc_tm::get_metrics()
@@ -199,4 +206,9 @@ void rlc_tm::write_pdu(uint8_t* payload, uint32_t nof_bytes)
   }
 }
 
+void rlc_tm::set_bsr_callback(bsr_callback_t callback)
+{
+  bsr_callback = std::move(callback);
+}
+
 } // namespace srsran

lib/src/rlc/rlc_um_base.cc

@@ -128,16 +128,19 @@ bool rlc_um_base::has_data()
 
 uint32_t rlc_um_base::get_buffer_state()
 {
-  if (tx) {
-    return tx->get_buffer_state();
-  }
-  return 0;
+  uint32_t newtx_queue   = 0;
+  uint32_t prio_tx_queue = 0;
+  get_buffer_state(newtx_queue, prio_tx_queue);
+  return newtx_queue + prio_tx_queue;
 }
 
 void rlc_um_base::get_buffer_state(uint32_t& newtx_queue, uint32_t& prio_tx_queue)
 {
-  newtx_queue   = get_buffer_state();
+  newtx_queue   = 0;
   prio_tx_queue = 0;
+  if (tx) {
+    newtx_queue = tx->get_buffer_state();
+  }
 }
 
 uint32_t rlc_um_base::read_pdu(uint8_t* payload, uint32_t nof_bytes)
@@ -178,6 +181,11 @@ void rlc_um_base::reset_metrics()
   metrics = {};
 }
 
+void rlc_um_base::set_bsr_callback(bsr_callback_t callback)
+{
+  tx->set_bsr_callback(std::move(callback));
+}
+
 /****************************************************************************
  * Helper functions
  ***************************************************************************/
@@ -250,6 +258,11 @@ bool rlc_um_base::rlc_um_base_tx::has_data()
   return (tx_sdu != nullptr || !tx_sdu_queue.is_empty());
 }
 
+void rlc_um_base::rlc_um_base_tx::set_bsr_callback(bsr_callback_t callback)
+{
+  bsr_callback = callback;
+}
+
 void rlc_um_base::rlc_um_base_tx::write_sdu(unique_byte_buffer_t sdu)
 {
   if (sdu) {

lib/src/rlc/rlc_um_lte.cc

@@ -90,6 +90,10 @@ uint32_t rlc_um_lte::rlc_um_lte_tx::get_buffer_state()
   if (n_bytes > 0)
     n_bytes += (cfg.um.is_mrb) ? 2 : 3;
 
+  if (bsr_callback) {
+    bsr_callback(parent->get_lcid(), n_bytes, 0);
+  }
+
   return n_bytes;
 }