#865

Committed 13 Jan 2023 11:46PM UTC coverage: 86.431% (+0.08%) from 86.354%

Build # #865

Build Type

push

Committed by StellarBot

Commit Message

Merge #6132

6132: Fixing to_non_par() for parallel simd policies r=hkaiser a=hkaiser



Co-authored-by: Hartmut Kaiser <hartmut.kaiser@gmail.com>

Run Details

64 of 64 new or added lines in 3 files covered. (100.0%)

174506 of 201901 relevant lines covered (86.43%)

1999640.87 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

95.74

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

//  Copyright (c) 2019-2020 ETH Zurich
//  Copyright (c) 2007-2023 Hartmut Kaiser
//  Copyright (c) 2019 Agustin Berge
//
//  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)

#pragma once

#include <hpx/config.hpp>
#include <hpx/allocator_support/internal_allocator.hpp>
#include <hpx/assert.hpp>
#include <hpx/async_base/launch_policy.hpp>
#include <hpx/execution/algorithms/detail/predicates.hpp>
#include <hpx/execution/detail/async_launch_policy_dispatch.hpp>
#include <hpx/execution/detail/future_exec.hpp>
#include <hpx/execution/detail/post_policy_dispatch.hpp>
#include <hpx/execution/detail/sync_launch_policy_dispatch.hpp>
#include <hpx/execution/executors/execution_parameters.hpp>
#include <hpx/execution/executors/fused_bulk_execute.hpp>
#include <hpx/execution/executors/static_chunk_size.hpp>
#include <hpx/execution_base/execution.hpp>
#include <hpx/execution_base/traits/is_executor.hpp>
#include <hpx/executors/detail/index_queue_spawning.hpp>
#include <hpx/executors/execution_policy_mappings.hpp>
#include <hpx/functional/bind_back.hpp>
#include <hpx/functional/deferred_call.hpp>
#include <hpx/functional/invoke.hpp>
#include <hpx/functional/one_shot.hpp>
#include <hpx/futures/future.hpp>
#include <hpx/futures/traits/future_traits.hpp>
#include <hpx/iterator_support/range.hpp>
#include <hpx/modules/concepts.hpp>
#include <hpx/modules/topology.hpp>
#include <hpx/serialization/serialize.hpp>
#include <hpx/threading_base/annotated_function.hpp>
#include <hpx/threading_base/scheduler_base.hpp>
#include <hpx/threading_base/thread_data.hpp>
#include <hpx/threading_base/thread_helpers.hpp>
#include <hpx/threading_base/thread_pool_base.hpp>
#include <hpx/timing/steady_clock.hpp>
#include <hpx/type_support/unused.hpp>

#include <algorithm>
#include <cstddef>
#include <string>
#include <type_traits>
#include <utility>
#include <vector>

namespace hpx::parallel::execution::detail {

    template <typename Policy>
    struct get_default_policy
    {
        static constexpr Policy call() noexcept
        {
            return Policy{};
        }
    };

    template <>
    struct get_default_policy<hpx::launch>
    {
        static constexpr hpx::launch::async_policy call() noexcept
        {
            return hpx::launch::async_policy{};
        }
    };

    ///////////////////////////////////////////////////////////////////////
    template <typename F, typename Shape, typename... Ts>
    struct bulk_function_result;

    ///////////////////////////////////////////////////////////////////////
    template <typename F, typename Shape, typename Future, typename... Ts>
    struct bulk_then_execute_result;

    template <typename F, typename Shape, typename Future, typename... Ts>
    struct then_bulk_function_result;
}    // namespace hpx::parallel::execution::detail

namespace hpx::execution {

    ///////////////////////////////////////////////////////////////////////////
    /// A \a parallel_executor creates groups of parallel execution agents
    /// which execute in threads implicitly created by the executor. This
    /// executor prefers continuing with the creating thread first before
    /// executing newly created threads.
    ///
    /// This executor conforms to the concepts of a TwoWayExecutor,
    /// and a BulkTwoWayExecutor
    template <typename Policy>
    struct parallel_policy_executor
    {
        /// Associate the parallel_execution_tag executor tag type as a default
        /// with this executor, except if the given launch policy is synch.
        using execution_category =
            std::conditional_t<std::is_same_v<Policy, launch::sync_policy>,
                sequenced_execution_tag, parallel_execution_tag>;

        /// Associate the static_chunk_size executor parameters type as a default
        /// with this executor.
        using executor_parameters_type = experimental::static_chunk_size;

        /// Create a new parallel executor
        constexpr explicit parallel_policy_executor(
            threads::thread_priority priority,
            threads::thread_stacksize stacksize =
                threads::thread_stacksize::default_,
            threads::thread_schedule_hint schedulehint = {},
            Policy l =
                parallel::execution::detail::get_default_policy<Policy>::call(),
            std::size_t hierarchical_threshold =
                hierarchical_threshold_default_)
          : pool_(nullptr)
          , policy_(l, priority, stacksize, schedulehint)
          , hierarchical_threshold_(hierarchical_threshold)
        {
        }

        constexpr explicit parallel_policy_executor(
            threads::thread_stacksize stacksize,
            threads::thread_schedule_hint schedulehint = {},
            Policy l =
                parallel::execution::detail::get_default_policy<Policy>::call())
          : pool_(nullptr)
          , policy_(l, l.priority(), stacksize, schedulehint)
        {
        }

        constexpr explicit parallel_policy_executor(
            threads::thread_schedule_hint schedulehint,
            Policy l =
                parallel::execution::detail::get_default_policy<Policy>::call())
          : pool_(nullptr)
          , policy_(l, l.priority(), l.stacksize(), schedulehint)
        {
        }

