ofh: reset the pool before processing new slot

Author: Pavel Harbanau

include/srsran/ofh/ethernet/ethernet_frame_pool.h

@@ -109,9 +109,9 @@ class frame_buffer_array
 
 public:
   // Constructor receives number of buffers stored/read at a time, reserves storage for all eAxCs.
-  frame_buffer_array(unsigned nof_buffers, unsigned buffer_size) :
-    increment_quant(nof_buffers),
-    storage_nof_buffers(ofh::MAX_NOF_SUPPORTED_EAXC * nof_buffers * NUM_OF_ENTRIES),
+  frame_buffer_array(unsigned nof_buffers_to_return, unsigned buffer_size, unsigned nof_antennas) :
+    increment_quant(nof_buffers_to_return),
+    storage_nof_buffers(nof_buffers_to_return * nof_antennas * NUM_OF_ENTRIES),
     buffers_array(storage_nof_buffers, frame_buffer{buffer_size}),
     write_position(storage_nof_buffers)
   {
@@ -162,6 +162,16 @@ class frame_buffer_array
     }
   }
 
+  // Changed state of all 'used' buffers to 'free'.
+  void reset_buffers()
+  {
+    for (auto& buffer : buffers_array) {
+      if (buffer.status != frame_buffer::frame_buffer_status::marked_to_send) {
+        buffer.status = frame_buffer::frame_buffer_status::free;
+      }
+    }
+  }
+
   // Returns a vector of pointers to the buffers ready for sending.
   span<const frame_buffer*> find_buffers_ready_for_sending()
   {
@@ -209,11 +219,11 @@ class eth_frame_pool
     pool_entry(units::bytes mtu, unsigned num_of_frames)
     {
       // DL C-Plane storage.
-      buffers.emplace_back(NUM_CP_MESSAGES_TO_RETURN, mtu.value());
+      buffers.emplace_back(NUM_CP_MESSAGES_TO_RETURN, mtu.value(), ofh::MAX_NOF_SUPPORTED_EAXC);
       // UL C-Plane storage.
-      buffers.emplace_back(NUM_CP_MESSAGES_TO_RETURN, mtu.value());
+      buffers.emplace_back(NUM_CP_MESSAGES_TO_RETURN, mtu.value(), ofh::MAX_NOF_SUPPORTED_EAXC * 2);
       // U-Plane storage.
-      buffers.emplace_back(num_of_frames, mtu.value());
+      buffers.emplace_back(num_of_frames, mtu.value(), ofh::MAX_NOF_SUPPORTED_EAXC);
     }
 
     /// Returns frame buffers for the given OFH type and given direction.
@@ -261,6 +271,12 @@ class eth_frame_pool
       entry_buf.clear_buffers();
     }
 
+    void reset_buffers(const ofh_pool_message_type& context)
+    {
+      frame_buffer_array& entry_buf = get_ofh_type_buffers(context.type, context.direction);
+      entry_buf.reset_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)
     {
@@ -378,17 +394,17 @@ class eth_frame_pool
     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.clear_buffers(msg_type);
+    cp_entry.reset_buffers(msg_type);
     // Clear buffers with UL Control-Plane messages.
     msg_type.direction = ofh::data_direction::uplink;
-    cp_entry.clear_buffers(msg_type);
+    cp_entry.reset_buffers(msg_type);
 
     // 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) {
       pool_entry& up_entry = get_pool_entry(slot_point, symbol);
-      up_entry.clear_buffers(msg_type);
+      up_entry.reset_buffers(msg_type);
     }
   }
 

lib/ofh/transmitter/ofh_downlink_handler_broadcast_impl.cpp

