f1ap-du: use blocking execute_on based on timers

Author: frankist

include/srsran/adt/noop_functor.h

@@ -10,6 +10,10 @@ struct noop_operation {
   {
     // Do nothing.
   }
+  void operator()() const
+  {
+    // Do nothing.
+  }
 };
 
 } // namespace srsran

include/srsran/f1ap/du/f1ap_du_factory.h

@@ -6,6 +6,7 @@
 namespace srsran {
 
 class du_high_ue_executor_mapper;
+class timer_manager;
 
 namespace srs_du {
 
@@ -16,7 +17,8 @@ std::unique_ptr<f1ap_du> create_f1ap(f1c_connection_client&      f1c_client_hand
                                      f1ap_du_configurator&       du_mng,
                                      task_executor&              ctrl_exec,
                                      du_high_ue_executor_mapper& ue_exec_mapper,
-                                     f1ap_du_paging_notifier&    paging_notifier);
+                                     f1ap_du_paging_notifier&    paging_notifier,
+                                     srsran::timer_manager&      timers);
 
 } // namespace srs_du
 } // namespace srsran

include/srsran/f1ap/du/f1c_bearer.h

@@ -77,10 +77,5 @@ class f1c_rx_pdu_handler
 class f1c_bearer : public f1c_tx_sdu_handler, public f1c_tx_delivery_handler, public f1c_rx_pdu_handler
 {};
 
-std::unique_ptr<f1ap_du> create_f1ap(f1c_connection_client&      f1c_client_handler,
-                                     f1ap_du_configurator&       du_mng,
-                                     task_executor&              ctrl_exec,
-                                     du_high_ue_executor_mapper& ue_exec_mapper,
-                                     f1ap_du_paging_notifier&    paging_notifier);
 } // namespace srs_du
 } // namespace srsran

include/srsran/support/async/execute_on_blocking.h

@@ -10,18 +10,22 @@
 
 #pragma once
 
+#include "srsran/adt/noop_functor.h"
 #include "srsran/support/async/execute_on.h"
 #include "srsran/support/timers.h"
 
 namespace srsran {
 
 namespace detail {
 
-template <typename TaskExecutor, bool IsExecute>
-auto dispatch_on_blocking(TaskExecutor& exec, timer_manager& timers)
+template <typename TaskExecutor, typename OnFailureToDispatch, bool IsExecute>
+auto dispatch_on_blocking(TaskExecutor& exec, timer_manager& timers, OnFailureToDispatch&& on_failure)
 {
   struct blocking_dispatch_on_awaiter {
-    blocking_dispatch_on_awaiter(TaskExecutor& exec_, timer_manager& timers_) : exec(exec_), timers(timers_) {}
+    blocking_dispatch_on_awaiter(TaskExecutor& exec_, timer_manager& timers_, OnFailureToDispatch&& on_failure_) :
+      exec(exec_), timers(timers_), on_failure(std::forward<OnFailureToDispatch>(on_failure_))
+    {
+    }
 
     bool await_ready() noexcept { return false; }
 
@@ -41,6 +45,8 @@ auto dispatch_on_blocking(TaskExecutor& exec, timer_manager& timers)
       }
 
       // Task execute/defer failed (potentially because task executor queue is full).
+      on_failure();
+
       // Leverage timer infrastructure to run task.
       // Note: Even if the timer expiry fails to invoke task in executor, it keeps trying on every tick.
       retry_timer = timers.create_unique_timer(exec);
@@ -54,56 +60,70 @@ auto dispatch_on_blocking(TaskExecutor& exec, timer_manager& timers)
     blocking_dispatch_on_awaiter& get_awaiter() { return *this; }
 
   private:
-    TaskExecutor&  exec;
-    timer_manager& timers;
-    unique_timer   retry_timer;
+    TaskExecutor&       exec;
+    timer_manager&      timers;
+    OnFailureToDispatch on_failure;
+    unique_timer        retry_timer;
   };
 
-  return blocking_dispatch_on_awaiter{exec, timers};
+  return blocking_dispatch_on_awaiter{exec, timers, std::forward<OnFailureToDispatch>(on_failure)};
 }
 
 } // namespace detail
 
 /// \brief Returns an awaitable that resumes the suspended coroutine in a different execution context. If the call
 /// to execute fails, the awaitable yields and will retry the dispatch at a later point, until it succeeds.
