fapi_adaptor: removed the mutex from the FAPI PHY adaptor and use an atomic to store the slot

 	      added warnings to the dummy downlink processor

Author: AlaiaL

lib/fapi/message_loggers.cpp

@@ -125,7 +125,8 @@ static void log_csi_rs_pdu(const dl_csi_rs_pdu& pdu, fmt::memory_buffer& buffer)
 void srsran::fapi::log_dl_tti_request(const dl_tti_request_message& msg, srslog::basic_logger& logger)
 {
   fmt::memory_buffer buffer;
-  fmt::format_to(buffer, "DL_TTI.request slot={}.{}", msg.sfn, msg.slot);
+  fmt::format_to(
+      buffer, "DL_TTI.request slot={}.{}, is_last_message_in_slot={}", msg.sfn, msg.slot, msg.is_last_message_in_slot);
 
   for (const auto& pdu : msg.pdus) {
     switch (pdu.pdu_type) {
@@ -430,7 +431,8 @@ void srsran::fapi::log_slot_indication(const slot_indication_message& msg, srslo
 void srsran::fapi::log_ul_dci_request(const ul_dci_request_message& msg, srslog::basic_logger& logger)
 {
   fmt::memory_buffer buffer;
-  fmt::format_to(buffer, "UL_DCI.request slot={}.{}", msg.sfn, msg.slot);
+  fmt::format_to(
+      buffer, "UL_DCI.request slot={}.{}, is_last_message_in_slot={}", msg.sfn, msg.slot, msg.is_last_message_in_slot);
 
   for (const auto& pdu : msg.pdus) {
     switch (pdu.pdu_type) {

lib/fapi_adaptor/phy/fapi_to_phy_translator.cpp

@@ -33,14 +33,24 @@ namespace {
 class downlink_processor_dummy : public downlink_processor
 {
 public:
-  void process_pdcch(const pdcch_processor::pdu_t& pdu) override {}
+  void process_pdcch(const pdcch_processor::pdu_t& pdu) override
+  {
+    srslog::fetch_basic_logger("FAPI").warning("Could not enqueue PDCCH PDU in the downlink processor");
+  }
   void process_pdsch(unique_tx_buffer                                                                     softbuffer,
                      const static_vector<span<const uint8_t>, pdsch_processor::MAX_NOF_TRANSPORT_BLOCKS>& data,
                      const pdsch_processor::pdu_t&                                                        pdu) override
   {
+    srslog::fetch_basic_logger("FAPI").warning("Could not enqueue PDSCH PDU in the downlink processor");
+  }
+  void process_ssb(const ssb_processor::pdu_t& pdu) override
+  {
+    srslog::fetch_basic_logger("FAPI").warning("Could not enqueue SSB PDU in the downlink processor");
+  }
+  void process_nzp_csi_rs(const nzp_csi_rs_generator::config_t& config) override
+  {
+    srslog::fetch_basic_logger("FAPI").warning("Could not enqueue NZP-CSI-RS PDU in the downlink processor");
   }
-  void process_ssb(const ssb_processor::pdu_t& pdu) override {}
-  void process_nzp_csi_rs(const nzp_csi_rs_generator::config_t& config) override {}
   bool configure_resource_grid(const resource_grid_context& context, resource_grid& grid) override { return true; }
   void finish_processing_pdus() override {}
 };
@@ -270,9 +280,13 @@ void fapi_to_phy_translator::dl_tti_request(const fapi::dl_tti_request_message&
 {
   // :TODO: check the current slot matches the DL_TTI.request slot. Do this in a different class.
   // :TODO: check the messages order. Do this in a different class.
-  std::lock_guard<std::mutex> lock(mutex);
-
   slot_point slot(scs, msg.sfn, msg.slot);
+  slot_point current_slot = get_current_slot();
+
+  if (!pdsch_repository.empty()) {
+    logger.warning(
+        "Could not process '{}' PDSCH PDUs from the slot '{}'", pdsch_repository.pdus.size(), pdsch_repository.slot);
+  }
 
   // Reset the repository.
   pdsch_repository.reset(slot);
@@ -428,9 +442,11 @@ static expected<uplink_pdus> translate_ul_tti_pdus_to_phy_pdus(const fapi::ul_tt
 void fapi_to_phy_translator::ul_tti_request(const fapi::ul_tti_request_message& msg)
 {
   // :TODO: check the messages order. Do this in a different class.
-  std::lock_guard<std::mutex> lock(mutex);
-
   slot_point slot(scs, msg.sfn, msg.slot);
+  slot_point current_slot = get_current_slot();
+
+  // Clear the repository for the message slot.
+  ul_pdu_repository.clear_slot(slot);
 
   // Release the controller of the previous slot in case that it has not been released before. In case that it already
   // is released, this call will do nothing.
@@ -488,7 +504,7 @@ void fapi_to_phy_translator::ul_tti_request(const fapi::ul_tti_request_message&
 
 void fapi_to_phy_translator::ul_dci_request(const fapi::ul_dci_request_message& msg)
 {
-  std::lock_guard<std::mutex> lock(mutex);
+  slot_point current_slot = get_current_slot();
 
   // Ignore messages that do not correspond to the current slot.
   if (!is_message_in_time(msg)) {
@@ -535,7 +551,7 @@ void fapi_to_phy_translator::ul_dci_request(const fapi::ul_dci_request_message&
 
 void fapi_to_phy_translator::tx_data_request(const fapi::tx_data_request_message& msg)
 {
-  std::lock_guard<std::mutex> lock(mutex);
+  slot_point current_slot = get_current_slot();
 
   // Ignore messages that do not correspond to the current slot.
   if (!is_message_in_time(msg)) {
@@ -544,6 +560,9 @@ void fapi_to_phy_translator::tx_data_request(const fapi::tx_data_request_message
     error_notifier.get().on_error_indication(fapi::build_invalid_sfn_error_indication(
         msg.sfn, msg.slot, fapi::message_type_id::tx_data_request, current_slot.sfn(), current_slot.slot_index()));
     l2_tracer << instant_trace_event{"tx_data_req_late", instant_trace_event::cpu_scope::global};
+
+    pdsch_repository.clear();
+
     return;
   }
 
@@ -553,6 +572,9 @@ void fapi_to_phy_translator::tx_data_request(const fapi::tx_data_request_message
                    pdsch_repository.pdus.size());
     // Raise invalid format error.
     error_notifier.get().on_error_indication(fapi::build_msg_tx_error_indication(msg.sfn, msg.slot));
+
+    pdsch_repository.clear();
+
     return;
   }
 
@@ -611,15 +633,18 @@ void fapi_to_phy_translator::tx_data_request(const fapi::tx_data_request_message
   }
 
   slot_controller_mngr.release_controller(slot);
+
+  // All the PDSCH PDUs have been processed. Clear the repository.
+  pdsch_repository.clear();
 }
 
 void fapi_to_phy_translator::handle_new_slot(slot_point slot)
 {
-  std::lock_guard<std::mutex> lock(mutex);
+  // Update the atomic variable that holds the slot point.
+  current_slot_count_val.store(slot.system_slot(), std::memory_order_release);
 
-  current_slot = slot;
+  // Release old controllers.
   slot_controller_mngr.handle_new_slot(slot);
-  ul_pdu_repository.clear_slot(slot);
 
   // Enqueue soft buffer run slot.
   if (!asynchronous_executor.execute([this, slot]() { buffer_pool.run_slot(slot); })) {
@@ -634,6 +659,7 @@ template <typename T>
 bool fapi_to_phy_translator::is_message_in_time(const T& msg) const
 {
   slot_point msg_slot(scs, msg.sfn, msg.slot);
+  slot_point current_slot         = get_current_slot();
   slot_point last_allowed_message = current_slot - int(nof_slots_request_headroom);
 
   return last_allowed_message <= msg_slot && msg_slot <= current_slot;
@@ -647,7 +673,7 @@ fapi_to_phy_translator::slot_based_upper_phy_controller_manager::get_controller(
 
 void fapi_to_phy_translator::slot_based_upper_phy_controller_manager::handle_new_slot(slot_point slot)
 {
-  // Remove the oldest controller.
+  // Release the oldest controller.
   slot_point oldest_slot  = slot - int(controllers.size() - 1);
   controller(oldest_slot) = slot_based_upper_phy_controller();
 }
@@ -673,7 +699,13 @@ fapi_to_phy_translator::slot_based_upper_phy_controller_manager::slot_based_uppe
   dl_processor_pool(dl_processor_pool_),
   rg_pool(rg_pool_),
   sector_id(sector_id_),
-  // The manager should be able to manage the current slot plus the requests headroom size.
-  controllers(nof_slots_request_headroom + 1U)
+  // The manager should be able to manage the current slot plus the requests headroom size. Also, as it should clean the
+  // oldest slot when a new SLOT.indication is received, it should have two extra positions for these oldest slots.
+  // Adding two extra positions makes that if a new SLOT.indication arrives in the middle of a valid processing PDU, the
+  // controller for that slot will not be released in the middle of processing the messages.
+  // NOTE: if we can ensure that no messages will be lost, we can remove this extra positions in the manager, as it
+  // would not be required to release controllers when the slot changes, Also, this would be valid when the time when a
+  // controller is released does not matter for incomplete/lost messages.
+  controllers(nof_slots_request_headroom + 3U)
 {
 }

lib/fapi_adaptor/phy/fapi_to_phy_translator.h

@@ -231,6 +231,12 @@ class fapi_to_phy_translator : public fapi::slot_message_gateway
   template <typename T>
   bool is_message_in_time(const T& msg) const;
 
+  /// Returns this adaptor current slot.
+  slot_point get_current_slot() const
+  {
+    return slot_point(scs, current_slot_count_val.load(std::memory_order_acquire));
+  }
+
 private:
   /// Sector identifier.
   const unsigned sector_id;
@@ -252,8 +258,8 @@ class fapi_to_phy_translator : public fapi::slot_message_gateway
   uplink_slot_pdu_repository& ul_pdu_repository;
   /// Asynchronous task executor.
   task_executor& asynchronous_executor;
-  /// Current slot.
-  slot_point current_slot;
+  /// Current slot count value.
+  std::atomic<uint32_t> current_slot_count_val;
   /// Slot controller manager.
   slot_based_upper_phy_controller_manager slot_controller_mngr;
   /// Precoding matrix repository.
@@ -262,9 +268,6 @@ class fapi_to_phy_translator : public fapi::slot_message_gateway
   std::unique_ptr<uci_part2_correspondence_repository> part2_repo;
   /// Error indication notifier.
   std::reference_wrapper<fapi::slot_error_message_notifier> error_notifier;
-  /// Protects concurrent access to shared variables.
-  // :TODO: make this lock free.
-  std::mutex mutex;
   /// Subcarrier spacing as per TS38.211 Section 4.2.
   const subcarrier_spacing scs;
   /// Common subcarrier spacing as per TS38.331 Section 6.2.2.

lib/ofh/receiver/ofh_uplane_prach_symbol_data_flow_writer.cpp

@@ -21,7 +21,10 @@ void uplane_prach_symbol_data_flow_writer::write_to_prach_buffer(unsigned
 
   prach_context prach_context = prach_context_repo.get(slot);
   if (prach_context.empty()) {
-    logger.info("Dropped received Open Fronthaul message as no uplink PRACH context was found for slot '{}'", slot);
+    logger.info(
+        "Dropped received Open Fronthaul message as no uplink PRACH context was found for slot '{}' and eAxC '{}'",
+        slot,
+        eaxc);
     return;
   }
 

lib/ofh/receiver/ofh_uplane_rx_symbol_data_flow_writer.cpp

@@ -21,10 +21,11 @@ void uplane_rx_symbol_data_flow_writer::write_to_resource_grid(unsigned
   unsigned       symbol     = results.params.symbol_id;
   uplink_context ul_context = ul_context_repo.get(slot, symbol);
   if (ul_context.empty()) {
-    logger.warning(
-        "Dropped received Open Fronthaul message as no uplink slot context was found for slot '{}' and symbol '{}'",
-        results.params.slot,
-        results.params.symbol_id);
+    logger.warning("Dropped received Open Fronthaul message as no uplink slot context was found for slot '{}', symbol "
+                   "'{}' and eAxC '{}'",
+                   results.params.slot,
+                   results.params.symbol_id,
+                   eaxc);
 
     return;
   }

tests/unittests/fapi/validators/helpers.cpp

@@ -319,6 +319,8 @@ dl_tti_request_message unittest::build_valid_dl_tti_request()
   msg.pdus.back().pdu_type   = dl_pdu_type::CSI_RS;
   msg.pdus.back().csi_rs_pdu = build_valid_dl_csi_pdu();
 
+  msg.is_last_message_in_slot = false;
+
   return msg;
 }
 
@@ -337,6 +339,8 @@ ul_dci_request_message unittest::build_valid_ul_dci_request()
   msg.pdus.back().pdu_type = ul_dci_pdu_type::PDCCH;
   msg.pdus.back().pdu      = build_valid_dl_pdcch_pdu();
 
+  msg.is_last_message_in_slot = true;
+
   return msg;
 }
 

tests/unittests/fapi_adaptor/phy/fapi_to_phy_translator_test.cpp

@@ -109,19 +109,18 @@ class fapi_to_phy_translator_fixture : public ::testing::Test
   resource_grid_pool_dummy        rg_pool;
   uplink_request_processor_dummy  ul_request_processor;
   uplink_slot_pdu_repository      pdu_repo;
-  const unsigned                  sector_id           = 0;
-  const unsigned                  allowed_delay_slots = 2;
-  const subcarrier_spacing        scs                 = subcarrier_spacing::kHz15;
-  const slot_point                slot                = {scs, 1, 0};
+  const unsigned                  sector_id         = 0;
+  const unsigned                  headroom_in_slots = 2;
+  const subcarrier_spacing        scs               = subcarrier_spacing::kHz15;
+  const slot_point                slot              = {scs, 1, 0};
   fapi::prach_config              prach_cfg;
   fapi::carrier_config            carrier_cfg = {0, 0, {}, {11, 51, 106, 0, 0}, 0, 0, 0, {}, {}, 0, 0, 0, 0};
   downlink_pdu_validator_dummy    dl_pdu_validator;
   uplink_pdu_validator_dummy      ul_pdu_validator;
   slot_error_message_notifier_spy error_notifier_spy;
   manual_task_worker              worker;
   tx_softbuffer_pool_spy          softbuffer_pool_spy;
-  fapi_to_phy_translator_config   config =
-      {sector_id, allowed_delay_slots, scs, scs_common, &prach_cfg, &carrier_cfg, {0}};
+  fapi_to_phy_translator_config config = {sector_id, headroom_in_slots, scs, scs_common, &prach_cfg, &carrier_cfg, {0}};
   fapi_to_phy_translator_dependencies dependencies = {
       &srslog::fetch_basic_logger("FAPI"),
       &dl_processor_pool,
@@ -151,10 +150,6 @@ TEST_F(fapi_to_phy_translator_fixture, downlink_processor_is_not_configured_on_n
   ASSERT_FALSE(dl_processor_pool.processor(slot).has_configure_resource_grid_method_been_called());
   ASSERT_FALSE(grid.has_set_all_zero_method_been_called());
 
-  // Assert that the downlink processor is configured.
-  ASSERT_FALSE(dl_processor_pool.processor(slot).has_configure_resource_grid_method_been_called());
-  ASSERT_FALSE(grid.has_set_all_zero_method_been_called());
-
   // Assert that the softbuffer pool ran the slot.
   auto& run_slot_entries = softbuffer_pool_spy.get_run_slot_entries();
   ASSERT_EQ(run_slot_entries.size(), 1);
@@ -246,7 +241,7 @@ TEST_F(fapi_to_phy_translator_fixture, dl_ssb_pdu_within_allowed_delay_is_proces
   translator.handle_new_slot(msg_slot);
 
   // Increase the slots.
-  for (unsigned i = 1; i != allowed_delay_slots; ++i) {
+  for (unsigned i = 1; i != headroom_in_slots; ++i) {
     translator.handle_new_slot(msg_slot + i);
   }
 
@@ -272,7 +267,7 @@ TEST_F(fapi_to_phy_translator_fixture, receiving_a_dl_tti_request_sends_previous
   msg.is_last_message_in_slot = false;
 
   // Increase the slots.
-  for (unsigned i = 1; i != allowed_delay_slots; ++i) {
+  for (unsigned i = 1; i != headroom_in_slots; ++i) {
     translator.handle_new_slot(slot + i);
   }
 
@@ -305,7 +300,7 @@ TEST_F(fapi_to_phy_translator_fixture, receiving_a_dl_tti_request_from_a_slot_de
   msg.is_last_message_in_slot = false;
 
   // Increase the slots.
-  for (unsigned i = 0, e = allowed_delay_slots + 1; i != e; ++i) {
+  for (unsigned i = 0, e = headroom_in_slots + 1; i != e; ++i) {
     current_slot += 1;
     translator.handle_new_slot(current_slot);
   }
@@ -324,6 +319,39 @@ TEST_F(fapi_to_phy_translator_fixture, receiving_a_dl_tti_request_from_a_slot_de
   ASSERT_EQ(error_msg.expected_slot, current_slot.slot_index());
 }
 
+TEST_F(fapi_to_phy_translator_fixture, message_received_is_sended_when_a_message_for_the_next_slot_is_received)
+{
+  ASSERT_FALSE(dl_processor_pool.processor(slot).has_configure_resource_grid_method_been_called());
+  ASSERT_FALSE(grid.has_set_all_zero_method_been_called());
+
+  fapi::dl_tti_request_message msg;
+  msg.sfn                     = slot.sfn();
+  msg.slot                    = slot.slot_index();
+  msg.is_last_message_in_slot = false;
+
+  // Send a DL_TTI.request.
+  translator.dl_tti_request(msg);
+
+  // Assert that the downlink processor is configured.
+  ASSERT_TRUE(dl_processor_pool.processor(slot).has_configure_resource_grid_method_been_called());
+  // Assert that the resource grid has NOT been set to zero.
+  ASSERT_FALSE(grid.has_set_all_zero_method_been_called());
+
+  // Increase the slots.
+  ASSERT_FALSE(dl_processor_pool.processor(slot).has_finish_processing_pdus_method_been_called());
+
+  translator.handle_new_slot(slot + headroom_in_slots);
+
+  ASSERT_FALSE(dl_processor_pool.processor(slot).has_finish_processing_pdus_method_been_called());
+
+  ++msg.slot;
+  translator.dl_tti_request(msg);
+
+  // Assert that the finish processing PDUs method of the previous slot downlink_processor has been called.
+  ASSERT_TRUE(dl_processor_pool.processor(slot).has_finish_processing_pdus_method_been_called());
+  ASSERT_FALSE(error_notifier_spy.has_on_error_indication_been_called());
+}
+
 TEST_F(fapi_to_phy_translator_fixture, message_received_is_sended_when_the_current_slot_is_bigger_than_the_allowed)
 {
   ASSERT_FALSE(dl_processor_pool.processor(slot).has_configure_resource_grid_method_been_called());
@@ -342,13 +370,15 @@ TEST_F(fapi_to_phy_translator_fixture, message_received_is_sended_when_the_curre
   // Assert that the resource grid has NOT been set to zero.
   ASSERT_FALSE(grid.has_set_all_zero_method_been_called());
 
+  // The manager headroom size is the number of headroom in slots plus 3.
+  unsigned manager_headroom_size = headroom_in_slots + 3U;
   // Increase the slots.
-  for (unsigned i = 0; i != allowed_delay_slots; ++i) {
+  for (unsigned i = 1; i != (manager_headroom_size - 1U); ++i) {
     translator.handle_new_slot(slot + i);
     ASSERT_FALSE(dl_processor_pool.processor(slot).has_finish_processing_pdus_method_been_called());
   }
 
-  translator.handle_new_slot(slot + allowed_delay_slots);
+  translator.handle_new_slot(slot + (manager_headroom_size - 1U));
 
   // Assert that the finish processing PDUs method of the previous slot downlink_processor has been called.
   ASSERT_TRUE(dl_processor_pool.processor(slot).has_finish_processing_pdus_method_been_called());