Starting to add discard functionality to PDCP LTE entity.

This includes:
  - Adding a queue (implemented with std::map) for undelivered PDUs.
    This queue uses the SN used for TX as the key.
  - Added discard timer that is started upon reception of the SDU. Upon
    expiry of the timeout a discard callback removes undelivered PDUs
from the queue.
  - Added the mechanisms to the notify_delivery to remove PDUs from the
    undelivered queue when the PDU is ACK'ed.
  - Added test case for both timer expiry and acknowledgment.
  - Fix up the getter for buffered SDUs to return the undelivered SDUs
  - Changed default PDCP discard timer, so AM has a discard timer by
    default.

lib/include/srslte/upper/pdcp_entity_lte.h

@@ -58,26 +58,60 @@ public:
   // Config helpers
   bool check_valid_config();
 
+  // TX SDU queue helper
+  bool store_sdu(uint32_t tx_count, const unique_byte_buffer_t& pdu);
+
+  // Getter for unacknowledged PDUs. Used for handover
+  std::map<uint32_t, srslte::unique_byte_buffer_t> get_buffered_pdus() override;
+
+  // Internal state getters/setters
   void get_bearer_state(pdcp_lte_state_t* state) override;
   void set_bearer_state(const pdcp_lte_state_t& state) override;
 
-  std::map<uint32_t, srslte::unique_byte_buffer_t> get_buffered_pdus() override;
+  // Getter for the number of discard timers. Used for debugging.
+  uint32_t nof_discard_timers() { return discard_timers_map.size(); }
 
 private:
   srsue::rlc_interface_pdcp* rlc = nullptr;
   srsue::rrc_interface_pdcp* rrc = nullptr;
   srsue::gw_interface_pdcp*  gw  = nullptr;
 
+  byte_buffer_pool* pool = nullptr;
+
   // State variables, as defined in TS 36 323, section 7.1
   pdcp_lte_state_t st = {};
 
   uint32_t reordering_window = 0;
   uint32_t maximum_pdcp_sn   = 0;
 
+  // Discard callback (discardTimer)
+  class discard_callback;
+  std::map<uint32_t, timer_handler::unique_timer> discard_timers_map;
+
+  // TX Queue
+  uint32_t                                 tx_queue_capacity = 512;
+  std::map<uint32_t, unique_byte_buffer_t> undelivered_sdus_queue;
+
   void handle_srb_pdu(srslte::unique_byte_buffer_t pdu);
   void handle_um_drb_pdu(srslte::unique_byte_buffer_t pdu);
   void handle_am_drb_pdu(srslte::unique_byte_buffer_t pdu);
 };
 
+// Discard callback (discardTimer)
+class pdcp_entity_lte::discard_callback
+{
+public:
+  discard_callback(pdcp_entity_lte* parent_, uint32_t sn_)
+  {
+    parent     = parent_;
+    discard_sn = sn_;
+  };
+  void operator()(uint32_t timer_id);
+
+private:
+  pdcp_entity_lte* parent;
+  uint32_t         discard_sn;
+};
+
 } // namespace srslte
 #endif // SRSLTE_PDCP_ENTITY_LTE_H

lib/src/upper/pdcp_entity_lte.cc

@@ -103,6 +103,15 @@ void pdcp_entity_lte::write_sdu(unique_byte_buffer_t sdu)
   // Get COUNT to be used with this packet
   uint32_t tx_count = COUNT(st.tx_hfn, st.next_pdcp_tx_sn);
 
+  // If the bearer is mapped to RLC AM, save TX_COUNT and a copy of the PDU.
+  // This will be used for reestablishment, where unack'ed PDUs will be re-transmitted.
+  // PDUs will be removed from the queue, either when the lower layers will report
+  // a succesfull transmission or when the discard timer expires.
+  // Status report will also use this queue, to know the First Missing SDU (FMS).
+  if (!rlc->rb_is_um(lcid)) {
+    store_sdu(st.next_pdcp_tx_sn, sdu);
+  }
+
   // check for pending security config in transmit direction
   if (enable_security_tx_sn != -1 && enable_security_tx_sn == static_cast<int32_t>(tx_count)) {
     enable_integrity(DIRECTION_TX);
@@ -112,6 +121,16 @@ void pdcp_entity_lte::write_sdu(unique_byte_buffer_t sdu)
 
   write_data_header(sdu, tx_count);
 
+  // Start discard timer
+  if (cfg.discard_timer != pdcp_discard_timer_t::infinity) {
+    timer_handler::unique_timer discard_timer = task_sched.get_unique_timer();
+    discard_callback            discard_fnc(this, st.next_pdcp_tx_sn);
+    discard_timer.set(static_cast<uint32_t>(cfg.discard_timer), discard_fnc);
+    discard_timer.run();
+    discard_timers_map.insert(std::make_pair(tx_count, std::move(discard_timer)));
+    log->debug("Discard Timer set for SN %u. Timeout: %ums\n", tx_count, static_cast<uint32_t>(cfg.discard_timer));
+  }
+
   // Append MAC (SRBs only)
   uint8_t mac[4]       = {};
   bool    do_integrity = integrity_direction == DIRECTION_TX || integrity_direction == DIRECTION_TXRX;
@@ -146,6 +165,7 @@ void pdcp_entity_lte::write_sdu(unique_byte_buffer_t sdu)
     st.next_pdcp_tx_sn = 0;
   }
 
