#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

0.0

/libs/core/executors/examples/executor_with_thread_hooks.cpp

//  Copyright (c) 2020-2024 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)

///////////////////////////////////////////////////////////////////////////////
// The purpose of this example is to show how to rite an executor that wraps
// any other executor and adds a hook into thread start and thread exit allowing
// to associate custom thread data with the tasks that are created by the
// underlying executor.

#include <hpx/algorithm.hpp>
#include <hpx/assert.hpp>
#include <hpx/execution.hpp>
#include <hpx/experimental/scope.hpp>
#include <hpx/init.hpp>

#include <algorithm>
#include <atomic>
#include <cstddef>
#include <iostream>
#include <type_traits>
#include <utility>
#include <vector>

namespace executor_example {

    template <typename BaseExecutor>
    class executor_with_thread_hooks
    {
    private:
        template <typename F>
        struct hook_wrapper
        {
            template <typename... Ts>
            decltype(auto) operator()(Ts&&... ts)
            {
                exec_.on_start_();
                auto on_exit = hpx::experimental::scope_exit(
                    [this]() mutable { exec_.on_stop_(); });

                return hpx::invoke(f_, std::forward<Ts>(ts)...);
            }

            executor_with_thread_hooks const& exec_;
            F f_;
        };

    public:
        using execution_category =
            hpx::traits::executor_execution_category_t<BaseExecutor>;
        using executor_parameters_type =
            hpx::traits::executor_parameters_type_t<BaseExecutor>;

        template <typename Executor, typename OnStart, typename OnStop>
        executor_with_thread_hooks(
            Executor&& exec, OnStart&& start, OnStop&& stop)
          : exec_(std::forward<Executor>(exec))
          , on_start_(std::forward<OnStart>(start))
          , on_stop_(std::forward<OnStop>(stop))
        {
        }

        bool operator==(executor_with_thread_hooks const& rhs) const noexcept
        {
            return exec_ == rhs.exec_;
        }

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

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

    private:
        // OneWayExecutor interface
        template <typename F, typename... Ts>
        friend decltype(auto) tag_invoke(
            hpx::parallel::execution::sync_execute_t,
            executor_with_thread_hooks const& exec, F&& f, Ts&&... ts)
        {
            return hpx::parallel::execution::sync_execute(exec.exec_,
                hook_wrapper<F>{exec, std::forward<F>(f)},
                std::forward<Ts>(ts)...);
        }

        // TwoWayExecutor interface
        template <typename F, typename... Ts>
        friend decltype(auto) tag_invoke(
            hpx::parallel::execution::async_execute_t,
            executor_with_thread_hooks const& exec, F&& f, Ts&&... ts)
        {
            return hpx::parallel::execution::async_execute(exec.exec_,
                hook_wrapper<F>{exec, std::forward<F>(f)},
                std::forward<Ts>(ts)...);
        }

        template <typename F, typename Future, typename... Ts>
        friend decltype(auto) tag_invoke(
            hpx::parallel::execution::then_execute_t,
            executor_with_thread_hooks const& exec, F&& f, Future&& predecessor,
            Ts&&... ts)
        {
            return hpx::parallel::execution::then_execute(exec.exec_,
                hook_wrapper<F>{exec, std::forward<F>(f)},
                std::forward<Future>(predecessor), std::forward<Ts>(ts)...);
        }

        // NonBlockingOneWayExecutor (adapted) interface
        template <typename F, typename... Ts>
        friend decltype(auto) tag_invoke(hpx::parallel::execution::post_t,
            executor_with_thread_hooks const& exec, F&& f, Ts&&... ts)
        {
            hpx::parallel::execution::post(exec.exec_,
                hook_wrapper<F>{exec, std::forward<F>(f)},
                std::forward<Ts>(ts)...);
        }

        // BulkOneWayExecutor interface
        template <typename F, typename S, typename... Ts>
        friend decltype(auto) tag_invoke(
            hpx::parallel::execution::bulk_sync_execute_t,
            executor_with_thread_hooks const& exec, F&& f, S const& shape,
            Ts&&... ts)
        {
            return hpx::parallel::execution::bulk_sync_execute(exec.exec_,
                hook_wrapper<F>{exec, std::forward<F>(f)}, shape,
                std::forward<Ts>(ts)...);
        }

