Skip to content

Fixing problems with scheduler fast-idle mode #6795

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wants to merge 2 commits into
base: master
Choose a base branch
from
Open

Fixing problems with scheduler fast-idle mode #6795

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
2be3baf
Fixing problems with scheduler fast-idle mode
hkaiser Oct 29, 2025
a29b1c4
Prevent staggler thread from blocking progress in index_queue_spawning
Pansysk75 Nov 3, 2025
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
8 changes: 5 additions & 3 deletions cmake/HPX_AddModule.cmake
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -69,7 +69,9 @@ function(add_hpx_module libname modulename)
FORCE
)

if(${modulename}_GLOBAL_HEADER_MODULE_GEN OR ${modulename}_MODULE_SOURCE)
if(HPX_WITH_CXX_MODULES AND (${modulename}_GLOBAL_HEADER_MODULE_GEN
OR ${modulename}_MODULE_SOURCE)
)
# Mark the module as exposing C++ modules
set(cxx_modules ${HPX_ENABLED_CXX_MODULES})
list(APPEND cxx_modules ${modulename})
Expand Down Expand Up @@ -193,7 +195,7 @@ function(add_hpx_module libname modulename)
set(global_header
"${CMAKE_CURRENT_BINARY_DIR}/include/hpx/modules/${modulename}.hpp"
)
if(${modulename}_GLOBAL_HEADER_MODULE_GEN)
if(HPX_WITH_CXX_MODULES AND ${modulename}_GLOBAL_HEADER_MODULE_GEN)
# generate list of macro headers to #include
list(LENGTH ${modulename}_MACRO_HEADERS macro_headers)
if(macro_headers GREATER 0)
Expand Down Expand Up @@ -236,7 +238,7 @@ function(add_hpx_module libname modulename)
)
set(generated_headers ${global_header})

