#882

Committed 31 Aug 2023 07:44PM UTC coverage: 41.798% (-44.7%) from 86.546%

Build # #882

Build Type

push

Committed by

Commit Message

Run Details

19442 of 46514 relevant lines covered (41.8%)

126375.38 hits per line

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

90.0

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

//  Copyright (c) 2021-2025 Hartmut Kaiser
//
//  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)

/// \file task_group.hpp
/// \page hpx::experimental::task_group
/// \headerfile hpx/experimental/task_group.hpp

#pragma once

#include <hpx/config.hpp>
#include <hpx/modules/concepts.hpp>
#include <hpx/modules/datastructures.hpp>
#include <hpx/modules/errors.hpp>
#include <hpx/modules/execution_base.hpp>
#include <hpx/modules/executors.hpp>
#include <hpx/modules/functional.hpp>
#include <hpx/modules/futures.hpp>
#include <hpx/modules/memory.hpp>
#include <hpx/modules/serialization.hpp>
#include <hpx/modules/synchronization.hpp>

#include <atomic>
#include <exception>
#include <type_traits>
#include <utility>

/// Top-level namespace
namespace hpx::experimental {

    /// A \c task_group represents concurrent execution of a group of tasks.
    /// Tasks can be dynamically added to the group while it is executing.
    HPX_CXX_EXPORT class task_group
    {
    public:
        HPX_CORE_EXPORT task_group();
        HPX_CORE_EXPORT ~task_group();

        task_group(task_group const&) = delete;
        task_group(task_group&&) = delete;

        task_group& operator=(task_group const&) = delete;
        task_group& operator=(task_group&&) = delete;

    public:
        /// \brief Adds a task to compute \c f() and returns immediately.
        ///
        /// \tparam Executor  The type of the executor to associate with this
        ///                   execution policy.
        /// \tparam F         The type of the user defined function to invoke.
        /// \tparam Ts        The type of additional arguments used to invoke \c f().
        ///
        /// \param exec       The executor to use for the execution of the
        ///                   parallel algorithm the returned execution
        ///                   policy is used with.
        /// \param f          The user defined function to invoke inside the task
        ///                   group.
        /// \param ts         Additional arguments to use to invoke \c f().

        template <typename Executor, typename F, typename... Ts>
        // clang-format off
            requires (
                hpx::traits::is_executor_any_v<std::decay_t<Executor>>
            )
        // clang-format on
        void run(Executor&& exec, F&& f, Ts&&... ts)
        {
            // make sure exceptions don't leave the latch in the wrong state
            if (latch_.reset_if_needed_and_count_up(1, 1))
            {
                has_arrived_.store(false, std::memory_order_release);
            }

            auto on_exit =
                hpx::experimental::scope_exit([this] { latch_.count_down(1); });

            hpx::parallel::execution::post(HPX_FORWARD(Executor, exec),
                [this, on_exit = HPX_MOVE(on_exit), f = HPX_FORWARD(F, f),
                    t = hpx::make_tuple(HPX_FORWARD(Ts, ts)...)]() mutable {
                    // latch needs to be released before the lambda exits
                    auto _(HPX_MOVE(on_exit));

                    hpx::detail::try_catch_exception_ptr(
                        [&]() { hpx::invoke_fused(HPX_MOVE(f), HPX_MOVE(t)); },
                        [this](std::exception_ptr e) {
                            add_exception(HPX_MOVE(e));
                        });
                });
        }

        /// \brief Adds a task to compute \c f() and returns immediately.
        ///
        /// \tparam F  The type of the user defined function to invoke.
        /// \tparam Ts The type of additional arguments used to invoke \c f().
        ///
        /// \param f   The user defined function to invoke inside the task
        ///            group.
        /// \param ts  Additional arguments to use to invoke \c f().

        template <typename F, typename... Ts>
        // clang-format off
            requires (
                !hpx::traits::is_executor_any_v<std::decay_t<F>>
            )
        // clang-format on
        void run(F&& f, Ts&&... ts)
        {
            run(execution::parallel_executor{}, HPX_FORWARD(F, f),
                HPX_FORWARD(Ts, ts)...);
        }

        /// \brief Waits for all tasks in the group to complete or be cancelled.
        HPX_CORE_EXPORT void wait();

        /// \brief Adds an exception to this \c task_group
        HPX_CORE_EXPORT void add_exception(std::exception_ptr p);

    private:
        friend class serialization::access;

        static constexpr void serialize(
            serialization::input_archive&, unsigned const) noexcept
        {
        }
        HPX_CORE_EXPORT void serialize(
            serialization::output_archive&, unsigned const);

    private:
        using shared_state_type = lcos::detail::future_data<void>;

        hpx::lcos::local::latch latch_;
        hpx::intrusive_ptr<shared_state_type> state_;
        hpx::exception_list errors_;
        std::atomic<bool> has_arrived_;
    };
}    // namespace hpx::experimental

namespace hpx::execution::experimental {

    using task_group HPX_DEPRECATED_V(1, 9,
        "hpx::execution:experimental::task_group is deprecated, use "
        "hpx::experimental::task_group instead") =
        hpx::experimental::task_group;
}    // namespace hpx::execution::experimental

