diff --git a/libs/core/executors/CMakeLists.txt b/libs/core/executors/CMakeLists.txt index 8deb1494338..2a3c42e15ae 100644 --- a/libs/core/executors/CMakeLists.txt +++ b/libs/core/executors/CMakeLists.txt @@ -26,6 +26,7 @@ set(executors_headers hpx/executors/execution_policy_parameters.hpp hpx/executors/execution_policy_scheduling_property.hpp hpx/executors/execution_policy.hpp + hpx/executors/executor_scheduler.hpp hpx/executors/explicit_scheduler_executor.hpp hpx/executors/fork_join_executor.hpp hpx/executors/limiting_executor.hpp diff --git a/libs/core/executors/include/hpx/executors/executor_scheduler.hpp b/libs/core/executors/include/hpx/executors/executor_scheduler.hpp new file mode 100644 index 00000000000..2c05a01989c --- /dev/null +++ b/libs/core/executors/include/hpx/executors/executor_scheduler.hpp @@ -0,0 +1,141 @@ +// Copyright (c) 2007-2025 Hartmut Kaiser +// Copyright (c) 2026 The STE||AR-Group +// +// SPDX-License-Identifier: BSL-1.0 +// Distributed under the Boost Software License, Version 1.0. (See accompanying +// file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt) + +/// \file parallel/executors/executor_scheduler.hpp + +#pragma once + +#include +#include +#include +#include +#include + +#include +#include +#include + +namespace hpx::execution::experimental { + + // Forward declarations + template + struct executor_scheduler; + + template + struct executor_sender; + + /////////////////////////////////////////////////////////////////////////// + template + struct executor_operation_state + { + HPX_NO_UNIQUE_ADDRESS std::decay_t exec_; + HPX_NO_UNIQUE_ADDRESS std::decay_t receiver_; + + template + executor_operation_state(Exec&& exec, Recv&& recv) + : exec_(HPX_FORWARD(Exec, exec)) + , receiver_(HPX_FORWARD(Recv, recv)) + { + } + + executor_operation_state(executor_operation_state&&) = delete; + executor_operation_state(executor_operation_state const&) = delete; + executor_operation_state& operator=( + executor_operation_state&&) = delete; + executor_operation_state& operator=( + executor_operation_state const&) = delete; + + ~executor_operation_state() = default; + + friend void tag_invoke(start_t, executor_operation_state& os) noexcept + { + hpx::detail::try_catch_exception_ptr( + [&]() { + hpx::parallel::execution::post(os.exec_, + [receiver = HPX_MOVE(os.receiver_)]() mutable { + hpx::execution::experimental::set_value( + HPX_MOVE(receiver)); + }); + }, + [&](std::exception_ptr ep) { + hpx::execution::experimental::set_error( + HPX_MOVE(os.receiver_), HPX_MOVE(ep)); + }); + } + }; + + /////////////////////////////////////////////////////////////////////////// + template + struct executor_sender + { + HPX_NO_UNIQUE_ADDRESS std::decay_t exec_; + + using completion_signatures = + hpx::execution::experimental::completion_signatures< + hpx::execution::experimental::set_value_t(), + hpx::execution::experimental::set_error_t(std::exception_ptr)>; + + template + friend auto tag_invoke( + hpx::execution::experimental::get_completion_signatures_t, + executor_sender const&, Env) noexcept -> completion_signatures; + + template + friend executor_operation_state tag_invoke( + connect_t, executor_sender&& s, Receiver&& receiver) + { + return {HPX_MOVE(s.exec_), HPX_FORWARD(Receiver, receiver)}; + } + + template + friend executor_operation_state tag_invoke( + connect_t, executor_sender const& s, Receiver&& receiver) + { + return {s.exec_, HPX_FORWARD(Receiver, receiver)}; + } + }; + + /////////////////////////////////////////////////////////////////////////// + template + struct executor_scheduler + { + using executor_type = std::decay_t; + + HPX_NO_UNIQUE_ADDRESS executor_type exec_; + + constexpr executor_scheduler() = default; + + template + requires(!std::is_same_v, executor_scheduler>) + explicit executor_scheduler(Exec&& exec) + : exec_(HPX_FORWARD(Exec, exec)) + { + } + + constexpr bool operator==(executor_scheduler const& rhs) const noexcept + { + return exec_ == rhs.exec_; + } + + constexpr bool operator!=(executor_scheduler const& rhs) const noexcept + { + return !(*this == rhs); + } + + friend executor_sender tag_invoke( + schedule_t, executor_scheduler&& sched) + { + return {HPX_MOVE(sched.exec_)}; + } + + friend executor_sender tag_invoke( + schedule_t, executor_scheduler const& sched) + { + return {sched.exec_}; + } + }; +} // namespace hpx::execution::experimental diff --git a/libs/core/executors/include/hpx/executors/parallel_executor.hpp b/libs/core/executors/include/hpx/executors/parallel_executor.hpp index be279bc9865..7e2807b4fcb 100644 --- a/libs/core/executors/include/hpx/executors/parallel_executor.hpp +++ b/libs/core/executors/include/hpx/executors/parallel_executor.hpp @@ -1210,3 +1210,20 @@ namespace hpx::execution::experimental { }; /// \endcond } // namespace hpx::execution::experimental + +// Break circular dependency: executor_scheduler.hpp includes post.hpp which +// references parallel_executor. Include it here after the class is complete. +#include + +namespace hpx::execution { + + HPX_CXX_CORE_EXPORT template + hpx::execution::experimental::executor_scheduler< + parallel_policy_executor> + tag_invoke(hpx::execution::experimental::get_scheduler_t, + parallel_policy_executor const& exec) + { + return hpx::execution::experimental::executor_scheduler< + parallel_policy_executor>(exec); + } +} // namespace hpx::execution diff --git a/libs/core/executors/include/hpx/executors/restricted_thread_pool_executor.hpp b/libs/core/executors/include/hpx/executors/restricted_thread_pool_executor.hpp index 75e9430bfe0..e86804d7d66 100644 --- a/libs/core/executors/include/hpx/executors/restricted_thread_pool_executor.hpp +++ b/libs/core/executors/include/hpx/executors/restricted_thread_pool_executor.hpp @@ -11,6 +11,7 @@ #include #include +#include #include #include #include @@ -226,6 +227,15 @@ namespace hpx::execution::experimental { } /// \endcond + friend hpx::execution::experimental::executor_scheduler< + restricted_policy_executor> + tag_invoke(hpx::execution::experimental::get_scheduler_t, + restricted_policy_executor const& exec) + { + return hpx::execution::experimental::executor_scheduler< + restricted_policy_executor>(exec); + } + private: std::uint16_t first_thread_; mutable std::atomic os_thread_; diff --git a/libs/core/executors/include/hpx/executors/sequenced_executor.hpp b/libs/core/executors/include/hpx/executors/sequenced_executor.hpp index 6d61e286ba3..20ec20afa16 100644 --- a/libs/core/executors/include/hpx/executors/sequenced_executor.hpp +++ b/libs/core/executors/include/hpx/executors/sequenced_executor.hpp @@ -10,6 +10,7 @@ #include #include +#include #include #include #include @@ -236,6 +237,15 @@ namespace hpx::execution { #endif } + friend hpx::execution::experimental::executor_scheduler< + sequenced_executor> + tag_invoke(hpx::execution::experimental::get_scheduler_t, + sequenced_executor const& exec) + { + return hpx::execution::experimental::executor_scheduler< + sequenced_executor>(exec); + } + private: friend class hpx::serialization::access; diff --git a/libs/core/executors/tests/unit/CMakeLists.txt b/libs/core/executors/tests/unit/CMakeLists.txt index 7248e7401a8..7e3c816e647 100644 --- a/libs/core/executors/tests/unit/CMakeLists.txt +++ b/libs/core/executors/tests/unit/CMakeLists.txt @@ -9,6 +9,7 @@ set(tests annotation_property created_executor execution_policy_mappings + executor_scheduler explicit_scheduler_executor fork_join_executor fork_join_executor_from diff --git a/libs/core/executors/tests/unit/executor_scheduler.cpp b/libs/core/executors/tests/unit/executor_scheduler.cpp new file mode 100644 index 00000000000..1059ba77157 --- /dev/null +++ b/libs/core/executors/tests/unit/executor_scheduler.cpp @@ -0,0 +1,113 @@ +// Copyright (c) 2026 The STE||AR-Group +// +// SPDX-License-Identifier: BSL-1.0 +// Distributed under the Boost Software License, Version 1.0. (See accompanying +// file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt) + +#include +#include +#include +#include + +#include +#include + +#include +#include +#include +#include + +/////////////////////////////////////////////////////////////////////////////// +void test_executor_scheduler_schedule() +{ + using namespace hpx::execution::experimental; + + hpx::execution::sequenced_executor exec; + + // Retrieve a P2300-compliant scheduler from the legacy executor + auto sched = get_scheduler(exec); + + // Verify the scheduler satisfies the is_scheduler trait + static_assert(is_scheduler_v, + "executor_scheduler must satisfy is_scheduler"); + + // Create a sender pipeline: schedule | then + auto s = then(schedule(sched), []() { return 42; }); + + // Execute synchronously and verify the result + auto result = hpx::this_thread::experimental::sync_wait(std::move(s)); + + HPX_TEST(result.has_value()); + HPX_TEST_EQ(hpx::get<0>(*result), 42); +} + +/////////////////////////////////////////////////////////////////////////////// +void test_executor_scheduler_schedule_parallel() +{ + using namespace hpx::execution::experimental; + + hpx::execution::parallel_executor exec; + + // Retrieve a P2300-compliant scheduler from the legacy executor + auto sched = get_scheduler(exec); + + // Verify the scheduler satisfies the is_scheduler trait + static_assert(is_scheduler_v, + "executor_scheduler must satisfy is_scheduler"); + + // Create a sender pipeline: schedule | then + auto s = then(schedule(sched), []() { return 42; }); + + // Execute synchronously and verify the result + auto result = hpx::this_thread::experimental::sync_wait(std::move(s)); + + HPX_TEST(result.has_value()); + HPX_TEST_EQ(hpx::get<0>(*result), 42); +} + +/////////////////////////////////////////////////////////////////////////////// +void test_executor_scheduler_schedule_restricted() +{ + using namespace hpx::execution::experimental; + + hpx::execution::experimental::restricted_thread_pool_executor exec; + + // Retrieve a P2300-compliant scheduler from the legacy executor + auto sched = get_scheduler(exec); + + // Verify the scheduler satisfies the is_scheduler trait + static_assert(is_scheduler_v, + "executor_scheduler must satisfy is_scheduler"); + + // Create a sender pipeline: schedule | then + auto s = then(schedule(sched), []() { return 42; }); + + // Execute synchronously and verify the result + auto result = hpx::this_thread::experimental::sync_wait(std::move(s)); + + HPX_TEST(result.has_value()); + HPX_TEST_EQ(hpx::get<0>(*result), 42); +} + +/////////////////////////////////////////////////////////////////////////////// +int hpx_main() +{ + test_executor_scheduler_schedule(); + test_executor_scheduler_schedule_parallel(); + test_executor_scheduler_schedule_restricted(); + + return hpx::local::finalize(); +} + +int main(int argc, char* argv[]) +{ + std::vector const cfg = {"hpx.os_threads=all"}; + + hpx::local::init_params init_args; + init_args.cfg = cfg; + + HPX_TEST_EQ_MSG(hpx::local::init(hpx_main, argc, argv, init_args), 0, + "HPX main exited with non-zero status"); + + return hpx::util::report_errors(); +}