if(${modulename}_GLOBAL_HEADER_MODULE_GEN)
if(HPX_WITH_CXX_MODULES AND ${modulename}_GLOBAL_HEADER_MODULE_GEN)
# collect all standard header files used by this module
set(found_includes)
hpx_collect_std_headers(
Expand Down
45 changes: 34 additions & 11 deletions libs/core/algorithms/include/hpx/parallel/util/foreach_partitioner.hpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -69,9 +69,18 @@ namespace hpx::parallel::util::detail {
auto&& shape =
detail::get_bulk_iteration_shape_idx(policy, first, count);

return execution::bulk_async_execute(policy.executor(),
partitioner_iteration<Result, F>{HPX_FORWARD(F, f)},
reshape(HPX_MOVE(shape)));
if constexpr (hpx::is_async_execution_policy_v<ExPolicy>)
{
return execution::bulk_async_execute(policy.executor(),
partitioner_iteration<Result, F>{HPX_FORWARD(F, f)},
reshape(HPX_MOVE(shape)));
}
else
{
return execution::bulk_sync_execute(policy.executor(),
partitioner_iteration<Result, F>{HPX_FORWARD(F, f)},
reshape(HPX_MOVE(shape)));
}
}
else
{
Expand Down Expand Up @@ -101,8 +110,8 @@ namespace hpx::parallel::util::detail {

template <typename ExPolicy_, typename FwdIter, typename F1,
typename F2, typename ReShape = hpx::identity>
static decltype(auto) call(ExPolicy_&& policy, FwdIter first,
std::size_t count, F1&& f1, F2&& f2, ReShape&& reshape = ReShape{})
static auto call(ExPolicy_&& policy, FwdIter first, std::size_t count,
F1&& f1, F2&& f2, ReShape&& reshape = ReShape{})
{
// inform parameter traits
using scoped_executor_parameters =
Expand All @@ -115,14 +124,28 @@ namespace hpx::parallel::util::detail {
FwdIter last = parallel::detail::next(first, count);
try
{
auto&& items = detail::foreach_partition<Result>(
HPX_FORWARD(ExPolicy_, policy), first, count,
HPX_FORWARD(F1, f1), HPX_FORWARD(ReShape, reshape));
if constexpr (std::is_void_v<decltype(foreach_partition<Result>(
policy, first, count, f1, reshape))>)
{
detail::foreach_partition<Result>(
HPX_FORWARD(ExPolicy_, policy), first, count,
HPX_FORWARD(F1, f1), HPX_FORWARD(ReShape, reshape));

scoped_params.mark_end_of_scheduling();
scoped_params.mark_end_of_scheduling();

return reduce(
HPX_MOVE(items), HPX_FORWARD(F2, f2), HPX_MOVE(last));
return HPX_INVOKE(f2, HPX_MOVE(last));
}
else
{
auto&& items = foreach_partition<Result>(
HPX_FORWARD(ExPolicy_, policy), first, count,
HPX_FORWARD(F1, f1), HPX_FORWARD(ReShape, reshape));

scoped_params.mark_end_of_scheduling();

return reduce(
HPX_MOVE(items), HPX_FORWARD(F2, f2), HPX_MOVE(last));
}
}
catch (...)
{
Expand Down
34 changes: 15 additions & 19 deletions libs/core/algorithms/include/hpx/parallel/util/result_types.hpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -416,13 +416,15 @@ namespace hpx::parallel::util {
namespace detail {

template <typename ZipIter>
in_out_result<typename hpx::tuple_element<0,
typename ZipIter::iterator_tuple_type>::type,
in_out_result<
typename hpx::tuple_element<0,
typename std::decay_t<ZipIter>::iterator_tuple_type>::type,
typename hpx::tuple_element<1,
typename ZipIter::iterator_tuple_type>::type>
typename std::decay_t<ZipIter>::iterator_tuple_type>::type>
get_in_out_result(ZipIter&& zipiter)
{
using iterator_tuple_type = typename ZipIter::iterator_tuple_type;
using iterator_tuple_type =
typename std::decay_t<ZipIter>::iterator_tuple_type;

using result_type = in_out_result<
typename hpx::tuple_element<0, iterator_tuple_type>::type,
Expand All @@ -433,11 +435,7 @@ namespace hpx::parallel::util {
}

template <typename ZipIterSender>
// clang-format off
requires (
hpx::execution::experimental::is_sender_v<ZipIterSender>
)
// clang-format on
requires(hpx::execution::experimental::is_sender_v<ZipIterSender>)
decltype(auto) get_in_out_result(ZipIterSender&& zipiter_sender)
{
return hpx::execution::experimental::then(
Expand Down Expand Up @@ -498,15 +496,17 @@ namespace hpx::parallel::util {
}

template <typename ZipIter>
in_in_out_result<typename hpx::tuple_element<0,
typename ZipIter::iterator_tuple_type>::type,
in_in_out_result<
typename hpx::tuple_element<0,
typename std::decay_t<ZipIter>::iterator_tuple_type>::type,
typename hpx::tuple_element<1,
typename ZipIter::iterator_tuple_type>::type,
typename std::decay_t<ZipIter>::iterator_tuple_type>::type,
typename hpx::tuple_element<2,
typename ZipIter::iterator_tuple_type>::type>
typename std::decay_t<ZipIter>::iterator_tuple_type>::type>
get_in_in_out_result(ZipIter&& zipiter)
{
using iterator_tuple_type = typename ZipIter::iterator_tuple_type;
using iterator_tuple_type =
typename std::decay_t<ZipIter>::iterator_tuple_type;

using result_type = in_in_out_result<
typename hpx::tuple_element<0, iterator_tuple_type>::type,
Expand All @@ -518,11 +518,7 @@ namespace hpx::parallel::util {
}

template <typename ZipIterSender>
// clang-format off
requires (
hpx::execution::experimental::is_sender_v<ZipIterSender>
)
// clang-format on
requires(hpx::execution::experimental::is_sender_v<ZipIterSender>)
decltype(auto) get_in_in_out_result(ZipIterSender&& zipiter_sender)
{
return hpx::execution::experimental::then(
Expand Down
58 changes: 54 additions & 4 deletions libs/core/algorithms/tests/performance/benchmark_merge.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -133,6 +133,33 @@ struct hpx::execution::experimental::is_executor_parameters<compute_chunk_size>
{
};

struct enable_fast_idle_mode
{
template <typename Executor>
friend void tag_override_invoke(
hpx::execution::experimental::mark_begin_execution_t,
enable_fast_idle_mode, Executor&&)
{
hpx::threads::add_scheduler_mode(
hpx::threads::policies::scheduler_mode::fast_idle_mode);
}

template <typename Executor>
friend void tag_override_invoke(
hpx::execution::experimental::mark_end_execution_t,
enable_fast_idle_mode, Executor&&)
{
hpx::threads::remove_scheduler_mode(
hpx::threads::policies::scheduler_mode::fast_idle_mode);
}
};

template <>
struct hpx::execution::experimental::is_executor_parameters<
enable_fast_idle_mode> : std::true_type
{
};

///////////////////////////////////////////////////////////////////////////////
template <typename T>
struct random_to_item_t
Expand Down Expand Up @@ -234,7 +261,7 @@ void run_benchmark(std::size_t vector_size1, std::size_t vector_size2,
double const time_seq = run_merge_benchmark_hpx(
test_count, seq, first1, last1, first2, last2, dest);

hpx::this_thread::sleep_for(std::chrono::seconds(1));
hpx::this_thread::sleep_for(std::chrono::milliseconds(200));

std::cout << "--- run_merge_benchmark_par ---" << std::endl;

Expand All @@ -251,11 +278,11 @@ void run_benchmark(std::size_t vector_size1, std::size_t vector_size2,
double const time_par = run_merge_benchmark_hpx(
test_count, policy.with(ccs), first1, last1, first2, last2, dest);

std::cout << "--- run_merge_benchmark_par_stackless ---" << std::endl;

HPX_ITT_PAUSE();

hpx::this_thread::sleep_for(std::chrono::seconds(1));
std::cout << "--- run_merge_benchmark_par_stackless ---" << std::endl;

hpx::this_thread::sleep_for(std::chrono::milliseconds(200));

HPX_ITT_RESUME();

Expand All @@ -270,6 +297,26 @@ void run_benchmark(std::size_t vector_size1, std::size_t vector_size2,

HPX_ITT_PAUSE();

std::cout << "--- run_merge_benchmark_par_stackless_fast_idle ---"
<< std::endl;

hpx::this_thread::sleep_for(std::chrono::milliseconds(200));

HPX_ITT_RESUME();

double time_par_stackless_fast_idle = 0;
{
enable_fast_idle_mode efim;
auto const stackless_policy =
hpx::execution::experimental::with_stacksize(
policy, hpx::threads::thread_stacksize::nostack);
time_par_stackless_fast_idle = run_merge_benchmark_hpx(test_count,
stackless_policy.with(ccs, efim), first1, last1, first2, last2,
dest);
}

HPX_ITT_PAUSE();

std::cout << "--- run_merge_benchmark_par_fork_join ---" << std::endl;
double time_par_fork_join = 0;
{
Expand All @@ -290,6 +337,9 @@ void run_benchmark(std::size_t vector_size1, std::size_t vector_size2,
hpx::util::format_to(std::cout, fmt, "par", time_par) << std::endl;
hpx::util::format_to(std::cout, fmt, "par_stackless", time_par_stackless)
<< std::endl;
hpx::util::format_to(
std::cout, fmt, "par_stackless_fast_idle", time_par_stackless_fast_idle)
<< std::endl;
hpx::util::format_to(std::cout, fmt, "par_fork_join", time_par_fork_join)
<< std::endl;
hpx::util::format_to(std::cout, fmt, "par_unseq", time_par_unseq)
Expand Down
37 changes: 37 additions & 0 deletions libs/core/algorithms/tests/performance/benchmark_merge_sweep.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -164,6 +164,33 @@ struct hpx::execution::experimental::is_executor_parameters<adaptive_chunk_size>
{
};

struct enable_fast_idle_mode
{
template <typename Executor>
friend void tag_override_invoke(
hpx::execution::experimental::mark_begin_execution_t,
enable_fast_idle_mode, Executor&&)
{
hpx::threads::add_scheduler_mode(
hpx::threads::policies::scheduler_mode::fast_idle_mode);
}

template <typename Executor>
friend void tag_override_invoke(
hpx::execution::experimental::mark_end_execution_t,
enable_fast_idle_mode, Executor&&)
{
hpx::threads::remove_scheduler_mode(
hpx::threads::policies::scheduler_mode::fast_idle_mode);
}
};

template <>
struct hpx::execution::experimental::is_executor_parameters<
enable_fast_idle_mode> : std::true_type
{
};

///////////////////////////////////////////////////////////////////////////////
template <typename T>
struct random_to_item_t
Expand Down Expand Up @@ -461,6 +488,11 @@ int hpx_main(hpx::program_options::variables_map& vm)
run_benchmark(stackless_policy, vector_size1, vector_size2, test_count,
std::random_access_iterator_tag(), alloc, "std::vector (stackless)",
entropy);

enable_fast_idle_mode efim;
run_benchmark(stackless_policy.with(efim), vector_size1, vector_size2,
test_count, std::random_access_iterator_tag(), alloc,
"std::vector (stackless, fast-idle mode)", entropy);
}

{
Expand All @@ -481,6 +513,11 @@ int hpx_main(hpx::program_options::variables_map& vm)
run_benchmark(stackless_policy, vector_size1, vector_size2, test_count,
std::random_access_iterator_tag(), alloc,
"hpx::compute::vector (stackless)", entropy);

enable_fast_idle_mode efim;
run_benchmark(stackless_policy.with(efim), vector_size1, vector_size2,
test_count, std::random_access_iterator_tag(), alloc,
"hpx::compute::vector (stackless, fast-idle mode)", entropy);
}

return hpx::local::finalize();
Expand Down
2 changes: 1 addition & 1 deletion libs/core/execution_base/include/hpx/execution_base/agent_base.hpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,7 +24,7 @@ namespace hpx::execution_base {
[[nodiscard]] virtual context_base const& context() const noexcept = 0;

virtual void yield(char const* desc) = 0;
virtual void yield_k(std::size_t k, char const* desc) = 0;
virtual bool yield_k(std::size_t k, char const* desc) = 0;
virtual void suspend(char const* desc) = 0;
virtual void resume(
hpx::threads::thread_priority priority, char const* desc) = 0;
Expand Down
2 changes: 1 addition & 1 deletion libs/core/execution_base/include/hpx/execution_base/agent_ref.hpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -47,7 +47,7 @@ namespace hpx::execution_base {

void yield(
char const* desc = "hpx::execution_base::agent_ref::yield") const;
void yield_k(std::size_t k,
bool yield_k(std::size_t k,
char const* desc = "hpx::execution_base::agent_ref::yield_k") const;
void suspend(
char const* desc = "hpx::execution_base::agent_ref::suspend") const;
Expand Down
6 changes: 3 additions & 3 deletions libs/core/execution_base/include/hpx/execution_base/this_thread.hpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -61,7 +61,7 @@ namespace hpx::execution_base {

HPX_CORE_EXPORT void yield(
char const* desc = "hpx::execution_base::this_thread::yield");
HPX_CORE_EXPORT void yield_k(std::size_t k,
HPX_CORE_EXPORT bool yield_k(std::size_t k,
char const* desc = "hpx::execution_base::this_thread::yield_k");
HPX_CORE_EXPORT void suspend(
char const* desc = "hpx::execution_base::this_thread::suspend");
Expand Down Expand Up @@ -148,7 +148,7 @@ namespace hpx::util {

namespace detail {

inline void yield_k(std::size_t k, char const* thread_name)
inline bool yield_k(std::size_t k, char const* thread_name)
{
#ifdef HPX_HAVE_SPINLOCK_DEADLOCK_DETECTION
if (k > 32 && get_spinlock_break_on_deadlock_enabled() &&
Expand All @@ -158,7 +158,7 @@ namespace hpx::util {
"possible deadlock detected");
}
#endif
hpx::execution_base::this_thread::yield_k(k, thread_name);
return hpx::execution_base::this_thread::yield_k(k, thread_name);
}
} // namespace detail

Expand Down
4 changes: 2 additions & 2 deletions libs/core/execution_base/src/agent_ref.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,13 +25,13 @@ namespace hpx::execution_base {
impl_->yield(desc);
}

void agent_ref::yield_k(std::size_t k, const char* desc) const
bool agent_ref::yield_k(std::size_t k, const char* desc) const
{
HPX_ASSERT(*this == hpx::execution_base::this_thread::agent());

// verify that there are no more registered locks for this OS-thread
util::verify_no_locks();
impl_->yield_k(k, desc);
return impl_->yield_k(k, desc);
}

void agent_ref::suspend(const char* desc) const
Expand Down
Loading
Loading