From 222c57e1712d1c082630e49c8f79cda697321b24 Mon Sep 17 00:00:00 2001 From: yulin25 Date: Fri, 21 Aug 2026 15:13:00 +0800 Subject: [PATCH 1/3] Fix start_scheduler query of __receiver_proxy_base. When the actual start scheduler cannot be used to construct a task_scheduler, fall back to inline_scheduler instead of failing compilation. When constructing the task_scheduler throws an exception, fall back to inline_scheduler instead of terminating. --- include/stdexec/__detail/__task_scheduler.hpp | 54 ++++++++++++------- 1 file changed, 36 insertions(+), 18 deletions(-) diff --git a/include/stdexec/__detail/__task_scheduler.hpp b/include/stdexec/__detail/__task_scheduler.hpp index 476a2591d..409aef279 100644 --- a/include/stdexec/__detail/__task_scheduler.hpp +++ b/include/stdexec/__detail/__task_scheduler.hpp @@ -1,4 +1,4 @@ -/* +/* * Copyright (c) 2026 NVIDIA Corporation * * Licensed under the Apache License Version 2.0 with LLVM Exceptions @@ -308,7 +308,7 @@ namespace STDEXEC _Rcvr __rcvr) noexcept : __rcvr_proxy_(std::move(__rcvr)) , __backend_(std::move(__backend)) - {} + { } constexpr void start() noexcept { @@ -329,7 +329,7 @@ namespace STDEXEC constexpr explicit __sender(task_scheduler __sch) noexcept : __attrs_{std::move(__sch)} - {} + { } template [[nodiscard]] @@ -507,7 +507,7 @@ namespace STDEXEC , __fn_(std::move(__fn)) , __shape_(__shape) , __backend_(std::move(__backend)) - {} + { } constexpr void set_value() noexcept final { @@ -566,7 +566,7 @@ namespace STDEXEC __any_task_scheduler_backend __backend) : __state_{std::move(__rcvr), __shape, std::move(__fn), std::move(__backend)} , __opstate1_(STDEXEC::connect(static_cast<_Sndr&&>(__sndr), __rcvr_t{&__state_})) - {} + { } constexpr void start() noexcept { @@ -591,7 +591,7 @@ namespace STDEXEC constexpr explicit __bulk_sender(_Sndr __sndr, task_scheduler __sch) : __sndr_(std::move(__sndr)) , __attrs_{std::move(__sch)} - {} + { } template constexpr auto connect(_Rcvr __rcvr) && @@ -690,7 +690,7 @@ namespace STDEXEC : _Alloc(std::move(__alloc)) , __opstate_(STDEXEC::connect(std::move(__sndr), __receiver_t{__rcvr, this, __destroy_pfn_[__in_situ]})) - {} + { } __opstate(__opstate&&) = delete; @@ -751,11 +751,11 @@ namespace STDEXEC bool const __in_situ = __storage.size() >= sizeof(__opstate_t); _Alloc const & __alloc = *this; auto& __opstate = __task::__emplace_into<__opstate_t>(__storage, - __alloc, - __alloc, - static_cast<_Sndr&&>(__sndr), - __rcvr_proxy, - __in_situ); + __alloc, + __alloc, + static_cast<_Sndr&&>(__sndr), + __rcvr_proxy, + __in_situ); STDEXEC::start(__opstate); } STDEXEC_CATCH_ALL @@ -775,7 +775,7 @@ namespace STDEXEC constexpr explicit __backend_for(_Sch __sch, _Alloc __alloc) noexcept : _Alloc(std::move(__alloc)) , __sch_(std::move(__sch)) - {} + { } constexpr void schedule(parallel_scheduler_replacement::receiver_proxy& __rcvr_proxy, std::span __storage) noexcept @@ -847,14 +847,32 @@ namespace STDEXEC if (__value_type == __mtypeid) { auto& __val = *static_cast*>(__dest); - if constexpr (__callable>) + + constexpr bool may_as_task_scheduler = [] { - __val.emplace(get_start_scheduler(get_env(__rcvr_))); - } - else + if constexpr (__callable>) + { + return __std::constructible_from< + task_scheduler, + __call_result_t>>; + } + else + { + return false; + } + }(); + + if constexpr (may_as_task_scheduler) { - __val.emplace(inline_scheduler{}); + try + { + __val.emplace(get_start_scheduler(get_env(__rcvr_))); + return; + } + catch (...) + { } } + __val.emplace(inline_scheduler{}); } } } // namespace __detail From 921b1983bdf6399242cf2435a4757469d6ac2839 Mon Sep 17 00:00:00 2001 From: yulin Date: Fri, 21 Aug 2026 19:03:37 +0800 Subject: [PATCH 2/3] Add a regression test. Parallel scheduler in an environment with itself as the start scheduler. --- .../schedulers/test_parallel_scheduler.cpp | 67 ++++++++++--------- 1 file changed, 37 insertions(+), 30 deletions(-) diff --git a/test/stdexec/schedulers/test_parallel_scheduler.cpp b/test/stdexec/schedulers/test_parallel_scheduler.cpp index 76bf85cc6..f713925bf 100644 --- a/test/stdexec/schedulers/test_parallel_scheduler.cpp +++ b/test/stdexec/schedulers/test_parallel_scheduler.cpp @@ -140,6 +140,13 @@ TEST_CASE("simple chain task on parallel scheduler", "[scheduler][parallel_sched (void) snd2; } +TEST_CASE("parallel scheduler used in an already scheduled environment", + "[scheduler][parallel_scheduler]") +{ + auto snd = ex::schedule(ex::get_parallel_scheduler()); + ex::sync_wait(ex::starts_on(ex::get_parallel_scheduler(), snd)); +} + TEST_CASE("checks stop_token before starting the work", "[scheduler][parallel_scheduler]") { STDEXEC::parallel_scheduler sched = STDEXEC::get_parallel_scheduler(); @@ -303,15 +310,15 @@ TEST_CASE("bulk_chunked on parallel_scheduler performs chunking", "[scheduler][p STDEXEC::parallel_scheduler sched = STDEXEC::get_parallel_scheduler(); auto bulk_snd = ex::bulk_chunked(ex::schedule(sched), - ex::par, - 10'000, - [&](int b, int e) - { + ex::par, + 10'000, + [&](int b, int e) + { if (e - b > 1) { has_chunking = true; } - }); + }); ex::sync_wait(std::move(bulk_snd)); REQUIRE(has_chunking.load()); @@ -325,15 +332,15 @@ TEST_CASE("bulk_chunked on parallel_scheduler covers the entire range", STDEXEC::parallel_scheduler sched = STDEXEC::get_parallel_scheduler(); auto bulk_snd = ex::bulk_chunked(ex::schedule(sched), - ex::par, - num_tasks, - [&](size_t b, size_t e) - { + ex::par, + num_tasks, + [&](size_t b, size_t e) + { for (auto i = b; i < e; ++i) { covered[i] = true; } - }); + }); ex::sync_wait(std::move(bulk_snd)); for (size_t i = 0; i < num_tasks; ++i) @@ -350,14 +357,14 @@ TEST_CASE("bulk_chunked with seq on parallel_scheduler doesn't do chunking", STDEXEC::parallel_scheduler sched = STDEXEC::get_parallel_scheduler(); auto bulk_snd = ex::bulk_chunked(ex::schedule(sched), - ex::seq, - num_tasks, - [&](size_t b, size_t e) - { + ex::seq, + num_tasks, + [&](size_t b, size_t e) + { REQUIRE(b == 0); REQUIRE(e == num_tasks); execution_count++; - }); + }); ex::sync_wait(std::move(bulk_snd)); REQUIRE(execution_count.load() == 1); @@ -422,7 +429,7 @@ struct terminal_bulk_scheduler_backend_impl : scr::parallel_scheduler_backend { explicit terminal_bulk_scheduler_backend_impl(bulk_completion_kind completion) noexcept : completion_(completion) - {} + { } void schedule(scr::receiver_proxy& r, std::span) noexcept override { @@ -475,7 +482,7 @@ struct backend_factory_guard { explicit backend_factory_guard(scr::__parallel_scheduler_backend_factory_t factory) : old_factory_(scr::set_parallel_scheduler_backend(factory)) - {} + { } ~backend_factory_guard() { @@ -555,7 +562,7 @@ struct tracked_value_sender tracked_value_sender(STDEXEC::parallel_scheduler sched, std::shared_ptr> live) : sched_(std::move(sched)) , value_(std::move(live)) - {} + { } auto get_env() const noexcept -> env { @@ -624,7 +631,7 @@ TEST_CASE("bulk on parallel_scheduler destroys stored predecessor values", { auto sched = STDEXEC::get_parallel_scheduler(); auto snd = tracked_value_sender{sched, live} - | ex::bulk(ex::par, 16, [](std::size_t, destructor_tracked_value&) noexcept {}); + | ex::bulk(ex::par, 16, [](std::size_t, destructor_tracked_value&) noexcept { }); auto result = ex::sync_wait(std::move(snd)); REQUIRE(result.has_value()); @@ -642,8 +649,8 @@ TEST_CASE("bulk on parallel_scheduler destroys stored predecessor values after e { STDEXEC::parallel_scheduler sched = STDEXEC::get_parallel_scheduler(); - auto snd = tracked_value_sender{sched, live} - | ex::bulk(ex::par, 16, [](std::size_t, destructor_tracked_value&) noexcept {}); + auto snd = tracked_value_sender{sched, live} + | ex::bulk(ex::par, 16, [](std::size_t, destructor_tracked_value&) noexcept { }); CHECK_THROWS_AS(ex::sync_wait(std::move(snd)), std::runtime_error); } @@ -660,8 +667,8 @@ TEST_CASE("bulk on parallel_scheduler destroys stored predecessor values after s { STDEXEC::parallel_scheduler sched = STDEXEC::get_parallel_scheduler(); - auto snd = tracked_value_sender{sched, live} - | ex::bulk(ex::par, 16, [](std::size_t, destructor_tracked_value&) noexcept {}); + auto snd = tracked_value_sender{sched, live} + | ex::bulk(ex::par, 16, [](std::size_t, destructor_tracked_value&) noexcept { }); auto result = ex::sync_wait(std::move(snd)); CHECK_FALSE(result.has_value()); @@ -675,13 +682,13 @@ TEST_CASE("empty environment always returns nullopt for any query", { struct my_receiver : scr::receiver_proxy { - void __query_env(ex::__type_index, ex::__type_index, void*) const noexcept override {} + void __query_env(ex::__type_index, ex::__type_index, void*) const noexcept override { } - void set_value() noexcept override {} + void set_value() noexcept override { } - void set_error(std::exception_ptr) noexcept override {} + void set_error(std::exception_ptr) noexcept override { } - void set_stopped() noexcept override {} + void set_stopped() noexcept override { } }; my_receiver rcvr{}; @@ -696,11 +703,11 @@ TEST_CASE("environment with a stop token can expose its stop token", { struct my_receiver : ex::parallel_scheduler_replacement::receiver_proxy { - void set_value() noexcept override {} + void set_value() noexcept override { } - void set_error(std::exception_ptr) noexcept override {} + void set_error(std::exception_ptr) noexcept override { } - void set_stopped() noexcept override {} + void set_stopped() noexcept override { } protected: void From 10da83e30f3023a0fc88314431223d352689cbae Mon Sep 17 00:00:00 2001 From: yulin Date: Fri, 21 Aug 2026 19:43:48 +0800 Subject: [PATCH 3/3] Run clang-format-21. --- include/stdexec/__detail/__task_scheduler.hpp | 26 ++++---- .../schedulers/test_parallel_scheduler.cpp | 60 +++++++++---------- 2 files changed, 43 insertions(+), 43 deletions(-) diff --git a/include/stdexec/__detail/__task_scheduler.hpp b/include/stdexec/__detail/__task_scheduler.hpp index 409aef279..5dc58b95d 100644 --- a/include/stdexec/__detail/__task_scheduler.hpp +++ b/include/stdexec/__detail/__task_scheduler.hpp @@ -308,7 +308,7 @@ namespace STDEXEC _Rcvr __rcvr) noexcept : __rcvr_proxy_(std::move(__rcvr)) , __backend_(std::move(__backend)) - { } + {} constexpr void start() noexcept { @@ -329,7 +329,7 @@ namespace STDEXEC constexpr explicit __sender(task_scheduler __sch) noexcept : __attrs_{std::move(__sch)} - { } + {} template [[nodiscard]] @@ -507,7 +507,7 @@ namespace STDEXEC , __fn_(std::move(__fn)) , __shape_(__shape) , __backend_(std::move(__backend)) - { } + {} constexpr void set_value() noexcept final { @@ -566,7 +566,7 @@ namespace STDEXEC __any_task_scheduler_backend __backend) : __state_{std::move(__rcvr), __shape, std::move(__fn), std::move(__backend)} , __opstate1_(STDEXEC::connect(static_cast<_Sndr&&>(__sndr), __rcvr_t{&__state_})) - { } + {} constexpr void start() noexcept { @@ -591,7 +591,7 @@ namespace STDEXEC constexpr explicit __bulk_sender(_Sndr __sndr, task_scheduler __sch) : __sndr_(std::move(__sndr)) , __attrs_{std::move(__sch)} - { } + {} template constexpr auto connect(_Rcvr __rcvr) && @@ -690,7 +690,7 @@ namespace STDEXEC : _Alloc(std::move(__alloc)) , __opstate_(STDEXEC::connect(std::move(__sndr), __receiver_t{__rcvr, this, __destroy_pfn_[__in_situ]})) - { } + {} __opstate(__opstate&&) = delete; @@ -751,11 +751,11 @@ namespace STDEXEC bool const __in_situ = __storage.size() >= sizeof(__opstate_t); _Alloc const & __alloc = *this; auto& __opstate = __task::__emplace_into<__opstate_t>(__storage, - __alloc, - __alloc, - static_cast<_Sndr&&>(__sndr), - __rcvr_proxy, - __in_situ); + __alloc, + __alloc, + static_cast<_Sndr&&>(__sndr), + __rcvr_proxy, + __in_situ); STDEXEC::start(__opstate); } STDEXEC_CATCH_ALL @@ -775,7 +775,7 @@ namespace STDEXEC constexpr explicit __backend_for(_Sch __sch, _Alloc __alloc) noexcept : _Alloc(std::move(__alloc)) , __sch_(std::move(__sch)) - { } + {} constexpr void schedule(parallel_scheduler_replacement::receiver_proxy& __rcvr_proxy, std::span __storage) noexcept @@ -870,7 +870,7 @@ namespace STDEXEC return; } catch (...) - { } + {} } __val.emplace(inline_scheduler{}); } diff --git a/test/stdexec/schedulers/test_parallel_scheduler.cpp b/test/stdexec/schedulers/test_parallel_scheduler.cpp index f713925bf..a206aa3e8 100644 --- a/test/stdexec/schedulers/test_parallel_scheduler.cpp +++ b/test/stdexec/schedulers/test_parallel_scheduler.cpp @@ -310,15 +310,15 @@ TEST_CASE("bulk_chunked on parallel_scheduler performs chunking", "[scheduler][p STDEXEC::parallel_scheduler sched = STDEXEC::get_parallel_scheduler(); auto bulk_snd = ex::bulk_chunked(ex::schedule(sched), - ex::par, - 10'000, - [&](int b, int e) - { + ex::par, + 10'000, + [&](int b, int e) + { if (e - b > 1) { has_chunking = true; } - }); + }); ex::sync_wait(std::move(bulk_snd)); REQUIRE(has_chunking.load()); @@ -332,15 +332,15 @@ TEST_CASE("bulk_chunked on parallel_scheduler covers the entire range", STDEXEC::parallel_scheduler sched = STDEXEC::get_parallel_scheduler(); auto bulk_snd = ex::bulk_chunked(ex::schedule(sched), - ex::par, - num_tasks, - [&](size_t b, size_t e) - { + ex::par, + num_tasks, + [&](size_t b, size_t e) + { for (auto i = b; i < e; ++i) { covered[i] = true; } - }); + }); ex::sync_wait(std::move(bulk_snd)); for (size_t i = 0; i < num_tasks; ++i) @@ -357,14 +357,14 @@ TEST_CASE("bulk_chunked with seq on parallel_scheduler doesn't do chunking", STDEXEC::parallel_scheduler sched = STDEXEC::get_parallel_scheduler(); auto bulk_snd = ex::bulk_chunked(ex::schedule(sched), - ex::seq, - num_tasks, - [&](size_t b, size_t e) - { + ex::seq, + num_tasks, + [&](size_t b, size_t e) + { REQUIRE(b == 0); REQUIRE(e == num_tasks); execution_count++; - }); + }); ex::sync_wait(std::move(bulk_snd)); REQUIRE(execution_count.load() == 1); @@ -429,7 +429,7 @@ struct terminal_bulk_scheduler_backend_impl : scr::parallel_scheduler_backend { explicit terminal_bulk_scheduler_backend_impl(bulk_completion_kind completion) noexcept : completion_(completion) - { } + {} void schedule(scr::receiver_proxy& r, std::span) noexcept override { @@ -482,7 +482,7 @@ struct backend_factory_guard { explicit backend_factory_guard(scr::__parallel_scheduler_backend_factory_t factory) : old_factory_(scr::set_parallel_scheduler_backend(factory)) - { } + {} ~backend_factory_guard() { @@ -562,7 +562,7 @@ struct tracked_value_sender tracked_value_sender(STDEXEC::parallel_scheduler sched, std::shared_ptr> live) : sched_(std::move(sched)) , value_(std::move(live)) - { } + {} auto get_env() const noexcept -> env { @@ -631,7 +631,7 @@ TEST_CASE("bulk on parallel_scheduler destroys stored predecessor values", { auto sched = STDEXEC::get_parallel_scheduler(); auto snd = tracked_value_sender{sched, live} - | ex::bulk(ex::par, 16, [](std::size_t, destructor_tracked_value&) noexcept { }); + | ex::bulk(ex::par, 16, [](std::size_t, destructor_tracked_value&) noexcept {}); auto result = ex::sync_wait(std::move(snd)); REQUIRE(result.has_value()); @@ -649,8 +649,8 @@ TEST_CASE("bulk on parallel_scheduler destroys stored predecessor values after e { STDEXEC::parallel_scheduler sched = STDEXEC::get_parallel_scheduler(); - auto snd = tracked_value_sender{sched, live} - | ex::bulk(ex::par, 16, [](std::size_t, destructor_tracked_value&) noexcept { }); + auto snd = tracked_value_sender{sched, live} + | ex::bulk(ex::par, 16, [](std::size_t, destructor_tracked_value&) noexcept {}); CHECK_THROWS_AS(ex::sync_wait(std::move(snd)), std::runtime_error); } @@ -667,8 +667,8 @@ TEST_CASE("bulk on parallel_scheduler destroys stored predecessor values after s { STDEXEC::parallel_scheduler sched = STDEXEC::get_parallel_scheduler(); - auto snd = tracked_value_sender{sched, live} - | ex::bulk(ex::par, 16, [](std::size_t, destructor_tracked_value&) noexcept { }); + auto snd = tracked_value_sender{sched, live} + | ex::bulk(ex::par, 16, [](std::size_t, destructor_tracked_value&) noexcept {}); auto result = ex::sync_wait(std::move(snd)); CHECK_FALSE(result.has_value()); @@ -682,13 +682,13 @@ TEST_CASE("empty environment always returns nullopt for any query", { struct my_receiver : scr::receiver_proxy { - void __query_env(ex::__type_index, ex::__type_index, void*) const noexcept override { } + void __query_env(ex::__type_index, ex::__type_index, void*) const noexcept override {} - void set_value() noexcept override { } + void set_value() noexcept override {} - void set_error(std::exception_ptr) noexcept override { } + void set_error(std::exception_ptr) noexcept override {} - void set_stopped() noexcept override { } + void set_stopped() noexcept override {} }; my_receiver rcvr{}; @@ -703,11 +703,11 @@ TEST_CASE("environment with a stop token can expose its stop token", { struct my_receiver : ex::parallel_scheduler_replacement::receiver_proxy { - void set_value() noexcept override { } + void set_value() noexcept override {} - void set_error(std::exception_ptr) noexcept override { } + void set_error(std::exception_ptr) noexcept override {} - void set_stopped() noexcept override { } + void set_stopped() noexcept override {} protected: void