ofh: removed the vector of executors for downlink as it uses a pool executor

AlaiaL · codebot · commit 7ad68902e437 · 2024-01-12T16:04:55.000Z
diff --git a/apps/gnb/gnb.cpp b/apps/gnb/gnb.cpp
@@ -143,7 +143,7 @@ static void configure_ru_ofh_executors_and_notifiers(ru_ofh_configuration&
     sector_deps.logger                      = dependencies.logger;
     sector_deps.receiver_executor           = workers.ru_rx_exec[i];
     sector_deps.transmitter_executor        = workers.ru_tx_exec[i];
-    sector_deps.downlink_executors          = workers.ru_dl_exec[i];
+    sector_deps.downlink_executor           = workers.ru_dl_exec[i];
   }
 }
 
diff --git a/apps/gnb/gnb_worker_manager.cpp b/apps/gnb/gnb_worker_manager.cpp
@@ -404,7 +404,7 @@ void worker_manager::create_ofh_executors(span<const cell_appconfig> cells, bool
         cpu_masks.push_back(affinity_mng.calcute_affinity_mask(gnb_sched_affinity_mask_types::ru));
       }
       create_worker_pool(name, nof_ofh_dl_workers, task_worker_queue_size, {{exec_name}}, prio, cpu_masks);
-      ru_dl_exec[i].push_back(exec_mng.executors().at(exec_name));
+      ru_dl_exec[i] = exec_mng.executors().at(exec_name);
     }
 
     // Executor for Open Fronthaul messages transmission.