        // BulkTwoWayExecutor interface
        template <typename F, typename S, typename... Ts>
        friend decltype(auto) tag_invoke(
            hpx::parallel::execution::bulk_async_execute_t,
            executor_with_thread_hooks const& exec, F&& f, S const& shape,
            Ts&&... ts)
        {
            return hpx::parallel::execution::bulk_async_execute(exec.exec_,
                hook_wrapper<F>{exec, std::forward<F>(f)}, shape,
                std::forward<Ts>(ts)...);
        }

        template <typename F, typename S, typename Future, typename... Ts>
        friend decltype(auto) tag_invoke(
            hpx::parallel::execution::bulk_then_execute_t,
            executor_with_thread_hooks const& exec, F&& f, S const& shape,
            Future&& predecessor, Ts&&... ts)
        {
            return hpx::parallel::execution::bulk_then_execute(exec.exec_,
                hook_wrapper<F>{exec, std::forward<F>(f)}, shape,
                std::forward<Future>(predecessor), std::forward<Ts>(ts)...);
        }

        [[nodiscard]] constexpr std::decay_t<BaseExecutor> const& get_executor()
            const noexcept
        {
            return exec_;
        }

    private:
        using thread_hook = hpx::function<void()>;

        BaseExecutor exec_;
        thread_hook on_start_;
        thread_hook on_stop_;
    };

    // clang-format off
    template <typename Executor, typename OnStart, typename OnStop>
    executor_with_thread_hooks(Executor&&, OnStart&&, OnStop&&)
        -> executor_with_thread_hooks<std::decay_t<Executor>>;
    // clang-format on
}    // namespace executor_example

///////////////////////////////////////////////////////////////////////////////
// simple forwarding implementations of executor traits
namespace hpx::execution::experimental {

    template <typename BaseExecutor>
    struct is_one_way_executor<
        executor_example::executor_with_thread_hooks<BaseExecutor>>
      : is_one_way_executor<std::decay_t<BaseExecutor>>
    {
    };

    template <typename BaseExecutor>
    struct is_never_blocking_one_way_executor<
        executor_example::executor_with_thread_hooks<BaseExecutor>>
      : is_never_blocking_one_way_executor<std::decay_t<BaseExecutor>>
    {
    };

    template <typename BaseExecutor>
    struct is_two_way_executor<
        executor_example::executor_with_thread_hooks<BaseExecutor>>
      : is_two_way_executor<std::decay_t<BaseExecutor>>
    {
    };

    template <typename BaseExecutor>
    struct is_bulk_one_way_executor<
        executor_example::executor_with_thread_hooks<BaseExecutor>>
      : is_bulk_one_way_executor<std::decay_t<BaseExecutor>>
    {
    };

    template <typename BaseExecutor>
    struct is_bulk_two_way_executor<
        executor_example::executor_with_thread_hooks<BaseExecutor>>
      : is_bulk_two_way_executor<std::decay_t<BaseExecutor>>
    {
    };
}    // namespace hpx::execution::experimental

int hpx_main()
{
    std::vector<double> v(1000);
    std::iota(v.begin(), v.end(), 0.0);

    std::atomic<std::size_t> starts(0);
    std::atomic<std::size_t> stops(0);

    auto on_start = [&]() { ++starts; };
    auto on_stop = [&]() { ++stops; };

    auto exec = executor_example::executor_with_thread_hooks(
        hpx::execution::par.executor(), on_start, on_stop);

    hpx::experimental::for_loop(
        hpx::execution::par.on(exec), 0, v.size(), [](std::size_t) {});

    std::cout << "Executed " << starts.load() << " starts and " << stops.load()
              << " stops\n";

    HPX_ASSERT(starts.load() != 0);
    HPX_ASSERT(stops.load() != 0);

    return hpx::local::finalize();
}

int main(int argc, char* argv[])
{
    return hpx::local::init(hpx_main, argc, argv);
}