        constexpr explicit parallel_policy_executor(
            Policy l =
                parallel::execution::detail::get_default_policy<Policy>::call())
          : pool_(nullptr)
          , policy_(l)
        {
        }

        constexpr explicit parallel_policy_executor(
            threads::thread_pool_base* pool, Policy l,
            std::size_t hierarchical_threshold =
                hierarchical_threshold_default_)
          : pool_(pool)
          , policy_(l)
          , hierarchical_threshold_(hierarchical_threshold)
        {
        }

        constexpr explicit parallel_policy_executor(
            threads::thread_pool_base* pool,
            threads::thread_priority priority =
                threads::thread_priority::default_,
            threads::thread_stacksize stacksize =
                threads::thread_stacksize::default_,
            threads::thread_schedule_hint schedulehint = {},
            Policy l =
                parallel::execution::detail::get_default_policy<Policy>::call(),
            std::size_t hierarchical_threshold =
                hierarchical_threshold_default_)
          : pool_(pool)
          , policy_(l, priority, stacksize, schedulehint)
          , hierarchical_threshold_(hierarchical_threshold)
        {
        }

        void set_hierarchical_threshold(std::size_t threshold)
        {
            hierarchical_threshold_ = threshold;
        }

    private:
        // property implementations

#if defined(HPX_HAVE_THREAD_DESCRIPTION)
        friend constexpr parallel_policy_executor tag_invoke(
            hpx::execution::experimental::with_annotation_t,
            parallel_policy_executor const& exec, char const* annotation)
        {
            auto exec_with_annotation = exec;
            exec_with_annotation.annotation_ = annotation;
            return exec_with_annotation;
        }

        friend parallel_policy_executor tag_invoke(
            hpx::execution::experimental::with_annotation_t,
            parallel_policy_executor const& exec, std::string annotation)
        {
            auto exec_with_annotation = exec;
            exec_with_annotation.annotation_ =
                hpx::detail::store_function_annotation(HPX_MOVE(annotation));
            return exec_with_annotation;
        }

        friend constexpr char const* tag_invoke(
            hpx::execution::experimental::get_annotation_t,
            parallel_policy_executor const& exec) noexcept
        {
            return exec.annotation_;
        }
#endif

        friend constexpr parallel_policy_executor tag_invoke(
            hpx::parallel::execution::with_processing_units_count_t,
            parallel_policy_executor const& exec, std::size_t num_cores)
        {
            auto exec_with_num_cores = exec;
            exec_with_num_cores.num_cores_ = num_cores;
            return exec_with_num_cores;
        }

        friend constexpr std::size_t tag_invoke(
            hpx::parallel::execution::processing_units_count_t,
            parallel_policy_executor const& exec,
            hpx::chrono::steady_duration const& = hpx::chrono::null_duration,
            std::size_t = 0)
        {
            return exec.get_num_cores();
        }

        friend auto tag_invoke(
            hpx::execution::experimental::get_processing_units_mask_t,
            parallel_policy_executor const& exec)
        {
            auto pool = exec.pool_ ?
                exec.pool_ :
                threads::detail::get_self_or_default_pool();
            return pool->get_used_processing_units(exec.get_num_cores(), false);
        }

        friend auto tag_invoke(hpx::execution::experimental::get_cores_mask_t,
            parallel_policy_executor const& exec)
        {
            auto pool = exec.pool_ ?
                exec.pool_ :
                threads::detail::get_self_or_default_pool();
            return pool->get_used_processing_units(exec.get_num_cores(), true);
        }

    public:
        // backwards compatibility support, will be removed in the future
        template <typename Parameters>
        std::size_t processing_units_count(Parameters&&,
            hpx::chrono::steady_duration const& = hpx::chrono::null_duration,
            std::size_t = 0) const
        {
            return get_num_cores();
        }

    public:
        /// \cond NOINTERNAL
        constexpr bool operator==(
            parallel_policy_executor const& rhs) const noexcept
        {
            return policy_ == rhs.policy_ && pool_ == rhs.pool_ &&
                hierarchical_threshold_ == rhs.hierarchical_threshold_;
        }

        constexpr bool operator!=(
            parallel_policy_executor const& rhs) const noexcept
        {
            return !(*this == rhs);
        }

        constexpr parallel_policy_executor const& context() const noexcept
        {
            return *this;
        }

        void policy(Policy policy) noexcept
        {
            policy_ = HPX_MOVE(policy);
        }

        constexpr Policy const& policy() const noexcept
        {
            return policy_;
        }
        /// \endcond

    private:
        /// \cond NOINTERNAL

        // OneWayExecutor interface
        template <typename F, typename... Ts>
        friend decltype(auto) tag_invoke(
            hpx::parallel::execution::sync_execute_t,
            parallel_policy_executor const& exec, F&& f, Ts&&... ts)
        {
#if defined(HPX_HAVE_THREAD_DESCRIPTION)
            hpx::scoped_annotation annotate(exec.annotation_ ?
                    exec.annotation_ :
                    "parallel_policy_executor::sync_execute");
#endif
            HPX_UNUSED(exec);
            return hpx::detail::sync_launch_policy_dispatch<
                launch::sync_policy>::call(exec.policy_, HPX_FORWARD(F, f),
                HPX_FORWARD(Ts, ts)...);
        }

