gnb: add configurable thread affinity profile

Author: sauka

apps/gnb/gnb_appconfig.h

@@ -755,6 +755,16 @@ struct buffer_pool_appconfig {
   std::size_t segment_size = 1024;
 };
 
+/// Expert configuration of the gNB app.
+struct expert_appconfig {
+  /// Enables usage of affinity profile tuned for higher performance.
+  bool enable_tuned_affinity_profile = false;
+  /// Number of threads per physical CPU.
+  unsigned nof_threads_per_cpu = 2;
+  /// Number of CPU cores reserved for non-priority tasks.
+  unsigned nof_cores_for_non_prio_workers = 4;
+};
+
 /// Monolithic gnb application configuration.
 struct gnb_appconfig {
   /// Logging configuration.
@@ -802,6 +812,9 @@ struct gnb_appconfig {
 
   /// \brief Buffer pool configuration.
   buffer_pool_appconfig buffer_pool_config;
+
+  /// \brief Expert configuration.
+  expert_appconfig expert_config;
 };
 
 } // namespace srsran

apps/gnb/gnb_appconfig_cli11_schema.cpp

@@ -1402,6 +1402,22 @@ static void parse_buffer_pool_config(CLI::App& app, buffer_pool_appconfig& confi
       ->capture_default_str();
 }
 
+static void parse_expert_config(CLI::App& app, expert_appconfig& config)
+{
+  app.add_option("--enable_tuned_affinity_profile",
+                 config.enable_tuned_affinity_profile,
+                 "Enable usage of tuned affinity profile")
+      ->capture_default_str();
+  app.add_option("--number_of_threads_per_cpu", config.nof_threads_per_cpu, "Number of threads per physical CPU")
+      ->capture_default_str()
+      ->check(CLI::Range(1, 2));
+  app.add_option("--number_of_reserved_cores",
+                 config.nof_cores_for_non_prio_workers,
+                 "Number of CPU cores reserved for non-priority tasks")
+      ->capture_default_str()
+      ->check(CLI::Range(0, 1024));
+}
+
 void srsran::configure_cli11_with_gnb_appconfig_schema(CLI::App& app, gnb_appconfig& gnb_cfg)
 {
   app.add_option("--gnb_id", gnb_cfg.gnb_id, "gNodeB identifier")->capture_default_str();
@@ -1513,4 +1529,8 @@ void srsran::configure_cli11_with_gnb_appconfig_schema(CLI::App& app, gnb_appcon
   // Test mode section.
   CLI::App* test_mode_subcmd = app.add_subcommand("test_mode", "Test mode configuration")->configurable();
   configure_cli11_test_mode_args(*test_mode_subcmd, gnb_cfg.test_mode_cfg);
+
+  // Expert section.
+  CLI::App* expert_subcmd = app.add_subcommand("expert", "Expert configuration")->configurable();
+  parse_expert_config(*expert_subcmd, gnb_cfg.expert_config);
 }

apps/gnb/gnb_worker_manager.cpp

@@ -27,6 +27,19 @@ worker_manager::worker_manager(const gnb_appconfig& appcfg)
     is_blocking_mode_active         = sdr_cfg.device_driver == "zmq";
   }
 
+  if (appcfg.expert_config.enable_tuned_affinity_profile) {
+    use_tuned_profile = true;
+    affinity_manager  = std::make_unique<affinity_mask_manager>(appcfg.expert_config.nof_threads_per_cpu,
+                                                               appcfg.expert_config.nof_cores_for_non_prio_workers);
+  } else {
+    affinity_manager = std::make_unique<affinity_mask_manager>();
+  }
+
+  // If an OFH RU is configured, create its executors first.
+  if (variant_holds_alternative<ru_ofh_appconfig>(appcfg.ru_cfg)) {
+    create_ofh_executors(appcfg.cells_cfg, variant_get<ru_ofh_appconfig>(appcfg.ru_cfg).is_downlink_parallelized);
+  }
+
   create_du_cu_executors(is_blocking_mode_active,
                          appcfg.expert_phy_cfg.nof_ul_threads,
                          appcfg.expert_phy_cfg.nof_dl_threads,
@@ -62,16 +75,26 @@ void worker_manager::create_worker(const std::string& name, Args&&... args)
   srsran_assert(ret.second, "Unable to create worker {}.", name);
 }
 
-void worker_manager::create_worker_pool(const std::string&          name,
-                                        size_t                      nof_workers,
-                                        size_t                      queue_size,
-                                        os_thread_realtime_priority prio)
+void worker_manager::create_worker_pool(const std::string&                    name,
+                                        size_t                                nof_workers,
+                                        size_t                                queue_size,
+                                        os_thread_realtime_priority           prio,
+                                        span<const os_sched_affinity_bitmask> cpu_masks)
 {
   auto ret = worker_pools.insert(
-      std::make_pair(name, std::make_unique<task_worker_pool>(nof_workers, queue_size, name, prio)));
+      std::make_pair(name, std::make_unique<task_worker_pool>(nof_workers, queue_size, name, prio, cpu_masks)));
   srsran_assert(ret.second, "Unable to create worker pool {}.", name);
 }
 
+void worker_manager::create_prio_worker(const std::string& name, size_t queue_size, os_thread_realtime_priority prio)
+{
+  os_sched_affinity_bitmask cpu_mask;
+  if (use_tuned_profile) {
+    cpu_mask = affinity_manager->reserve_cpu(name, prio);
+  }
+  create_worker(name, queue_size, prio, cpu_mask);
+}
+
 void worker_manager::create_du_cu_executors(bool                       is_blocking_mode_active,
                                             unsigned                   nof_ul_workers,
                                             unsigned                   nof_dl_workers,
@@ -80,17 +103,24 @@ void worker_manager::create_du_cu_executors(bool                       is_blocki
                                             unsigned                   pipeline_depth)
 {
   // Instantiate workers
-  create_worker("gnb_ue", 512);
-  gnb_ctrl_worker = std::make_unique<gnb_ctrl_worker_type>("gnb_ctrl",
-                                                           std::array<unsigned, 2>{64, task_worker_queue_size},
-                                                           std::chrono::microseconds{200},
-                                                           os_thread_realtime_priority::max() - 20,
-                                                           os_sched_affinity_bitmask{});
-  du_cell_worker  = std::make_unique<du_cell_worker_type>("du_cell",
+  create_prio_worker("gnb_ue", 512);
+  os_sched_affinity_bitmask cpu_mask;
+  if (use_tuned_profile) {
+    cpu_mask = affinity_manager->reserve_cpu("gnb_ctrl", os_thread_realtime_priority::max() - 20);
+  }
+  gnb_ctrl_worker = std::make_unique<du_cell_worker_type>("gnb_ctrl",
+                                                          std::array<unsigned, 2>{64, task_worker_queue_size},
+                                                          std::chrono::microseconds{200},
+                                                          os_thread_realtime_priority::max() - 20,
+                                                          cpu_mask);
+  if (use_tuned_profile) {
+    cpu_mask = affinity_manager->reserve_cpu("du_cell", os_thread_realtime_priority::max() - 2);
+  }
+  du_cell_worker = std::make_unique<du_cell_worker_type>("du_cell",
                                                          std::array<unsigned, 2>{8, task_worker_queue_size},
                                                          std::chrono::microseconds{10},
                                                          os_thread_realtime_priority::max() - 2,
-                                                         os_sched_affinity_bitmask{});
+                                                         cpu_mask);
 
   // Instantiate task executors
   cu_cp_exec    = make_priority_task_executor_ptr<task_queue_priority::min>(*gnb_ctrl_worker);
@@ -151,26 +181,33 @@ void worker_manager::create_du_low_executors(bool                       is_block
   for (unsigned cell_id = 0, cell_end = cells_cfg.size(); cell_id != cell_end; ++cell_id) {
     if (is_blocking_mode_active) {
       if (cell_id == 0) {
-        create_worker("phy_worker", task_worker_queue_size, os_thread_realtime_priority::max());
+        create_prio_worker("phy_worker", task_worker_queue_size, os_thread_realtime_priority::max());
       }
       task_worker& phy_worker = *workers.at("phy_worker");
       upper_pusch_exec.push_back(std::make_unique<task_worker_executor>(phy_worker));
       upper_pucch_exec.push_back(std::make_unique<task_worker_executor>(phy_worker));
       upper_prach_exec.push_back(std::make_unique<task_worker_executor>(phy_worker));
       du_low_dl_executors[cell_id].emplace_back(std::make_unique<task_worker_executor>(phy_worker));
     } else {
-      const std::string& name_ul = "up_phy_ul#" + std::to_string(cell_id);
-      create_worker_pool(name_ul, nof_ul_workers, task_worker_queue_size, os_thread_realtime_priority::max() - 20);
+      const std::string&                     name_ul = "up_phy_ul#" + std::to_string(cell_id);
+      const auto                             prio    = os_thread_realtime_priority::max() - 20;
+      std::vector<os_sched_affinity_bitmask> cpu_masks;
+      if (use_tuned_profile) {
+        for (unsigned w = 0; w != nof_ul_workers; ++w) {
+          cpu_masks.push_back(affinity_manager->reserve_cpu(name_ul, prio));
+        }
+      }
+      create_worker_pool(name_ul, nof_ul_workers, task_worker_queue_size, prio, cpu_masks);
       upper_pusch_exec.push_back(std::make_unique<task_worker_pool_executor>(*worker_pools.at(name_ul)));
       upper_pucch_exec.push_back(std::make_unique<task_worker_pool_executor>(*worker_pools.at(name_ul)));
 
       const std::string& name_prach = "phy_prach#" + std::to_string(cell_id);
-      create_worker(name_prach, task_worker_queue_size, os_thread_realtime_priority::max() - 2);
+      create_prio_worker(name_prach, task_worker_queue_size, os_thread_realtime_priority::max() - 2);
       upper_prach_exec.push_back(std::make_unique<task_worker_executor>(*workers.at(name_prach)));
       for (unsigned i_dl_worker = 0; i_dl_worker != nof_dl_workers; ++i_dl_worker) {
         // Create upper PHY DL executors.
         std::string worker_name = "up_phy_dl#" + std::to_string(cell_id) + "#" + std::to_string(i_dl_worker);
-        create_worker(worker_name, task_worker_queue_size, os_thread_realtime_priority::max() - 10);
+        create_prio_worker(worker_name, task_worker_queue_size, os_thread_realtime_priority::max() - 10);
         du_low_dl_executors[cell_id].emplace_back(std::make_unique<task_worker_executor>(*workers.at(worker_name)));
       }
     }
@@ -182,22 +219,27 @@ void worker_manager::create_du_low_executors(bool                       is_block
             cells_cfg[cell_id].cell.nof_antennas_dl) /
            ldpc::MAX_MESSAGE_SIZE) *
           pipeline_depth;
-      create_worker_pool(name_pdsch, nof_pdsch_workers, max_nof_pdsch_cb_slot, os_thread_realtime_priority::max() - 10);
+
+      const auto                             prio = os_thread_realtime_priority::max() - 10;
+      std::vector<os_sched_affinity_bitmask> cpu_masks;
+      if (use_tuned_profile) {
+        for (unsigned w = 0; w != nof_pdsch_workers; ++w) {
+          cpu_masks.push_back(affinity_manager->reserve_cpu(name_pdsch, prio));
+        }
+      }
+      create_worker_pool(name_pdsch, nof_pdsch_workers, max_nof_pdsch_cb_slot, prio, cpu_masks);
 
       upper_pdsch_exec[cell_id] = std::make_unique<task_worker_pool_executor>(*worker_pools.at(name_pdsch));
     }
   }
 }
 
 /// Returns an affinity bitmask using the given affinity mask manager.
-static os_sched_affinity_bitmask get_affinity_mask(affinity_mask_manager& manager, const std::string& name)
+static os_sched_affinity_bitmask
+get_affinity_mask(affinity_mask_manager& manager, const std::string& name, unsigned priority_from_max)
 {
-  auto cpu_idx = manager.reserve_cpu_index();
-  if (cpu_idx.is_error()) {
-    fmt::print("Could not set the affinity for the {} executor\n", name);
-  }
-
-  return (cpu_idx.has_value()) ? os_sched_affinity_bitmask(cpu_idx.value()) : os_sched_affinity_bitmask();
+  const os_thread_realtime_priority priority = os_thread_realtime_priority::max() - priority_from_max;
+  return manager.reserve_cpu(name, priority);
 }
 
 void worker_manager::create_ofh_executors(span<const cell_appconfig> cells, bool is_downlink_parallelized)
@@ -222,7 +264,7 @@ void worker_manager::create_ofh_executors(span<const cell_appconfig> cells, bool
                                                                     4,
                                                                     std::chrono::microseconds{0},
                                                                     os_thread_realtime_priority::max() - 0,
-                                                                    get_affinity_mask(affinity_manager, name)));
+                                                                    get_affinity_mask(*affinity_manager, name, 0)));
     ru_timing_exec = std::make_unique<
         general_task_worker_executor<concurrent_queue_policy::lockfree_spsc, concurrent_queue_wait_policy::sleep>>(
         *ru_spsc_workers.back());
@@ -237,7 +279,7 @@ void worker_manager::create_ofh_executors(span<const cell_appconfig> cells, bool
       ru_mpsc_workers.push_back(std::make_unique<ru_mpsc_worker_type>(name,
                                                                       task_worker_queue_size,
                                                                       os_thread_realtime_priority::max() - 5,
-                                                                      get_affinity_mask(affinity_manager, name)));
+                                                                      get_affinity_mask(*affinity_manager, name, 5)));
       ru_dl_exec[i].push_back(make_task_executor(*ru_mpsc_workers.back()));
     }
 
@@ -248,7 +290,7 @@ void worker_manager::create_ofh_executors(span<const cell_appconfig> cells, bool
                                                                       task_worker_queue_size,
                                                                       std::chrono::microseconds{1},
                                                                       os_thread_realtime_priority::max() - 1,
-                                                                      get_affinity_mask(affinity_manager, name)));
+                                                                      get_affinity_mask(*affinity_manager, name, 1)));
       ru_tx_exec.push_back(make_task_executor(*ru_spsc_workers.back()));
     }
 
@@ -259,7 +301,7 @@ void worker_manager::create_ofh_executors(span<const cell_appconfig> cells, bool
                                                                       2,
                                                                       std::chrono::microseconds{1},
                                                                       os_thread_realtime_priority::max() - 1,
-                                                                      get_affinity_mask(affinity_manager, name)));
+                                                                      get_affinity_mask(*affinity_manager, name, 1)));
       ru_rx_exec.push_back(make_task_executor(*ru_spsc_workers.back()));
     }
   }
@@ -268,11 +310,11 @@ void worker_manager::create_ofh_executors(span<const cell_appconfig> cells, bool
 void worker_manager::create_lower_phy_executors(lower_phy_thread_profile lower_phy_profile, unsigned nof_cells)
 {
   // Radio Unit worker and executor.
-  create_worker("radio", task_worker_queue_size);
+  create_prio_worker("radio", task_worker_queue_size);
   radio_exec = std::make_unique<task_worker_executor>(*workers.at("radio"));
 
   // Radio Unit statistics worker and executor.
-  create_worker("ru_stats_worker", 1);
+  create_prio_worker("ru_stats_worker", 1);
   ru_printer_exec = std::make_unique<task_worker_executor>(*workers.at("ru_stats_worker"));
 
   for (unsigned cell_id = 0; cell_id != nof_cells; ++cell_id) {
@@ -296,7 +338,7 @@ void worker_manager::create_lower_phy_executors(lower_phy_thread_profile lower_p
       case lower_phy_thread_profile::single: {
         fmt::print("Lower PHY in single executor mode.\n");
         const std::string& name = "lower_phy#" + std::to_string(cell_id);
-        create_worker(name, 128, os_thread_realtime_priority::max());
+        create_prio_worker(name, 128, os_thread_realtime_priority::max());
         task_worker& lower_phy_worker = *workers.at(name);
         lower_phy_tx_exec.push_back(std::make_unique<task_worker_executor>(lower_phy_worker));
         lower_phy_rx_exec.push_back(std::make_unique<task_worker_executor>(lower_phy_worker));
@@ -311,8 +353,8 @@ void worker_manager::create_lower_phy_executors(lower_phy_thread_profile lower_p
         fmt::print("Lower PHY in dual executor mode.\n");
         const std::string& name_dl = "lower_phy_dl#" + std::to_string(cell_id);
         const std::string& name_ul = "lower_phy_ul#" + std::to_string(cell_id);
-        create_worker(name_dl, 128, os_thread_realtime_priority::max());
-        create_worker(name_ul, 2, os_thread_realtime_priority::max() - 1);
+        create_prio_worker(name_dl, 128, os_thread_realtime_priority::max());
+        create_prio_worker(name_ul, 2, os_thread_realtime_priority::max() - 1);
         lower_phy_tx_exec.push_back(std::make_unique<task_worker_executor>(*workers.at(name_dl)));
         lower_phy_rx_exec.push_back(std::make_unique<task_worker_executor>(*workers.at(name_ul)));
         lower_phy_dl_exec.push_back(std::make_unique<task_worker_executor>(*workers.at(name_dl)));
@@ -328,10 +370,10 @@ void worker_manager::create_lower_phy_executors(lower_phy_thread_profile lower_p
         const std::string& name_ul = "lower_phy_ul#" + std::to_string(cell_id);
         const std::string& name_tx = "lower_phy_tx#" + std::to_string(cell_id);
         const std::string& name_rx = "lower_phy_rx#" + std::to_string(cell_id);
-        create_worker(name_tx, 128, os_thread_realtime_priority::max());
-        create_worker(name_rx, 1, os_thread_realtime_priority::max() - 2);
-        create_worker(name_dl, 128, os_thread_realtime_priority::max() - 1);
-        create_worker(name_ul, 128, os_thread_realtime_priority::max() - 3);
+        create_prio_worker(name_tx, 128, os_thread_realtime_priority::max());
+        create_prio_worker(name_rx, 1, os_thread_realtime_priority::max() - 2);
+        create_prio_worker(name_dl, 128, os_thread_realtime_priority::max() - 1);
+        create_prio_worker(name_ul, 128, os_thread_realtime_priority::max() - 3);
         lower_phy_tx_exec.push_back(std::make_unique<task_worker_executor>(*workers.at(name_tx)));
         lower_phy_rx_exec.push_back(std::make_unique<task_worker_executor>(*workers.at(name_rx)));
         lower_phy_dl_exec.push_back(std::make_unique<task_worker_executor>(*workers.at(name_dl)));
@@ -348,8 +390,6 @@ void worker_manager::create_lower_phy_executors(lower_phy_thread_profile lower_p
 void worker_manager::create_ru_executors(const gnb_appconfig& appcfg)
 {
   if (variant_holds_alternative<ru_ofh_appconfig>(appcfg.ru_cfg)) {
-    create_ofh_executors(appcfg.cells_cfg, variant_get<ru_ofh_appconfig>(appcfg.ru_cfg).is_downlink_parallelized);
-
     return;
   }