-template <typename TaskExecutor>
-auto execute_on_blocking(TaskExecutor& exec, timer_manager& timers)
+/// \param[in] exec Executor used to dispatch coroutine to a new execution context.
+/// \param[in] timers Timer service used to handle reattempts to dispatch task to new execution context.
+/// \param[in] on_failure Callback invoked in case the dispatch to executor fails at first attempt.
+template <typename TaskExecutor, typename OnFailureToDispatch = noop_operation>
+auto execute_on_blocking(TaskExecutor& exec, timer_manager& timers, OnFailureToDispatch&& on_failure = noop_operation{})
 {
-  return detail::dispatch_on_blocking<TaskExecutor, true>(exec, timers);
+  return detail::dispatch_on_blocking<TaskExecutor, OnFailureToDispatch, true>(
+      exec, timers, std::forward<OnFailureToDispatch>(on_failure));
 }
 
 /// \brief Returns an awaitable that resumes the suspended coroutine in a different execution context. If the call
 /// to defer fails, the awaitable yields and will retry the dispatch at a later point, until it succeeds.
-template <typename TaskExecutor>
-auto defer_on_blocking(TaskExecutor& exec, timer_manager& timers)
+/// \param[in] exec Executor used to dispatch coroutine to a new execution context.
+/// \param[in] timers Timer service used to handle reattempts to dispatch task to new execution context.
+/// \param[in] on_failure Callback invoked in case the dispatch to executor fails at first attempt.
+template <typename TaskExecutor, typename OnFailureToDispatch = noop_operation>
+auto defer_on_blocking(TaskExecutor& exec, timer_manager& timers, OnFailureToDispatch&& on_failure = noop_operation{})
 {
-  return detail::dispatch_on_blocking<TaskExecutor, false>(exec, timers);
+  return detail::dispatch_on_blocking<TaskExecutor, OnFailureToDispatch, false>(
+      exec, timers, std::forward<OnFailureToDispatch>(on_failure));
 }
 
 /// \brief Returns an async_task<void> that runs a given invocable task in a \c dispatch_exec executor, and once the
 /// task is complete, it resumes the suspended coroutine in a \c return_exec executor.
 template <typename DispatchTaskExecutor,
           typename CurrentTaskExecutor,
           typename Callable,
