ofh: change logging of lates in the transmitter queue

Author: Pavel Harbanau

include/srsran/ofh/ethernet/ethernet_frame_pool.h

@@ -108,6 +108,11 @@ class frame_buffer_array
   };
 
 public:
+  struct used_buffer {
+    frame_buffer*          buffer;
+    ofh::slot_symbol_point timestamp;
+  };
+
   // Constructor receives number of buffers stored/read at a time, reserves storage for all eAxCs.
   frame_buffer_array(unsigned nof_buffers_to_return, unsigned buffer_size, unsigned nof_antennas) :
     increment_quant(nof_buffers_to_return),
@@ -140,14 +145,14 @@ class frame_buffer_array
 
   // Stores actually used buffers in a list of buffers ready for sending.
   // Unused buffers state is changed to 'free'.
-  void push_buffers(span<frame_buffer> prepared_buffers)
+  void push_buffers(span<frame_buffer> prepared_buffers, ofh::slot_symbol_point symbol_point)
   {
     for (auto& buffer : prepared_buffers) {
       if (buffer.empty()) {
         buffer.status = frame_buffer::frame_buffer_status::free;
       } else {
         buffer.status = frame_buffer::frame_buffer_status::used;
-        used_buffers.push_back(&buffer);
+        used_buffers.push_back({&buffer, symbol_point});
       }
     }
   }
@@ -176,14 +181,16 @@ class frame_buffer_array
   span<const frame_buffer*> find_buffers_ready_for_sending()
   {
     aux_array.clear();
-    for (auto& buffer : used_buffers) {
-      buffer->status = frame_buffer::frame_buffer_status::marked_to_send;
-      aux_array.push_back(buffer);
+    for (auto& used_buf : used_buffers) {
+      used_buf.buffer->status = frame_buffer::frame_buffer_status::marked_to_send;
+      aux_array.push_back(used_buf.buffer);
     }
     used_buffers.clear();
     return aux_array;
   }
 
+  span<const used_buffer> get_prepared_buffers() const { return used_buffers; }
+
 private:
   // Number of buffers accessed at a time.
   unsigned increment_quant;
@@ -192,7 +199,7 @@ class frame_buffer_array
   // Data buffers.
   storage_array_type buffers_array;
   // Used buffers are added to this list.
-  static_vector<frame_buffer*, 128> used_buffers;
+  static_vector<used_buffer, 128> used_buffers;
   // Auxiliary array used as a list of ready-to-send buffers returned to a reader.
   std::vector<const frame_buffer*> aux_array;
   // Keeps track of the current write position.
@@ -202,8 +209,11 @@ class frame_buffer_array
 /// Pool of Ethernet frames pre-allocated for each slot symbol.
 class eth_frame_pool
 {
+  /// Number of slots the pool can accommodate.
+  static constexpr size_t NUM_SLOTS = 20L;
+
   /// Maximum number of entries contained by the pool, one entry per OFDM symbol, sized to accommodate 20 slots.
-  static constexpr size_t NUM_ENTRIES = NOF_OFDM_SYM_PER_SLOT_NORMAL_CP * 20L;
+  static constexpr size_t NUM_ENTRIES = NOF_OFDM_SYM_PER_SLOT_NORMAL_CP * NUM_SLOTS;
 
   /// Number of symbols in an interval for which an auxiliary vector is pre-allocated to store buffer pointers.
   static constexpr size_t NUM_INTERVAL_SYMBOL = 14;
@@ -259,10 +269,10 @@ class eth_frame_pool
     }
 
     /// Push span of ready buffers to the array associated with the given OFH type.
-    void push_buffers(const ofh_pool_message_type& context, span<frame_buffer> prepared_buffers)
+    void push_buffers(const frame_pool_context& context, span<frame_buffer> prepared_buffers)
     {
-      frame_buffer_array& entry_buf = get_ofh_type_buffers(context.type, context.direction);
-      entry_buf.push_buffers(prepared_buffers);
+      frame_buffer_array& entry_buf = get_ofh_type_buffers(context.type.type, context.type.direction);
+      entry_buf.push_buffers(prepared_buffers, context.symbol_point);
     }
 
     void clear_buffers(const ofh_pool_message_type& context)
@@ -277,6 +287,12 @@ class eth_frame_pool
       entry_buf.reset_buffers();
     }
 
