Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions libs/core/executors/CMakeLists.txt
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
176 changes: 176 additions & 0 deletions libs/core/executors/include/hpx/executors/executor_scheduler.hpp
Original file line number Diff line number Diff line change
@@ -0,0 +1,176 @@
// 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 <hpx/config.hpp>

#include <hpx/modules/errors.hpp>
#include <hpx/modules/execution.hpp>
#include <hpx/modules/execution_base.hpp>

#include <exception>
#include <type_traits>
#include <utility>

namespace hpx::execution::experimental {

// Forward declaration
template <typename Executor>
struct executor_scheduler;

namespace detail {

///////////////////////////////////////////////////////////////////////
template <typename Executor, typename Receiver>
struct executor_operation_state
{
HPX_NO_UNIQUE_ADDRESS std::decay_t<Executor> exec_;
HPX_NO_UNIQUE_ADDRESS std::decay_t<Receiver> receiver_;

template <typename Exec, typename Recv>
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;

void start() & noexcept
{
hpx::detail::try_catch_exception_ptr(
[&]() {
if constexpr (
std::is_same_v<
hpx::traits::executor_execution_category_t<
std::decay_t<Executor>>,
hpx::execution::sequenced_execution_tag>)
{
// For sequenced executors, invoke inline
hpx::execution::experimental::set_value(
HPX_MOVE(receiver_));
}
else
{
hpx::parallel::execution::post(exec_,
[receiver = HPX_MOVE(receiver_)]() mutable {
hpx::execution::experimental::set_value(
HPX_MOVE(receiver));
});
}
},
[&](std::exception_ptr ep) {
hpx::execution::experimental::set_error(
HPX_MOVE(receiver_), HPX_MOVE(ep));
});
}
};

///////////////////////////////////////////////////////////////////////
template <typename Executor>
struct executor_sender
{
using sender_concept = hpx::execution::experimental::sender_t;

HPX_NO_UNIQUE_ADDRESS std::decay_t<Executor> 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 <typename Self, typename... Env>
static consteval auto get_completion_signatures() noexcept
-> completion_signatures
{
return {};
}

constexpr auto get_env() const noexcept
{
struct env
{
std::decay_t<Executor> const& exec_;

constexpr auto query(hpx::execution::experimental::
get_completion_scheduler_t<
hpx::execution::experimental::set_value_t>)
const noexcept
{
return executor_scheduler<Executor>{exec_};
}
};
return env{exec_};
}

template <typename Receiver>
auto connect(Receiver&& receiver) &&
{
return executor_operation_state<Executor,
std::decay_t<Receiver>>{
HPX_MOVE(exec_), HPX_FORWARD(Receiver, receiver)};
}

template <typename Receiver>
auto connect(Receiver&& receiver) const&
{
return executor_operation_state<Executor,
std::decay_t<Receiver>>{
exec_, HPX_FORWARD(Receiver, receiver)};
}
};
} // namespace detail

///////////////////////////////////////////////////////////////////////////
template <typename Executor>
struct executor_scheduler
{
using executor_type = std::decay_t<Executor>;

HPX_NO_UNIQUE_ADDRESS executor_type exec_;

constexpr executor_scheduler() = default;

template <typename Exec>
requires(!std::is_same_v<std::decay_t<Exec>, executor_scheduler>)
constexpr explicit executor_scheduler(Exec&& exec) noexcept
: 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);
}

constexpr detail::executor_sender<Executor> schedule() const& noexcept
{
return {exec_};
Comment thread
hkaiser marked this conversation as resolved.
}
Comment thread
hkaiser marked this conversation as resolved.

constexpr detail::executor_sender<Executor> schedule() && noexcept
{
return {HPX_MOVE(exec_)};
}
};
} // namespace hpx::execution::experimental
41 changes: 38 additions & 3 deletions libs/core/executors/include/hpx/executors/parallel_executor.hpp
Original file line number Diff line number Diff line change
Expand Up @@ -68,6 +68,16 @@ namespace hpx::parallel::execution::detail {
struct then_bulk_function_result;
} // namespace hpx::parallel::execution::detail

namespace hpx::execution::experimental {
template <typename Executor>
struct executor_scheduler;

namespace detail {
template <typename Executor>
struct executor_sender;
} // namespace detail
} // namespace hpx::execution::experimental