+  // Pass PDU to lower layers
   rlc->write_sdu(lcid, std::move(sdu));
 }
 
@@ -340,11 +360,75 @@ void pdcp_entity_lte::handle_am_drb_pdu(srslte::unique_byte_buffer_t pdu)
 }
 
 /****************************************************************************
- * Delivery notifications from RLC
+ * TX PDUs Queue Helper
+ ***************************************************************************/
+bool pdcp_entity_lte::store_sdu(uint32_t tx_count, const unique_byte_buffer_t& sdu)
+{
+  // Check capacity
+  if (undelivered_sdus_queue.size() >= tx_queue_capacity) {
+    log->warning("The undelivered PDU queue is growing large. TX_COUNT=%d, Queue size=%ld\n",
+                 tx_count,
+                 undelivered_sdus_queue.size());
+  }
+
+  // Check wether PDU is already in the queue
+  if (undelivered_sdus_queue.find(tx_count) != undelivered_sdus_queue.end()) {
+    log->error("PDU already exists in the queue. TX_COUNT=%d\n", tx_count);
+    return false;
+  }
+
+  // Copy PDU contents into queue
+  unique_byte_buffer_t sdu_copy = allocate_unique_buffer(*pool);
+  memcpy(sdu_copy->msg, sdu->msg, sdu->N_bytes);
+  sdu_copy->N_bytes = sdu->N_bytes;
+
+  undelivered_sdus_queue.insert(std::make_pair(tx_count, std::move(sdu_copy)));
+  return true;
+}
+
+/****************************************************************************
+ * Discard functionality
+ ***************************************************************************/
+// Discard Timer Callback (discardTimer)
+void pdcp_entity_lte::discard_callback::operator()(uint32_t timer_id)
+{
+  parent->log->debug("Discard timer expired for PDU with SN = %d\n", discard_sn);
+
+  // Discard PDU if unacknowledged
+  if (parent->undelivered_sdus_queue.find(discard_sn) != parent->undelivered_sdus_queue.end()) {
+    parent->undelivered_sdus_queue.erase(discard_sn);
+    parent->log->debug("Removed undelivered PDU with TX_COUNT=%d\n", discard_sn);
+  } else {
+    parent->log->debug("Could not find PDU to discard. TX_COUNT=%d\n", discard_sn);
+  }
+
+  // Notify the RLC of the discard. It's the RLC to actually discard, if no segment was transmitted yet.
+  parent->rlc->discard_sdu(parent->lcid, discard_sn);
+
+  // Remove timer from map
+  // NOTE: this will delete the callback. It *must* be the last instruction.
+  parent->discard_timers_map.erase(discard_sn);
+}
+
+/****************************************************************************
+ * Handle delivery notifications from RLC
  ***************************************************************************/
 void pdcp_entity_lte::notify_delivery(const std::vector<uint32_t>& pdcp_sns)
 {
-  logger.debug("Received delivery notification from RLC. Number of PDU notified=%ld", pdcp_sns.size());
+  log->debug("Received delivery notification from RLC. Number of PDU notified=%ld", pdcp_sns.size());
+
+  for (uint32_t sn : pdcp_sns) {
+    // Find undelivered PDU info
+    std::map<uint32_t, unique_byte_buffer_t>::iterator it = undelivered_sdus_queue.find(sn);
+    if (it == undelivered_sdus_queue.end()) {
+      log->warning("Could not find PDU for delivery notification. Notified SN=%d\n", sn);
+      return;
+    }
+
+    // If ACK'ed bytes are equal to (or exceed) PDU size, remove PDU and disarm timer.
+    undelivered_sdus_queue.erase(sn);
+    discard_timers_map.erase(sn);
+  }
 }
 
 /****************************************************************************
@@ -386,7 +470,15 @@ void pdcp_entity_lte::set_bearer_state(const pdcp_lte_state_t& state)
 
 std::map<uint32_t, srslte::unique_byte_buffer_t> pdcp_entity_lte::get_buffered_pdus()
 {
-  return {};
+  std::map<uint32_t, srslte::unique_byte_buffer_t> cpy{};
+  // Deep copy undelivered SDUs
+  // TODO: investigate wheter the deep copy can be avoided by moving the undelivered SDU queue.
+  // That can only be done just before the PDCP is disabled though.
+  for (auto it = undelivered_sdus_queue.begin(); it != undelivered_sdus_queue.end(); it++) {
+    cpy[it->first]    = allocate_unique_buffer(*pool);
+    (*cpy[it->first]) = *(it->second);
+  }
+  return cpy;
 }
 
 } // namespace srslte

lib/test/upper/CMakeLists.txt

@@ -76,6 +76,10 @@ add_executable(pdcp_lte_test_rx pdcp_lte_test_rx.cc)
 target_link_libraries(pdcp_lte_test_rx srslte_upper srslte_common)
 add_test(pdcp_lte_test_rx pdcp_lte_test_rx)
 
+add_executable(pdcp_lte_test_discard_sdu pdcp_lte_test_discard_sdu.cc)
+target_link_libraries(pdcp_lte_test_discard_sdu srslte_upper srslte_common)
+add_test(pdcp_lte_test_discard_sdu pdcp_lte_test_discard_sdu)
+
 ########################################################################
 # Option to run command after build (useful for remote builds)
 ########################################################################

lib/test/upper/pdcp_lte_test_discard_sdu.cc

@@ -0,0 +1,151 @@
+/**
+ *
+ * \section COPYRIGHT
+ *
+ * Copyright 2013-2021 Software Radio Systems Limited
+ *
+ * By using this file, you agree to the terms and conditions set
+ * forth in the LICENSE file which can be found at the top level of
+ * the distribution.
+ *
+ */
+#include "pdcp_lte_test.h"
+#include <numeric>
+
+/*
+ * Test of imediate notification from RLC
+ */
+int test_tx_sdu_notify(const srslte::pdcp_lte_state_t& init_state,
+                       srslte::pdcp_discard_timer_t    discard_timeout,
+                       srslte::byte_buffer_pool*       pool,
+                       srslte::log_ref                 log)
+{
+  srslte::pdcp_config_t cfg = {1,
+                               srslte::PDCP_RB_IS_DRB,
+                               srslte::SECURITY_DIRECTION_UPLINK,
+                               srslte::SECURITY_DIRECTION_DOWNLINK,
+                               srslte::PDCP_SN_LEN_12,
+                               srslte::pdcp_t_reordering_t::ms500,
+                               discard_timeout};
+
+  pdcp_lte_test_helper     pdcp_hlp(cfg, sec_cfg, log);
+  srslte::pdcp_entity_lte* pdcp  = &pdcp_hlp.pdcp;
+  rlc_dummy*               rlc   = &pdcp_hlp.rlc;
+  srsue::stack_test_dummy* stack = &pdcp_hlp.stack;
+
+  pdcp_hlp.set_pdcp_initial_state(init_state);
+
+  // Write test SDU
+  srslte::unique_byte_buffer_t sdu = allocate_unique_buffer(*pool);
+  sdu->append_bytes(sdu1, sizeof(sdu1));
+  pdcp->write_sdu(std::move(sdu));
+
+  srslte::unique_byte_buffer_t out_pdu = srslte::allocate_unique_buffer(*pool);
+  rlc->get_last_sdu(out_pdu);
+  TESTASSERT(out_pdu->N_bytes == 4);
+
+  TESTASSERT(pdcp->nof_discard_timers() == 1); // One timer should be running
+  std::vector<uint32_t> sns_notified = {0};
+  pdcp->notify_delivery(sns_notified);
+  TESTASSERT(pdcp->nof_discard_timers() == 0); // Timer should have been difused after
+
+  // RLC should not be notified of SDU discard
+  TESTASSERT(rlc->discard_count == 0);
+
+  // Make sure there are no timers still left on the map
+  TESTASSERT(pdcp->nof_discard_timers() == 0);
+
+  return 0;
+}
+
+/*
+ * Test discard timer expiry
+ */
+int test_tx_sdu_discard(const srslte::pdcp_lte_state_t& init_state,
+                        srslte::pdcp_discard_timer_t    discard_timeout,
+                        srslte::byte_buffer_pool*       pool,
+                        srslte::log_ref                 log)
+{
+  srslte::pdcp_config_t cfg = {1,
+                               srslte::PDCP_RB_IS_DRB,
+                               srslte::SECURITY_DIRECTION_UPLINK,
+                               srslte::SECURITY_DIRECTION_DOWNLINK,
+                               srslte::PDCP_SN_LEN_12,
+                               srslte::pdcp_t_reordering_t::ms500,
+                               discard_timeout};
+
+  pdcp_lte_test_helper     pdcp_hlp(cfg, sec_cfg, log);
+  srslte::pdcp_entity_lte* pdcp  = &pdcp_hlp.pdcp;
+  rlc_dummy*               rlc   = &pdcp_hlp.rlc;
+  srsue::stack_test_dummy* stack = &pdcp_hlp.stack;
+
+  pdcp_hlp.set_pdcp_initial_state(init_state);
+
+  // Write test SDU
+  srslte::unique_byte_buffer_t sdu = allocate_unique_buffer(*pool);
+  sdu->append_bytes(sdu1, sizeof(sdu1));
+  pdcp->write_sdu(std::move(sdu));
+
+  srslte::unique_byte_buffer_t out_pdu = srslte::allocate_unique_buffer(*pool);
+  rlc->get_last_sdu(out_pdu);
+  TESTASSERT(out_pdu->N_bytes == 4);
+
+  // Run TTIs for Discard timers
+  for (uint32_t i = 0; i < static_cast<uint32_t>(cfg.discard_timer) - 1; ++i) {
+    stack->run_tti();
+  }
+  TESTASSERT(pdcp->nof_discard_timers() == 1); // One timer should be running
+  TESTASSERT(rlc->discard_count == 0);         // No discard yet
+
+  // Last timer step
+  stack->run_tti();
+
+  TESTASSERT(pdcp->nof_discard_timers() == 0); // Timer should have been difused after expiry
+  TESTASSERT(rlc->discard_count == 1);         // RLC should be notified of discard
+
+  std::vector<uint32_t> sns_notified = {0};
+  pdcp->notify_delivery(sns_notified); // PDCP should not find PDU to notify.
+  return 0;
+}
+/*
+ * TX Test: PDCP Entity with SN LEN = 12 and 18.
+ * PDCP entity configured with EIA2 and EEA2
+ */
+int test_tx_discard_all(srslte::byte_buffer_pool* pool, srslte::log_ref log)
+{
+  /*
+   * TX Test 1: PDCP Entity with SN LEN = 12
+   * Test TX PDU discard.
+   */
+  TESTASSERT(test_tx_sdu_notify(normal_init_state, srslte::pdcp_discard_timer_t::ms50, pool, log) == 0);
+
+  /*
+   * TX Test 2: PDCP Entity with SN LEN = 12
+   * Test TX PDU discard.
+   */
+  TESTASSERT(test_tx_sdu_discard(normal_init_state, srslte::pdcp_discard_timer_t::ms50, pool, log) == 0);
+  return 0;
+}
+
+// Setup all tests
+int run_all_tests(srslte::byte_buffer_pool* pool)
+{
+  // Setup log
+  srslte::log_ref log("PDCP LTE Test");
+  log->set_level(srslte::LOG_LEVEL_DEBUG);
+  log->set_hex_limit(128);
+
+  TESTASSERT(test_tx_discard_all(pool, log) == 0);
+  return 0;
+}
+
+int main()
+{
+  if (run_all_tests(srslte::byte_buffer_pool::get_instance()) != SRSLTE_SUCCESS) {
+    fprintf(stderr, "pdcp_lte_tests() failed\n");
+    return SRSLTE_ERROR;
+  }
+  srslte::byte_buffer_pool::cleanup();
+
+  return SRSLTE_SUCCESS;
+}

srsenb/drb.conf.example

@@ -6,7 +6,7 @@ qci_config = (
 {
   qci=7;
   pdcp_config = {
-    discard_timer = 100;                
+    discard_timer = -1;                
     pdcp_sn_size = 12;                  
   }
   rlc_config = {
@@ -28,7 +28,7 @@ qci_config = (
 {
   qci=9;
   pdcp_config = {
-    discard_timer = -1;
+    discard_timer = 200;
     status_report_required = true;
   }
   rlc_config = {