[parallel] Add projection support to hpx::is_sorted, hpx::is_sorted_until, and hpx::is_partitioned CPOs

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

@@ -125,7 +125,6 @@ namespace hpx {
 
 #include <algorithm>
 #include <cstddef>
-#include <cstdint>
 #include <iterator>
 #include <type_traits>
 #include <utility>
@@ -136,9 +135,16 @@ namespace hpx::parallel {
     ////////////////////////////////////////////////////////////////////////////
     // is_partitioned
     namespace detail {
+        enum class partition_status : std::uint8_t
+        {
+            is_partitioned = 0,
+            is_not_partitioned = 1,
+            mixed = 2,
+            cancelled = 3
+        };
 
         /// \cond NOINTERNAL
-        HPX_CXX_CORE_EXPORT template <typename T>
+        template <typename T>
         inline bool sequential_is_partitioned(std::vector<T>&& res)
         {
             auto first = res.begin();
@@ -160,7 +166,7 @@ namespace hpx::parallel {
             return true;
         }
 
-        HPX_CXX_CORE_EXPORT template <typename Iter, typename Sent>
+        template <typename Iter, typename Sent>
         struct is_partitioned
           : public algorithm<is_partitioned<Iter, Sent>, bool>
         {
@@ -184,18 +190,11 @@ namespace hpx::parallel {
             static decltype(auto) parallel(ExPolicy&& policy, Iter_ first,
                 Sent_ last, Pred&& pred, Proj&& proj)
             {
-                using difference_type =
-                    typename std::iterator_traits<Iter_>::difference_type;
+                using difference_type = hpx::traits::iter_difference_t<Iter_>;
                 using result = util::detail::algorithm_result<ExPolicy, bool>;
                 constexpr bool has_scheduler_executor =
                     hpx::execution_policy_has_scheduler_executor_v<ExPolicy>;
 
-                if constexpr (!has_scheduler_executor)
-                {
-                    if (first == last)
-                        return result::get(true);
-                }
-
                 difference_type count =
                     hpx::parallel::detail::distance(first, last);
 
@@ -208,36 +207,64 @@ namespace hpx::parallel {
                 util::invoke_projected<Pred, Proj> pred_projected(
                     HPX_FORWARD(Pred, pred), HPX_FORWARD(Proj, proj));
                 util::cancellation_token<> tok;
-                using intermediate_result_t = std::uint8_t;
+                using intermediate_result_t = partition_status;
 
                 auto f1 = [tok, pred_projected = HPX_MOVE(pred_projected)](
                               Iter_ part_begin, std::size_t part_count) mutable
                     -> intermediate_result_t {
                     bool fst_bool = HPX_INVOKE(pred_projected, *part_begin);
                     if (part_count == 1)
-                        return fst_bool;
+                        return fst_bool ? partition_status::is_partitioned :
+                                          partition_status::is_not_partitioned;
+
+                    bool is_mixed = false;
 
                     util::const_loop_n<std::decay_t<ExPolicy>>(++part_begin,
                         --part_count, tok,
-                        [&fst_bool, &pred_projected, &tok](
+                        [&fst_bool, &is_mixed, &pred_projected, &tok](
                             Iter_ const& a) mutable -> void {
                             if (fst_bool != hpx::invoke(pred_projected, *a))
                             {
                                 if (fst_bool)
+                                {
                                     fst_bool = false;
+                                    is_mixed = true;
+                                }
                                 else
+                                {
                                     tok.cancel();
+                                }
                             }
                         });
 
-                    return fst_bool;
+                    if (tok.was_cancelled())
+                        return partition_status::cancelled;
+
+                    return is_mixed ?
+                        partition_status::mixed :
+                        (fst_bool ? partition_status::is_partitioned :
+                                    partition_status::is_not_partitioned);
                 };
 
                 auto f2 = [tok](auto&& results) -> bool {
                     if (tok.was_cancelled())
                         return false;
-                    return sequential_is_partitioned(
-                        HPX_FORWARD(decltype(results), results));
+
+                    auto it = std::find_if(hpx::util::begin(results),
+                        hpx::util::end(results), [](partition_status x) {
+                            return x != partition_status::is_partitioned;
+                        });
+
+                    if (it == hpx::util::end(results))
+                        return true;
+
+                    if (*it == partition_status::mixed)
+                        ++it;
+
+                    return std::all_of(
+                        it, hpx::util::end(results), [](partition_status x) {
+                            return x == partition_status::is_not_partitioned;
+                        });
                 };
 
                 return util::partitioner<ExPolicy, bool,

libs/core/algorithms/tests/unit/algorithms/CMakeLists.txt

@@ -80,6 +80,7 @@ set(tests
     inplace_merge
     is_partitioned
     is_sorted
+    is_sorted_projection
     is_sorted_until
     lexicographical_compare
     make_heap

libs/core/algorithms/tests/unit/algorithms/is_sorted_projection.cpp

@@ -0,0 +1,153 @@
+//  Copyright (c) 2024-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 <hpx/algorithm.hpp>
+#include <hpx/init.hpp>
+#include <hpx/modules/testing.hpp>
+
+#include <cstddef>
+#include <functional>
+#include <utility>
+#include <vector>
+
+void test_is_sorted_projection()
+{
+    using namespace hpx::execution;
+    using element = std::pair<int, int>;
+
+    std::vector<element> sorted_c = {{10, 1}, {20, 3}, {30, 5}, {40, 8}};
+    std::vector<element> unsorted_c = {{10, 1}, {20, 5}, {30, 3}, {40, 8}};
+
+    auto proj = [](element const& p) { return p.second; };
+
+    HPX_TEST(hpx::is_sorted(
+        sorted_c.begin(), sorted_c.end(), std::less<int>{}, proj));
+    HPX_TEST(!hpx::is_sorted(
+        unsorted_c.begin(), unsorted_c.end(), std::less<int>{}, proj));
+
+    HPX_TEST(hpx::is_sorted(
+        seq, sorted_c.begin(), sorted_c.end(), std::less<int>{}, proj));
+    HPX_TEST(!hpx::is_sorted(
+        seq, unsorted_c.begin(), unsorted_c.end(), std::less<int>{}, proj));
+
+    HPX_TEST(hpx::is_sorted(
+        par, sorted_c.begin(), sorted_c.end(), std::less<int>{}, proj));
+    HPX_TEST(!hpx::is_sorted(
+        par, unsorted_c.begin(), unsorted_c.end(), std::less<int>{}, proj));
+
+    HPX_TEST(hpx::is_sorted(
+        par(task), sorted_c.begin(), sorted_c.end(), std::less<int>{}, proj)
+            .get());
+    HPX_TEST(!hpx::is_sorted(
+        par(task), unsorted_c.begin(), unsorted_c.end(), std::less<int>{}, proj)
+            .get());
+
+    HPX_TEST(hpx::is_sorted(
+        par, sorted_c.begin(), sorted_c.begin(), std::less<int>{}, proj));
+    HPX_TEST(hpx::is_sorted(
+        par, sorted_c.begin(), sorted_c.begin() + 1, std::less<int>{}, proj));
+}
+
+void test_is_sorted_until_projection()
+{
+    using namespace hpx::execution;
+    using element = std::pair<int, int>;
+
+    std::vector<element> c = {{10, 1}, {20, 3}, {30, 8}, {40, 5}, {50, 9}};
+
+    auto proj = [](element const& p) { return p.second; };
+
+    auto it_seq =
+        hpx::is_sorted_until(c.begin(), c.end(), std::less<int>{}, proj);
+    HPX_TEST(it_seq != c.end());
+    HPX_TEST_EQ(it_seq->second, 5);
+
+    auto it_seq2 =
+        hpx::is_sorted_until(seq, c.begin(), c.end(), std::less<int>{}, proj);
+    HPX_TEST(it_seq2 != c.end());
+    HPX_TEST_EQ(it_seq2->second, 5);
+
+    auto it_par = hpx::is_sorted_until(
+        hpx::execution::par, c.begin(), c.end(), std::less<int>{}, proj);
+    HPX_TEST(it_par != c.end() && it_par->second == 5);
+
+    auto it_task =
+        hpx::is_sorted_until(hpx::execution::par(hpx::execution::task),
+            c.begin(), c.end(), std::less<int>{}, proj)
+            .get();
+    HPX_TEST(it_task != c.end() && it_task->second == 5);
+
+    std::vector<element> fully_sorted = {{10, 1}, {20, 3}, {30, 5}};
+    auto it_end = hpx::is_sorted_until(
+        par, fully_sorted.begin(), fully_sorted.end(), std::less<int>{}, proj);
+    HPX_TEST(it_end == fully_sorted.end());
+
+    auto it_empty = hpx::is_sorted_until(par, fully_sorted.begin(),
+        fully_sorted.begin(), std::less<int>{}, proj);
+    HPX_TEST(it_empty == fully_sorted.begin());
+}
+
+void test_is_partitioned_projection()
+{
+    using namespace hpx::execution;
+    using element = std::pair<int, int>;
+
+    std::vector<element> partitioned_c = {{10, 2}, {20, 4}, {30, 1}, {40, 3}};
+    std::vector<element> not_partitioned_c = {
+        {10, 2}, {20, 1}, {30, 4}, {40, 3}};
+
+    auto proj = [](element const& p) { return p.second; };
+    auto pred = [](int v) { return v % 2 == 0; };
+
+    HPX_TEST(hpx::is_partitioned(
+        partitioned_c.begin(), partitioned_c.end(), pred, proj));
+    HPX_TEST(!hpx::is_partitioned(
+        not_partitioned_c.begin(), not_partitioned_c.end(), pred, proj));
+
+    HPX_TEST(hpx::is_partitioned(
+        seq, partitioned_c.begin(), partitioned_c.end(), pred, proj));
+    HPX_TEST(!hpx::is_partitioned(
+        seq, not_partitioned_c.begin(), not_partitioned_c.end(), pred, proj));
+
+    HPX_TEST(hpx::is_partitioned(
+        par, partitioned_c.begin(), partitioned_c.end(), pred, proj));
+    HPX_TEST(!hpx::is_partitioned(
+        par, not_partitioned_c.begin(), not_partitioned_c.end(), pred, proj));
+
+    HPX_TEST(hpx::is_partitioned(
+        par(task), partitioned_c.begin(), partitioned_c.end(), pred, proj)
+            .get());
+    HPX_TEST(!hpx::is_partitioned(par(task), not_partitioned_c.begin(),
+        not_partitioned_c.end(), pred, proj)
+            .get());
+
+    HPX_TEST(hpx::is_partitioned(
+        par, partitioned_c.begin(), partitioned_c.begin(), pred, proj));
+    HPX_TEST(hpx::is_partitioned(
+        par, partitioned_c.begin(), partitioned_c.begin() + 1, pred, proj));
+}
+
+int hpx_main(hpx::program_options::variables_map&)
+{
+    test_is_sorted_projection();
+    test_is_sorted_until_projection();
+    test_is_partitioned_projection();
+
+    return hpx::local::finalize();
+}
+
+int main(int argc, char* argv[])
+{
+    std::vector<std::string> 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();
+}