diff --git a/apps/gnb/gnb_worker_manager.h b/apps/gnb/gnb_worker_manager.h
@@ -38,29 +38,29 @@ struct worker_manager {
   /// - e1ap_cu_cp::handle_message calls cu-cp ctrl exec
   /// - e1ap_cu_up::handle_message calls cu-up ue exec
 
-  task_executor*                           cu_cp_exec      = nullptr;
-  task_executor*                           cu_up_ctrl_exec = nullptr; ///< CU-UP executor for control
-  task_executor*                           cu_up_dl_exec   = nullptr; ///< CU-UP executor for DL data flow
-  task_executor*                           cu_up_ul_exec   = nullptr; ///< CU-UP executor for UL data flow
-  std::vector<task_executor*>              lower_phy_tx_exec;
-  std::vector<task_executor*>              lower_phy_rx_exec;
-  std::vector<task_executor*>              lower_phy_dl_exec;
-  std::vector<task_executor*>              lower_phy_ul_exec;
-  std::vector<task_executor*>              lower_prach_exec;
-  std::vector<task_executor*>              upper_pusch_exec;
-  std::vector<task_executor*>              upper_pusch_decoder_exec;
-  std::vector<task_executor*>              upper_pucch_exec;
-  std::vector<task_executor*>              upper_prach_exec;
-  std::vector<task_executor*>              upper_pdsch_exec;
-  task_executor*                           radio_exec      = nullptr;
-  task_executor*                           ru_printer_exec = nullptr;
-  task_executor*                           ru_timing_exec  = nullptr;
-  std::vector<std::vector<task_executor*>> ru_dl_exec;
-  std::vector<task_executor*>              ru_tx_exec;
-  std::vector<task_executor*>              ru_rx_exec;
-  task_executor*                           cu_cp_e2_exec    = nullptr;
-  task_executor*                           cu_up_e2_exec    = nullptr;
-  task_executor*                           metrics_hub_exec = nullptr;
+  task_executor*              cu_cp_exec      = nullptr;
+  task_executor*              cu_up_ctrl_exec = nullptr; ///< CU-UP executor for control
+  task_executor*              cu_up_dl_exec   = nullptr; ///< CU-UP executor for DL data flow
+  task_executor*              cu_up_ul_exec   = nullptr; ///< CU-UP executor for UL data flow
+  std::vector<task_executor*> lower_phy_tx_exec;
+  std::vector<task_executor*> lower_phy_rx_exec;
+  std::vector<task_executor*> lower_phy_dl_exec;
+  std::vector<task_executor*> lower_phy_ul_exec;
+  std::vector<task_executor*> lower_prach_exec;
+  std::vector<task_executor*> upper_pusch_exec;
+  std::vector<task_executor*> upper_pusch_decoder_exec;
+  std::vector<task_executor*> upper_pucch_exec;
+  std::vector<task_executor*> upper_prach_exec;
+  std::vector<task_executor*> upper_pdsch_exec;
+  task_executor*              radio_exec      = nullptr;
+  task_executor*              ru_printer_exec = nullptr;
+  task_executor*              ru_timing_exec  = nullptr;
+  std::vector<task_executor*> ru_dl_exec;
+  std::vector<task_executor*> ru_tx_exec;
+  std::vector<task_executor*> ru_rx_exec;
+  task_executor*              cu_cp_e2_exec    = nullptr;
+  task_executor*              cu_up_e2_exec    = nullptr;
+  task_executor*              metrics_hub_exec = nullptr;
 
   std::unique_ptr<cu_up_executor_pool> cu_up_exec_mapper;
 
diff --git a/include/srsran/ofh/ofh_sector_config.h b/include/srsran/ofh/ofh_sector_config.h
@@ -106,8 +106,8 @@ struct sector_configuration {
 struct sector_dependencies {
   /// Logger.
   srslog::basic_logger* logger = nullptr;
-  /// Downlink task executors.
-  std::vector<task_executor*> downlink_executors;
+  /// Downlink task executor.
+  task_executor* downlink_executor;
   /// Transmitter task executor.
   task_executor* transmitter_executor = nullptr;
   /// Receiver task executor.
diff --git a/include/srsran/ru/ru_ofh_configuration.h b/include/srsran/ru/ru_ofh_configuration.h
@@ -110,8 +110,8 @@ struct ru_ofh_configuration {
 struct ru_ofh_sector_dependencies {
   /// Logger.
   srslog::basic_logger* logger = nullptr;
-  /// Downlink task executors.
-  std::vector<task_executor*> downlink_executors;
+  /// Downlink task executor.
+  task_executor* downlink_executor;
   /// Receiver task executor.
   task_executor* receiver_executor = nullptr;
   /// Transmitter task executor.
diff --git a/lib/ofh/ofh_factories.cpp b/lib/ofh/ofh_factories.cpp
@@ -106,7 +106,7 @@ std::unique_ptr<sector> srsran::ofh::create_ofh_sector(const sector_configuratio
   auto transmitter = create_transmitter(tx_config,
                                         *sector_deps.logger,
                                         *sector_deps.transmitter_executor,
-                                        sector_deps.downlink_executors,
+                                        *sector_deps.downlink_executor,
                                         std::move(sector_deps.eth_gateway),
                                         prach_repo,
                                         slot_repo,
diff --git a/lib/ofh/transmitter/ofh_data_flow_cplane_scheduling_commands_task_dispatcher.h b/lib/ofh/transmitter/ofh_data_flow_cplane_scheduling_commands_task_dispatcher.h
@@ -18,38 +18,21 @@
 namespace srsran {
 namespace ofh {
 
-/// Open Fronthaul Control-Plane scheduling and beamforming commands data flow task dispatcher implementation entry.
-struct data_flow_cplane_downlink_task_dispatcher_entry {
-  data_flow_cplane_downlink_task_dispatcher_entry(
-      std::unique_ptr<data_flow_cplane_scheduling_commands> data_flow_cplane_,
-      task_executor&                                        executor_) :
-    data_flow_cplane(std::move(data_flow_cplane_)), executor(executor_)
-  {
-    srsran_assert(data_flow_cplane, "Invalid data flow");
-  }
-
-  std::unique_ptr<data_flow_cplane_scheduling_commands> data_flow_cplane;
-  task_executor&                                        executor;
-};
-
 /// Open Fronthaul Control-Plane scheduling and beamforming commands data flow task dispatcher implementation.
 class data_flow_cplane_downlink_task_dispatcher : public data_flow_cplane_scheduling_commands
 {
 public:
-  data_flow_cplane_downlink_task_dispatcher(std::vector<data_flow_cplane_downlink_task_dispatcher_entry>&& config_) :
-    dispatchers(std::move(config_))
+  data_flow_cplane_downlink_task_dispatcher(std::unique_ptr<data_flow_cplane_scheduling_commands> data_flow_cplane_,
+                                            task_executor&                                        executor_) :
+    data_flow_cplane(std::move(data_flow_cplane_)), executor(executor_)
   {
+    srsran_assert(data_flow_cplane, "Invalid data flow");
   }
 
   // See interface for documentation.
   void enqueue_section_type_1_message(const data_flow_cplane_type_1_context& context) override
   {
-    unsigned                                         index            = last_used++ % dispatchers.size();
-    data_flow_cplane_downlink_task_dispatcher_entry& dispatcher       = dispatchers[index];
-    data_flow_cplane_scheduling_commands&            data_flow_cplane = *dispatcher.data_flow_cplane;
-
-    if (not dispatcher.executor.execute(
-            [&data_flow_cplane, context]() { data_flow_cplane.enqueue_section_type_1_message(context); })) {
+    if (!executor.execute([this, context]() { data_flow_cplane->enqueue_section_type_1_message(context); })) {
       srslog::fetch_basic_logger("OFH").warning("Failed to dispatch Control-Plane type 1 message for slot={}",
                                                 context.slot);
     }
@@ -58,20 +41,15 @@ class data_flow_cplane_downlink_task_dispatcher : public data_flow_cplane_schedu
   // See interface for documentation.
   void enqueue_section_type_3_prach_message(const data_flow_cplane_scheduling_prach_context& context) override
   {
-    unsigned                                         index            = last_used++ % dispatchers.size();
-    data_flow_cplane_downlink_task_dispatcher_entry& dispatcher       = dispatchers[index];
-    data_flow_cplane_scheduling_commands&            data_flow_cplane = *dispatcher.data_flow_cplane;
-
-    if (not dispatcher.executor.execute(
-            [&data_flow_cplane, context]() { data_flow_cplane.enqueue_section_type_3_prach_message(context); })) {
+    if (!executor.execute([this, context]() { data_flow_cplane->enqueue_section_type_3_prach_message(context); })) {
       srslog::fetch_basic_logger("OFH").warning("Failed to dispatch Control-Plane type 3 message for slot={}",
                                                 context.slot);
     }
   }
 
 private:
-  std::atomic<unsigned>                                        last_used{0};
-  std::vector<data_flow_cplane_downlink_task_dispatcher_entry> dispatchers;
+  std::unique_ptr<data_flow_cplane_scheduling_commands> data_flow_cplane;
+  task_executor&                                        executor;
 };
 
 } // namespace ofh
diff --git a/lib/ofh/transmitter/ofh_data_flow_uplane_downlink_task_dispatcher.h b/lib/ofh/transmitter/ofh_data_flow_uplane_downlink_task_dispatcher.h
@@ -17,46 +17,31 @@
 namespace srsran {
 namespace ofh {
 
-/// Open Fronthaul User-Plane downlink data flow task dispatcher entry.
-struct data_flow_uplane_downlink_task_dispatcher_entry {
-  data_flow_uplane_downlink_task_dispatcher_entry(std::unique_ptr<data_flow_uplane_downlink_data> data_flow_uplane_,
-                                                  task_executor&                                  executor_) :
-    data_flow_uplane(std::move(data_flow_uplane_)), executor(executor_)
-  {
-    srsran_assert(data_flow_uplane, "Invalid data flow");
-  }
-
-  std::unique_ptr<data_flow_uplane_downlink_data> data_flow_uplane;
-  task_executor&                                  executor;
-};
-
 /// Open Fronthaul User-Plane downlink data flow task dispatcher implementation.
 class data_flow_uplane_downlink_task_dispatcher : public data_flow_uplane_downlink_data
 {
 public:
-  data_flow_uplane_downlink_task_dispatcher(std::vector<data_flow_uplane_downlink_task_dispatcher_entry>&& config_) :
-    dispatchers(std::move(config_))
+  data_flow_uplane_downlink_task_dispatcher(std::unique_ptr<data_flow_uplane_downlink_data> data_flow_uplane_,
+                                            task_executor&                                  executor_) :
+    data_flow_uplane(std::move(data_flow_uplane_)), executor(executor_)
   {
+    srsran_assert(data_flow_uplane, "Invalid data flow");
   }
 
   // See interface for documentation.
   void enqueue_section_type_1_message(const data_flow_uplane_resource_grid_context& context,
                                       const resource_grid_reader&                   grid) override
   {
-    unsigned                                         index            = context.port % dispatchers.size();
-    data_flow_uplane_downlink_task_dispatcher_entry& dispatcher       = dispatchers[index];
-    data_flow_uplane_downlink_data&                  data_flow_uplane = *dispatcher.data_flow_uplane;
-
-    if (not dispatcher.executor.execute([&data_flow_uplane, context, &grid]() {
-          data_flow_uplane.enqueue_section_type_1_message(context, grid);
-        })) {
+    if (!executor.execute(
+            [this, context, &grid]() { data_flow_uplane->enqueue_section_type_1_message(context, grid); })) {
       srslog::fetch_basic_logger("OFH").warning(
           "Failed to dispatch message in the downlink data flow User-Plane for slot={}", context.slot);
     }
   }
 
 private:
-  std::vector<data_flow_uplane_downlink_task_dispatcher_entry> dispatchers;
+  std::unique_ptr<data_flow_uplane_downlink_data> data_flow_uplane;
+  task_executor&                                  executor;
 };
 
 } // namespace ofh
diff --git a/lib/ofh/transmitter/ofh_transmitter_factories.cpp b/lib/ofh/transmitter/ofh_transmitter_factories.cpp
@@ -103,20 +103,12 @@ create_downlink_manager(const transmitter_config&                         tx_con
                         srslog::basic_logger&                             logger,
                         std::shared_ptr<ether::eth_frame_pool>            frame_pool,
                         std::shared_ptr<uplink_cplane_context_repository> ul_cp_context_repo,
-                        const std::vector<task_executor*>&                executors)
+                        task_executor&                                    executor)
 {
-  std::vector<data_flow_uplane_downlink_task_dispatcher_entry> df_uplane_task_dispatcher_cfg;
-  std::vector<data_flow_cplane_downlink_task_dispatcher_entry> df_cplane_task_dispatcher_cfg;
-  for (auto* executor : executors) {
-    df_cplane_task_dispatcher_cfg.emplace_back(
-        create_data_flow_cplane_sched(tx_config, logger, frame_pool, ul_cp_context_repo), *executor);
-    df_uplane_task_dispatcher_cfg.emplace_back(create_data_flow_uplane_data(tx_config, logger, frame_pool), *executor);
-  }
-
-  auto data_flow_cplane =
-      std::make_unique<data_flow_cplane_downlink_task_dispatcher>(std::move(df_cplane_task_dispatcher_cfg));
-  auto data_flow_uplane =
-      std::make_unique<data_flow_uplane_downlink_task_dispatcher>(std::move(df_uplane_task_dispatcher_cfg));
+  auto data_flow_cplane = std::make_unique<data_flow_cplane_downlink_task_dispatcher>(
+      create_data_flow_cplane_sched(tx_config, logger, frame_pool, ul_cp_context_repo), executor);
+  auto data_flow_uplane = std::make_unique<data_flow_uplane_downlink_task_dispatcher>(
+      create_data_flow_uplane_data(tx_config, logger, frame_pool), executor);
 
   if (tx_config.downlink_broadcast) {
     downlink_handler_broadcast_impl_config dl_config;
@@ -210,7 +202,7 @@ static transmitter_impl_dependencies
 resolve_transmitter_dependencies(const transmitter_config&                         tx_config,
                                  srslog::basic_logger&                             logger,
                                  task_executor&                                    tx_executor,
-                                 const std::vector<task_executor*>&                downlink_executors,
+                                 task_executor&                                    downlink_executor,
                                  std::unique_ptr<ether::gateway>                   eth_gateway,
                                  std::shared_ptr<prach_context_repository>         prach_context_repo,
                                  std::shared_ptr<uplink_context_repository>        ul_slot_context_repo,
@@ -224,12 +216,12 @@ resolve_transmitter_dependencies(const transmitter_config&
   auto frame_pool = create_eth_frame_pool(tx_config, logger);
 
   dependencies.dl_manager =
-      create_downlink_manager(tx_config, logger, frame_pool, ul_cp_context_repo, downlink_executors);
+      create_downlink_manager(tx_config, logger, frame_pool, ul_cp_context_repo, downlink_executor);
 
   dependencies.ul_request_handler = std::make_unique<uplink_request_handler_task_dispatcher>(
       create_uplink_request_handler(
           tx_config, logger, frame_pool, prach_context_repo, ul_slot_context_repo, ul_cp_context_repo),
-      *downlink_executors.front());
+      downlink_executor);
 
   ether::gw_config eth_cfg;
   eth_cfg.interface                   = tx_config.interface;
@@ -259,7 +251,7 @@ std::unique_ptr<transmitter>
 srsran::ofh::create_transmitter(const transmitter_config&                         transmitter_cfg,
                                 srslog::basic_logger&                             logger,
                                 task_executor&                                    tx_executor,
-                                const std::vector<task_executor*>&                downlink_executors,
+                                task_executor&                                    downlink_executor,
                                 std::unique_ptr<ether::gateway>                   eth_gateway,
                                 std::shared_ptr<prach_context_repository>         prach_context_repo,
                                 std::shared_ptr<uplink_context_repository>        ul_slot_context_repo,
@@ -269,7 +261,7 @@ srsran::ofh::create_transmitter(const transmitter_config&
                                             resolve_transmitter_dependencies(transmitter_cfg,
                                                                              logger,
                                                                              tx_executor,
-                                                                             downlink_executors,
+                                                                             downlink_executor,
                                                                              std::move(eth_gateway),
                                                                              prach_context_repo,
                                                                              ul_slot_context_repo,
diff --git a/lib/ofh/transmitter/ofh_transmitter_factories.h b/lib/ofh/transmitter/ofh_transmitter_factories.h
@@ -30,7 +30,7 @@ namespace ofh {
 std::unique_ptr<transmitter> create_transmitter(const transmitter_config&                         transmitter_cfg,
                                                 srslog::basic_logger&                             logger,
                                                 task_executor&                                    tx_executor,
-                                                const std::vector<task_executor*>&                downlink_executors,
+                                                task_executor&                                    downlink_executor,
                                                 std::unique_ptr<ether::gateway>                   eth_gateway,
                                                 std::shared_ptr<prach_context_repository>         prach_context_repo,
                                                 std::shared_ptr<uplink_context_repository>        ul_slot_context_repo,
diff --git a/lib/ru/ofh/ru_ofh_factory.cpp b/lib/ru/ofh/ru_ofh_factory.cpp
@@ -66,7 +66,7 @@ static ofh::sector_dependencies generate_sector_dependencies(ru_ofh_sector_depen
   ofh_sector_dependencies.logger               = dependencies.logger;
   ofh_sector_dependencies.receiver_executor    = dependencies.receiver_executor;
   ofh_sector_dependencies.transmitter_executor = dependencies.transmitter_executor;
-  ofh_sector_dependencies.downlink_executors   = dependencies.downlink_executors;
+  ofh_sector_dependencies.downlink_executor    = dependencies.downlink_executor;
   ofh_sector_dependencies.notifier             = notifier;
   ofh_sector_dependencies.eth_gateway          = std::move(dependencies.eth_gateway);
 
diff --git a/tests/integrationtests/ofh/ofh_integration_test.cpp b/tests/integrationtests/ofh/ofh_integration_test.cpp
@@ -811,7 +811,7 @@ struct worker_manager {
       if (not exec_mng.add_execution_context(create_execution_context(ru_worker))) {
         report_fatal_error("Failed to instantiate {} execution context", ru_worker.name);
       }
-      ru_dl_exec.push_back(exec_mng.executors().at(exec_name));
+      ru_dl_exec = exec_mng.executors().at(exec_name);
     }
 
     // Executor for Open Fronthaul messages transmission.
@@ -879,13 +879,13 @@ struct worker_manager {
     }
   }
 
-  task_execution_manager      exec_mng;
-  task_executor*              ru_timing_exec = nullptr;
-  std::vector<task_executor*> ru_dl_exec;
-  task_executor*              ru_tx_exec;
-  task_executor*              ru_rx_exec;
-  task_executor*              test_du_sim_exec;
-  task_executor*              test_ru_sim_exec;
+  task_execution_manager exec_mng;
+  task_executor*         ru_timing_exec = nullptr;
+  task_executor*         ru_dl_exec;
+  task_executor*         ru_tx_exec;
+  task_executor*         ru_rx_exec;
+  task_executor*         test_du_sim_exec;
+  task_executor*         test_ru_sim_exec;
 };
 } // namespace
 
@@ -964,7 +964,7 @@ static ru_ofh_dependencies generate_ru_dependencies(srslog::basic_logger&
   tx_gateway              = gateway.get();
   sector_deps.eth_gateway = std::move(gateway);
   // Configure executors.
-  sector_deps.downlink_executors   = workers.ru_dl_exec;
+  sector_deps.downlink_executor    = workers.ru_dl_exec;
   sector_deps.transmitter_executor = workers.ru_tx_exec;
   sector_deps.receiver_executor    = workers.ru_rx_exec;
 

Original file line number	Diff line number	Diff line change
`@@ -143,7 +143,7 @@ static void configure_ru_ofh_executors_and_notifiers(ru_ofh_configuration&`
`143`	`143`	`sector_deps.logger = dependencies.logger;`
`144`	`144`	`sector_deps.receiver_executor = workers.ru_rx_exec[i];`
`145`	`145`	`sector_deps.transmitter_executor = workers.ru_tx_exec[i];`
`146`		`- sector_deps.downlink_executors = workers.ru_dl_exec[i];`
	`146`	`+ sector_deps.downlink_executor = workers.ru_dl_exec[i];`
`147`	`147`	`}`
`148`	`148`	`}`
`149`	`149`
Original file line number	Diff line number	Diff line change
`@@ -404,7 +404,7 @@ void worker_manager::create_ofh_executors(span<const cell_appconfig> cells, bool`
`404`	`404`	`cpu_masks.push_back(affinity_mng.calcute_affinity_mask(gnb_sched_affinity_mask_types::ru));`
`405`	`405`	`}`
`406`	`406`	`create_worker_pool(name, nof_ofh_dl_workers, task_worker_queue_size, {{exec_name}}, prio, cpu_masks);`
`407`		`- ru_dl_exec[i].push_back(exec_mng.executors().at(exec_name));`
	`407`	`+ ru_dl_exec[i] = exec_mng.executors().at(exec_name);`
`408`	`408`	`}`
`409`	`409`
`410`	`410`	`// Executor for Open Fronthaul messages transmission.`