1	// Copyright (c) 2020-2024 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	///////////////////////////////////////////////////////////////////////////////
8	// The purpose of this example is to show how to rite an executor that wraps
9	// any other executor and adds a hook into thread start and thread exit allowing
10	// to associate custom thread data with the tasks that are created by the
11	// underlying executor.
12
13	#include <hpx/algorithm.hpp>
14	#include <hpx/assert.hpp>
15	#include <hpx/execution.hpp>
16	#include <hpx/experimental/scope.hpp>
17	#include <hpx/init.hpp>
18
19	#include <algorithm>
20	#include <atomic>
21	#include <cstddef>
22	#include <iostream>
23	#include <type_traits>
24	#include <utility>
25	#include <vector>
26
27	namespace executor_example {
28
29	template <typename BaseExecutor>	×
30	class executor_with_thread_hooks
31	{
32	private:
33	template <typename F>
34	struct hook_wrapper
35	{
36	template <typename... Ts>
37	decltype(auto) operator()(Ts&&... ts)
38	{
39	exec_.on_start_();
40	auto on_exit = hpx::experimental::scope_exit(
41	[this]() mutable { exec_.on_stop_(); });
42
43	return hpx::invoke(f_, std::forward<Ts>(ts)...);
44	}
45
46	executor_with_thread_hooks const& exec_;
47	F f_;
48	};
49
50	public:
51	using execution_category =
52	hpx::traits::executor_execution_category_t<BaseExecutor>;	×
53	using executor_parameters_type =
54	hpx::traits::executor_parameters_type_t<BaseExecutor>;	×
55		×
56	template <typename Executor, typename OnStart, typename OnStop>
57	executor_with_thread_hooks(
58	Executor&& exec, OnStart&& start, OnStop&& stop)
59	: exec_(std::forward<Executor>(exec))
60	, on_start_(std::forward<OnStart>(start))
61	, on_stop_(std::forward<OnStop>(stop))
62	{
63	}
64
65	bool operator==(executor_with_thread_hooks const& rhs) const noexcept
66	{
67	return exec_ == rhs.exec_;
68	}
69		×
70	bool operator!=(executor_with_thread_hooks const& rhs) const noexcept
71	{	×
72	return !(*this == rhs);
73	}
74
75	executor_with_thread_hooks const& context() const noexcept	×
76	{
77	return *this;
78	}
79
80	private:
81	// OneWayExecutor interface
82	template <typename F, typename... Ts>
83	friend decltype(auto) tag_invoke(
84	hpx::parallel::execution::sync_execute_t,
85	executor_with_thread_hooks const& exec, F&& f, Ts&&... ts)
86	{
87	return hpx::parallel::execution::sync_execute(exec.exec_,
88	hook_wrapper<F>{exec, std::forward<F>(f)},
89	std::forward<Ts>(ts)...);
90	}
91
92	// TwoWayExecutor interface
93	template <typename F, typename... Ts>
94	friend decltype(auto) tag_invoke(
95	hpx::parallel::execution::async_execute_t,
96	executor_with_thread_hooks const& exec, F&& f, Ts&&... ts)
97	{
98	return hpx::parallel::execution::async_execute(exec.exec_,
99	hook_wrapper<F>{exec, std::forward<F>(f)},
100	std::forward<Ts>(ts)...);
101	}
102
103	template <typename F, typename Future, typename... Ts>
104	friend decltype(auto) tag_invoke(
105	hpx::parallel::execution::then_execute_t,
106	executor_with_thread_hooks const& exec, F&& f, Future&& predecessor,
107	Ts&&... ts)
108	{
109	return hpx::parallel::execution::then_execute(exec.exec_,
110	hook_wrapper<F>{exec, std::forward<F>(f)},
111	std::forward<Future>(predecessor), std::forward<Ts>(ts)...);
112	}
113
114	// NonBlockingOneWayExecutor (adapted) interface
115	template <typename F, typename... Ts>
116	friend decltype(auto) tag_invoke(hpx::parallel::execution::post_t,
117	executor_with_thread_hooks const& exec, F&& f, Ts&&... ts)
118	{
119	hpx::parallel::execution::post(exec.exec_,
120	hook_wrapper<F>{exec, std::forward<F>(f)},
121	std::forward<Ts>(ts)...);
122	}
123
124	// BulkOneWayExecutor interface
125	template <typename F, typename S, typename... Ts>
126	friend decltype(auto) tag_invoke(
127	hpx::parallel::execution::bulk_sync_execute_t,
128	executor_with_thread_hooks const& exec, F&& f, S const& shape,
129	Ts&&... ts)
130	{
131	return hpx::parallel::execution::bulk_sync_execute(exec.exec_,
132	hook_wrapper<F>{exec, std::forward<F>(f)}, shape,
133	std::forward<Ts>(ts)...);
134	}
135
136	// BulkTwoWayExecutor interface
137	template <typename F, typename S, typename... Ts>
138	friend decltype(auto) tag_invoke(
139	hpx::parallel::execution::bulk_async_execute_t,
140	executor_with_thread_hooks const& exec, F&& f, S const& shape,
141	Ts&&... ts)
142	{
143	return hpx::parallel::execution::bulk_async_execute(exec.exec_,
144	hook_wrapper<F>{exec, std::forward<F>(f)}, shape,
145	std::forward<Ts>(ts)...);
146	}
147
148	template <typename F, typename S, typename Future, typename... Ts>
149	friend decltype(auto) tag_invoke(
150	hpx::parallel::execution::bulk_then_execute_t,
151	executor_with_thread_hooks const& exec, F&& f, S const& shape,
152	Future&& predecessor, Ts&&... ts)
153	{
154	return hpx::parallel::execution::bulk_then_execute(exec.exec_,
155	hook_wrapper<F>{exec, std::forward<F>(f)}, shape,	×
156	std::forward<Future>(predecessor), std::forward<Ts>(ts)...);
157	}	×
158
159	[[nodiscard]] constexpr std::decay_t<BaseExecutor> const& get_executor()
160	const noexcept
161	{
162	return exec_;
163	}
164
165	private:
166	using thread_hook = hpx::function<void()>;
167
168	BaseExecutor exec_;
169	thread_hook on_start_;
170	thread_hook on_stop_;
171	};
172
173	// clang-format off
174	template <typename Executor, typename OnStart, typename OnStop>
175	executor_with_thread_hooks(Executor&&, OnStart&&, OnStop&&)
176	-> executor_with_thread_hooks<std::decay_t<Executor>>;
177	// clang-format on
178	} // namespace executor_example
179
180	///////////////////////////////////////////////////////////////////////////////
181	// simple forwarding implementations of executor traits
182	namespace hpx::execution::experimental {
183
184	template <typename BaseExecutor>
185	struct is_one_way_executor<
186	executor_example::executor_with_thread_hooks<BaseExecutor>>
187	: is_one_way_executor<std::decay_t<BaseExecutor>>
188	{
189	};
190
191	template <typename BaseExecutor>
192	struct is_never_blocking_one_way_executor<
193	executor_example::executor_with_thread_hooks<BaseExecutor>>
194	: is_never_blocking_one_way_executor<std::decay_t<BaseExecutor>>
195	{
196	};
197
198	template <typename BaseExecutor>
199	struct is_two_way_executor<
200	executor_example::executor_with_thread_hooks<BaseExecutor>>
201	: is_two_way_executor<std::decay_t<BaseExecutor>>
202	{
203	};
204
205	template <typename BaseExecutor>
206	struct is_bulk_one_way_executor<
207	executor_example::executor_with_thread_hooks<BaseExecutor>>
208	: is_bulk_one_way_executor<std::decay_t<BaseExecutor>>
209	{
210	};
211
212	template <typename BaseExecutor>
213	struct is_bulk_two_way_executor<
214	executor_example::executor_with_thread_hooks<BaseExecutor>>
215	: is_bulk_two_way_executor<std::decay_t<BaseExecutor>>
216	{
217	};
218	} // namespace hpx::execution::experimental
219
220	int hpx_main()
221	{
222	std::vector<double> v(1000);
223	std::iota(v.begin(), v.end(), 0.0);
224
225	std::atomic<std::size_t> starts(0);
226	std::atomic<std::size_t> stops(0);
227
228	auto on_start = [&]() { ++starts; };
229	auto on_stop = [&]() { ++stops; };
230		×
231	auto exec = executor_example::executor_with_thread_hooks(
232	hpx::execution::par.executor(), on_start, on_stop);	×
233
234	hpx::experimental::for_loop(
235	hpx::execution::par.on(exec), 0, v.size(), [](std::size_t) {});	×
236		×
237	std::cout << "Executed " << starts.load() << " starts and " << stops.load()
238	<< " stops\n";	×
239		×
240	HPX_ASSERT(starts.load() != 0);
241	HPX_ASSERT(stops.load() != 0);
242		×
243	return hpx::local::finalize();
244	}
245		×
246	int main(int argc, char* argv[])
247	{
248	return hpx::local::init(hpx_main, argc, argv);	×
249	}

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