105 #include <type_traits>
153 kOptimizationStrategy = 0,
165 kNormalPriority = 100,
166 kHighestPriority = std::numeric_limits<uint16_t>::max()
179 std::atomic_bool alive{
true};
379 template <
typename CallbackT,
typename = std::enable_if_t<!std::is_convertible_v<CallbackT, Schedule::RetCallback>>>
397 template <
typename CallbackT,
typename = std::enable_if_t<std::is_convertible_v<CallbackT, Schedule::RetCallback>>>
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);
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,
585 static uint64_t get_current_nano_time();
588 std::shared_ptr<Schedule::Status::Impl> impl);
590 bool add_timer(
Timer* timer);
592 bool remove_timer(
Timer* timer);
594 bool push_task(
Callback&& callback, uint16_t priority,
bool droppable =
true,
598 bool drop_one_normal_task();
600 bool drop_one_lockfree_task(
bool keep_reserved =
false);
602 bool drop_one_priority_task();
604 bool reserve_lockfree_task();
606 void release_lockfree_task();
608 void push_normal_task(
Callback&& callback,
bool droppable =
true);
610 bool push_lockfree_task(
Callback&& callback);
612 void push_priority_task(
Callback&& callback, uint16_t priority,
bool droppable =
true);
616 bool process_normal_task(
bool block);
618 bool process_lockfree_task(
bool block);
620 bool process_priority_task(
bool block);
622 bool process_timer_task(int64_t& next_sleep_time);
624 void drop_pending_tasks();
628 std::unique_ptr<Impl> impl_;
638 std::shared_ptr<Schedule::Status::Impl> impl) {
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);
645 if VUNLIKELY (!alive_state->alive.load(std::memory_order_acquire)) {
646 impl->is_valid.store(
false, std::memory_order_relaxed);
650 bool post_ret =
false;
657 post_ret =
post_task(std::move(wrapper));
660 impl->is_valid.store(post_ret, std::memory_order_relaxed);
664 template <
typename CallbackT,
typename>
671 status.launcher_ = make_launcher(config, std::move(wrapper_callback), status.impl_);
676 template <
typename CallbackT,
typename>
683 status.launcher_ = make_launcher(config, std::move(wrapper_callback), status.impl_);
688 template <
class FunctionT,
class... ArgsT,
typename ResultT>
690 auto bound = [
function = std::forward<FunctionT>(
function),
691 args = std::make_tuple(std::forward<ArgsT>(args)...)]()
mutable -> ResultT {
693 [&
function](
auto&&... unpacked_args) -> ResultT {
694 return std::invoke(
function, std::forward<decltype(unpacked_args)>(unpacked_args)...);
699 if constexpr (kIsSupportMoveFunction) {
700 std::packaged_task<ResultT()> task(std::move(bound));
701 auto res = task.get_future();
709 auto res = task->get_future();
719 template <
class FunctionT,
class... ArgsT,
typename ResultT>
722 auto bound = [
function = std::forward<FunctionT>(
function),
723 args = std::make_tuple(std::forward<ArgsT>(args)...)]()
mutable -> ResultT {
725 [&
function](
auto&&... unpacked_args) -> ResultT {
726 return std::invoke(
function, std::forward<decltype(unpacked_args)>(unpacked_args)...);
731 if constexpr (kIsSupportMoveFunction) {
732 std::packaged_task<ResultT()> task(std::move(bound));
733 auto res = task.get_future();
741 auto res = task->get_future();
Serial task dispatcher with selectable queue backend and bounded timer registry.
Definition: message_loop.h:126
MessageLoop(Type type)
Constructs a loop with the given queue type.
void register_begin_handler(Callback &&callback)
Registers a callback fired once at loop thread startup.
bool spin_once(bool block=true)
Processes one batch of pending tasks and timers on the calling thread.
MessageLoop()
Constructs a loop with the default kNormalType queue.
virtual void on_task_changed(Callback &&callback, uint32_t start_time)
Dispatches a ready task on the loop thread.
bool async_run()
Starts the loop on a new background thread.
bool wait_for_quit(int ms=Timer::kInfinite, bool check=true)
Waits until the loop has fully exited.
bool spin()
Alias of run blocking the calling thread.
bool wakeup()
Wakes the loop thread if it is suspended in its idle wait.
void set_name(const std::string &name)
Sets a human-readable name visible to profiling tools.
virtual bool is_in_same_thread() const
Reports whether the calling thread is owned by this loop.
std::shared_ptr< AliveState > get_alive_state() const
Returns the shared lifetime flag used by cross-thread bridges.
bool post_task(Callback &&callback)
Posts a task for execution on the loop thread.
TaskHandle post_task_with_priority_handle(Callback &&callback, uint16_t priority, const PostTaskOptions &options={})
Tracked variant of post_task_with_priority returning a TaskHandle.
virtual void on_end()
Hook invoked once on the loop thread after the last task runs.
std::future< ResultT > invoke_task(FunctionT &&function, ArgsT &&... args)
Dispatches a callable to the loop thread and returns a std::future for the result.
Definition: message_loop.h:689
Type
Internal queue implementation type.
Definition: message_loop.h:140
@ kPriorityType
Priority-ordered queue with droppable / protected split.
Definition: message_loop.h:143
std::future< ResultT > invoke_task_with_priority(FunctionT &&function, uint16_t priority, ArgsT &&... args)
Priority variant of invoke_task; requires a kPriorityType loop.
Definition: message_loop.h:720
virtual void on_idle()
Hook invoked on the loop thread each time the queue becomes empty.
Strategy get_strategy() const
Returns the active back-pressure strategy.
Schedule::Status exec_task(const Schedule::Config &config, CallbackT &&callback)
Schedules a void-returning callable and returns a chainable Schedule::Status.
Definition: message_loop.h:665
Type get_type() const
Returns the queue type this loop was constructed with.
bool run()
Runs the loop on the calling thread until quit is requested.
bool is_running() const
Reports whether the loop is currently running.
bool is_busy() const
Reports whether the loop is currently executing a task.
Priority
Built-in priority levels for kPriorityType loops; higher values dispatch first.
Definition: message_loop.h:161
@ kNoPriority
FIFO sentinel.
Definition: message_loop.h:162
virtual uint32_t get_max_elapsed_time() const
Returns the maximum allowed task execution time in milliseconds.
TaskHandle post_task_handle(Callback &&callback, const PostTaskOptions &options={})
Tracked variant of post_task returning a TaskHandle.
void set_strategy(Strategy strategy)
Replaces the back-pressure strategy.
virtual size_t get_max_timer_count() const
Returns the maximum number of timers that can be attached.
bool is_ready_to_quit() const
Reports whether quit has been requested and the loop is winding down.
const std::string & get_name() const
Returns the loop name set via set_name.
bool post_task_with_priority(Callback &&callback, uint16_t priority)
Posts a task with an explicit priority on a kPriorityType loop.
virtual size_t get_max_task_count() const
Returns the maximum queue depth.
void reset_lockfree_capacity()
Recreates the lock-free queue, clearing all queued tasks and counters.
void register_idle_handler(Callback &&callback)
Registers a callback fired every time the queue becomes empty.
virtual ~MessageLoop()
Destructor; requests quit and joins the dispatcher thread if needed.
size_t get_task_count() const
Returns the current pending task count.
Strategy
Back-pressure strategy applied when the bounded queue is at capacity.
Definition: message_loop.h:152
virtual void on_task_timeout(Callback &&callback, uint32_t elapsed_time)
Hook invoked when a task exceeds get_max_elapsed_time().
void register_end_handler(Callback &&callback)
Registers a callback fired once when the loop thread exits.
virtual bool wait_for_idle(int ms=Timer::kInfinite, bool check=true)
Waits until the loop has drained its queue and is not executing a task.
bool quit(bool force=false)
Requests the loop to exit.
virtual void on_begin()
Hook invoked once on the loop thread before the first task runs.
RAII handle returned by exec_task() when the wrapped callback returns bool.
Definition: schedule.h:303
RAII handle returned by exec_task() when the wrapped callback returns void.
Definition: schedule.h:164
Shared observable handle returned by tracked posting APIs.
Definition: task_handle.h:189
Periodic or one-shot timer that fires on an associated MessageLoop thread.
Definition: timer.h:84
static bool call_once(class MessageLoop *message_loop, uint32_t interval_ms, Callback &&callback, uint16_t priority=0)
Posts a fire-and-forget one-shot timer to a loop.
static constexpr int kInfinite
Sentinel loop_count value meaning fire forever.
Definition: timer.h:94
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.
std::shared_ptr< T > make_shared(Args &&... args)
Fallback make_shared forwarding to std::make_shared.
Definition: memory_resource.h:276
TaskOverflowPolicy
Per-post override for the bounded-queue overflow strategy.
Definition: task_handle.h:128
@ kUseDispatcherStrategy
Inherit the dispatcher-configured overflow strategy.
Fluent task-scheduling wrapper used by MessageLoop::exec_task() and family.
Lifetime gate shared with cross-thread observers.
Definition: message_loop.h:177
std::mutex mtx
Definition: message_loop.h:178
Bundle of optional knobs accepted by tracked task-posting APIs.
Definition: task_handle.h:156
Scheduling parameters captured at the call to exec_task().
Definition: schedule.h:129
uint32_t delay_ms
Delay before posting; 0 posts immediately.
Definition: schedule.h:146
uint16_t priority
Dispatch priority hint; higher fires sooner.
Definition: schedule.h:147
static Status process(const Config &config, Callback &&callback, Callback &wrapper_callback)
Wraps a void callback in a Config envelope and produces the task wrapper for the dispatcher.
static RetStatus process_with_ret(const Config &config, RetCallback &&callback, Callback &wrapper_callback)
Wraps a bool-returning callback in a Config envelope and produces the task wrapper for the dispatcher...
Observable handle returned by tracked task-posting APIs of MessageLoop and ThreadPool.
MessageLoop-driven periodic or one-shot timer with priority support.