Merge pull request #295 from elbeno/urgent-triggers

✨ Add `urgent_trigger_scheduler`

Author: elbeno

docs/schedulers.adoc

@@ -407,3 +407,30 @@ NOTE: It is possible to use a `trigger_scheduler` that takes arguments as a
 "normal" scheduler, i.e. functions like `start_on` will work; however the
 arguments passed to `run_triggers` will be discarded when used with constructs
 like `start_on(trigger_scheduler<"name", int>{}, just(42))`.
+
+Normally, calling `run_triggers` runs all the work in order -- and there can be
+multiple senders waiting to run on a given trigger. Occasionally it is useful to
+have a sender jump to the head of the queue. To do this use an `urgent_trigger_scheduler`:
+
+[source,cpp]
+----
+int x{};
+async::trigger_scheduler<"name", int>{}.schedule()
+  | async::then([&] (auto i) { x = i; })
+  | async::start_detached();
+
+// this sender will be queued before the one above
+async::urgent_trigger_scheduler<"name", int>{}.schedule()
+  | async::then([&] (auto i) { x = i*2; })
+  | async::start_detached();
+
+// when event occurs
+async::run_one_trigger<"name">(42);
+// x is now 84
+async::run_triggers<"name">(12);
+// x is now 12
+----
+
+This is not a general-purpose priority management mechanism, but it can be
+useful to prevent priority inversion when integrating triggers with other work
+that uses fixed interrupt priorities.

include/async/incite_on.hpp

@@ -178,7 +178,8 @@ template <sender S, typename Sched>
 namespace _incite_on {
 template <typename Uniq>
 class trigger_scheduler
-    : public trigger_mgr::scheduler<trigger_scheduler<Uniq>, Uniq> {
+    : public trigger_mgr::scheduler<trigger_scheduler<Uniq>, Uniq,
+                                    trigger_mgr::queue_at_back> {
     [[nodiscard]] friend constexpr auto operator==(trigger_scheduler,
                                                    trigger_scheduler)
         -> bool = default;

include/async/schedulers/trigger_manager.hpp

@@ -77,6 +77,20 @@ struct all {
 };
 } // namespace run_policy
 
+namespace trigger_mgr {
+struct queue_at_front {
+    constexpr static auto push(auto &list, auto &task) -> void {
+        list.push_front(std::addressof(task));
+    }
+};
+
+struct queue_at_back {
+    constexpr static auto push(auto &list, auto &task) -> void {
+        list.push_back(std::addressof(task));
+    }
+};
+} // namespace trigger_mgr
+
 template <typename Name, typename... Args> struct trigger_manager {
     using task_t = trigger_task<Args...>;
 
@@ -86,12 +100,12 @@ template <typename Name, typename... Args> struct trigger_manager {
     stdx::atomic<int> task_count;
 
   public:
-    auto enqueue(task_t &t) -> bool {
+    template <typename QueuePolicy> auto enqueue(task_t &t) -> bool {
         return conc::call_in_critical_section<mutex>([&]() -> bool {
             auto const added = not std::exchange(t.pending, true);
             if (added) {
                 ++task_count;
-                tasks[0].push_back(std::addressof(t));
+                QueuePolicy::push(tasks[0], t);
             }
             return added;
         });

include/async/schedulers/trigger_scheduler.hpp

@@ -67,9 +67,10 @@ struct op_state_base<Rcvr, Ops> {
     auto clear_stop_cb() -> void {}
 };
 
-template <typename Name, typename Rcvr, typename... Args>
-struct op_state final : op_state_base<Rcvr, op_state<Name, Rcvr, Args...>>,
-                        trigger_task<Args...> {
+template <typename Name, typename Rcvr, typename QueuePolicy, typename... Args>
+struct op_state final
+    : op_state_base<Rcvr, op_state<Name, Rcvr, QueuePolicy, Args...>>,
+      trigger_task<Args...> {
     template <stdx::same_as_unqualified<Rcvr> R>
     // NOLINTNEXTLINE(bugprone-forwarding-reference-overload)
     constexpr explicit(true) op_state(R &&r) : rcvr{std::forward<R>(r)} {}
@@ -93,7 +94,7 @@ struct op_state final : op_state_base<Rcvr, op_state<Name, Rcvr, Args...>>,
             complete_stopped();
             return;
         }
-        triggers<Name, Args...>.enqueue(*this);
+        triggers<Name, Args...>.template enqueue<QueuePolicy>(*this);
         this->emplace_stop_cb();
     }
 
@@ -111,7 +112,8 @@ struct op_state final : op_state_base<Rcvr, op_state<Name, Rcvr, Args...>>,
     [[no_unique_address]] Rcvr rcvr;
 };
 
-template <typename S, typename Name, typename... Args> class scheduler {
+template <typename S, typename Name, typename QueuePolicy, typename... Args>
+class scheduler {
     struct sender {
         using is_sender = void;
 
@@ -127,7 +129,8 @@ template <typename S, typename Name, typename... Args> class scheduler {
         template <receiver R>
         [[nodiscard]] constexpr auto connect(R &&r) const {
             check_connect<sender, R>();
-            return trigger_mgr::op_state<Name, std::remove_cvref_t<R>, Args...>{
+            return trigger_mgr::op_state<Name, std::remove_cvref_t<R>,
+                                         QueuePolicy, Args...>{
                 std::forward<R>(r)};
         }
     };
@@ -140,6 +143,7 @@ template <typename S, typename Name, typename... Args> class scheduler {
 };
 } // namespace trigger_mgr
 
+namespace detail {
 template <stdx::ct_string Name, typename... Args>
 class trigger_scheduler
     : public trigger_mgr::scheduler<trigger_scheduler<Name, Args...>,
@@ -148,6 +152,15 @@ class trigger_scheduler
                                                    trigger_scheduler)
         -> bool = default;
 };
+} // namespace detail
+
+template <stdx::ct_string Name, typename... Args>
+using trigger_scheduler =
+    detail::trigger_scheduler<Name, trigger_mgr::queue_at_back, Args...>;
+
+template <stdx::ct_string Name, typename... Args>
+using urgent_trigger_scheduler =
+    detail::trigger_scheduler<Name, trigger_mgr::queue_at_front, Args...>;
 
 struct trigger_scheduler_sender_t;
 

test/schedulers/trigger_scheduler.cpp

@@ -355,3 +355,22 @@ TEST_CASE("thread safety for immediate execution", "[trigger_scheduler]") {
     CHECK(var2 == 42);
     CHECK(async::triggers<stdx::cts_t<"rqp_imm">>.empty());
 }
+
+TEMPLATE_TEST_CASE(
+    "urgent_trigger_scheduler puts a task at the front of the queue",
+    "[trigger_scheduler]", decltype([] {})) {
+    constexpr auto name = type_string<TestType>;
+    auto s = async::urgent_trigger_scheduler<name>{};
+
+    int var1{};
+    async::sender auto sndr1 =
+        async::start_on(s, async::just_result_of([&] { var1 *= 2; }));
+    CHECK(async::start_detached(sndr1));
+    async::sender auto sndr2 =
+        async::start_on(s, async::just_result_of([&] { var1 += 42; }));
+    CHECK(async::start_detached(sndr2));
+
+    async::run_triggers<name>();
+    CHECK(var1 == 84);
+    CHECK(async::triggers<stdx::cts_t<name>>.empty());
+}