        // TwoWayExecutor interface
        template <typename F, typename... Ts>
        friend decltype(auto) tag_invoke(
            hpx::parallel::execution::async_execute_t,
            parallel_policy_executor const& exec, F&& f, Ts&&... ts)
        {
#if defined(HPX_HAVE_THREAD_DESCRIPTION)
            hpx::threads::thread_description desc(f, exec.annotation_);
#else
            hpx::threads::thread_description desc(f);
#endif
            auto pool = exec.pool_ ?
                exec.pool_ :
                threads::detail::get_self_or_default_pool();
            return hpx::detail::async_launch_policy_dispatch<Policy>::call(
                exec.policy_, desc, pool, HPX_FORWARD(F, f),
                HPX_FORWARD(Ts, ts)...);
        }

        template <typename F, typename Future, typename... Ts>
        friend decltype(auto) tag_invoke(
            hpx::parallel::execution::then_execute_t,
            parallel_policy_executor const& exec, F&& f, Future&& predecessor,
            Ts&&... ts)
        {
            using result_type =
                hpx::util::detail::invoke_deferred_result_t<F, Future, Ts...>;

#if defined(HPX_HAVE_THREAD_DESCRIPTION)
            auto&& func = hpx::util::one_shot(hpx::bind_back(
                hpx::annotated_function(HPX_FORWARD(F, f), exec.annotation_),
                HPX_FORWARD(Ts, ts)...));
#else
            auto&& func = hpx::util::one_shot(
                hpx::bind_back(HPX_FORWARD(F, f), HPX_FORWARD(Ts, ts)...));
#endif

            hpx::traits::detail::shared_state_ptr_t<result_type> p =
                lcos::detail::make_continuation_alloc_nounwrap<result_type>(
                    hpx::util::internal_allocator<>{},
                    HPX_FORWARD(Future, predecessor), exec.policy_,
                    HPX_MOVE(func));

            return hpx::traits::future_access<hpx::future<result_type>>::create(
                HPX_MOVE(p));
        }

        // NonBlockingOneWayExecutor (adapted) interface
        template <typename F, typename... Ts>
        void post_impl(F&& f, Ts&&... ts) const
        {
#if defined(HPX_HAVE_THREAD_DESCRIPTION)
            hpx::threads::thread_description desc(f, annotation_);
#else
            hpx::threads::thread_description desc(f);
#endif
            auto pool =
                pool_ ? pool_ : threads::detail::get_self_or_default_pool();
            hpx::detail::post_policy_dispatch<Policy>::call(
                policy_, desc, pool, HPX_FORWARD(F, f), HPX_FORWARD(Ts, ts)...);
        }

        template <typename F, typename... Ts>
        friend void tag_invoke(hpx::parallel::execution::post_t,
            parallel_policy_executor const& exec, F&& f, Ts&&... ts)
        {
            exec.post_impl(HPX_FORWARD(F, f), HPX_FORWARD(Ts, ts)...);
        }

        // BulkTwoWayExecutor interface
        // clang-format off
        template <typename F, typename S, typename... Ts,
            HPX_CONCEPT_REQUIRES_(
                !std::is_integral_v<S>
            )>
        // clang-format on
        friend decltype(auto) tag_invoke(
            hpx::parallel::execution::bulk_async_execute_t,
            parallel_policy_executor const& exec, F&& f, S const& shape,
            Ts&&... ts)
        {
#if defined(HPX_HAVE_THREAD_DESCRIPTION)
            hpx::threads::thread_description desc(f, exec.annotation_);
#else
            hpx::threads::thread_description desc(f);
#endif
            auto pool = exec.pool_ ?
                exec.pool_ :
                threads::detail::get_self_or_default_pool();

            bool do_not_combine_tasks = hpx::threads::do_not_combine_tasks(
                exec.policy().get_hint().sharing_mode());

            // use scheduling based on index_queue if no hierarchical threshold
            // is given
            if (exec.hierarchical_threshold_ == 0 && !do_not_combine_tasks)
            {
                return parallel::execution::detail::
                    index_queue_bulk_async_execute(desc, pool,
                        exec.get_first_core(), exec.get_num_cores(),
                        exec.hierarchical_threshold_, exec.policy_,
                        HPX_FORWARD(F, f), shape, HPX_FORWARD(Ts, ts)...);
            }

            return parallel::execution::detail::hierarchical_bulk_async_execute(
                desc, pool, exec.get_first_core(), exec.get_num_cores(),
                exec.hierarchical_threshold_, exec.policy_, HPX_FORWARD(F, f),
                shape, HPX_FORWARD(Ts, ts)...);
        }

        // clang-format off
        template <typename F, typename S, typename Future, typename... Ts,
            HPX_CONCEPT_REQUIRES_(
                !std::is_integral_v<S>
            )>
        // clang-format on
        friend decltype(auto) tag_invoke(
            hpx::parallel::execution::bulk_then_execute_t,
            parallel_policy_executor const& exec, F&& f, S const& shape,
            Future&& predecessor, Ts&&... ts)
        {
#if defined(HPX_HAVE_THREAD_DESCRIPTION)
            return parallel::execution::detail::
                hierarchical_bulk_then_execute_helper(exec, exec.policy_,
                    hpx::annotated_function(
                        HPX_FORWARD(F, f), exec.annotation_),
                    shape, HPX_FORWARD(Future, predecessor),
                    HPX_FORWARD(Ts, ts)...);
#else
            return parallel::execution::detail::
                hierarchical_bulk_then_execute_helper(exec, exec.policy_,
                    HPX_FORWARD(F, f), shape, HPX_FORWARD(Future, predecessor),
                    HPX_FORWARD(Ts, ts)...);
#endif
        }

