VLink  2.0.0
A high-performance communication middleware
message_loop.h
Go to the documentation of this file.
1 /*
2  * Copyright (C) 2026 by Thun Lu. All rights reserved.
3  * Author: Thun Lu <thun.lu@zohomail.cn>
4  * Repo: https://github.com/thun-res/vlink
5  * _ __ __ _ __
6  * | | / / / / (_) ____ / /__
7  * | | / / / / / / / __ \ / //_/
8  * | |/ / / /___ / / / / / / / ,<
9  * |___/ /_____/ /_/ /_/ /_/ /_/|_|
10  *
11  * Licensed under the Apache License, Version 2.0 (the "License");
12  * you may not use this file except in compliance with the License.
13  * You may obtain a copy of the License at
14  *
15  * http://www.apache.org/licenses/LICENSE-2.0
16  *
17  * Unless required by applicable law or agreed to in writing, software
18  * distributed under the License is distributed on an "AS IS" BASIS,
19  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
20  * See the License for the specific language governing permissions and
21  * limitations under the License.
22  */
23 
24 /**
25  * @file message_loop.h
26  * @brief Single-threaded task dispatcher with three queue backends, timers and scheduling envelopes.
27  *
28  * @details
29  * @c MessageLoop is VLink's serial task executor. It owns a bounded task queue and the timer
30  * registry attached to it; tasks posted from any thread execute in order on the loop's own
31  * thread. Three queue implementations are available; the choice is fixed at construction.
32  *
33  * @par Dispatch mode table
34  *
35  * | Queue type | Backend | Capacity | Properties |
36  * | ----------------- | ------------------------------------ | -------- | --------------------------- |
37  * | @c kNormalType | @c mutex @c + @c std::deque (or pmr) | 10000 | FIFO, lock-based (default) |
38  * | @c kLockfreeType | @c MpmcQueue<Callback> | 10000 | Lock-free MPMC |
39  * | @c kPriorityType | Two pmr priority queues | 10000 | Droppable / protected split |
40  *
41  * Back-pressure on a full queue is selected by @c Strategy: @c kOptimizationStrategy retries
42  * up to ten times with 1 ms sleeps before dropping an eligible task, @c kPopStrategy drops
43  * immediately, @c kBlockStrategy retries indefinitely. Idle dispatch is always condition-variable
44  * driven and independent of @c Strategy.
45  *
46  * @par Lifecycle diagram
47  *
48  * @verbatim
49  * run() / async_run()
50  * +--------+ |
51  * | idle | ----- async_run -----> | spawn thread
52  * +--------+ |
53  * ^ v
54  * | +--------+
55  * | wait_for_quit | active | <-- post_task / exec_task
56  * | (joined) +--------+
57  * | |
58  * | quit() | quit(true)
59  * | v
60  * | +--------+
61  * +------- on_end() -------- | drain | -- pending tasks --> dropped
62  * +--------+
63  * @endverbatim
64  *
65  * @par Priority levels
66  *
67  * | Symbol | Value | Notes |
68  * | ----------------------- | ----- | ------------------------------ |
69  * | @c kNoPriority | 0 | FIFO sentinel |
70  * | @c kLowestPriority | 1 | Lowest real priority |
71  * | @c kTimerPriority | 50 | Reserved for timer callbacks |
72  * | @c kNormalPriority | 100 | Default user-task priority |
73  * | @c kHighestPriority | 65535 | Highest priority |
74  *
75  * @par Example
76  * @code
77  * vlink::MessageLoop loop;
78  * loop.async_run();
79  *
80  * loop.post_task([] { do_work(); });
81  * loop.post_task_with_priority([] { urgent(); }, vlink::MessageLoop::kHighestPriority);
82  *
83  * auto fut = loop.invoke_task([]() -> int { return compute(); });
84  * int result = fut.get(); // wait from a different thread
85  *
86  * loop.quit();
87  * loop.wait_for_quit();
88  * @endcode
89  *
90  * @note Maximum queue depth is @c 10000 (@c kMaxTaskSize) and maximum timer count is @c 100
91  * (@c kMaxTimerSize). Blocking @c .get() on the future returned by @c invoke_task from
92  * inside the loop thread deadlocks; always call it from another thread.
93  */
94 
95 #pragma once
96 
97 #include <atomic>
98 #include <functional>
99 #include <future>
100 #include <limits>
101 #include <memory>
102 #include <mutex>
103 #include <string>
104 #include <tuple>
105 #include <type_traits>
106 #include <utility>
107 
108 #include "./functional.h"
109 #include "./memory_resource.h"
110 #include "./schedule.h"
111 #include "./task_handle.h"
112 #include "./timer.h"
113 
114 namespace vlink {
115 
116 /**
117  * @class MessageLoop
118  * @brief Serial task dispatcher with selectable queue backend and bounded timer registry.
119  *
120  * @details
121  * Owns a thread that pulls tasks from a bounded queue and fires timer callbacks. Other threads
122  * may post via @c post_task, @c post_task_with_priority, @c invoke_task or @c exec_task from
123  * any context. The class is the building block of @c MultiLoop and the engine target of
124  * @c GraphTask::execute.
125  */
127  public:
128  /**
129  * @brief Callback type for tasks and event handlers.
130  *
131  * @details
132  * Move-only so move-only targets such as @c std::packaged_task or lambdas holding
133  * @c std::unique_ptr can be posted directly without a shared trampoline.
134  */
135  using Callback = MoveFunction<void()>;
136 
137  /**
138  * @brief Internal queue implementation type.
139  */
140  enum Type : uint8_t {
141  kNormalType = 0, ///< Mutex-protected FIFO queue (default).
142  kLockfreeType = 1, ///< Lock-free MPMC queue.
143  kPriorityType = 2, ///< Priority-ordered queue with droppable / protected split.
144  };
145 
146  /**
147  * @brief Back-pressure strategy applied when the bounded queue is at capacity.
148  *
149  * @details
150  * Idle dispatch is unconditionally cv-driven; this enum only affects the push side.
151  */
152  enum Strategy : uint8_t {
153  kOptimizationStrategy = 0, ///< Retry up to ten times, then drop one eligible task and push.
154  kPopStrategy = 1, ///< Drop one eligible task immediately and push.
155  kBlockStrategy = 2, ///< Retry indefinitely until space appears.
156  };
157 
158  /**
159  * @brief Built-in priority levels for @c kPriorityType loops; higher values dispatch first.
160  */
161  enum Priority : uint16_t {
162  kNoPriority = 0, ///< FIFO sentinel.
163  kLowestPriority = 1, ///< Lowest real priority.
164  kTimerPriority = 50, ///< Reserved for timer callbacks.
165  kNormalPriority = 100, ///< Default user-task priority.
166  kHighestPriority = std::numeric_limits<uint16_t>::max() ///< Highest priority.
167  };
168 
169  /**
170  * @brief Lifetime gate shared with cross-thread observers.
171  *
172  * @details
173  * The destructor of @c MessageLoop flips @c alive to @c false under @c mtx as its very first
174  * step, so a caller that holds @c mtx and observes @c alive @c == @c true is guaranteed the
175  * loop is still safe to touch. Obtain via @c get_alive_state.
176  */
177  struct AliveState final {
178  std::mutex mtx;
179  std::atomic_bool alive{true};
180  };
181 
182  /**
183  * @brief Constructs a loop with the default @c kNormalType queue.
184  */
186 
187  /**
188  * @brief Constructs a loop with the given queue type.
189  *
190  * @param type Queue implementation.
191  */
192  explicit MessageLoop(Type type);
193 
194  /**
195  * @brief Destructor; requests quit and joins the dispatcher thread if needed.
196  */
197  virtual ~MessageLoop();
198 
199  /**
200  * @brief Sets a human-readable name visible to profiling tools.
201  *
202  * @param name Loop name.
203  */
204  void set_name(const std::string& name);
205 
206  /**
207  * @brief Returns the loop name set via @c set_name.
208  *
209  * @return Reference to the stored name.
210  */
211  [[nodiscard]] const std::string& get_name() const;
212 
213  /**
214  * @brief Returns the queue type this loop was constructed with.
215  *
216  * @return Queue type.
217  */
218  [[nodiscard]] Type get_type() const;
219 
220  /**
221  * @brief Returns the active back-pressure strategy.
222  *
223  * @return Current strategy.
224  */
225  [[nodiscard]] Strategy get_strategy() const;
226 
227  /**
228  * @brief Replaces the back-pressure strategy.
229  *
230  * @param strategy New strategy; takes effect on the next full-queue push.
231  */
232  void set_strategy(Strategy strategy);
233 
234  /**
235  * @brief Registers a callback fired once at loop thread startup.
236  *
237  * @details
238  * Must be registered before @c run / @c async_run; later calls are ignored.
239  *
240  * @param callback Startup handler.
241  */
243 
244  /**
245  * @brief Registers a callback fired once when the loop thread exits.
246  *
247  * @details
248  * Must be registered before @c run / @c async_run; later calls are ignored.
249  *
250  * @param callback Shutdown handler.
251  */
252  void register_end_handler(Callback&& callback);
253 
254  /**
255  * @brief Registers a callback fired every time the queue becomes empty.
256  *
257  * @details
258  * Must be registered before @c run / @c async_run; later calls are ignored.
259  *
260  * @param callback Idle handler.
261  */
262  void register_idle_handler(Callback&& callback);
263 
264  /**
265  * @brief Runs the loop on the calling thread until @c quit is requested.
266  *
267  * @return @c true after a normal exit; @c false when the loop is already running.
268  */
269  bool run();
270 
271  /**
272  * @brief Starts the loop on a new background thread.
273  *
274  * @return @c true when the thread was started; @c false when the loop is already running.
275  */
276  bool async_run();
277 
278  /**
279  * @brief Alias of @c run blocking the calling thread.
280  *
281  * @return Same return value as @c run.
282  */
283  bool spin();
284 
285  /**
286  * @brief Processes one batch of pending tasks and timers on the calling thread.
287  *
288  * @param block When @c true (default) blocks if the queue is empty; when @c false returns immediately.
289  * @return @c true after a normal processing cycle; @c false when quitting or the call came from a
290  * thread that does not own the loop and is unrelated to @c run / @c async_run.
291  */
292  bool spin_once(bool block = true);
293 
294  /**
295  * @brief Requests the loop to exit.
296  *
297  * @details
298  * With @p force @c == @c false the current batch finishes and the loop exits afterwards. With
299  * @p force @c == @c true the in-flight batch is also aborted. Tasks queued after the request
300  * are rejected. Returns @c false when @c quit had already been called (only meaningful when
301  * @p force is @c false).
302  *
303  * @param force When @c true, also discards the in-flight batch. Default: @c false.
304  * @return @c true when the quit signal was accepted.
305  */
306  bool quit(bool force = false);
307 
308  /**
309  * @brief Waits until the loop has fully exited.
310  *
311  * @param ms Maximum wait in milliseconds; @c Timer::kInfinite means no limit.
312  * @param check When @c true (default) rejects calls from the loop's own thread.
313  * @return @c true if the loop exited before the timeout.
314  */
315  bool wait_for_quit(int ms = Timer::kInfinite, bool check = true);
316 
317  /**
318  * @brief Posts a task for execution on the loop thread.
319  *
320  * @details
321  * Thread-safe. Returns @c false when the loop is quitting or the configured @c Strategy
322  * cannot make room for the new task.
323  *
324  * @param callback Task to post.
325  * @return @c true when the task was enqueued.
326  */
327  bool post_task(Callback&& callback);
328 
329  /**
330  * @brief Tracked variant of @c post_task returning a @c TaskHandle.
331  *
332  * @param callback Task to post.
333  * @param options Optional overflow / drop / cancellation policy.
334  * @return Handle bound to the posted task; valid even when posting was rejected.
335  */
336  [[nodiscard]] TaskHandle post_task_handle(Callback&& callback, const PostTaskOptions& options = {});
337 
338  /**
339  * @brief Posts a task with an explicit priority on a @c kPriorityType loop.
340  *
341  * @details
342  * Higher @p priority values dispatch first. Non-priority loops reject the call and return
343  * @c false.
344  *
345  * @param callback Task to post.
346  * @param priority Dispatch priority in @c [kLowestPriority, @c kHighestPriority].
347  * @return @c true when the task was enqueued.
348  */
349  bool post_task_with_priority(Callback&& callback, uint16_t priority);
350 
351  /**
352  * @brief Tracked variant of @c post_task_with_priority returning a @c TaskHandle.
353  *
354  * @param callback Task to post.
355  * @param priority Dispatch priority.
356  * @param options Optional overflow / drop / cancellation policy.
357  * @return Handle bound to the posted task.
358  */
359  [[nodiscard]] TaskHandle post_task_with_priority_handle(Callback&& callback, uint16_t priority,
360  const PostTaskOptions& options = {});
361 
362  /**
363  * @brief Schedules a @c void-returning callable and returns a chainable @c Schedule::Status.
364  *
365  * @details
366  * @c Schedule::Config carries the delay, priority and timeout fields. The returned status
367  * supports @c on_catch / @c on_schedule_timeout / @c on_execution_timeout chaining. Posting is
368  * deferred until the handle is committed (when the temporary handle is destroyed at the end of
369  * the fluent expression), so every continuation is registered before the task runs. A caller
370  * that stores the handle instead of consuming it inline must call @c Schedule::Status::dispatch()
371  * to post the task and learn whether it was accepted.
372  *
373  * @tparam CallbackT Callable returning @c void.
374  * @param config Scheduling envelope.
375  * @param callback Callable to schedule.
376  * @return Chainable status object.
377  */
378  // NOLINTNEXTLINE(modernize-use-constraints)
379  template <typename CallbackT, typename = std::enable_if_t<!std::is_convertible_v<CallbackT, Schedule::RetCallback>>>
380  Schedule::Status exec_task(const Schedule::Config& config, CallbackT&& callback);
381 
382  /**
383  * @brief Schedules a @c bool-returning callable and returns a chainable @c Schedule::RetStatus.
384  *
385  * @details
386  * @c on_then fires when the callable returns @c true; @c on_else fires when it returns @c false.
387  * Posting is deferred until the handle is committed (destroyed at the end of the fluent
388  * expression), guaranteeing every continuation is registered first; a stored handle must call
389  * @c Schedule::Status::dispatch() to post it.
390  *
391  * @tparam CallbackT Callable returning @c bool.
392  * @param config Scheduling envelope.
393  * @param callback Callable to schedule.
394  * @return Chainable status object.
395  */
396  // NOLINTNEXTLINE(modernize-use-constraints)
397  template <typename CallbackT, typename = std::enable_if_t<std::is_convertible_v<CallbackT, Schedule::RetCallback>>>
398  Schedule::RetStatus exec_task(const Schedule::Config& config, CallbackT&& callback);
399 
400  /**
401  * @brief Wakes the loop thread if it is suspended in its idle wait.
402  *
403  * @details
404  * Concurrent calls coalesce; only the first one after the previous wakeup actually signals.
405  *
406  * @return @c true when a wakeup is pending (newly issued or already in flight).
407  */
408  bool wakeup();
409 
410  /**
411  * @brief Recreates the lock-free queue, clearing all queued tasks and counters.
412  *
413  * @details
414  * Applies only to @c kLockfreeType loops; on other types the call is a no-op. Must be invoked
415  * while the loop is stopped; calls made while it is running are logged and skipped.
416  */
418 
419  /**
420  * @brief Reports whether the loop is currently running.
421  *
422  * @return @c true between a successful start and the corresponding @c quit.
423  */
424  [[nodiscard]] bool is_running() const;
425 
426  /**
427  * @brief Reports whether @c quit has been requested and the loop is winding down.
428  *
429  * @return @c true once @c quit has been observed.
430  */
431  [[nodiscard]] bool is_ready_to_quit() const;
432 
433  /**
434  * @brief Reports whether the loop is currently executing a task.
435  *
436  * @return @c true while a callback is on the call stack inside the loop.
437  */
438  [[nodiscard]] bool is_busy() const;
439 
440  /**
441  * @brief Returns the current pending task count.
442  *
443  * @return Number of tasks waiting in the queue.
444  */
445  [[nodiscard]] size_t get_task_count() const;
446 
447  /**
448  * @brief Waits until the loop has drained its queue and is not executing a task.
449  *
450  * @param ms Maximum wait in milliseconds; @c Timer::kInfinite means no limit.
451  * @param check When @c true (default) rejects calls from the loop's own thread.
452  * @return @c true when the idle condition was reached within @p ms.
453  */
454  virtual bool wait_for_idle(int ms = Timer::kInfinite, bool check = true);
455 
456  /**
457  * @brief Returns the maximum queue depth.
458  *
459  * @return @c kMaxTaskSize (@c 10000) by default.
460  */
461  [[nodiscard]] virtual size_t get_max_task_count() const;
462 
463  /**
464  * @brief Returns the maximum number of timers that can be attached.
465  *
466  * @return @c kMaxTimerSize (@c 100) by default.
467  */
468  [[nodiscard]] virtual size_t get_max_timer_count() const;
469 
470  /**
471  * @brief Returns the maximum allowed task execution time in milliseconds.
472  *
473  * @details
474  * Tasks exceeding this duration trigger @c on_task_timeout. Zero disables the check.
475  *
476  * @return Maximum execution time in milliseconds.
477  */
478  [[nodiscard]] virtual uint32_t get_max_elapsed_time() const;
479 
480  /**
481  * @brief Reports whether the calling thread is owned by this loop.
482  *
483  * @details
484  * Detects callbacks that re-enter the loop synchronously. For @c MultiLoop this also covers
485  * worker threads of the underlying pool.
486  *
487  * @return @c true when called from a thread belonging to this loop.
488  */
489  [[nodiscard]] virtual bool is_in_same_thread() const;
490 
491  /**
492  * @brief Returns the shared lifetime flag used by cross-thread bridges.
493  *
494  * @details
495  * The returned @c AliveState outlives this loop. Adapters that need to post continuations
496  * back to this loop should lock @c mtx, re-check @c alive, and only call back while still
497  * holding the lock.
498  *
499  * @return Shared handle; never null while the loop object is alive.
500  */
501  [[nodiscard]] std::shared_ptr<AliveState> get_alive_state() const;
502 
503  /**
504  * @brief Dispatches a callable to the loop thread and returns a @c std::future for the result.
505  *
506  * @details
507  * Thread-safe. When posting fails the returned future becomes ready with
508  * @c std::future_error / @c broken_promise.
509  *
510  * @warning Do not call @c .get() on the future from the loop's own thread; doing so deadlocks.
511  *
512  * @tparam FunctionT Callable type.
513  * @tparam ArgsT Argument types.
514  * @tparam ResultT Return type (deduced).
515  * @param function Callable to dispatch.
516  * @param args Arguments forwarded to @p function.
517  * @return Future that becomes ready after the callable completes.
518  */
519  template <class FunctionT, class... ArgsT, typename ResultT = std::invoke_result_t<FunctionT, ArgsT...>>
520  [[nodiscard]] std::future<ResultT> invoke_task(FunctionT&& function, ArgsT&&... args);
521 
522  /**
523  * @brief Priority variant of @c invoke_task; requires a @c kPriorityType loop.
524  *
525  * @tparam FunctionT Callable type.
526  * @tparam ArgsT Argument types.
527  * @tparam ResultT Return type (deduced).
528  * @param function Callable to dispatch.
529  * @param priority Dispatch priority.
530  * @param args Arguments forwarded to @p function.
531  * @return Future that becomes ready after the callable completes.
532  */
533  template <class FunctionT, class... ArgsT, typename ResultT = std::invoke_result_t<FunctionT, ArgsT...>>
534  [[nodiscard]] std::future<ResultT> invoke_task_with_priority(FunctionT&& function, uint16_t priority,
535  ArgsT&&... args);
536 
537  protected:
538  /**
539  * @brief Hook invoked once on the loop thread before the first task runs.
540  *
541  * @details
542  * Subclasses override to perform per-thread initialisation.
543  */
544  virtual void on_begin();
545 
546  /**
547  * @brief Hook invoked once on the loop thread after the last task runs.
548  *
549  * @details
550  * Subclasses override to perform per-thread cleanup.
551  */
552  virtual void on_end();
553 
554  /**
555  * @brief Hook invoked on the loop thread each time the queue becomes empty.
556  *
557  * @details
558  * Subclasses override to perform idle bookkeeping.
559  */
560  virtual void on_idle();
561 
562  /**
563  * @brief Dispatches a ready task on the loop thread.
564  *
565  * @details
566  * The default implementation invokes @p callback inline. Subclasses such as @c MultiLoop
567  * override to redirect the call to a worker pool; overrides that do not forward must invoke
568  * @p callback themselves or the task is silently dropped.
569  *
570  * @param callback Task to dispatch.
571  * @param start_time Millisecond @c steady_clock timestamp captured at enqueue time, or @c 0
572  * when elapsed tracking is disabled.
573  */
574  virtual void on_task_changed(Callback&& callback, uint32_t start_time);
575 
576  /**
577  * @brief Hook invoked when a task exceeds @c get_max_elapsed_time().
578  *
579  * @param callback Task that timed out.
580  * @param elapsed_time Actual execution time in milliseconds.
581  */
582  virtual void on_task_timeout(Callback&& callback, uint32_t elapsed_time);
583 
584  private:
585  static uint64_t get_current_nano_time();
586 
587  Schedule::Callback make_launcher(const Schedule::Config& config, Schedule::Callback&& wrapper,
588  std::shared_ptr<Schedule::Status::Impl> impl);
589 
590  bool add_timer(Timer* timer);
591 
592  bool remove_timer(Timer* timer);
593 
594  bool push_task(Callback&& callback, uint16_t priority, bool droppable = true,
596  const TaskHandle* submit_handle = nullptr);
597 
598  bool drop_one_normal_task();
599 
600  bool drop_one_lockfree_task(bool keep_reserved = false);
601 
602  bool drop_one_priority_task();
603 
604  bool reserve_lockfree_task();
605 
606  void release_lockfree_task();
607 
608  void push_normal_task(Callback&& callback, bool droppable = true);
609 
610  bool push_lockfree_task(Callback&& callback);
611 
612  void push_priority_task(Callback&& callback, uint16_t priority, bool droppable = true);
613 
614  void do_consume();
615 
616  bool process_normal_task(bool block);
617 
618  bool process_lockfree_task(bool block);
619 
620  bool process_priority_task(bool block);
621 
622  bool process_timer_task(int64_t& next_sleep_time);
623 
624  void drop_pending_tasks();
625 
626  friend Timer;
627  struct Impl;
628  std::unique_ptr<Impl> impl_;
629 
631 };
632 
633 ////////////////////////////////////////////////////////////////
634 /// Details
635 ////////////////////////////////////////////////////////////////
636 
637 inline Schedule::Callback MessageLoop::make_launcher(const Schedule::Config& config, Schedule::Callback&& wrapper,
638  std::shared_ptr<Schedule::Status::Impl> impl) {
639  auto alive_state = get_alive_state();
640 
641  return [this, alive_state = std::move(alive_state), config, impl = std::move(impl),
642  wrapper = std::move(wrapper)]() mutable {
643  std::lock_guard alive_lock(alive_state->mtx);
644 
645  if VUNLIKELY (!alive_state->alive.load(std::memory_order_acquire)) {
646  impl->is_valid.store(false, std::memory_order_relaxed); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
647  return; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
648  }
649 
650  bool post_ret = false;
651 
652  if (config.delay_ms > 0) {
653  post_ret = Timer::call_once(this, config.delay_ms, std::move(wrapper), config.priority);
654  } else if (get_type() == kPriorityType && config.priority != kNoPriority) {
655  post_ret = post_task_with_priority(std::move(wrapper), config.priority);
656  } else {
657  post_ret = post_task(std::move(wrapper));
658  }
659 
660  impl->is_valid.store(post_ret, std::memory_order_relaxed);
661  };
662 }
663 
664 template <typename CallbackT, typename>
665 Schedule::Status MessageLoop::exec_task(const Schedule::Config& config, CallbackT&& callback) {
666  Schedule::Callback wrapper_callback;
667 
668  // NOLINTNEXTLINE(bugprone-move-forwarding-reference)
669  auto status = Schedule::process(config, std::move(callback), wrapper_callback);
670 
671  status.launcher_ = make_launcher(config, std::move(wrapper_callback), status.impl_);
672 
673  return status;
674 }
675 
676 template <typename CallbackT, typename>
677 Schedule::RetStatus MessageLoop::exec_task(const Schedule::Config& config, CallbackT&& callback) {
678  Schedule::Callback wrapper_callback;
679 
680  // NOLINTNEXTLINE(bugprone-move-forwarding-reference)
681  auto status = Schedule::process_with_ret(config, std::move(callback), wrapper_callback);
682 
683  status.launcher_ = make_launcher(config, std::move(wrapper_callback), status.impl_);
684 
685  return status;
686 }
687 
688 template <class FunctionT, class... ArgsT, typename ResultT>
689 inline std::future<ResultT> MessageLoop::invoke_task(FunctionT&& function, ArgsT&&... args) {
690  auto bound = [function = std::forward<FunctionT>(function),
691  args = std::make_tuple(std::forward<ArgsT>(args)...)]() mutable -> ResultT {
692  return std::apply(
693  [&function](auto&&... unpacked_args) -> ResultT {
694  return std::invoke(function, std::forward<decltype(unpacked_args)>(unpacked_args)...);
695  },
696  args);
697  };
698 
699  if constexpr (kIsSupportMoveFunction) {
700  std::packaged_task<ResultT()> task(std::move(bound));
701  auto res = task.get_future();
702 
703  if VUNLIKELY (!post_task([task = std::move(task)]() mutable { task(); })) {
704  }
705 
706  return res;
707  } else {
708  auto task = MemoryResource::make_shared<std::packaged_task<ResultT()>>(std::move(bound));
709  auto res = task->get_future();
710 
711  if VUNLIKELY (!post_task([task]() mutable { (*task)(); })) {
712  task.reset();
713  }
714 
715  return res;
716  }
717 }
718 
719 template <class FunctionT, class... ArgsT, typename ResultT>
720 inline std::future<ResultT> MessageLoop::invoke_task_with_priority(FunctionT&& function, uint16_t priority,
721  ArgsT&&... args) {
722  auto bound = [function = std::forward<FunctionT>(function),
723  args = std::make_tuple(std::forward<ArgsT>(args)...)]() mutable -> ResultT {
724  return std::apply(
725  [&function](auto&&... unpacked_args) -> ResultT {
726  return std::invoke(function, std::forward<decltype(unpacked_args)>(unpacked_args)...);
727  },
728  args);
729  };
730 
731  if constexpr (kIsSupportMoveFunction) {
732  std::packaged_task<ResultT()> task(std::move(bound));
733  auto res = task.get_future();
734 
735  if VUNLIKELY (!post_task_with_priority([task = std::move(task)]() mutable { task(); }, priority)) {
736  }
737 
738  return res;
739  } else {
740  auto task = MemoryResource::make_shared<std::packaged_task<ResultT()>>(std::move(bound));
741  auto res = task->get_future();
742 
743  if VUNLIKELY (!post_task_with_priority([task]() mutable { (*task)(); }, priority)) {
744  task.reset();
745  }
746 
747  return res;
748  }
749 }
750 
751 } // namespace vlink
Pool-backed type-erased callables: copyable vlink::Function and move-only vlink::MoveFunction.
#define VUNLIKELY(...)
Short alias for VLINK_UNLIKELY.
Definition: macros.h:289
#define VLINK_EXPORT
Definition: macros.h:81
#define VLINK_DISALLOW_COPY_AND_ASSIGN(classname)
Deletes the copy constructor and copy-assignment operator of classname.
Definition: macros.h:174
PMR adapter that lets standard pmr-aware containers allocate through vlink::MemoryPool.
Fluent task-scheduling wrapper used by MessageLoop::exec_task() and family.
Observable handle returned by tracked task-posting APIs of MessageLoop and ThreadPool.
MessageLoop-driven periodic or one-shot timer with priority support.