du-high: make timer executor synchronous when in zmq mode

Author: frankist

apps/services/worker_manager.cpp

@@ -323,7 +323,7 @@ void worker_manager::create_low_prio_executors(const expert_execution_appconfig&
   // Used for PCAP writing.
   non_rt_pool.executors.emplace_back("low_prio_exec", task_priority::max - 1);
   // Used for control plane and timer management.
-  non_rt_pool.executors.push_back({"high_prio_exec", task_priority::max, {}, std::nullopt, not rt_mode});
+  non_rt_pool.executors.push_back({"high_prio_exec", task_priority::max});
   // Used to serialize all CU-UP tasks, while CU-UP does not support multithreading.
   non_rt_pool.executors.push_back({"cu_up_strand",
                                    task_priority::max - 1,
@@ -337,9 +337,11 @@ void worker_manager::create_low_prio_executors(const expert_execution_appconfig&
   // Configuration of strands for PCAP writing. These strands will use the low priority executor.
   append_pcap_strands(low_prio_strands, cu_cp_pcaps, cu_up_pcaps, du_pcaps);
 
-  // Configuration of strand for the control plane handling (CU-CP and DU-high control plane). This strand will
-  // support two priority levels, the highest being for timer management.
-  strand cp_strand{{{"timer_exec", concurrent_queue_policy::lockfree_spsc, task_worker_queue_size},
+  // Configuration of strand for the control plane handling (CU-CP and DU-high control plane).
+  // This strand will support two priority levels, the highest being for timer management.
+  // Note: In case of non-RT operation, we make the timer_exec synchronous. This will have the effect of stopping
+  // the lower layers from running faster than this strand.
+  strand cp_strand{{{"timer_exec", concurrent_queue_policy::lockfree_spsc, task_worker_queue_size, not rt_mode},
                     {"ctrl_exec", concurrent_queue_policy::lockfree_mpmc, task_worker_queue_size}}};
   high_prio_strands.push_back(cp_strand);
 

include/srsran/support/executors/task_execution_manager.h

@@ -38,6 +38,8 @@ struct strand {
     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;

lib/support/executors/task_execution_manager.cpp

@@ -196,15 +196,22 @@ class common_task_execution_context : public task_execution_context
 
         // Create executors that own the strand through reference counting.
         for (unsigned i = 0; i != strand_cfg.queues.size(); ++i) {
-          enqueue_priority prio = detail::queue_index_to_enqueue_priority(i, strand_cfg.queues.size());
-          execs.emplace_back(strand_cfg.queues[i].name, make_priority_task_executor_ptr(prio, shared_strand));
+          enqueue_priority prio      = detail::queue_index_to_enqueue_priority(i, strand_cfg.queues.size());
+          auto             prio_exec = make_priority_task_executor_ptr(prio, shared_strand);
+          if (strand_cfg.queues[i].synchronous) {
+            prio_exec = make_sync_executor(std::move(prio_exec));
+          }
+          execs.emplace_back(strand_cfg.queues[i].name, std::move(prio_exec));
         }
       } else {
         // Single priority level case.
         concurrent_queue_params qparams;
         qparams.policy  = strand_cfg.queues[0].policy;
         qparams.size    = strand_cfg.queues[0].size;
         auto strand_ptr = make_task_strand_ptr(exec_type{basic_exec}, qparams);
+        if (strand_cfg.queues[0].synchronous) {
+          strand_ptr = make_sync_executor(std::move(strand_ptr));
+        }
 
         // Strand becomes the executor.
         execs.emplace_back(strand_cfg.queues[0].name, std::move(strand_ptr));