        // map execution policy categories to proper executor
        friend decltype(auto) tag_invoke(
            hpx::execution::experimental::to_non_par_t,
            parallel_policy_executor const& exec)
        {
            if constexpr (std::is_same_v<Policy, launch::sync_policy>)
            {
                return exec;
            }
            else
            {
                auto non_par_exec =
                    parallel_policy_executor<launch::sync_policy>(exec.pool_,
                        launch::sync_policy(exec.policy_.priority(),
                            exec.policy_.stacksize(), exec.policy_.hint()),
                        exec.hierarchical_threshold_);

#if defined(HPX_HAVE_THREAD_DESCRIPTION)
                return hpx::execution::experimental::with_annotation(
                    HPX_MOVE(non_par_exec), exec.annotation_);
#else
                return non_par_exec;
#endif
            }
        }
        /// \endcond

    private:
        /// \cond NOINTERNAL
        std::size_t get_num_cores() const
        {
            if (num_cores_ != 0)
                return num_cores_;

            if constexpr (std::is_same_v<Policy, launch::sync_policy>)
            {
                return 1;
            }
            else
            {
                auto pool =
                    pool_ ? pool_ : threads::detail::get_self_or_default_pool();
                return pool->get_os_thread_count();
            }
        }

        std::size_t get_first_core() const
        {
            if (policy_.hint().mode !=
                    hpx::threads::thread_schedule_hint_mode::none &&
                policy_.hint().hint != -1)
            {
                return policy_.hint().hint;
            }
            return 0;
        }

        friend class hpx::serialization::access;

        template <typename Archive>
        void serialize(Archive& ar, unsigned int const /* version */)
        {
            // clang-format off
            ar & policy_ & hierarchical_threshold_ & num_cores_;
            // clang-format on
        }
        /// \endcond

    private:
        /// \cond NOINTERNAL
        static constexpr std::size_t hierarchical_threshold_default_ = 0;

        threads::thread_pool_base* pool_;
        Policy policy_;
        std::size_t hierarchical_threshold_ = hierarchical_threshold_default_;
        std::size_t num_cores_ = 0;
#if defined(HPX_HAVE_THREAD_DESCRIPTION)
        char const* annotation_ = nullptr;
#endif
        /// \endcond
    };

    // support all properties exposed by the embedded policy
    // clang-format off
    template <typename Tag, typename Policy, typename Property,
        HPX_CONCEPT_REQUIRES_(
            hpx::execution::experimental::is_scheduling_property_v<Tag>
        )>
    // clang-format on
    auto tag_invoke(
        Tag tag, parallel_policy_executor<Policy> const& exec, Property&& prop)
        -> decltype(std::declval<parallel_policy_executor<Policy>>().policy(
                        std::declval<Tag>()(
                            std::declval<Policy>(), std::declval<Property>())),
            parallel_policy_executor<Policy>())
    {
        auto exec_with_prop = exec;
        exec_with_prop.policy(tag(exec.policy(), HPX_FORWARD(Property, prop)));
        return exec_with_prop;
    }

    // clang-format off
    template <typename Tag,typename Policy,
        HPX_CONCEPT_REQUIRES_(
            hpx::execution::experimental::is_scheduling_property_v<Tag>
        )>
    // clang-format on
    auto tag_invoke(Tag tag, parallel_policy_executor<Policy> const& exec)
        -> decltype(std::declval<Tag>()(std::declval<Policy>()))
    {
        return tag(exec.policy());
    }

    using parallel_executor = parallel_policy_executor<hpx::launch>;
}    // namespace hpx::execution

namespace hpx::parallel::execution {
    /// \cond NOINTERNAL
    template <typename Policy>
    struct is_one_way_executor<hpx::execution::parallel_policy_executor<Policy>>
      : std::true_type
    {
    };

    template <typename Policy>
    struct is_never_blocking_one_way_executor<
        hpx::execution::parallel_policy_executor<Policy>> : std::true_type
    {
    };

    template <typename Policy>
    struct is_two_way_executor<hpx::execution::parallel_policy_executor<Policy>>
      : std::true_type
    {
    };

    template <typename Policy>
    struct is_bulk_two_way_executor<
        hpx::execution::parallel_policy_executor<Policy>> : std::true_type
    {
    };
    /// \endcond
}    // namespace hpx::parallel::execution