-          typename ReturnType = detail::function_return_t<decltype(&Callable::operator())>>
+          typename OnFailureToDispatch = noop_operation,
+          typename ReturnType          = detail::function_return_t<decltype(&Callable::operator())>>
 std::enable_if_t<std::is_same_v<ReturnType, void>, async_task<void>>
 execute_and_continue_on_blocking(DispatchTaskExecutor& dispatch_exec,
                                  CurrentTaskExecutor&  return_exec,
                                  timer_manager&        timers,
-                                 Callable&&            callable)
+                                 Callable&&            callable,
+                                 OnFailureToDispatch&& on_failure = noop_operation{})
 {
-  return launch_async([&return_exec, &dispatch_exec, task = std::forward<Callable>(callable), &timers](
-                          coro_context<async_task<void>>& ctx) mutable {
+  return launch_async([&return_exec,
+                       &dispatch_exec,
+                       task       = std::forward<Callable>(callable),
+                       on_failure = std::forward<OnFailureToDispatch>(on_failure),
+                       &timers](coro_context<async_task<void>>& ctx) mutable {
     CORO_BEGIN(ctx);
 
     // Dispatch execution context switch.
-    CORO_AWAIT(execute_on_blocking(dispatch_exec, timers));
+    CORO_AWAIT(execute_on_blocking(dispatch_exec, timers, on_failure));
 
     // Run task.
     task();
 
     // Continuation in the original executor.
-    CORO_AWAIT(execute_on_blocking(return_exec, timers));
+    CORO_AWAIT(execute_on_blocking(return_exec, timers, on_failure));
 
     CORO_RETURN();
   });
@@ -114,26 +134,32 @@ execute_and_continue_on_blocking(DispatchTaskExecutor& dispatch_exec,
 template <typename DispatchTaskExecutor,
           typename CurrentTaskExecutor,
           typename Callable,
-          typename ReturnType = detail::function_return_t<decltype(&Callable::operator())>>
+          typename OnFailureToDispatch = noop_operation,
+          typename ReturnType          = detail::function_return_t<decltype(&Callable::operator())>>
 std::enable_if_t<not std::is_same_v<ReturnType, void>, async_task<ReturnType>>
 execute_and_continue_on_blocking(DispatchTaskExecutor& dispatch_exec,
                                  CurrentTaskExecutor&  return_exec,
                                  timer_manager&        timers,
-                                 Callable&&            callable)
+                                 Callable&&            callable,
+                                 OnFailureToDispatch&& on_failure = noop_operation{})
 {
   ReturnType ret{};
-  return launch_async([&return_exec, &dispatch_exec, task = std::forward<Callable>(callable), &timers, ret](
-                          coro_context<async_task<ReturnType>>& ctx) mutable {
+  return launch_async([&return_exec,
+                       &dispatch_exec,
+                       task       = std::forward<Callable>(callable),
+                       on_failure = std::forward<OnFailureToDispatch>(on_failure),
+                       &timers,
+                       ret](coro_context<async_task<ReturnType>>& ctx) mutable {
     CORO_BEGIN(ctx);
 
     // Dispatch execution context switch.
-    CORO_AWAIT(execute_on_blocking(dispatch_exec, timers));
+    CORO_AWAIT(execute_on_blocking(dispatch_exec, timers, on_failure));
 
     // Run task.
     ret = task();
 
     // Continuation in the original executor.
-    CORO_AWAIT(execute_on_blocking(return_exec, timers));
+    CORO_AWAIT(execute_on_blocking(return_exec, timers, on_failure));
 
     CORO_RETURN(ret);
   });

lib/du/du_high/adapters/f1ap_test_mode_adapter.cpp

@@ -234,14 +234,15 @@ std::unique_ptr<f1ap_du> srsran::srs_du::create_du_high_f1ap(f1c_connection_clie
                                                              task_executor&              ctrl_exec,
                                                              du_high_ue_executor_mapper& ue_exec_mapper,
                                                              f1ap_du_paging_notifier&    paging_notifier,
+                                                             timer_manager&              timers,
                                                              const du_test_mode_config&  test_cfg)
 {
   if (not test_cfg.test_ue.has_value()) {
-    return create_f1ap(f1c_client_handler, du_mng, ctrl_exec, ue_exec_mapper, paging_notifier);
+    return create_f1ap(f1c_client_handler, du_mng, ctrl_exec, ue_exec_mapper, paging_notifier, timers);
   }
 
   // Create a F1AP test mode adapter that wraps the real F1AP and intercepts messages to the F1-C client.
   auto f1ap_testmode = std::make_unique<f1ap_test_mode_adapter>(*test_cfg.test_ue, f1c_client_handler, ctrl_exec);
-  f1ap_testmode->connect(create_f1ap(*f1ap_testmode, du_mng, ctrl_exec, ue_exec_mapper, paging_notifier));
+  f1ap_testmode->connect(create_f1ap(*f1ap_testmode, du_mng, ctrl_exec, ue_exec_mapper, paging_notifier, timers));
   return f1ap_testmode;
 }

lib/du/du_high/adapters/f1ap_test_mode_adapter.h

@@ -25,6 +25,7 @@ std::unique_ptr<f1ap_du> create_du_high_f1ap(f1c_connection_client&      f1c_cli
                                              task_executor&              ctrl_exec,
                                              du_high_ue_executor_mapper& ue_exec_mapper,
                                              f1ap_du_paging_notifier&    paging_notifier,
+                                             timer_manager&              timers,
                                              const du_test_mode_config&  test_cfg);
 
 } // namespace srs_du

lib/du/du_high/du_high_impl.cpp

@@ -120,6 +120,7 @@ du_high_impl::du_high_impl(const du_high_configuration& config_) :
                              cfg.exec_mapper->du_control_executor(),
                              cfg.exec_mapper->ue_mapper(),
                              adapters->f1ap_paging_notifier,
+                             timers,
                              cfg.test_cfg);
 
   du_manager = create_du_manager(du_manager_params{

lib/du/du_high/du_manager/du_ue/du_ue_controller_impl.cpp

@@ -318,20 +318,27 @@ async_task<void> du_ue_controller_impl::create_stop_traffic_task()
 
 async_task<void> du_ue_controller_impl::run_in_ue_executor(unique_task task)
 {
-  return launch_async([this, task = std::move(task)](coro_context<async_task<void>>& ctx) {
+  auto log_dispatch_retry = [this]() {
+    logger.warning("ue={}: Postpone dispatching of control task to executor. Cause: Task queue is full", ue_index);
+  };
+
+  return launch_async([this, task = std::move(task), log_dispatch_retry](coro_context<async_task<void>>& ctx) {
     CORO_BEGIN(ctx);
 
     // Sync with UE control executor to run provided task.
-    CORO_AWAIT(defer_on_blocking(cfg.services.ue_execs.ctrl_executor(ue_index), cfg.services.timers));
+    CORO_AWAIT(
+        defer_on_blocking(cfg.services.ue_execs.ctrl_executor(ue_index), cfg.services.timers, log_dispatch_retry));
     task();
-    CORO_AWAIT(defer_on_blocking(cfg.services.du_mng_exec, cfg.services.timers));
+    CORO_AWAIT(execute_on_blocking(cfg.services.du_mng_exec, cfg.services.timers, log_dispatch_retry));
 
     // Sync with remaining UE executors, as there might be still pending tasks dispatched to those.
     // TODO: use when_all awaiter
-    CORO_AWAIT(defer_on_blocking(cfg.services.ue_execs.mac_ul_pdu_executor(ue_index), cfg.services.timers));
-    CORO_AWAIT(defer_on_blocking(cfg.services.du_mng_exec, cfg.services.timers));
-    CORO_AWAIT(defer_on_blocking(cfg.services.ue_execs.f1u_dl_pdu_executor(ue_index), cfg.services.timers));
-    CORO_AWAIT(defer_on_blocking(cfg.services.du_mng_exec, cfg.services.timers));
+    CORO_AWAIT(defer_on_blocking(
+        cfg.services.ue_execs.mac_ul_pdu_executor(ue_index), cfg.services.timers, log_dispatch_retry));
+    CORO_AWAIT(execute_on_blocking(cfg.services.du_mng_exec, cfg.services.timers, log_dispatch_retry));
+    CORO_AWAIT(defer_on_blocking(
+        cfg.services.ue_execs.f1u_dl_pdu_executor(ue_index), cfg.services.timers, log_dispatch_retry));
+    CORO_AWAIT(execute_on_blocking(cfg.services.du_mng_exec, cfg.services.timers, log_dispatch_retry));
 
     CORO_RETURN();
   });

lib/f1ap/du/f1ap_du_factory.cpp

@@ -10,9 +10,6 @@
 
 #include "srsran/f1ap/du/f1ap_du_factory.h"
 #include "f1ap_du_impl.h"
-#include "srsran/du/du_high/du_manager/du_manager.h"
-
-/// Notice this would be the only place were we include concrete class implementation files.
 
 using namespace srsran;
 using namespace srs_du;
@@ -21,8 +18,10 @@ std::unique_ptr<f1ap_du> srsran::srs_du::create_f1ap(f1c_connection_client&
                                                      f1ap_du_configurator&       du_mng,
                                                      task_executor&              ctrl_exec,
                                                      du_high_ue_executor_mapper& ue_exec_mapper,
-                                                     f1ap_du_paging_notifier&    paging_notifier)
+                                                     f1ap_du_paging_notifier&    paging_notifier,
+                                                     srsran::timer_manager&      timers)
 {
-  auto f1ap_du = std::make_unique<f1ap_du_impl>(f1c_client_handler, du_mng, ctrl_exec, ue_exec_mapper, paging_notifier);
+  auto f1ap_du =
+      std::make_unique<f1ap_du_impl>(f1c_client_handler, du_mng, ctrl_exec, ue_exec_mapper, paging_notifier, timers);
   return f1ap_du;
 }

lib/f1ap/du/f1ap_du_impl.cpp

@@ -71,13 +71,14 @@ f1ap_du_impl::f1ap_du_impl(f1c_connection_client&      f1c_client_handler_,
                            f1ap_du_configurator&       du_mng_,
                            task_executor&              ctrl_exec_,
                            du_high_ue_executor_mapper& ue_exec_mapper_,
-                           f1ap_du_paging_notifier&    paging_notifier_) :
+                           f1ap_du_paging_notifier&    paging_notifier_,
+                           timer_manager&              timers_) :
   logger(srslog::fetch_basic_logger("DU-F1")),
   ctrl_exec(ctrl_exec_),
   du_mng(du_mng_),
   paging_notifier(paging_notifier_),
   connection_handler(f1c_client_handler_, *this, du_mng, ctrl_exec),
-  ues(du_mng, ctrl_exec, ue_exec_mapper_),
+  ues(du_mng, ctrl_exec, ue_exec_mapper_, timers_),
   events(std::make_unique<f1ap_event_manager>(du_mng.get_timer_factory()))
 {
 }