✨ Use senders to achieve callback request-response semantics

Author: elbeno

CMakeLists.txt

@@ -24,14 +24,15 @@ include(cmake/string_catalog.cmake)
 
 add_versioned_package("gh:boostorg/mp11#boost-1.83.0")
 fmt_recipe(10.2.1)
-add_versioned_package("gh:intel/cpp-std-extensions#6c4fb17")
-add_versioned_package("gh:intel/cpp-baremetal-concurrency#8d49b6d")
+add_versioned_package("gh:intel/cpp-std-extensions#646bdbe")
+add_versioned_package("gh:intel/cpp-baremetal-concurrency#fef18ca")
+add_versioned_package("gh:intel/cpp-baremetal-senders-and-receivers#113eeff")
 
 add_library(cib INTERFACE)
 target_compile_features(cib INTERFACE cxx_std_20)
 target_include_directories(cib INTERFACE include)
-target_link_libraries_system(cib INTERFACE concurrency fmt::fmt-header-only
-                             stdx)
+target_link_libraries_system(cib INTERFACE async concurrency
+                             fmt::fmt-header-only stdx)
 
 target_compile_options(
     cib

include/msg/send.hpp

@@ -0,0 +1,124 @@
+#pragma once
+
+#include <async/completion_tags.hpp>
+#include <async/concepts.hpp>
+#include <async/connect.hpp>
+#include <async/schedulers/trigger_scheduler.hpp>
+#include <async/start.hpp>
+#include <async/then.hpp>
+
+#include <stdx/concepts.hpp>
+#include <stdx/ct_string.hpp>
+
+#include <type_traits>
+#include <utility>
+
+namespace msg {
+namespace _send_recv {
+template <typename Sched>
+using scheduler_sender = decltype(std::declval<Sched>().schedule());
+
+template <typename S>
+concept valid_send_action =
+    requires { typename std::remove_cvref_t<S>::is_send_action; };
+
+template <valid_send_action SA, typename Sched, typename Rcvr>
+// NOLINTNEXTLINE(cppcoreguidelines-special-member-functions)
+struct op_state {
+    template <stdx::same_as_unqualified<SA> S, typename Sch, typename R>
+    constexpr op_state(S &&s, Sch &&sch, R &&r)
+        : send(s), ops{async::connect(std::forward<Sch>(sch).schedule(),
+                                      std::forward<R>(r))} {}
+    constexpr op_state(op_state &&) = delete;
+
+    using ops_t = async::connect_result_t<scheduler_sender<Sched>, Rcvr>;
+
+    constexpr auto start() & -> void {
+        async::start(ops);
+        send();
+    }
+
+    SA send;
+    ops_t ops;
+};
+
+template <valid_send_action SA, typename Sched> struct sender {
+    using is_sender = void;
+
+    [[no_unique_address]] SA send;
+    [[no_unique_address]] Sched sched;
+
+  public:
+    template <async::receiver R>
+    [[nodiscard]] constexpr auto
+    connect(R &&r) && -> op_state<SA, Sched, std::remove_cvref_t<R>> {
+        async::check_connect<sender &&, R>();
+        return {std::move(send), std::move(sched), std::forward<R>(r)};
+    }
+
+    template <async::receiver R>
+    [[nodiscard]] constexpr auto
+    connect(R &&r) const & -> op_state<SA, Sched, std::remove_cvref_t<R>> {
+        async::check_connect<sender, R>();
+        return {send, sched, std::forward<R>(r)};
+    }
+
+    template <typename Env>
+    [[nodiscard]] constexpr static auto get_completion_signatures(Env const &)
+        -> async::completion_signatures_of_t<scheduler_sender<Sched>, Env> {
+        return {};
+    }
+};
+
+template <typename Sndr, typename Sched>
+sender(Sndr, Sched) -> sender<Sndr, Sched>;
+
+template <typename F> struct send_action : F {
+    using is_send_action = void;
+};
+template <typename F> send_action(F) -> send_action<F>;
+
+template <stdx::ct_string Name, typename... Args> struct pipeable {
+    template <typename Adaptor> struct type {
+        [[no_unique_address]] Adaptor a;
+
+      private:
+        template <valid_send_action S, stdx::same_as_unqualified<type> Self>
+        friend constexpr auto operator|(S &&s, Self &&self) -> async::sender
+            auto {
+            return _send_recv::sender{
+                       std::forward<S>(s),
+                       async::trigger_scheduler<Name, Args...>{}} |
+                   std::forward<Self>(self).a;
+        }
+    };
+
+    template <typename T> type(T) -> type<T>;
+};
+} // namespace _send_recv
+
+template <typename F, typename... Args>
+constexpr auto send(F &&f, Args &&...args) {
+    return _send_recv::send_action{
+        [f = std::forward<F>(f), ... args = std::forward<Args>(args)]() {
+            return f(args...);
+        }};
+}
+
+template <stdx::ct_string Name, typename... RecvArgs, typename F,
+          typename... Args>
+[[nodiscard]] constexpr auto then_receive(F &&f, Args &&...args) {
+    return typename _send_recv::pipeable<Name, RecvArgs...>::type{async::then(
+        [f = std::forward<F>(f), ... args = std::forward<Args>(args)](
+            RecvArgs const &...as) { return f(as..., args...); })};
+}
+
+template <stdx::ct_string Name, _send_recv::valid_send_action S, typename F,
+          typename... Args>
+[[nodiscard]] constexpr auto then_receive(S &&s, F &&f,
+                                          Args &&...args) -> async::sender
+    auto {
+    return std::forward<S>(s) |
+           then_receive<Name>(std::forward<F>(f), std::forward<Args>(args)...);
+}
+} // namespace msg

test/CMakeLists.txt

@@ -67,6 +67,7 @@ add_tests(
     msg/indexed_callback
     msg/indexed_handler
     msg/message
+    msg/send
     sc/format
     sc/string_constant
     seq/sequencer)

test/msg/send.cpp

@@ -0,0 +1,81 @@
+#include <async/schedulers/trigger_manager.hpp>
+#include <async/start_detached.hpp>
+#include <cib/cib.hpp>
+#include <msg/callback.hpp>
+#include <msg/field.hpp>
+#include <msg/message.hpp>
+#include <msg/send.hpp>
+#include <msg/service.hpp>
+
+#include <stdx/ct_conversions.hpp>
+#include <stdx/ct_string.hpp>
+
+#include <catch2/catch_template_test_macros.hpp>
+#include <catch2/catch_test_macros.hpp>
+
+namespace {
+template <typename T>
+constexpr auto type_string =
+    stdx::ct_string<stdx::type_as_string<T>().size() + 1>{
+        stdx::type_as_string<T>()};
+} // namespace
+
+TEMPLATE_TEST_CASE("request-response", "[send]", decltype([] {})) {
+    constexpr auto name = type_string<TestType>;
+    int var{};
+
+    auto s = msg::send([&](auto i) { async::run_triggers<name>(i); }, 42) |
+             msg::then_receive<name, int>(
+                 [&](auto recvd, auto x) { var = recvd + x; }, 17);
+    CHECK(var == 0);
+    CHECK(async::start_detached_unstoppable(s));
+    CHECK(var == 59);
+}
+
+namespace {
+using msg::at;
+using msg::operator""_dw;
+using msg::operator""_msb;
+using msg::operator""_lsb;
+using msg::operator""_f;
+
+using id_field =
+    msg::field<"id", std::uint32_t>::located<at{0_dw, 31_msb, 24_lsb}>;
+using field1 =
+    msg::field<"f1", std::uint32_t>::located<at{0_dw, 15_msb, 0_lsb}>;
+using field2 =
+    msg::field<"f2", std::uint32_t>::located<at{1_dw, 23_msb, 16_lsb}>;
+using field3 =
+    msg::field<"f3", std::uint32_t>::located<at{1_dw, 15_msb, 0_lsb}>;
+
+using msg_defn = msg::message<"msg", id_field, field1, field2, field3>;
+using test_msg_t = msg::owning<msg_defn>;
+using msg_view_t = msg::const_view<msg_defn>;
+
+constexpr auto test_callback = msg::callback<"cb", msg_defn>(
+    "id"_f == msg::constant<0x80>,
+    [](msg_view_t v) { async::run_triggers<"cb">(v); });
+
+struct test_service : msg::service<msg_view_t> {};
+struct test_project {
+    constexpr static auto config = cib::config(
+        cib::exports<test_service>, cib::extend<test_service>(test_callback));
+};
+} // namespace
+
+TEST_CASE("request-response through handler", "[send]") {
+    cib::nexus<test_project> test_nexus{};
+    test_nexus.init();
+
+    std::uint32_t var{};
+    auto s =
+        msg::send(
+            [&](auto id) {
+                cib::service<test_service>->handle(test_msg_t{"id"_f = id});
+            },
+            0x80) |
+        msg::then_receive<"cb", msg_view_t>(
+            [&](auto v) { var = v.get("id"_f); });
+    CHECK(async::start_detached_unstoppable(s));
+    CHECK(var == 0x80);
+}