Applying minor optimizations and fixes in various places

Signed-off-by: Hartmut Kaiser <hartmut.kaiser@gmail.com>

Author: hkaiser

docs/sphinx/manual/executors.rst

@@ -418,19 +418,6 @@ Full example code
         // The annotating executor exposes the same executor categories as its
         // underlying (wrapped) executor.
 
-        template <typename BaseExecutor>
-        struct is_never_blocking_one_way_executor<
-            simple_annotating_executor<BaseExecutor>>
-        : is_never_blocking_one_way_executor<BaseExecutor>
-        {
-        };
-
-        template <typename BaseExecutor>
-        struct is_one_way_executor<simple_annotating_executor<BaseExecutor>>
-        : is_one_way_executor<BaseExecutor>
-        {
-        };
-
         template <typename BaseExecutor>
         struct is_two_way_executor<simple_annotating_executor<BaseExecutor>>
         : is_two_way_executor<BaseExecutor>

libs/core/algorithms/include/hpx/parallel/util/foreach_partitioner.hpp

@@ -170,7 +170,9 @@ namespace hpx::parallel::util::detail {
             try
             {
                 if constexpr (std::is_void_v<decltype(foreach_partition<Result>(
-                                  policy, first, count, f1, reshape))>)
+                                  HPX_FORWARD(ExPolicy_, policy), first, count,
+                                  HPX_FORWARD(F1, f1),
+                                  HPX_FORWARD(ReShape, reshape)))>)
                 {
                     detail::foreach_partition<Result>(
                         HPX_FORWARD(ExPolicy_, policy), first, count,

libs/core/algorithms/tests/performance/benchmark_merge.cpp

@@ -212,6 +212,8 @@ struct random_to_item_t
     }
 };
 