1	// Copyright (c) 2019-2020 ETH Zurich
2	// Copyright (c) 2007-2023 Hartmut Kaiser
3	// Copyright (c) 2019 Agustin Berge
4	//
5	// SPDX-License-Identifier: BSL-1.0
6	// Distributed under the Boost Software License, Version 1.0. (See accompanying
7	// file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
8
9	#pragma once
10
11	#include <hpx/config.hpp>
12	#include <hpx/allocator_support/internal_allocator.hpp>
13	#include <hpx/assert.hpp>
14	#include <hpx/async_base/launch_policy.hpp>
15	#include <hpx/execution/algorithms/detail/predicates.hpp>
16	#include <hpx/execution/detail/async_launch_policy_dispatch.hpp>
17	#include <hpx/execution/detail/future_exec.hpp>
18	#include <hpx/execution/detail/post_policy_dispatch.hpp>
19	#include <hpx/execution/detail/sync_launch_policy_dispatch.hpp>
20	#include <hpx/execution/executors/execution_parameters.hpp>
21	#include <hpx/execution/executors/fused_bulk_execute.hpp>
22	#include <hpx/execution/executors/static_chunk_size.hpp>
23	#include <hpx/execution_base/execution.hpp>
24	#include <hpx/execution_base/traits/is_executor.hpp>
25	#include <hpx/executors/detail/index_queue_spawning.hpp>
26	#include <hpx/executors/execution_policy_mappings.hpp>
27	#include <hpx/functional/bind_back.hpp>
28	#include <hpx/functional/deferred_call.hpp>
29	#include <hpx/functional/invoke.hpp>
30	#include <hpx/functional/one_shot.hpp>
31	#include <hpx/futures/future.hpp>
32	#include <hpx/futures/traits/future_traits.hpp>
33	#include <hpx/iterator_support/range.hpp>
34	#include <hpx/modules/concepts.hpp>
35	#include <hpx/modules/topology.hpp>
36	#include <hpx/serialization/serialize.hpp>
37	#include <hpx/threading_base/annotated_function.hpp>
38	#include <hpx/threading_base/scheduler_base.hpp>
39	#include <hpx/threading_base/thread_data.hpp>
40	#include <hpx/threading_base/thread_helpers.hpp>
41	#include <hpx/threading_base/thread_pool_base.hpp>
42	#include <hpx/timing/steady_clock.hpp>
43	#include <hpx/type_support/unused.hpp>
44
45	#include <algorithm>
46	#include <cstddef>
47	#include <string>
48	#include <type_traits>
49	#include <utility>
50	#include <vector>
51
52	namespace hpx::parallel::execution::detail {
53
54	template <typename Policy>
55	struct get_default_policy
56	{
57	static constexpr Policy call() noexcept	24✔
58	{
59	return Policy{};	24✔
60	}
61	};
62
63	template <>
64	struct get_default_policy<hpx::launch>
65	{
66	static constexpr hpx::launch::async_policy call() noexcept	576,702✔
67	{
68	return hpx::launch::async_policy{};	576,702✔
69	}
70	};
71
72	///////////////////////////////////////////////////////////////////////
73	template <typename F, typename Shape, typename... Ts>
74	struct bulk_function_result;
75
76	///////////////////////////////////////////////////////////////////////
77	template <typename F, typename Shape, typename Future, typename... Ts>
78	struct bulk_then_execute_result;
79
80	template <typename F, typename Shape, typename Future, typename... Ts>
81	struct then_bulk_function_result;
82	} // namespace hpx::parallel::execution::detail
83
84	namespace hpx::execution {
85
86	///////////////////////////////////////////////////////////////////////////
87	/// A \a parallel_executor creates groups of parallel execution agents
88	/// which execute in threads implicitly created by the executor. This
89	/// executor prefers continuing with the creating thread first before
90	/// executing newly created threads.
91	///
92	/// This executor conforms to the concepts of a TwoWayExecutor,
93	/// and a BulkTwoWayExecutor
94	template <typename Policy>
95	struct parallel_policy_executor
96	{
97	/// Associate the parallel_execution_tag executor tag type as a default
98	/// with this executor, except if the given launch policy is synch.
99	using execution_category =
100	std::conditional_t<std::is_same_v<Policy, launch::sync_policy>,
101	sequenced_execution_tag, parallel_execution_tag>;
102
103	/// Associate the static_chunk_size executor parameters type as a default
104	/// with this executor.
105	using executor_parameters_type = experimental::static_chunk_size;
106
107	/// Create a new parallel executor
108	constexpr explicit parallel_policy_executor(	2,148✔
109	threads::thread_priority priority,
110	threads::thread_stacksize stacksize =
111	threads::thread_stacksize::default_,
112	threads::thread_schedule_hint schedulehint = {},
113	Policy l =
114	parallel::execution::detail::get_default_policy<Policy>::call(),
115	std::size_t hierarchical_threshold =
116	hierarchical_threshold_default_)
117	: pool_(nullptr)	2,148✔
118	, policy_(l, priority, stacksize, schedulehint)	2,148✔
119	, hierarchical_threshold_(hierarchical_threshold)	2,148✔
120	{
121	}	2,148✔
122
123	constexpr explicit parallel_policy_executor(	51✔
124	threads::thread_stacksize stacksize,
125	threads::thread_schedule_hint schedulehint = {},
126	Policy l =
127	parallel::execution::detail::get_default_policy<Policy>::call())
128	: pool_(nullptr)	51✔
129	, policy_(l, l.priority(), stacksize, schedulehint)	51✔
130	{
131	}	51✔
132
133	constexpr explicit parallel_policy_executor(	×
134	threads::thread_schedule_hint schedulehint,
135	Policy l =
136	parallel::execution::detail::get_default_policy<Policy>::call())
137	: pool_(nullptr)	×
138	, policy_(l, l.priority(), l.stacksize(), schedulehint)	×
139	{
140	}	×
141
142	constexpr explicit parallel_policy_executor(	589,829✔
143	Policy l =
144	parallel::execution::detail::get_default_policy<Policy>::call())
145	: pool_(nullptr)	589,833✔
146	, policy_(l)	589,833✔
147	{
148	}	589,833✔
149
150	constexpr explicit parallel_policy_executor(	4✔
151	threads::thread_pool_base* pool, Policy l,
152	std::size_t hierarchical_threshold =
153	hierarchical_threshold_default_)
154	: pool_(pool)	4✔
155	, policy_(l)	4✔
156	, hierarchical_threshold_(hierarchical_threshold)	4✔
157	{
158	}	4✔
159
160	constexpr explicit parallel_policy_executor(	99✔
161	threads::thread_pool_base* pool,
162	threads::thread_priority priority =
163	threads::thread_priority::default_,
164	threads::thread_stacksize stacksize =
165	threads::thread_stacksize::default_,
166	threads::thread_schedule_hint schedulehint = {},
167	Policy l =
168	parallel::execution::detail::get_default_policy<Policy>::call(),
169	std::size_t hierarchical_threshold =
170	hierarchical_threshold_default_)
171	: pool_(pool)	99✔
172	, policy_(l, priority, stacksize, schedulehint)	99✔
173	, hierarchical_threshold_(hierarchical_threshold)	99✔
174	{
175	}	99✔
176
177	void set_hierarchical_threshold(std::size_t threshold)
178	{
179	hierarchical_threshold_ = threshold;
180	}
181
182	private:
183	// property implementations
184
185	#if defined(HPX_HAVE_THREAD_DESCRIPTION)
186	friend constexpr parallel_policy_executor tag_invoke(
187	hpx::execution::experimental::with_annotation_t,
188	parallel_policy_executor const& exec, char const* annotation)
189	{
190	auto exec_with_annotation = exec;
191	exec_with_annotation.annotation_ = annotation;
192	return exec_with_annotation;
193	}
194
195	friend parallel_policy_executor tag_invoke(
196	hpx::execution::experimental::with_annotation_t,
197	parallel_policy_executor const& exec, std::string annotation)
198	{
199	auto exec_with_annotation = exec;
200	exec_with_annotation.annotation_ =
201	hpx::detail::store_function_annotation(HPX_MOVE(annotation));
202	return exec_with_annotation;
203	}
204
205	friend constexpr char const* tag_invoke(
206	hpx::execution::experimental::get_annotation_t,
207	parallel_policy_executor const& exec) noexcept
208	{
209	return exec.annotation_;
210	}
211	#endif
212
213	friend constexpr parallel_policy_executor tag_invoke(	141✔
214	hpx::parallel::execution::with_processing_units_count_t,
215	parallel_policy_executor const& exec, std::size_t num_cores)
216	{
217	auto exec_with_num_cores = exec;	141✔
218	exec_with_num_cores.num_cores_ = num_cores;	141✔
219	return exec_with_num_cores;	141✔
220	}
221
222	friend constexpr std::size_t tag_invoke(	4✔
223	hpx::parallel::execution::processing_units_count_t,
224	parallel_policy_executor const& exec,
225	hpx::chrono::steady_duration const& = hpx::chrono::null_duration,
226	std::size_t = 0)
227	{
228	return exec.get_num_cores();	4✔
229	}
230
231	friend auto tag_invoke(	1✔
232	hpx::execution::experimental::get_processing_units_mask_t,
233	parallel_policy_executor const& exec)
234	{
235	auto pool = exec.pool_ ?	1✔
236	exec.pool_ :	×
237	threads::detail::get_self_or_default_pool();	1✔
238	return pool->get_used_processing_units(exec.get_num_cores(), false);	1✔
239	}
240
241	friend auto tag_invoke(hpx::execution::experimental::get_cores_mask_t,	1✔
242	parallel_policy_executor const& exec)
243	{
244	auto pool = exec.pool_ ?	1✔
245	exec.pool_ :	×
246	threads::detail::get_self_or_default_pool();	1✔
247	return pool->get_used_processing_units(exec.get_num_cores(), true);	1✔
248	}
249
250	public:
251	// backwards compatibility support, will be removed in the future
252	template <typename Parameters>
253	std::size_t processing_units_count(Parameters&&,	313,471✔
254	hpx::chrono::steady_duration const& = hpx::chrono::null_duration,
255	std::size_t = 0) const
256	{
257	return get_num_cores();	313,471✔
258	}
259
260	public:
261	/// \cond NOINTERNAL
262	constexpr bool operator==(
263	parallel_policy_executor const& rhs) const noexcept
264	{
265	return policy_ == rhs.policy_ && pool_ == rhs.pool_ &&
266	hierarchical_threshold_ == rhs.hierarchical_threshold_;
267	}
268
269	constexpr bool operator!=(
270	parallel_policy_executor const& rhs) const noexcept
271	{
272	return !(*this == rhs);
273	}
274
275	constexpr parallel_policy_executor const& context() const noexcept
276	{
277	return *this;
278	}
279
280	void policy(Policy policy) noexcept	1,391✔
281	{
282	policy_ = HPX_MOVE(policy);	1,391✔
283	}	1,391✔
284
285	constexpr Policy const& policy() const noexcept	183,612✔
286	{
287	return policy_;	183,612✔
288	}
289	/// \endcond
290
291	private:
292	/// \cond NOINTERNAL
293
294	// OneWayExecutor interface
295	template <typename F, typename... Ts>
296	friend decltype(auto) tag_invoke(	88✔
297	hpx::parallel::execution::sync_execute_t,
298	parallel_policy_executor const& exec, F&& f, Ts&&... ts)
299	{
300	#if defined(HPX_HAVE_THREAD_DESCRIPTION)
301	hpx::scoped_annotation annotate(exec.annotation_ ?
302	exec.annotation_ :
303	"parallel_policy_executor::sync_execute");
304	#endif
305	HPX_UNUSED(exec);	88✔
306	return hpx::detail::sync_launch_policy_dispatch<	88✔
307	launch::sync_policy>::call(exec.policy_, HPX_FORWARD(F, f),	88✔
308	HPX_FORWARD(Ts, ts)...);	85✔
309	}
310
311	// TwoWayExecutor interface
312	template <typename F, typename... Ts>
313	friend decltype(auto) tag_invoke(	886,124✔
314	hpx::parallel::execution::async_execute_t,
315	parallel_policy_executor const& exec, F&& f, Ts&&... ts)
316	{
317	#if defined(HPX_HAVE_THREAD_DESCRIPTION)
318	hpx::threads::thread_description desc(f, exec.annotation_);
319	#else
320	hpx::threads::thread_description desc(f);	886,124✔
321	#endif
322	auto pool = exec.pool_ ?	886,124✔
323	exec.pool_ :	603,746✔
324	threads::detail::get_self_or_default_pool();	282,378✔
325	return hpx::detail::async_launch_policy_dispatch<Policy>::call(	886,124✔
326	exec.policy_, desc, pool, HPX_FORWARD(F, f),	886,124✔
327	HPX_FORWARD(Ts, ts)...);	51,601✔
328	}
329
330	template <typename F, typename Future, typename... Ts>
331	friend decltype(auto) tag_invoke(	28,028✔
332	hpx::parallel::execution::then_execute_t,
333	parallel_policy_executor const& exec, F&& f, Future&& predecessor,
334	Ts&&... ts)
335	{
336	using result_type =
337	hpx::util::detail::invoke_deferred_result_t<F, Future, Ts...>;
338
339	#if defined(HPX_HAVE_THREAD_DESCRIPTION)
340	auto&& func = hpx::util::one_shot(hpx::bind_back(
341	hpx::annotated_function(HPX_FORWARD(F, f), exec.annotation_),
342	HPX_FORWARD(Ts, ts)...));
343	#else
344	auto&& func = hpx::util::one_shot(	28,028✔
345	hpx::bind_back(HPX_FORWARD(F, f), HPX_FORWARD(Ts, ts)...));	28,028✔
346	#endif
347
348	hpx::traits::detail::shared_state_ptr_t<result_type> p =
349	lcos::detail::make_continuation_alloc_nounwrap<result_type>(	28,028✔
350	hpx::util::internal_allocator<>{},	28,028✔
351	HPX_FORWARD(Future, predecessor), exec.policy_,	28,028✔
352	HPX_MOVE(func));	28,028✔
353
354	return hpx::traits::future_access<hpx::future<result_type>>::create(	28,028✔
355	HPX_MOVE(p));
356	}	28,028✔
357
358	// NonBlockingOneWayExecutor (adapted) interface
359	template <typename F, typename... Ts>
360	void post_impl(F&& f, Ts&&... ts) const	639,090✔
361	{
362	#if defined(HPX_HAVE_THREAD_DESCRIPTION)
363	hpx::threads::thread_description desc(f, annotation_);
364	#else
365	hpx::threads::thread_description desc(f);	639,094✔
366	#endif
367	auto pool =	639,086✔
368	pool_ ? pool_ : threads::detail::get_self_or_default_pool();	639,090✔
369	hpx::detail::post_policy_dispatch<Policy>::call(	639,086✔
370	policy_, desc, pool, HPX_FORWARD(F, f), HPX_FORWARD(Ts, ts)...);	639,086✔
371	}	639,086✔
372
373	template <typename F, typename... Ts>
374	friend void tag_invoke(hpx::parallel::execution::post_t,	639,090✔
375	parallel_policy_executor const& exec, F&& f, Ts&&... ts)
376	{
377	exec.post_impl(HPX_FORWARD(F, f), HPX_FORWARD(Ts, ts)...);	639,090✔
378	}	639,090✔
379
380	// BulkTwoWayExecutor interface
381	// clang-format off
382	template <typename F, typename S, typename... Ts,
383	HPX_CONCEPT_REQUIRES_(
384	!std::is_integral_v<S>
385	)>
386	// clang-format on
387	friend decltype(auto) tag_invoke(	181,039✔
388	hpx::parallel::execution::bulk_async_execute_t,
389	parallel_policy_executor const& exec, F&& f, S const& shape,
390	Ts&&... ts)
391	{
392	#if defined(HPX_HAVE_THREAD_DESCRIPTION)
393	hpx::threads::thread_description desc(f, exec.annotation_);
394	#else
395	hpx::threads::thread_description desc(f);	181,039✔
396	#endif
397	auto pool = exec.pool_ ?	181,039✔
398	exec.pool_ :	10✔
399	threads::detail::get_self_or_default_pool();	181,029✔
400
401	bool do_not_combine_tasks = hpx::threads::do_not_combine_tasks(	181,039✔
402	exec.policy().get_hint().sharing_mode());	181,039✔
403
404	// use scheduling based on index_queue if no hierarchical threshold
405	// is given
406	if (exec.hierarchical_threshold_ == 0 && !do_not_combine_tasks)	181,039✔
407	{
408	return parallel::execution::detail::	180,810✔
409	index_queue_bulk_async_execute(desc, pool,	180,810✔
410	exec.get_first_core(), exec.get_num_cores(),	180,810✔
411	exec.hierarchical_threshold_, exec.policy_,	180,810✔
412	HPX_FORWARD(F, f), shape, HPX_FORWARD(Ts, ts)...);	180,810✔
413	}
414
415	return parallel::execution::detail::hierarchical_bulk_async_execute(	229✔
416	desc, pool, exec.get_first_core(), exec.get_num_cores(),	229✔
417	exec.hierarchical_threshold_, exec.policy_, HPX_FORWARD(F, f),	229✔
418	shape, HPX_FORWARD(Ts, ts)...);	229✔
419	}	181,039✔
420
421	// clang-format off
422	template <typename F, typename S, typename Future, typename... Ts,
423	HPX_CONCEPT_REQUIRES_(
424	!std::is_integral_v<S>
425	)>
426	// clang-format on
427	friend decltype(auto) tag_invoke(	20✔
428	hpx::parallel::execution::bulk_then_execute_t,
429	parallel_policy_executor const& exec, F&& f, S const& shape,
430	Future&& predecessor, Ts&&... ts)
431	{
432	#if defined(HPX_HAVE_THREAD_DESCRIPTION)
433	return parallel::execution::detail::
434	hierarchical_bulk_then_execute_helper(exec, exec.policy_,
435	hpx::annotated_function(
436	HPX_FORWARD(F, f), exec.annotation_),
437	shape, HPX_FORWARD(Future, predecessor),
438	HPX_FORWARD(Ts, ts)...);
439	#else
440	return parallel::execution::detail::	20✔
441	hierarchical_bulk_then_execute_helper(exec, exec.policy_,	20✔
442	HPX_FORWARD(F, f), shape, HPX_FORWARD(Future, predecessor),	20✔
443	HPX_FORWARD(Ts, ts)...);	18✔
444	#endif
445	}
446
447	// map execution policy categories to proper executor
448	friend decltype(auto) tag_invoke(	4✔
449	hpx::execution::experimental::to_non_par_t,
450	parallel_policy_executor const& exec)
451	{
452	if constexpr (std::is_same_v<Policy, launch::sync_policy>)
453	{
454	return exec;
455	}
456	else
457	{
458	auto non_par_exec =
459	parallel_policy_executor<launch::sync_policy>(exec.pool_,	8✔
460	launch::sync_policy(exec.policy_.priority(),	8✔
461	exec.policy_.stacksize(), exec.policy_.hint()),	4✔
462	exec.hierarchical_threshold_);	4✔
463
464	#if defined(HPX_HAVE_THREAD_DESCRIPTION)
465	return hpx::execution::experimental::with_annotation(
466	HPX_MOVE(non_par_exec), exec.annotation_);
467	#else
468	return non_par_exec;	4✔
469	#endif
470	}
471	}
472	/// \endcond
473
474	private:
475	/// \cond NOINTERNAL
476	std::size_t get_num_cores() const	494,516✔
477	{
478	if (num_cores_ != 0)	494,516✔
479	return num_cores_;	318✔
480
481	if constexpr (std::is_same_v<Policy, launch::sync_policy>)
482	{
483	return 1;	4✔
484	}
485	else
486	{
487	auto pool =	494,193✔
488	pool_ ? pool_ : threads::detail::get_self_or_default_pool();	494,194✔
489	return pool->get_os_thread_count();	494,193✔
490	}
491	}	494,515✔
492
493	std::size_t get_first_core() const	181,039✔
494	{
495	if (policy_.hint().mode !=	181,265✔
496	hpx::threads::thread_schedule_hint_mode::none &&	181,039✔
497	policy_.hint().hint != -1)	226✔
498	{
499	return policy_.hint().hint;	226✔
500	}
501	return 0;	180,813✔
502	}	181,039✔
503
504	friend class hpx::serialization::access;
505
506	template <typename Archive>
507	void serialize(Archive& ar, unsigned int const /* version */)	1,100✔
508	{
509	// clang-format off
510	ar & policy_ & hierarchical_threshold_ & num_cores_;	1,100✔
511	// clang-format on
512	}	1,100✔
513	/// \endcond
514
515	private:
516	/// \cond NOINTERNAL
517	static constexpr std::size_t hierarchical_threshold_default_ = 0;
518
519	threads::thread_pool_base* pool_;
520	Policy policy_;
521	std::size_t hierarchical_threshold_ = hierarchical_threshold_default_;	589,884✔
522	std::size_t num_cores_ = 0;	592,135✔
523	#if defined(HPX_HAVE_THREAD_DESCRIPTION)
524	char const* annotation_ = nullptr;
525	#endif
526	/// \endcond
527	};
528
529	// support all properties exposed by the embedded policy
530	// clang-format off
531	template <typename Tag, typename Policy, typename Property,
532	HPX_CONCEPT_REQUIRES_(
533	hpx::execution::experimental::is_scheduling_property_v<Tag>
534	)>
535	// clang-format on
536	auto tag_invoke(	1,391✔
537	Tag tag, parallel_policy_executor<Policy> const& exec, Property&& prop)
538	-> decltype(std::declval<parallel_policy_executor<Policy>>().policy(
539	std::declval<Tag>()(
540	std::declval<Policy>(), std::declval<Property>())),
541	parallel_policy_executor<Policy>())
542	{
543	auto exec_with_prop = exec;	1,391✔
544	exec_with_prop.policy(tag(exec.policy(), HPX_FORWARD(Property, prop)));	1,391✔
545	return exec_with_prop;	1,391✔
546	}
547
548	// clang-format off
549	template <typename Tag,typename Policy,
550	HPX_CONCEPT_REQUIRES_(
551	hpx::execution::experimental::is_scheduling_property_v<Tag>
552	)>
553	// clang-format on
554	auto tag_invoke(Tag tag, parallel_policy_executor<Policy> const& exec)	1,182✔
555	-> decltype(std::declval<Tag>()(std::declval<Policy>()))
556	{
557	return tag(exec.policy());	1,182✔
558	}
559
560	using parallel_executor = parallel_policy_executor<hpx::launch>;
561	} // namespace hpx::execution
562
563	namespace hpx::parallel::execution {
564	/// \cond NOINTERNAL
565	template <typename Policy>
566	struct is_one_way_executor<hpx::execution::parallel_policy_executor<Policy>>
567	: std::true_type
568	{
569	};
570
571	template <typename Policy>
572	struct is_never_blocking_one_way_executor<
573	hpx::execution::parallel_policy_executor<Policy>> : std::true_type
574	{
575	};
576
577	template <typename Policy>
578	struct is_two_way_executor<hpx::execution::parallel_policy_executor<Policy>>
579	: std::true_type
580	{
581	};
582
583	template <typename Policy>
584	struct is_bulk_two_way_executor<
585	hpx::execution::parallel_policy_executor<Policy>> : std::true_type
586	{
587	};
588	/// \endcond
589	} // namespace hpx::parallel::execution

STEllAR-GROUP / hpx / #865

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous