[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

@@ -1,6 +1,6 @@
 //  Copyright (c) 2020 ETH Zurich
 //  Copyright (c) 2015 Daniel Bourgeois
-//  Copyright (c) 2017-2025 Hartmut Kaiser
+//  Copyright (c) 2017-2026 Hartmut Kaiser
 //
 //  SPDX-License-Identifier: BSL-1.0
 //  Distributed under the Boost Software License, Version 1.0. (See accompanying
@@ -111,6 +111,8 @@ namespace hpx {
 #else
 
 #include <hpx/config.hpp>
+#include <hpx/algorithms/traits/projected.hpp>
+#include <hpx/modules/execution.hpp>
 #include <hpx/modules/executors.hpp>
 #include <hpx/modules/functional.hpp>
 #include <hpx/modules/iterator_support.hpp>
@@ -120,6 +122,7 @@ namespace hpx {
 #include <hpx/parallel/util/detail/algorithm_result.hpp>
 #include <hpx/parallel/util/detail/sender_util.hpp>
 #include <hpx/parallel/util/invoke_projected.hpp>
+
 #include <hpx/parallel/util/loop.hpp>
 #include <hpx/parallel/util/partitioner.hpp>
 
@@ -136,36 +139,22 @@ namespace hpx::parallel {
     ////////////////////////////////////////////////////////////////////////////
     // is_partitioned
     namespace detail {
+        enum class partition_status : std::uint8_t
+        {
+            false_ = 0,
+            true_ = 1,
+            mixed = 2,
+            cancelled = 3
+        };
 
         /// \cond NOINTERNAL
-        HPX_CXX_CORE_EXPORT template <typename T>
-        inline bool sequential_is_partitioned(std::vector<T>&& res)
-        {
-            auto first = res.begin();
-            auto const last = res.end();
-            while (first != last && hpx::unwrap(*first))
-            {
-                ++first;
-            }
-            if (first != last)
-            {
-                ++first;
-                while (first != last)
-                {
-                    if (hpx::unwrap(*first))
-                        return false;
-                    ++first;
-                }
-            }
-            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>
         {
             constexpr is_partitioned() noexcept
-              : algorithm<is_partitioned, bool>("is_partitioned")
+              : algorithm<is_partitioned<Iter, Sent>, bool>("is_partitioned")
             {
             }
 
@@ -184,8 +173,7 @@ 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>;
@@ -208,36 +196,68 @@ 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::true_ :
+                                          partition_status::false_;
+
+                    bool is_mixed = false;
+
+                    using local_policy = std::conditional_t<
+                        hpx::is_unsequenced_execution_policy_v<
+                            std::decay_t<ExPolicy>>,
+                        hpx::execution::unsequenced_policy,
+                        hpx::execution::sequenced_policy>;
 
-                    util::loop_n<std::decay_t<ExPolicy>>(++part_begin,
-                        --part_count, tok,
-                        [&fst_bool, &pred_projected, &tok](
-                            Iter_ const& a) mutable -> void {
+                    util::loop_n<local_policy>(++part_begin, --part_count, tok,
+                        [&fst_bool, &is_mixed, &pred_projected, &tok](
+                            auto 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::true_ :
+                                                  partition_status::false_);
                 };
 
                 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::true_;
+                        });
+
+                    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::false_;
+                        });
                 };
 
                 return util::partitioner<ExPolicy, bool,
@@ -258,7 +278,10 @@ namespace hpx {
         template <typename FwdIter, typename Pred>
         // clang-format off
             requires (
-                std::forward_iterator<FwdIter>
+                std::forward_iterator<FwdIter> &&
+                hpx::is_invocable_v<Pred,
+                    hpx::traits::iter_value_t<FwdIter>
+                >
             )
         // clang-format on
         friend bool tag_fallback_invoke(
@@ -273,7 +296,10 @@ namespace hpx {
         // clang-format off
             requires (
                 hpx::is_execution_policy_v<ExPolicy> &&
-                std::forward_iterator<FwdIter>
+                std::forward_iterator<FwdIter> &&
+                hpx::is_invocable_v<Pred,
+                    hpx::traits::iter_value_t<FwdIter>
+                >
             )
         // clang-format on
         friend decltype(auto) tag_fallback_invoke(hpx::is_partitioned_t,
@@ -283,6 +309,48 @@ namespace hpx {
                 .call(HPX_FORWARD(ExPolicy, policy), first, last,
                     HPX_MOVE(pred), hpx::identity_v);
         }
+
+        template <typename FwdIter, typename Pred,
+            typename Proj = hpx::identity>
+        // clang-format off
+            requires (
+                std::forward_iterator<FwdIter> &&
+                hpx::parallel::traits::is_projected_v<Proj, FwdIter> &&
+                hpx::is_invocable_v<Pred,
+                    hpx::util::invoke_result_t<Proj,
+                        hpx::traits::iter_value_t<FwdIter>>
+                >
+            )
+        // clang-format on
+        friend bool tag_fallback_invoke(hpx::is_partitioned_t, FwdIter first,
+            FwdIter last, Pred pred, Proj proj)
+        {
+            return hpx::parallel::detail::is_partitioned<FwdIter, FwdIter>()
+                .call(hpx::execution::seq, first, last, HPX_MOVE(pred),
+                    HPX_MOVE(proj));
+        }
+
+        template <typename ExPolicy, typename FwdIter, typename Pred,
+            typename Proj = hpx::identity>
+        // clang-format off
+            requires (
+                hpx::is_execution_policy_v<ExPolicy> &&
+                std::forward_iterator<FwdIter> &&
+                hpx::parallel::traits::is_projected_v<Proj, FwdIter> &&
+                hpx::is_invocable_v<Pred,
+                    hpx::util::invoke_result_t<Proj,
+                        hpx::traits::iter_value_t<FwdIter>>
+                >
+            )
+        // clang-format on
+        friend decltype(auto) tag_fallback_invoke(hpx::is_partitioned_t,
+            ExPolicy&& policy, FwdIter first, FwdIter last, Pred pred,
+            Proj proj)
+        {
+            return hpx::parallel::detail::is_partitioned<FwdIter, FwdIter>()
+                .call(HPX_FORWARD(ExPolicy, policy), first, last,
+                    HPX_MOVE(pred), HPX_MOVE(proj));
+        }
     } is_partitioned{};
 }    // namespace hpx