Handle reference error types in ensure_started and let_error

libs/pika/execution/include/pika/execution/algorithms/ensure_started.hpp

@@ -115,8 +115,10 @@ namespace pika::execution::experimental {
             template <template <typename...> class Variant>
             using error_types =
                 pika::util::detail::unique_t<pika::util::detail::prepend_t<
-                    typename pika::execution::experimental::sender_traits<
-                        Sender>::template error_types<Variant>,
+                    pika::util::detail::transform_t<
+                        typename pika::execution::experimental::sender_traits<
+                            Sender>::template error_types<Variant>,
+                        std::decay>,
                     std::exception_ptr>>;
 
             static constexpr bool sends_done = false;

libs/pika/execution/include/pika/execution/algorithms/let_error.hpp

@@ -59,10 +59,10 @@ namespace pika { namespace execution { namespace experimental {
 
             // Type of the potential errors returned from the predecessor sender
             template <template <typename...> class Variant>
-            using predecessor_error_types =
+            using predecessor_error_types = pika::util::detail::transform_t<
                 typename pika::execution::experimental::sender_traits<
-                    std::decay_t<PredecessorSender>>::
-                    template error_types<Variant>;
+                    PredecessorSender>::template error_types<Variant>,
+                std::decay>;
             static_assert(
                 !std::is_same<predecessor_error_types<pika::util::pack>,
                     pika::util::pack<>>::value,
@@ -210,7 +210,7 @@ namespace pika { namespace execution { namespace experimental {
                                 // TODO: receiver is moved before the visit, but
                                 // the invoke inside the visit may throw.
                                 r.op_state.predecessor_error
-                                    .template emplace<Error>(
+                                    .template emplace<std::decay_t<Error>>(
                                         PIKA_FORWARD(Error, error));
                                 pika::detail::visit(
                                     set_error_visitor{PIKA_MOVE(r.receiver),

libs/pika/execution/tests/unit/algorithm_bulk.cpp

@@ -136,6 +136,23 @@ int main()
         PIKA_TEST_EQ(set_value_count, 30);
     }
 
+    {
+        std::atomic<bool> set_value_called{false};
+        std::atomic<int> set_value_count{0};
+        int x = 42;
+        auto s = ex::bulk(ex::just(const_reference_sender<decltype(x)>{x}), 10,
+            [&](int n, auto&&) {
+                PIKA_TEST_EQ(n, set_value_count);
+                ++set_value_count;
+            });
+        auto f = [](auto x) { PIKA_TEST_EQ(x.x, 42); };
+        auto r = callback_receiver<decltype(f)>{f, set_value_called};
+        auto os = ex::connect(std::move(s), std::move(r));
+        ex::start(os);
+        PIKA_TEST(set_value_called);
+        PIKA_TEST_EQ(set_value_count, 10);
+    }
+
     // operator| overload
     {
         std::atomic<bool> set_value_called{false};
@@ -234,6 +251,16 @@ int main()
         PIKA_TEST_EQ(custom_bulk_call_count, 3);
     }
 
+    {
+        std::atomic<bool> set_error_called{false};
+        auto s = ex::bulk(const_reference_error_sender{}, 0, [](int) {});
+        auto r = error_callback_receiver<decltype(check_exception_ptr)>{
+            check_exception_ptr, set_error_called};
+        auto os = ex::connect(std::move(s), std::move(r));
+        ex::start(os);
+        PIKA_TEST(set_error_called);
+    }
+
     test_adl_isolation(ex::bulk(ex::just(), 1, my_namespace::my_type{}));
 #endif
 

libs/pika/execution/tests/unit/algorithm_drop_value.cpp

@@ -91,6 +91,16 @@ int main()
         PIKA_TEST(set_value_called);
     }
 
+    {
+        std::atomic<bool> set_value_called{false};
+        int x = 42;
+        auto s = ex::drop_value(const_reference_sender<decltype(x)>{x});
+        auto r = callback_receiver<decltype(f)>{f, set_value_called};
+        auto os = ex::connect(std::move(s), std::move(r));
+        ex::start(os);
+        PIKA_TEST(set_value_called);
+    }
+
     // operator| overload
     {
         std::atomic<bool> set_value_called{false};
@@ -113,6 +123,16 @@ int main()
         PIKA_TEST(set_error_called);
     }
 
+    {
+        std::atomic<bool> set_error_called{false};
+        auto s = ex::drop_value(const_reference_error_sender{});
+        auto r = error_callback_receiver<decltype(check_exception_ptr)>{
+            check_exception_ptr, set_error_called};
+        auto os = ex::connect(std::move(s), std::move(r));
+        ex::start(os);
+        PIKA_TEST(set_error_called);
+    }
+
     test_adl_isolation(ex::drop_value(my_namespace::my_sender{}));
 
     return pika::util::report_errors();

libs/pika/execution/tests/unit/algorithm_ensure_started.cpp

@@ -146,6 +146,16 @@ int main()
         PIKA_TEST(set_error_called);
     }
 
+    {
+        std::atomic<bool> set_error_called{false};
+        auto s = const_reference_error_sender{} | ex::ensure_started();
+        auto r = error_callback_receiver<decltype(check_exception_ptr)>{
+            check_exception_ptr, set_error_called};
+        auto os = ex::connect(std::move(s), std::move(r));
+        ex::start(os);
+        PIKA_TEST(set_error_called);
+    }
+
     {
         std::atomic<bool> set_error_called{false};
         auto s = error_sender{} | ex::ensure_started() | ex::ensure_started() |

libs/pika/execution/tests/unit/algorithm_let_error.cpp

@@ -47,6 +47,25 @@ int main()
         PIKA_TEST(let_error_callback_called);
     }
 
+#if !defined(PIKA_HAVE_P2300_REFERENCE_IMPLEMENTATION)
+    {
+        std::atomic<bool> set_value_called{false};
+        std::atomic<bool> let_error_callback_called{false};
+        auto s1 = const_reference_error_sender{};
+        auto s2 = ex::let_error(std::move(s1), [&](std::exception_ptr ep) {
+            check_exception_ptr(ep);
+            let_error_callback_called = true;
+            return void_sender();
+        });
+        auto f = [] {};
+        auto r = callback_receiver<decltype(f)>{f, set_value_called};
+        auto os = ex::connect(std::move(s2), std::move(r));
+        ex::start(os);
+        PIKA_TEST(set_value_called);
+        PIKA_TEST(let_error_callback_called);
+    }
+#endif
+
     {
         std::atomic<bool> set_value_called{false};
         std::atomic<bool> let_error_callback_called{false};

libs/pika/execution/tests/unit/algorithm_then.cpp

@@ -75,6 +75,18 @@ int main()
         PIKA_TEST(set_value_called);
     }
 
+    {
+        std::atomic<bool> set_value_called{false};
+        int x = 42;
+        auto s = ex::then(const_reference_sender<decltype(x)>{x},
+            [](int i) { return i + 1; });
+        auto f = [](int x) { PIKA_TEST_EQ(x, 43); };
+        auto r = callback_receiver<decltype(f)>{f, set_value_called};
+        auto os = ex::connect(std::move(s), std::move(r));
+        ex::start(os);
+        PIKA_TEST(set_value_called);
+    }
+
     {
         std::atomic<bool> set_value_called{false};
         auto s = ex::then(ex::just(custom_type_non_default_constructible{0}),
@@ -174,6 +186,16 @@ int main()
         PIKA_TEST(set_error_called);
     }
 
+    {
+        std::atomic<bool> set_error_called{false};
+        auto s = ex::then(const_reference_error_sender{}, [] {});
+        auto r = error_callback_receiver<decltype(check_exception_ptr)>{
+            check_exception_ptr, set_error_called};
+        auto os = ex::connect(std::move(s), std::move(r));
+        ex::start(os);
+        PIKA_TEST(set_error_called);
+    }
+
     {
         std::atomic<bool> set_error_called{false};
         auto s1 = ex::then(ex::just(0), [](int x) { return ++x; });

libs/pika/execution/tests/unit/algorithm_transfer.cpp

@@ -256,6 +256,25 @@ int main()
         PIKA_TEST(!scheduler_execute_called);
     }
 
+    {
+        std::atomic<bool> set_value_called{false};
+        std::atomic<bool> scheduler_schedule_called{false};
+        std::atomic<bool> scheduler_execute_called{false};
+        std::atomic<bool> tag_invoke_overload_called{false};
+        int x = 3;
+        auto s = ex::transfer(const_reference_sender<decltype(x)>{x},
+            scheduler{scheduler_schedule_called, scheduler_execute_called,
+                tag_invoke_overload_called});
+        auto f = [](int x) { PIKA_TEST_EQ(x, 3); };
+        auto r = callback_receiver<decltype(f)>{f, set_value_called};
+        auto os = ex::connect(std::move(s), std::move(r));
+        ex::start(os);
+        PIKA_TEST(set_value_called);
+        PIKA_TEST(!tag_invoke_overload_called);
+        PIKA_TEST(scheduler_schedule_called);
+        PIKA_TEST(!scheduler_execute_called);
+    }
+
     {
         std::atomic<bool> set_value_called{false};
         std::atomic<bool> scheduler_schedule_called{false};
@@ -472,6 +491,34 @@ int main()
         PIKA_TEST(!scheduler_execute_called);
     }
 
+    {
+        std::atomic<bool> set_error_called{false};
+        std::atomic<bool> tag_invoke_overload_called{false};
+        std::atomic<bool> scheduler_schedule_called{false};
+        std::atomic<bool> scheduler_execute_called{false};
+        auto s = ex::transfer(const_reference_error_sender{},
+            scheduler{scheduler_schedule_called, scheduler_execute_called,
+                tag_invoke_overload_called});
+        auto r = error_callback_receiver<decltype(check_exception_ptr)>{
+            check_exception_ptr, set_error_called};
+        auto os = ex::connect(std::move(s), std::move(r));
+        ex::start(os);
+        PIKA_TEST(set_error_called);
+        PIKA_TEST(!tag_invoke_overload_called);
+        // The reference implementation transfers to the given scheduler no
+        // matter the signal from the predecessor sender. Our implementation
+        // only transfers on set_value. In this particular case the reference
+        // implementation puts in more effort call set_error on the scheduler's
+        // context, but it can't be guaranteed in all cases which is why
+        // transfer doesn't provide a completion scheduler for set_error.
+#if defined(PIKA_HAVE_P2300_REFERENCE_IMPLEMENTATION)
+        PIKA_TEST(scheduler_schedule_called);
+#else
+        PIKA_TEST(!scheduler_schedule_called);
+#endif
+        PIKA_TEST(!scheduler_execute_called);
+    }
+
     test_adl_isolation(ex::transfer(ex::just(), my_namespace::my_scheduler{}));
 
     return pika::util::report_errors();