+    span<const frame_buffer_array::used_buffer> get_prepared_buffers(const ofh_pool_message_type& context) const
+    {
+      const frame_buffer_array& entry_buf = get_ofh_type_buffers(context.type, context.direction);
+      return entry_buf.get_prepared_buffers();
+    }
+
     /// Returns a view over a next stored frame buffer for a given OFH type.
     span<const frame_buffer*> read_buffers(const ofh_pool_message_type& context)
     {
@@ -320,7 +336,7 @@ class eth_frame_pool
     pool_entry& p_entry = get_pool_entry(context.symbol_point.get_slot(), context.symbol_point.get_symbol_index());
     // Lock and update the pool entry.
     std::lock_guard<std::mutex> lock(mutex);
-    p_entry.push_buffers(context.type, prepared_buffers);
+    p_entry.push_buffers(context, prepared_buffers);
   }
 
   /// Returns data buffers from the pool to a reader thread given a specific symbol context.
@@ -385,29 +401,76 @@ class eth_frame_pool
     }
   }
 
-  /// Clear stored buffers associated with the given slot.
-  void clear_slot(slot_point slot_point)
+  /// Clears stored buffers associated with the given slot and logs the messages that could not be sent.
+  void clear_downlink_slot(slot_point slot_point, srslog::basic_logger& logger)
   {
     // Lock before changing the pool entries.
     std::lock_guard<std::mutex> lock(mutex);
 
     pool_entry& cp_entry = get_pool_entry(slot_point, 0);
     // Clear buffers with DL Control-Plane messages.
     ofh_pool_message_type msg_type{ofh::message_type::control_plane, ofh::data_direction::downlink};
-    cp_entry.reset_buffers(msg_type);
-    // Clear buffers with UL Control-Plane messages.
-    msg_type.direction = ofh::data_direction::uplink;
-    cp_entry.reset_buffers(msg_type);
+
+    auto dl_cp_buffers = cp_entry.get_prepared_buffers(msg_type);
+    for (const auto& used_buf : dl_cp_buffers) {
+      if (used_buf.timestamp.get_slot() == slot_point) {
+        continue;
+      }
+      logger.warning("Detected '{}' late downlink C-Plane messages in the transmitter queue for slot '{}'",
+                     dl_cp_buffers.size(),
+                     used_buf.timestamp.get_slot());
+      cp_entry.reset_buffers(msg_type);
+      break;
+    }
 
     // Clear buffers with User-Plane messages.
     msg_type.type      = ofh::message_type::user_plane;
     msg_type.direction = ofh::data_direction::downlink;
-    for (unsigned symbol = 0; symbol != 14; ++symbol) {
+    for (unsigned symbol = 0; symbol != NOF_OFDM_SYM_PER_SLOT_NORMAL_CP; ++symbol) {
       pool_entry& up_entry = get_pool_entry(slot_point, symbol);
-      up_entry.reset_buffers(msg_type);
+
+      auto dl_up_buffers = up_entry.get_prepared_buffers(msg_type);
+      for (const auto& used_buf : dl_up_buffers) {
+        if (used_buf.timestamp.get_slot() == slot_point) {
+          continue;
+        }
+        logger.warning(
+            "Detected '{}' late downlink U-Plane messages in the transmitter queue for slot '{}', symbol '{}'",
+            dl_up_buffers.size(),
+            used_buf.timestamp.get_slot(),
+            used_buf.timestamp.get_symbol_index());
+        up_entry.reset_buffers(msg_type);
+        break;
+      }
     }
   }
 
+  /// Clears stored uplink C-Plane buffers associated with the given slot and logs the messages that could not be sent.
+  void clear_uplink_slot(slot_point slot_point, srslog::basic_logger& logger)
+  {
+    // Lock before changing the pool entries.
+    std::lock_guard<std::mutex> lock(mutex);
+
+    pool_entry& cp_entry = get_pool_entry(slot_point, 0);
+    // Clear buffers with UL Control-Plane messages.
+    ofh_pool_message_type msg_type{ofh::message_type::control_plane, ofh::data_direction::uplink};
+
+    auto ul_cp_buffers = cp_entry.get_prepared_buffers(msg_type);
+    for (const auto& used_buf : ul_cp_buffers) {
+      if (used_buf.timestamp.get_slot() == slot_point) {
+        continue;
+      }
+      logger.warning("Detected '{}' late uplink C-Plane messages in the transmitter queue for slot '{}'",
+                     ul_cp_buffers.size(),
+                     used_buf.timestamp.get_slot());
+      cp_entry.reset_buffers(msg_type);
+      break;
+    }
+  }
+
+  /// Returns number of slots the pool can accommodate.
+  size_t pool_size_in_slots() const { return NUM_SLOTS; }
+
 private:
   /// Buffer pool.
   std::vector<pool_entry> pool;

lib/ofh/transmitter/ofh_downlink_handler_broadcast_impl.cpp

@@ -59,7 +59,7 @@ void downlink_handler_broadcast_impl::handle_dl_data(const resource_grid_context
                                                      const resource_grid_reader&  grid)
 {
   // Clear any stale buffers associated with the context slot.
-  frame_pool.clear_slot(context.slot);
+  frame_pool.clear_downlink_slot(context.slot, logger);
 
   if (window_checker.is_late(context.slot)) {
     err_notifier.get().on_late_downlink_message({context.slot, sector_id});

lib/ofh/transmitter/ofh_downlink_handler_impl.cpp

@@ -62,7 +62,7 @@ void downlink_handler_impl::handle_dl_data(const resource_grid_context& context,
                 grid.get_nof_ports());
 
   // Clear any stale buffers associated with the context slot.
-  frame_pool.clear_slot(context.slot);
+  frame_pool.clear_downlink_slot(context.slot, logger);
 
   if (window_checker.is_late(context.slot)) {
     err_notifier.get().on_late_downlink_message({context.slot, sector_id});

lib/ofh/transmitter/ofh_message_transmitter_impl.cpp

@@ -76,33 +76,4 @@ void message_transmitter_impl::on_new_symbol(slot_symbol_point symbol_point)
   interval.start = symbol_point + timing_params.sym_up_dl_end;
   interval.end   = symbol_point + timing_params.sym_up_dl_start;
   transmit_enqueued_messages(interval);
-
-  // Log the late messages when the transmission window closes.
-  log_late_messages_on_tx_window_close(symbol_point);
-}
-
-void message_transmitter_impl::log_late_messages_on_tx_window_close(slot_symbol_point symbol_point)
-{
-  log_late_messages(symbol_point + timing_params.sym_cp_dl_end, message_type::control_plane, data_direction::downlink);
-  log_late_messages(symbol_point + timing_params.sym_cp_ul_end, message_type::control_plane, data_direction::uplink);
-  log_late_messages(symbol_point + timing_params.sym_up_dl_end, message_type::user_plane, data_direction::downlink);
-}
-
-void message_transmitter_impl::log_late_messages(slot_symbol_point late_point,
-                                                 message_type      type,
-                                                 data_direction    direction)
-{
-  const ether::frame_pool_context context{{type, direction}, late_point};
-  auto                            frame_buffers = pool.read_frame_buffers(context);
-
-  if (!frame_buffers.empty()) {
-    logger.warning("Detected '{}' late Ethernet frames in the transmitter queue for slot '{}', symbol '{}', type '{}', "
-                   "direction '{}'",
-                   frame_buffers.size(),
-                   late_point.get_slot(),
-                   late_point.get_symbol_index(),
-                   (type == message_type::control_plane) ? "control" : "user",
-                   (direction == data_direction::downlink) ? "downlink" : "uplink");
-    pool.clear_sent_frame_buffers(context);
-  }
 }

lib/ofh/transmitter/ofh_message_transmitter_impl.h

@@ -79,12 +79,6 @@ class message_transmitter_impl : public ota_symbol_boundary_notifier
 private:
   /// Transmits enqueued messages for the given interval of slot symbol points.
   void transmit_enqueued_messages(const ether::frame_pool_interval& interval);
-
-  /// Logs the messages that could not be sent due the transmission window closed.
-  void log_late_messages_on_tx_window_close(slot_symbol_point symbol_point);
-
-  /// Logs the late messages for the given late slot symbol point, message type and direction.
-  void log_late_messages(slot_symbol_point late_point, message_type type, data_direction direction);
 };
 
 } // namespace ofh

lib/ofh/transmitter/ofh_transmitter_factories.cpp

@@ -161,10 +161,11 @@ create_uplink_request_handler(const transmitter_config&
   config.cp                  = tx_config.cp;
 
   uplink_request_handler_impl_dependencies dependencies;
-  dependencies.logger        = &logger;
-  dependencies.ul_slot_repo  = ul_slot_context_repo;
-  dependencies.ul_prach_repo = prach_context_repo;
-  dependencies.data_flow     = create_data_flow_cplane_sched(tx_config, logger, frame_pool, ul_cp_context_repo);
+  dependencies.logger         = &logger;
+  dependencies.ul_slot_repo   = ul_slot_context_repo;
+  dependencies.ul_prach_repo  = prach_context_repo;
+  dependencies.frame_pool_ptr = frame_pool;
+  dependencies.data_flow      = create_data_flow_cplane_sched(tx_config, logger, frame_pool, ul_cp_context_repo);
 
   return std::make_unique<uplink_request_handler_impl>(config, std::move(dependencies));
 }

lib/ofh/transmitter/ofh_uplink_request_handler_impl.cpp

@@ -48,17 +48,22 @@ uplink_request_handler_impl::uplink_request_handler_impl(const uplink_request_ha
   ul_prach_repo_ptr(dependencies.ul_prach_repo),
   ul_slot_repo(*ul_slot_repo_ptr),
   ul_prach_repo(*ul_prach_repo_ptr),
-  data_flow(std::move(dependencies.data_flow))
+  data_flow(std::move(dependencies.data_flow)),
+  frame_pool_ptr(dependencies.frame_pool_ptr),
+  frame_pool(*frame_pool_ptr)
 {
   srsran_assert(ul_slot_repo_ptr, "Invalid uplink repository");
   srsran_assert(ul_prach_repo_ptr, "Invalid PRACH repository");
   srsran_assert(data_flow, "Invalid data flow");
+  srsran_assert(frame_pool_ptr, "Invalid frame pool");
 }
 
 void uplink_request_handler_impl::handle_prach_occasion(const prach_buffer_context& context, prach_buffer& buffer)
 {
   logger.debug("Registering PRACH context entry for slot '{}' and sector#{}", context.slot, context.sector);
 
+  frame_pool.clear_uplink_slot(context.slot, logger);
+
   // Sampling rate defining the \f$T_s = 1/(\Delta f_{ref} \times N_{f,ref})\f$ parameter, see 3GPP TS38.211,
   // clause 4.1.
   // Open Fronthaul parameters timeOffset and cpLength are expressed in multiple of \f$T_s\f$ units.
@@ -136,6 +141,8 @@ void uplink_request_handler_impl::handle_new_uplink_slot(const resource_grid_con
 {
   logger.debug("Registering UL context entry for slot '{}' and sector#{}", context.slot, context.sector);
 
+  frame_pool.clear_uplink_slot(context.slot, logger);
+
   // Store the context in the repository.
   ul_slot_repo.add(context, grid);
 

lib/ofh/transmitter/ofh_uplink_request_handler_impl.h

@@ -14,6 +14,7 @@
 #include "../support/uplink_context_repository.h"
 #include "ofh_data_flow_cplane_scheduling_commands.h"
 #include "srsran/adt/optional.h"
+#include "srsran/ofh/ethernet/ethernet_frame_pool.h"
 #include "srsran/ofh/transmitter/ofh_uplink_request_handler.h"
 #include "srsran/ran/tdd/tdd_ul_dl_config.h"
 
@@ -42,6 +43,8 @@ struct uplink_request_handler_impl_dependencies {
   std::shared_ptr<uplink_context_repository> ul_slot_repo;
   /// Uplink PRACH context repository.
   std::shared_ptr<prach_context_repository> ul_prach_repo;
+  /// Ethernet frame pool.
+  std::shared_ptr<ether::eth_frame_pool> frame_pool_ptr;
   /// Data flow for Control-Plane scheduling commands.
   std::unique_ptr<data_flow_cplane_scheduling_commands> data_flow;
 };
@@ -71,6 +74,8 @@ class uplink_request_handler_impl : public uplink_request_handler
   uplink_context_repository&                            ul_slot_repo;
   prach_context_repository&                             ul_prach_repo;
   std::unique_ptr<data_flow_cplane_scheduling_commands> data_flow;
+  std::shared_ptr<ether::eth_frame_pool>                frame_pool_ptr;
+  ether::eth_frame_pool&                                frame_pool;
 };
 
 } // namespace ofh

tests/unittests/ofh/ethernet/ethernet_frame_pool_test.cpp

@@ -457,6 +457,11 @@ TEST_P(EthFramePoolFixture, pool_should_have_enough_space)
 
 TEST_P(EthFramePoolFixture, clearing_full_pool_should_allow_adding_more_data)
 {
+  srslog::basic_logger& logger = srslog::fetch_basic_logger("OFH_TEST", false);
+  logger.set_level(srslog::basic_levels::debug);
+
+  size_t pool_size_slots = pool.pool_size_in_slots();
+
   slot_point slot(to_numerology_value(scs), 0);
   for (unsigned slot_count = 0; slot_count < TEST_NUM_SLOTS; ++slot_count) {
     for (unsigned symbol = 0; symbol < nof_symbols; ++symbol) {
@@ -478,8 +483,9 @@ TEST_P(EthFramePoolFixture, clearing_full_pool_should_allow_adding_more_data)
         }
         pool.push_frame_buffers(ctx, frame_buffers);
       }
-      // Clear full slot in the pool and try to get buffers again.
-      pool.clear_slot(slot);
+      // Increase slot by pool size, clear stale buffers in the pool and try to get buffers again.
+      auto wrapped_slot = slot + pool_size_slots;
+      pool.clear_downlink_slot(wrapped_slot, logger);
       for (unsigned i = 0; i != nof_requested_buffers; ++i) {
         span<frame_buffer> frame_buffers = pool.get_frame_buffers(ctx);
         ASSERT_TRUE(!frame_buffers.empty()) << "Non-empty span of buffers expected";
@@ -506,7 +512,7 @@ TEST_P(EthFramePoolFixture, clearing_full_pool_should_allow_adding_more_data)
         pool.push_frame_buffers(ctx, frame_buffers);
       }
       // Clear full slot in the pool and try to get buffers again.
-      pool.clear_slot(slot);
+      pool.clear_uplink_slot(wrapped_slot, logger);
       for (unsigned i = 0; i != nof_requested_buffers; ++i) {
         span<frame_buffer> frame_buffers = pool.get_frame_buffers(ctx);
         ASSERT_TRUE(!frame_buffers.empty()) << "Non-empty span of buffers expected";