1	// Copyright (c) 2021-2025 Hartmut Kaiser
2	//
3	// SPDX-License-Identifier: BSL-1.0
4	// Distributed under the Boost Software License, Version 1.0. (See accompanying
5	// file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
6
7	/// \file task_group.hpp
8	/// \page hpx::experimental::task_group
9	/// \headerfile hpx/experimental/task_group.hpp
10
11	#pragma once
12
13	#include <hpx/config.hpp>
14	#include <hpx/modules/concepts.hpp>
15	#include <hpx/modules/datastructures.hpp>
16	#include <hpx/modules/errors.hpp>
17	#include <hpx/modules/execution_base.hpp>
18	#include <hpx/modules/executors.hpp>
19	#include <hpx/modules/functional.hpp>
20	#include <hpx/modules/futures.hpp>
21	#include <hpx/modules/memory.hpp>
22	#include <hpx/modules/serialization.hpp>
23	#include <hpx/modules/synchronization.hpp>
24
25	#include <atomic>
26	#include <exception>
27	#include <type_traits>
28	#include <utility>
29
30	/// Top-level namespace
31	namespace hpx::experimental {
32
33	/// A \c task_group represents concurrent execution of a group of tasks.
34	/// Tasks can be dynamically added to the group while it is executing.
35	HPX_CXX_EXPORT class task_group
36	{
37	public:
38	HPX_CORE_EXPORT task_group();
39	HPX_CORE_EXPORT ~task_group();
40
41	task_group(task_group const&) = delete;
42	task_group(task_group&&) = delete;
43
44	task_group& operator=(task_group const&) = delete;
45	task_group& operator=(task_group&&) = delete;
46
47	public:
48	/// \brief Adds a task to compute \c f() and returns immediately.
49	///
50	/// \tparam Executor The type of the executor to associate with this
51	/// execution policy.
52	/// \tparam F The type of the user defined function to invoke.
53	/// \tparam Ts The type of additional arguments used to invoke \c f().
54	///
55	/// \param exec The executor to use for the execution of the
56	/// parallel algorithm the returned execution
57	/// policy is used with.
58	/// \param f The user defined function to invoke inside the task
59	/// group.
60	/// \param ts Additional arguments to use to invoke \c f().
61
62	template <typename Executor, typename F, typename... Ts>
63	// clang-format off
64	requires (
65	hpx::traits::is_executor_any_v<std::decay_t<Executor>>
66	)
67	// clang-format on
68	void run(Executor&& exec, F&& f, Ts&&... ts)
69	{
70	// make sure exceptions don't leave the latch in the wrong state
71	if (latch_.reset_if_needed_and_count_up(1, 1))
72	{
73	has_arrived_.store(false, std::memory_order_release);
74	}
75
76	auto on_exit =
77	hpx::experimental::scope_exit([this] { latch_.count_down(1); });
78
79	hpx::parallel::execution::post(HPX_FORWARD(Executor, exec),
80	[this, on_exit = HPX_MOVE(on_exit), f = HPX_FORWARD(F, f),
81	t = hpx::make_tuple(HPX_FORWARD(Ts, ts)...)]() mutable {
82	// latch needs to be released before the lambda exits	2✔
83	auto _(HPX_MOVE(on_exit));
84
85	hpx::detail::try_catch_exception_ptr(	2✔
86	[&]() { hpx::invoke_fused(HPX_MOVE(f), HPX_MOVE(t)); },
87	[this](std::exception_ptr e) {
88	add_exception(HPX_MOVE(e));	16✔
89	});
90	});
91	}	2✔
92
93	/// \brief Adds a task to compute \c f() and returns immediately.
94	///	2✔
95	/// \tparam F The type of the user defined function to invoke.	2✔
96	/// \tparam Ts The type of additional arguments used to invoke \c f().	×
97	///
98	/// \param f The user defined function to invoke inside the task	2✔
99	/// group.	2✔
100	/// \param ts Additional arguments to use to invoke \c f().
101
102	template <typename F, typename... Ts>
103	// clang-format off
104	requires (
105	!hpx::traits::is_executor_any_v<std::decay_t<F>>
106	)
107	// clang-format on
108	void run(F&& f, Ts&&... ts)
109	{
110	run(execution::parallel_executor{}, HPX_FORWARD(F, f),
111	HPX_FORWARD(Ts, ts)...);
112	}
113
114	/// \brief Waits for all tasks in the group to complete or be cancelled.
115	HPX_CORE_EXPORT void wait();
116
117	/// \brief Adds an exception to this \c task_group	1✔
118	HPX_CORE_EXPORT void add_exception(std::exception_ptr p);
119
120	private:
121	friend class serialization::access;
122
123	static constexpr void serialize(
124	serialization::input_archive&, unsigned const) noexcept
125	{
126	}
127	HPX_CORE_EXPORT void serialize(
128	serialization::output_archive&, unsigned const);
129
130	private:
131	using shared_state_type = lcos::detail::future_data<void>;
132
133	hpx::lcos::local::latch latch_;
134	hpx::intrusive_ptr<shared_state_type> state_;
135	hpx::exception_list errors_;
136	std::atomic<bool> has_arrived_;
137	};
138	} // namespace hpx::experimental
139
140	namespace hpx::execution::experimental {
141
142	using task_group HPX_DEPRECATED_V(1, 9,
143	"hpx::execution:experimental::task_group is deprecated, use "
144	"hpx::experimental::task_group instead") =
145	hpx::experimental::task_group;
146	} // namespace hpx::execution::experimental

STEllAR-GROUP / hpx / #882

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