app: remove unused features of task_executor_manager

Author: frankist

apps/examples/ofh/ru_emulator.cpp

@@ -851,8 +851,7 @@ struct worker_manager {
         const std::string exec_name = "ru_rx_exec_#" + std::to_string(i);
 
         const single_worker ru_worker{name,
-                                      {concurrent_queue_policy::lockfree_spsc, 2},
-                                      {{exec_name}},
+                                      {exec_name, concurrent_queue_policy::lockfree_spsc, 2},
                                       std::chrono::microseconds{1},
                                       os_thread_realtime_priority::max() - 1};
         if (!exec_mng.add_execution_context(create_execution_context(ru_worker))) {
@@ -866,8 +865,7 @@ struct worker_manager {
         const std::string   name      = "ru_emu_#" + std::to_string(i);
         const std::string   exec_name = "ru_emu_exec_#" + std::to_string(i);
         const single_worker ru_worker{name,
-                                      {concurrent_queue_policy::lockfree_spsc, task_worker_queue_size},
-                                      {{exec_name}},
+                                      {exec_name, concurrent_queue_policy::lockfree_spsc, task_worker_queue_size},
                                       std::chrono::microseconds{1},
                                       os_thread_realtime_priority::max() - 1};
         if (!exec_mng.add_execution_context(create_execution_context(ru_worker))) {
@@ -883,8 +881,7 @@ struct worker_manager {
       const std::string exec_name = "ru_timing_exec";
 
       const single_worker ru_worker{name,
-                                    {concurrent_queue_policy::lockfree_spsc, 4},
-                                    {{exec_name}},
+                                    {exec_name, concurrent_queue_policy::lockfree_spsc, 4},
                                     std::chrono::microseconds{1},
                                     os_thread_realtime_priority::max() - 0};
       if (!exec_mng.add_execution_context(create_execution_context(ru_worker))) {

apps/services/worker_manager/worker_manager.cpp

@@ -152,17 +152,17 @@ struct worker_manager {
   std::vector<os_sched_affinity_manager> affinity_mng;
 
   /// Helper method to create workers with non zero priority.
-  void create_prio_worker(const std::string&                                    name,
-                          unsigned                                              queue_size,
-                          const std::vector<execution_config_helper::executor>& execs,
-                          const os_sched_affinity_bitmask&                      mask,
-                          os_thread_realtime_priority prio = os_thread_realtime_priority::no_realtime());
+  void create_prio_worker(const std::string&               name,
+                          const std::string&               exec_name,
+                          unsigned                         queue_size,
+                          const os_sched_affinity_bitmask& mask,
+                          os_thread_realtime_priority      prio = os_thread_realtime_priority::no_realtime());
 
   /// Helper method to create worker pool.
-  void create_worker_pool(const std::string&                                    name,
-                          unsigned                                              nof_workers,
-                          unsigned                                              queue_size,
-                          const std::vector<execution_config_helper::executor>& execs,
+  void create_worker_pool(const std::string&                    name,
+                          unsigned                              nof_workers,
+                          const std::string&                    exec_name,
+                          unsigned                              queue_size,
                           os_thread_realtime_priority           prio      = os_thread_realtime_priority::no_realtime(),
                           span<const os_sched_affinity_bitmask> cpu_masks = {},
                           concurrent_queue_policy               queue_policy = concurrent_queue_policy::locking_mpmc);

apps/services/worker_manager/worker_manager.h

@@ -11,6 +11,7 @@
 #pragma once
 
 #include <limits>
+#include <optional>
 
 namespace srsran {
 
@@ -46,6 +47,8 @@ struct concurrent_queue_params {
   concurrent_queue_policy policy;
   /// Task queue size.
   unsigned size;
+  /// In case of moodycamel MPMC, the number of pre-reserved producers can be set.
+  unsigned nof_prereserved_producers = 2;
 };
 
 /// \brief Queue priority used to map to specific queue of the \c priority_multiqueue_task_worker. The higher the

include/srsran/adt/detail/concurrent_queue_params.h

@@ -160,6 +160,12 @@ class concurrent_queue<T, concurrent_queue_policy::moodycamel_lockfree_mpmc, con
     blocking_ext_base(sleep_time_), non_block_queue_base(qsize)
   {
   }
+  explicit concurrent_queue(size_t                    qsize,
+                            size_t                    nof_prereserved_producers = 2,
+                            std::chrono::microseconds sleep_time_               = std::chrono::microseconds{0}) :
+    blocking_ext_base(sleep_time_), non_block_queue_base(qsize, nof_prereserved_producers)
+  {
+  }
 
   /// Creates a sequential consumer for this queue.
   consumer_type create_consumer() { return consumer_type(*this); }

include/srsran/adt/moodycamel_mpmc_queue.h

@@ -26,58 +26,15 @@ namespace execution_config_helper {
 using task_priority = enqueue_priority;
 
 /// Parameters of a queue of tasks.
-using task_queue = concurrent_queue_params;
-
-/// Parameters of a strand executor.
-struct strand {
-  struct executor {
-    /// \brief Name of the strand executor.
-    std::string name;
-    /// \brief Queueing policy associated with this strand executor.
-    concurrent_queue_policy policy;
-    /// \brief Size of the queue used.
-    unsigned size;
-    /// \brief Whether the caller blocks waiting for task to complete.
-    bool synchronous = false;
-  };
-  /// Queues of different priorities. The lower the index, the higher the priority.
-  std::vector<executor> queues;
-};
-
-/// Parameters of the task executor, including name and decorators.
-struct executor {
-  /// Name of the executor.
+struct task_queue {
+  /// Name attributed to this task queue.
   std::string name;
-  /// Priority assigned to the tasks dispatched through this executor.
-  task_priority priority = task_priority::min;
-  /// Strands instantiated on top of this executor.
-  std::vector<strand> strands;
-  /// \brief Present if the executor works as a strand, serializing all the enqueued tasks. The value is the size of
-  /// the strand queue size.
-  std::optional<unsigned> strand_queue_size;
-  /// \brief Whether to make an executor synchronous. If true, the executor will be blocking, until the pushed task is
-  /// fully executed. This will have a negative impact on performance, but can be useful for debugging.
-  bool synchronous = false;
-
-  executor(const std::string&         name_,
-           const std::vector<strand>& strands_           = {},
-           std::optional<unsigned>    strand_queue_size_ = std::nullopt,
-           bool                       synchronous_       = false) :
-    name(name_), strands(strands_), strand_queue_size(strand_queue_size_), synchronous(synchronous_)
-  {
-  }
-  executor(const std::string&         name_,
-           task_priority              priority_,
-           const std::vector<strand>& strands_           = {},
-           std::optional<unsigned>    strand_queue_size_ = std::nullopt,
-           bool                       synchronous_       = false) :
-    name(name_),
-    priority(priority_),
-    strands(strands_),
-    strand_queue_size(strand_queue_size_),
-    synchronous(synchronous_)
-  {
-  }
+  /// Queueing policy used by this task queue.
+  concurrent_queue_policy policy;
+  /// Size of the queue used.
+  unsigned size;
+  /// Number of pre-reserved producers in the case of the moodycamel lockfree MPMC queue.
+  unsigned nof_prereserved_producers = 2;
 };
 
 /// Arguments for a single task worker creation.
@@ -86,17 +43,13 @@ struct single_worker {
   std::string name;
   /// Queue used by the task worker.
   task_queue queue;
-  /// Executors associated with this execution context.
-  std::vector<executor> executors;
   /// \brief Wait time in microseconds, when task queue has no pending tasks. If not set, a condition variable is
   /// used to wake up the worker when a new task is pushed.
   std::optional<std::chrono::microseconds> wait_sleep_time;
   /// OS priority of the worker thread.
   os_thread_realtime_priority prio = os_thread_realtime_priority::no_realtime();
   /// Bit mask to set worker cpu affinity.
   os_sched_affinity_bitmask mask = {};
-  /// Non null in case tracing of the worker executors is enabled.
-  file_event_tracer<true>* tracer = nullptr;
 };
 
 /// Arguments for a task worker pool creation.
@@ -107,16 +60,12 @@ struct worker_pool {
   unsigned nof_workers;
   /// Queue(s) used by the task worker. The lower the index, the higher the priority.
   std::vector<task_queue> queues;
-  /// Executors associated with this execution context.
-  std::vector<executor> executors;
   /// \brief Wait time in microseconds, when task queue has no pending tasks.
   std::chrono::microseconds sleep_time;
   /// OS priority of the worker thread.
   os_thread_realtime_priority prio = os_thread_realtime_priority::no_realtime();
   /// Array of CPU bitmasks to assign to each worker in the pool.
   std::vector<os_sched_affinity_bitmask> masks;
-  /// Non null in case tracing of the worker executors is enabled.
-  file_event_tracer<true>* tracer = nullptr;
 };
 
 /// Arguments for the creation of a priority multiqueue worker.
@@ -128,14 +77,10 @@ struct priority_multiqueue_worker {
   std::vector<task_queue> queues;
   /// \brief Wait time in microseconds, when task queue has no pending tasks.
   std::chrono::microseconds spin_sleep_time;
-  /// Executors associated with this execution context.
-  std::vector<executor> executors;
   /// OS priority of the worker thread.
   os_thread_realtime_priority prio = os_thread_realtime_priority::no_realtime();
   /// Bit mask to set worker cpu affinity.
   os_sched_affinity_bitmask mask = {};
-  /// Non null in case tracing of the worker executors is enabled.
-  file_event_tracer<true>* tracer = nullptr;
 };
 
 } // namespace execution_config_helper

include/srsran/support/executors/task_execution_manager.h

@@ -73,7 +73,10 @@ template <>
 class base_task_queue<concurrent_queue_policy::lockfree_mpmc>
 {
 protected:
-  base_task_queue(size_t qsize, std::chrono::microseconds wait_sleep_time) : queue(qsize, wait_sleep_time) {}
+  base_task_queue(size_t qsize, std::chrono::microseconds wait_sleep_time, unsigned /* unused */) :
+    queue(qsize, wait_sleep_time)
+  {
+  }
 
   // Queue of pending tasks.
   concurrent_queue<unique_task, concurrent_queue_policy::lockfree_mpmc, concurrent_queue_wait_policy::sleep> queue;
@@ -83,9 +86,7 @@ template <>
 class base_task_queue<concurrent_queue_policy::locking_mpmc>
 {
 protected:
-  base_task_queue(size_t qsize, std::chrono::microseconds wait_sleep_time = std::chrono::microseconds{0}) : queue(qsize)
-  {
-  }
+  base_task_queue(size_t qsize, std::chrono::microseconds wait_sleep_time, unsigned /* unused */) : queue(qsize) {}
 
   // Queue of pending tasks.
   concurrent_queue<unique_task, concurrent_queue_policy::locking_mpmc, concurrent_queue_wait_policy::condition_variable>
@@ -96,8 +97,8 @@ template <>
 class base_task_queue<concurrent_queue_policy::moodycamel_lockfree_mpmc>
 {
 protected:
-  base_task_queue(size_t qsize, std::chrono::microseconds wait_sleep_time = std::chrono::microseconds{0}) :
-    queue(qsize, wait_sleep_time)
+  base_task_queue(size_t qsize, std::chrono::microseconds wait_sleep_time, unsigned nof_prereserved_producers) :
+    queue(qsize, nof_prereserved_producers, wait_sleep_time)
   {
   }
 
@@ -175,9 +176,10 @@ class task_worker_pool : public detail::base_task_queue<QueuePolicy>, public det
   task_worker_pool(std::string                           worker_pool_name,
                    unsigned                              nof_workers_,
                    unsigned                              qsize_,
-                   std::chrono::microseconds             wait_sleep_time = std::chrono::microseconds{100},
-                   os_thread_realtime_priority           prio            = os_thread_realtime_priority::no_realtime(),
-                   span<const os_sched_affinity_bitmask> cpu_masks       = {});
+                   std::chrono::microseconds             wait_sleep_time           = std::chrono::microseconds{100},
+                   unsigned                              nof_prereserved_producers = 2,
+                   os_thread_realtime_priority           prio      = os_thread_realtime_priority::no_realtime(),
+                   span<const os_sched_affinity_bitmask> cpu_masks = {});
   ~task_worker_pool();
 
   /// \brief Push a new task to be processed by the worker pool. If the task queue is full, it skips the task and

include/srsran/support/executors/task_worker_pool.h

@@ -106,7 +106,6 @@ class any_task_queue_impl final : public detail::any_task_queue
   queue_impl q;
 };
 
-template <typename... Args>
 std::unique_ptr<detail::any_task_queue> make_any_task_queue(const concurrent_queue_params& params)
 {
   switch (params.policy) {
@@ -121,7 +120,8 @@ std::unique_ptr<detail::any_task_queue> make_any_task_queue(const concurrent_que
           any_task_queue_impl<concurrent_queue_policy::locking_mpmc, concurrent_queue_wait_policy::condition_variable>>(
           params.size);
     case concurrent_queue_policy::moodycamel_lockfree_mpmc:
-      return std::make_unique<any_task_queue_impl<concurrent_queue_policy::moodycamel_lockfree_mpmc>>(params.size);
+      return std::make_unique<any_task_queue_impl<concurrent_queue_policy::moodycamel_lockfree_mpmc>>(
+          params.size, params.nof_prereserved_producers);
     default:
       report_fatal_error("Unknown concurrent_queue_policy");
   }