@@ -29,15 +29,21 @@ downlink_handler_broadcast_impl::downlink_handler_broadcast_impl(
       *dependencies.logger,
       calculate_nof_symbols_before_ota(config.cp, config.scs, config.dl_processing_time, config.tx_timing_params),
       get_nsymb_per_slot(config.cp),
-      to_numerology_value(config.scs))
+      to_numerology_value(config.scs)),
+  frame_pool_ptr(dependencies.frame_pool_ptr),
+  frame_pool(*frame_pool_ptr)
 {
   srsran_assert(data_flow_cplane, "Invalid Control-Plane data flow");
   srsran_assert(data_flow_uplane, "Invalid User-Plane data flow");
+  srsran_assert(frame_pool_ptr, "Invalid frame pool");
 }
 
 void downlink_handler_broadcast_impl::handle_dl_data(const resource_grid_context& context,
                                                      const resource_grid_reader&  grid)
 {
+  // Clear any stale buffers associated with the context slot.
+  frame_pool.clear_slot(context.slot);
+
   if (window_checker.is_late(context.slot)) {
     logger.warning(
         "Dropped late downlink resource grid in slot '{}' and sector#{}. No OFH data will be transmitted for this slot",

lib/ofh/transmitter/ofh_downlink_handler_broadcast_impl.h

@@ -15,6 +15,7 @@
 #include "ofh_tx_window_checker.h"
 #include "srsran/adt/span.h"
 #include "srsran/adt/static_vector.h"
+#include "srsran/ofh/ethernet/ethernet_frame_pool.h"
 #include "srsran/ofh/ofh_constants.h"
 #include "srsran/ofh/transmitter/ofh_downlink_handler.h"
 #include "srsran/ofh/transmitter/ofh_transmitter_configuration.h"
@@ -47,6 +48,8 @@ struct downlink_handler_broadcast_impl_dependencies {
   std::unique_ptr<data_flow_cplane_scheduling_commands> data_flow_cplane;
   /// Data flow for User-Plane.
   std::unique_ptr<data_flow_uplane_downlink_data> data_flow_uplane;
+  /// Ethernet frame pool.
+  std::shared_ptr<ether::eth_frame_pool> frame_pool_ptr;
 };
 
 /// \brief Open Fronthaul downlink broadcast handler implementation.
@@ -72,6 +75,8 @@ class downlink_handler_broadcast_impl : public downlink_handler
   std::unique_ptr<data_flow_cplane_scheduling_commands> data_flow_cplane;
   std::unique_ptr<data_flow_uplane_downlink_data>       data_flow_uplane;
   tx_window_checker                                     window_checker;
+  std::shared_ptr<ether::eth_frame_pool>                frame_pool_ptr;
+  ether::eth_frame_pool&                                frame_pool;
 };
 
 } // namespace ofh

lib/ofh/transmitter/ofh_transmitter_factories.cpp

@@ -121,6 +121,7 @@ create_downlink_manager(const transmitter_config&                         tx_con
     dl_dependencies.logger           = &logger;
     dl_dependencies.data_flow_cplane = std::move(data_flow_cplane);
     dl_dependencies.data_flow_uplane = std::move(data_flow_uplane);
+    dl_dependencies.frame_pool_ptr   = frame_pool;
 
     return std::make_unique<downlink_manager_broadcast_impl>(dl_config, std::move(dl_dependencies));
   }

tests/integrationtests/ofh/ofh_integration_test.cpp

@@ -1104,11 +1104,9 @@ static ru_ofh_dependencies generate_ru_dependencies(srslog::basic_logger&
   sector_deps.eth_gateway = std::move(gateway);
 
   // Configure Ethernet receiver.
-  if (!test_params.use_loopback_receiver) {
-    auto dummy_receiver      = std::make_unique<dummy_eth_receiver>(logger, *workers.ru_rx_exec);
-    eth_receiver             = dummy_receiver.get();
-    sector_deps.eth_receiver = std::move(dummy_receiver);
-  }
+  auto dummy_receiver      = std::make_unique<dummy_eth_receiver>(logger, *workers.ru_rx_exec);
+  eth_receiver             = dummy_receiver.get();
+  sector_deps.eth_receiver = std::move(dummy_receiver);
 
   return dependencies;
 }
@@ -1138,20 +1136,26 @@ int main(int argc, char** argv)
   test_gateway*            tx_gateway;
   test_ether_receiver*     eth_receiver;
 
+  ru_ofh_configuration ru_cfg = generate_ru_config();
+  ru_ofh_dependencies  ru_deps =
+      generate_ru_dependencies(logger, workers, &timing_notifier, &rx_symbol_notifier, tx_gateway, eth_receiver);
+
   if (test_params.use_loopback_receiver) {
+    ru_deps.sector_dependencies[0].eth_receiver.reset();
     eth_receiver_ptr = std::make_unique<lo_eth_receiver>(logger);
     eth_receiver     = eth_receiver_ptr.get();
   }
-
-  ru_ofh_configuration ru_cfg = generate_ru_config();
-  ru_ofh_dependencies  ru_deps =
-      generate_ru_dependencies(logger, workers, &timing_notifier, &rx_symbol_notifier, tx_gateway, eth_receiver);
   std::unique_ptr<radio_unit> ru_object = create_ofh_ru(ru_cfg, std::move(ru_deps));
 
   // Get RU downlink plane handler.
   auto& ru_dl_handler = ru_object->get_downlink_plane_handler();
   auto& ru_ul_handler = ru_object->get_uplink_plane_handler();
 
+  if (test_params.use_loopback_receiver) {
+    eth_receiver_ptr = std::make_unique<lo_eth_receiver>(logger);
+    eth_receiver     = eth_receiver_ptr.get();
+  }
+
   // Create RU emulator instance.
   ru_compression_params ul_compression_params{to_compression_type(test_params.data_compr_method),
                                               test_params.data_bitwidth};

tests/unittests/ofh/ethernet/ethernet_frame_pool_test.cpp

@@ -412,6 +412,113 @@ TEST_P(EthFramePoolFixture, read_interval_should_return_correct_data)
   pool.clear_sent_frame_buffers(interval);
 }
 