namespace hpx::execution {

///////////////////////////////////////////////////////////////////////////
Expand Down Expand Up @@ -156,7 +166,8 @@ namespace hpx::execution {
}

public:
parallel_policy_executor_base(parallel_policy_executor_base const& rhs)
parallel_policy_executor_base(
parallel_policy_executor_base const& rhs) noexcept
: pool_(rhs.pool_)
, policy_(rhs.policy_)
, first_core_(rhs.first_core_)
Expand All @@ -169,7 +180,7 @@ namespace hpx::execution {
// NOLINTEND(bugprone-crtp-constructor-accessibility)

parallel_policy_executor_base& operator=(
parallel_policy_executor_base const& rhs)
parallel_policy_executor_base const& rhs) noexcept
{
if (this != &rhs)
{
Expand Down Expand Up @@ -673,11 +684,15 @@ namespace hpx::execution {

public:
/// \cond NOINTERNAL
constexpr hpx::execution::experimental::executor_scheduler<
parallel_policy_executor>
query(hpx::execution::experimental::get_scheduler_t) const noexcept;

constexpr bool operator==(
parallel_policy_executor const& rhs) const noexcept
{
return base_type::policy_ == rhs.policy_ &&
base_type::pool_ == rhs.pool;
base_type::pool_ == rhs.pool_;
}

constexpr bool operator!=(
Expand Down Expand Up @@ -1021,6 +1036,10 @@ namespace hpx::execution {
return *this;
}

constexpr hpx::execution::experimental::executor_scheduler<
parallel_policy_executor>
query(hpx::execution::experimental::get_scheduler_t) const noexcept;

private:
/// \cond NOINTERNAL
friend class hpx::serialization::access;
Expand Down Expand Up @@ -1192,3 +1211,19 @@ 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 <hpx/executors/executor_scheduler.hpp>

namespace hpx::execution {
template <typename Policy, bool HierarchicalSpawning>
constexpr hpx::execution::experimental::executor_scheduler<
parallel_policy_executor<Policy, HierarchicalSpawning>>
parallel_policy_executor<Policy, HierarchicalSpawning>::query(
hpx::execution::experimental::get_scheduler_t) const noexcept
{
return hpx::execution::experimental::executor_scheduler<
parallel_policy_executor>(*this);
}
} // namespace hpx::execution
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,7 @@

#include <hpx/config.hpp>
#include <hpx/assert.hpp>
#include <hpx/executors/executor_scheduler.hpp>
#include <hpx/executors/parallel_executor.hpp>
#include <hpx/modules/concepts.hpp>
#include <hpx/modules/execution.hpp>
Expand Down Expand Up @@ -62,15 +63,16 @@ namespace hpx::execution::experimental {
exec_, num_threads);
}

restricted_policy_executor(restricted_policy_executor const& other)
restricted_policy_executor(
restricted_policy_executor const& other) noexcept
: first_thread_(other.first_thread_)
, os_thread_(other.os_thread_.load())
, exec_(other.exec_)
{
}

restricted_policy_executor& operator=(
restricted_policy_executor const& rhs)
restricted_policy_executor const& rhs) noexcept
{
first_thread_ = rhs.first_thread_;
os_thread_ = rhs.os_thread_.load();
Expand Down Expand Up @@ -226,6 +228,15 @@ namespace hpx::execution::experimental {
}
/// \endcond

public:
constexpr hpx::execution::experimental::executor_scheduler<
restricted_policy_executor>
query(hpx::execution::experimental::get_scheduler_t) const noexcept
{
return hpx::execution::experimental::executor_scheduler<
restricted_policy_executor>(*this);
}

private:
std::uint16_t first_thread_;
mutable std::atomic<std::size_t> os_thread_;
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,7 @@

#include <hpx/config.hpp>
#include <hpx/executors/execution_policy_mappings.hpp>
#include <hpx/executors/executor_scheduler.hpp>
#include <hpx/executors/parallel_executor.hpp>
#include <hpx/modules/errors.hpp>
#include <hpx/modules/execution.hpp>
Expand Down Expand Up @@ -236,6 +237,17 @@ namespace hpx::execution {
#endif
}

public:
/// \cond NOINTERNAL
constexpr hpx::execution::experimental::executor_scheduler<
sequenced_executor>
query(hpx::execution::experimental::get_scheduler_t) const noexcept
{
return hpx::execution::experimental::executor_scheduler<
sequenced_executor>(*this);
}
/// \endcond

private:
friend class hpx::serialization::access;

Expand Down
3 changes: 2 additions & 1 deletion libs/core/executors/tests/unit/CMakeLists.txt
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
# Copyright (c) 2020-2025 The STE||AR-Group
# Copyright (c) 2020-2026 The STE||AR-Group
#
# SPDX-License-Identifier: BSL-1.0
# Distributed under the Boost Software License, Version 1.0. (See accompanying
Expand All @@ -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
Expand Down
Loading
Loading