Fixing issues in the implementation of memoizing_range

- flyby: optimize implementation of some functions in topology
- flyby: cache pu mask for parallel_executor

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

Author: hkaiser

docs/sphinx/manual/executors.rst

@@ -403,13 +403,13 @@ Full example code
         // Bulk async_execute (used by parallel algorithms)
         template <typename F, typename Shape, typename... Ts>
         friend auto tag_invoke(hpx::parallel::execution::bulk_async_execute_t,
-            simple_annotating_executor const& exec, F&& f, Shape const& shape,
+            simple_annotating_executor const& exec, F&& f, Shape && shape,
             Ts&&... ts)
         {
             return hpx::parallel::execution::bulk_async_execute(
                 exec.base_,
                 hpx::annotated_function(std::forward<F>(f), exec.annotation_),
-                shape, std::forward<Ts>(ts)...);
+                std::forward<Shape>(shape), std::forward<Ts>(ts)...);
         }
     };
 
@@ -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>
@@ -503,13 +490,13 @@ to integrate seamlessly with the |hpx| parallel algorithms infrastructure.
 
     template <typename F, typename Shape, typename... Ts>
     friend auto tag_invoke(hpx::parallel::execution::bulk_async_execute_t,
-        simple_annotating_executor const& exec, F&& f, Shape const& shape,
+        simple_annotating_executor const& exec, F&& f, Shape && shape,
         Ts&&... ts)
     {
         return hpx::parallel::execution::bulk_async_execute(
             exec.base_,
             hpx::annotated_function(std::forward<F>(f), exec.annotation_),
-            shape, std::forward<Ts>(ts)...);
+            std::forward<Shape>(shape), std::forward<Ts>(ts)...);
     }
 
 Note how we delegate the actual execution to `exec.base_`, the underlying

libs/core/algorithms/include/hpx/parallel/algorithms/merge.hpp

@@ -661,14 +661,30 @@ namespace hpx::parallel {
             using merge_region =
                 hpx::tuple<Iter1, std::size_t, Iter2, std::size_t, std::size_t>;
 
+#if HPX_HAVE_ITTNOTIFY != 0 && !defined(HPX_HAVE_APEX)
+            static hpx::util::itt::event notify_event("reshape");
+            hpx::util::itt::mark_event e(notify_event);
+#endif
+
 #if defined(HPX_HAVE_CXX20_COROUTINES)
-            auto reshape = [](auto&& shape, auto&& args)
+            auto reshape = [](auto&& shape, auto&& args,
+                               std::size_t const cores)
                 -> hpx::generator<merge_region&, merge_region> {
                 auto [len1, first2, last2, comp, proj1, proj2] = args;
+
+                if (std::size(shape) == 1 && cores == 1)
+                {
+                    // special case: one core, one chunk - the partitioner will
+                    // switch to sequential execution in this case
+                    auto [it1, size1, _] = *std::begin(shape);
+                    auto size2 = detail::distance(first2, last2);
+                    merge_region region(it1, size1, first2, size2, 0);
+                    co_yield region;
+                    co_return;
+                }
 #else
             auto reshape = [len1, first2, last2, comp, proj1, proj2](
-                               auto&& shape, std::size_t cores) {
-
+                               auto&& shape, std::size_t const cores) {
                 auto shape_size = std::size(shape);
                 if (shape_size == 1 && cores == 1)
                 {
@@ -684,11 +700,6 @@ namespace hpx::parallel {
                 reshaped.reserve(2 * shape_size);
 #endif
 
-#if HPX_HAVE_ITTNOTIFY != 0 && !defined(HPX_HAVE_APEX)
-                static hpx::util::itt::event notify_event("reshape");
-                hpx::util::itt::mark_event e(notify_event);
-#endif
-
                 Iter2 it2 = first2;
                 std::size_t dest_start = 0;
                 for (auto [it1, size1, base] : shape)
@@ -784,9 +795,9 @@ namespace hpx::parallel {
 #if defined(HPX_HAVE_CXX20_COROUTINES)
             return [reshape = HPX_MOVE(reshape),
                        data = hpx::make_tuple(len1, first2, last2, comp, proj1,
-                           proj2)](auto&& shape) {
+                           proj2)](auto&& shape, std::size_t cores) {
                 return util::memoizing_range(
-                    reshape(shape, data), 2 * std::size(shape));
+                    reshape(shape, data, cores), 2 * std::size(shape));
             };
 #else
             return reshape;

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,