+// Pool should have enough space to store C-Plane and U-Plane packets for all antennas.
+TEST_P(EthFramePoolFixture, pool_should_have_enough_space)
+{
+  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) {
+      ofh::slot_symbol_point    symbol_point(slot, symbol, nof_symbols);
+      ether::frame_pool_context ctx{{ofh_type, ofh::data_direction::downlink}, symbol_point};
+      unsigned                  nof_requested_buffers = ofh::MAX_NOF_SUPPORTED_EAXC;
+      // DL C-Plane and U-Plane
+      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";
+        for (auto& buffer : frame_buffers) {
+          buffer.set_size(64);
+        }
+        pool.push_frame_buffers(ctx, frame_buffers);
+      }
+      pool.read_frame_buffers(ctx);
+      pool.clear_sent_frame_buffers(ctx);
+
+      if (ofh_type == ofh::message_type::user_plane) {
+        continue;
+      }
+      // UL C-Plane
+      // We may need to write Type 1 and Type 3 C-Plane messages in the same slot and symbol.
+      nof_requested_buffers *= 2;
+      ctx = {{ofh_type, ofh::data_direction::uplink}, symbol_point};
+      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";
+        for (auto& buffer : frame_buffers) {
+          buffer.set_size(64);
+        }
+        pool.push_frame_buffers(ctx, frame_buffers);
+      }
+      pool.read_frame_buffers(ctx);
+      pool.clear_sent_frame_buffers(ctx);
+    }
+    ++slot;
+  }
+}
+
+TEST_P(EthFramePoolFixture, clearing_full_pool_should_allow_adding_more_data)
+{
+  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) {
+      ofh::slot_symbol_point symbol_point(slot, symbol, nof_symbols);
+
+      if (ofh_type == ofh::message_type::control_plane && symbol != 0) {
+        continue;
+      }
+
+      // DL C-Plane and U-Plane
+      bool                      pool_has_space = true;
+      ether::frame_pool_context ctx{{ofh_type, ofh::data_direction::downlink}, symbol_point};
+      unsigned                  nof_requested_buffers = ofh::MAX_NOF_SUPPORTED_EAXC;
+      while (pool_has_space) {
+        span<frame_buffer> frame_buffers = pool.get_frame_buffers(ctx);
+        pool_has_space                   = !frame_buffers.empty();
+        for (auto& buffer : frame_buffers) {
+          buffer.set_size(64);
+        }
+        pool.push_frame_buffers(ctx, frame_buffers);
+      }
+      // Clear full slot in the pool and try to get buffers again.
+      pool.clear_slot(slot);
+      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";
+        pool.push_frame_buffers(ctx, frame_buffers);
+      }
+      pool.read_frame_buffers(ctx);
+      pool.clear_sent_frame_buffers(ctx);
+
+      if (ofh_type == ofh::message_type::user_plane) {
+        continue;
+      }
+
+      // UL C-Plane
+      // We may need to write Type 1 and Type 3 C-Plane messages in the same slot and symbol.
+      nof_requested_buffers *= 2;
+      ctx            = {{ofh_type, ofh::data_direction::uplink}, symbol_point};
+      pool_has_space = true;
+      while (pool_has_space) {
+        span<frame_buffer> frame_buffers = pool.get_frame_buffers(ctx);
+        pool_has_space                   = !frame_buffers.empty();
+        for (auto& buffer : frame_buffers) {
+          buffer.set_size(64);
+        }
+        pool.push_frame_buffers(ctx, frame_buffers);
+      }
+      // Clear full slot in the pool and try to get buffers again.
+      pool.clear_slot(slot);
+      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";
+        pool.push_frame_buffers(ctx, frame_buffers);
+      }
+      pool.read_frame_buffers(ctx);
+      pool.clear_sent_frame_buffers(ctx);
+    }
+    ++slot;
+  }
+}
+
 INSTANTIATE_TEST_SUITE_P(EthFramePoolTestSuite,
                          EthFramePoolFixture,
                          ::testing::Combine(::testing::Values(mtu::MTU_9000, mtu::MTU_5000, mtu::MTU_1500),