+using data_type = int;
+
 ///////////////////////////////////////////////////////////////////////////////
 template <typename IteratorTag, typename Allocator>
 void run_benchmark(std::size_t vector_size1, std::size_t vector_size2,
@@ -231,7 +233,7 @@ void run_benchmark(std::size_t vector_size1, std::size_t vector_size2,
     hpx::generate(par, std::begin(uniform_distribution),
         std::end(uniform_distribution), [&] { return dist(re); });
 
-    using test_container = test_container<IteratorTag, int, Allocator>;
+    using test_container = test_container<IteratorTag, data_type, Allocator>;
     using container = typename test_container::type;
     using T = typename container::value_type;
 
@@ -416,7 +418,8 @@ int hpx_main(hpx::program_options::variables_map& vm)
     {
         auto policy = hpx::execution::par;
         using allocator_type =
-            hpx::compute::host::detail::policy_allocator<int, decltype(policy)>;
+            hpx::compute::host::detail::policy_allocator<data_type,
+                decltype(policy)>;
         allocator_type alloc(policy);
 
         run_benchmark(vector_size1, vector_size2, test_count,

libs/core/execution/include/hpx/execution/detail/post_policy_dispatch.hpp

@@ -40,7 +40,6 @@ namespace hpx::detail {
             {
                 hint.runs_as_child_mode(
                     hpx::threads::thread_execution_hint::none);
-                policy.set_hint(hint);
             }
 
             threads::thread_init_data data(

libs/core/execution_base/include/hpx/execution_base/execution.hpp

@@ -103,7 +103,7 @@ namespace hpx::parallel::execution {
     /// \returns f(ts...)'s result
     ///
     /// \note It will call tag_invoke(sync_execute_t, exec, f, ts...) if it
-    ///       exists. For two-way executors it will invoke asynch_execute_t
+    ///       exists. For two-way executors it will invoke async_execute_t
     ///       and wait for the task's completion before returning.
     ///
     HPX_CXX_EXPORT inline constexpr struct sync_execute_t final
@@ -148,7 +148,7 @@ namespace hpx::parallel::execution {
     ///
     /// \note This is valid for one way executors (calls
     ///       make_ready_future(exec.sync_execute(f, ts...) if it exists)
-    ///       and for two way executors (calls exec.async_execute(f, ts...)
+    ///       and for two-way executors (calls exec.async_execute(f, ts...)
     ///       if it exists).
     ///
     /// \returns f(ts...)'s result through a future
@@ -188,7 +188,7 @@ namespace hpx::parallel::execution {
     ///
     /// \returns f(ts...)'s result through a future
     ///
-    /// \note This is valid for two way executors (calls
+    /// \note This is valid for two-way executors (calls
     ///       exec.then_execute(f, predecessor, ts...) if it exists) and
     ///       for one way executors (calls predecessor.then(bind(f, ts...))).
     ///
@@ -227,10 +227,10 @@ namespace hpx::parallel::execution {
     ///             given executor.
     /// \param ts   [in] Additional arguments to use to invoke \a f.
     ///
-    /// \note This is valid for two way executors (calls
+    /// \note This is valid for two-way executors (calls
     ///       exec.post(f, ts...), if available, otherwise
     ///       it calls exec.async_execute(f, ts...) while discarding the
-    ///       returned future), and for non-blocking two way executors
+    ///       returned future), and for non-blocking two-way executors
     ///       (calls exec.post(f, ts...) if it exists).
     ///
     HPX_CXX_EXPORT inline constexpr struct post_t final
@@ -283,7 +283,7 @@ namespace hpx::parallel::execution {
     /// \param ts    [in] Additional arguments to use to invoke \a f.
     ///
     /// \returns The return type of \a executor_type::bulk_sync_execute
-    ///          if defined by \a executor_type. Otherwise a vector holding
+    ///          if defined by \a executor_type. Otherwise, a vector holding
     ///          the returned values of each invocation of \a f except when
     ///          \a f returns void, which case void is returned.
     ///
@@ -352,7 +352,7 @@ namespace hpx::parallel::execution {
     /// \param ts    [in] Additional arguments to use to invoke \a f.
     ///
     /// \returns The return type of \a executor_type::bulk_async_execute if
-    ///          defined by \a executor_type. Otherwise a vector
+    ///          defined by \a executor_type. Otherwise, a vector
     ///          of futures holding the returned values of each invocation
     ///          of \a f.
     ///
@@ -423,7 +423,7 @@ namespace hpx::parallel::execution {
     /// \param ts    [in] Additional arguments to use to invoke \a f.
     ///
     /// \returns The return type of \a executor_type::bulk_then_execute
-    ///          if defined by \a executor_type. Otherwise a vector holding
+    ///          if defined by \a executor_type. Otherwise, a vector holding
     ///          the returned values of each invocation of \a f.
     ///
     /// \note This calls exec.bulk_then_execute(f, shape, pred, ts...) if it
@@ -489,7 +489,7 @@ namespace hpx::parallel::execution {
     ///              given executor.
     ///
     /// \returns The return type of \a executor_type::async_invoke if defined by
-    ///          \a executor_type. Otherwise a future<void>
+    ///          \a executor_type. Otherwise, a future<void>
     ///          representing finishing the execution of all functions \a fs.
     ///
     /// \note This calls exec.async_invoke(fs...) if it exists; otherwise it

libs/core/execution_base/src/this_thread.cpp

@@ -25,6 +25,7 @@
 #include <utility>
 
 #if defined(HPX_WINDOWS)
+#include <ctime>
 #include <windows.h>
 #else
 #ifndef _AIX
@@ -40,12 +41,82 @@ namespace hpx::execution_base {
 
     namespace {
 
+#if defined(HPX_WINDOWS)
+        // Number of performance counter increments per nanosecond, or zero if
+        // it could not be determined.
+        struct ticks_per_nanosecond
+        {
+            ticks_per_nanosecond()
+            {
+                LARGE_INTEGER freq;
+                if (QueryPerformanceFrequency(&freq))
+                    ticks = static_cast<double>(freq.QuadPart) / 1e9;
+            }
+
+            double ticks = 0.0;
+        };
+        ticks_per_nanosecond ticks;
+
+        // our own (crude) implementation of nanosleep
+        int win_nanosleep(std::timespec const& delay)
+        {
+            if (delay.tv_nsec < 0 || delay.tv_nsec >= 999999999)
+            {
+                return -1;
+            }
+
+            // for small delays we busy-wait
+            if (delay.tv_sec == 0 && ticks.ticks != 0.0)
+            {
+                // compensate for fluctuations introduced by Sleep()
+                auto const sleep_for = delay.tv_nsec / 1000000 - 10;
+
+                // overall number of ticks to delay
+                auto const ticks_to_delay = static_cast<long>(
+                    static_cast<double>(delay.tv_nsec) * ticks.ticks);
+
+                LARGE_INTEGER counter;
+                if (QueryPerformanceCounter(&counter))
+                {
+                    // wait until the performance counter has reached this value
+                    auto const wait_until = counter.QuadPart + ticks_to_delay;
+
+                    // use Sleep() if appropriate
+                    if (sleep_for > 0)
+                    {
+                        Sleep(sleep_for);
+                    }
+
+                    // simply busy-wait for the remaining amount of time, don't
+                    // wait if delay is zero
+                    while (counter.QuadPart < wait_until &&
+                        QueryPerformanceCounter(&counter))
+                    {
+                    }
+                    return 0;
+                }
+            }
+
+            // Fallback and longer delays
+            Sleep(static_cast<long>(delay.tv_sec) * 1000 +
+                delay.tv_nsec / 1000000);
+
+            return 0;
+        }
+
+        constexpr std::timespec wait_0ns = {.tv_sec = 0, .tv_nsec = 0};
+        constexpr std::timespec wait_1000ns = {.tv_sec = 0, .tv_nsec = 1000};
+#endif
+
+        ///////////////////////////////////////////////////////////////////////
         struct default_context final : execution_base::context_base
         {
-            resource_base const& resource() const noexcept override
+            [[nodiscard]] resource_base const& resource()
+                const noexcept override
             {
                 return resource_;
             }
+
             resource_base resource_;
         };
 
@@ -96,7 +167,7 @@ namespace hpx::execution_base {
         void default_agent::yield(char const* /* desc */)
         {
 #if defined(HPX_WINDOWS)
-            Sleep(0);
+            win_nanosleep(wait_0ns);
 #else
             sched_yield();
 #endif
@@ -116,7 +187,7 @@ namespace hpx::execution_base {
             else if (k < 32 || k & 1)    //-V112
             {
 #if defined(HPX_WINDOWS)
-                Sleep(0);
+                win_nanosleep(wait_0ns);
                 return true;
 #else
                 sched_yield();
@@ -126,7 +197,7 @@ namespace hpx::execution_base {
             else
             {
 #if defined(HPX_WINDOWS)
-                Sleep(1);
+                win_nanosleep(wait_1000ns);
                 return true;
 #else
                 // g++ -Wextra warns on {} or {0}

libs/core/executors/include/hpx/executors/parallel_executor.hpp

@@ -65,6 +65,11 @@ namespace hpx::parallel::execution::detail {
     struct then_bulk_function_result;
 }    // namespace hpx::parallel::execution::detail
 
+#if !defined(HPX_HAVE_MORE_THAN_64_THREADS) ||                                 \
+    (defined(HPX_HAVE_MAX_CPU_COUNT) && HPX_HAVE_MAX_CPU_COUNT <= 64)
+#define HPX_MASK_TYPE_IS_CONSTEXPR_CONSTRUCTIBLE
+#endif
+
 namespace hpx::execution {
 
     ///////////////////////////////////////////////////////////////////////////
@@ -223,6 +228,11 @@ namespace hpx::execution {
         {
             auto exec_with_num_cores = exec;
             exec_with_num_cores.num_cores_ = num_cores;
+
+#if defined(HPX_MASK_TYPE_IS_CONSTEXPR_CONSTRUCTIBLE)
+            // force recomputing cached pu mask
+            exec_with_num_cores.mask_ = hpx::threads::mask_type();
+#endif
             return exec_with_num_cores;
         }
 
@@ -594,6 +604,12 @@ namespace hpx::execution {
 
         hpx::threads::mask_type pu_mask() const noexcept
         {
+#if defined(HPX_MASK_TYPE_IS_CONSTEXPR_CONSTRUCTIBLE)
+            if (hpx::threads::any(mask_))
+            {
+                return mask_;
+            }
+#endif
             auto const num_threads = get_num_cores();
             auto const* pool =
                 pool_ ? pool_ : threads::detail::get_self_or_default_pool();
@@ -621,6 +637,10 @@ namespace hpx::execution {
                     }
                 }
             }
+
+#if defined(HPX_MASK_TYPE_IS_CONSTEXPR_CONSTRUCTIBLE)
+            mask_ = mask;
+#endif
             return mask;
         }
 
@@ -644,6 +664,9 @@ namespace hpx::execution {
         std::size_t hierarchical_threshold_ = hierarchical_threshold_default_;
         std::size_t first_core_ = 0;
         std::size_t num_cores_ = 0;
+#if defined(HPX_MASK_TYPE_IS_CONSTEXPR_CONSTRUCTIBLE)
+        mutable hpx::threads::mask_type mask_ = hpx::threads::mask_type();
+#endif
 #if defined(HPX_HAVE_THREAD_DESCRIPTION)
         char const* annotation_ = nullptr;
 #endif
@@ -685,6 +708,8 @@ namespace hpx::execution {
         parallel_policy_executor<hpx::launch>;
 }    // namespace hpx::execution
 
+#undef HPX_MASK_TYPE_IS_CONSTEXPR_CONSTRUCTIBLE
+
 namespace hpx::execution::experimental {
 
     /// \cond NOINTERNAL

libs/core/topology/include/hpx/topology/topology.hpp

@@ -320,8 +320,8 @@ namespace hpx::threads {
         mask_type init_thread_affinity_mask(
             std::size_t num_core, std::size_t num_pu) const;
 
-        hwloc_bitmap_t mask_to_bitmap(
-            mask_cref_type mask, hwloc_obj_type_t htype) const;
+        hwloc_bitmap_t mask_to_bitmap(mask_cref_type mask,
+            hwloc_obj_type_t htype, unsigned* count = nullptr) const;
         mask_type bitmap_to_mask(
             hwloc_bitmap_t bitmap, hwloc_obj_type_t htype) const;