#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/resource_partitioner/examples/guided_pool_test.cpp

//  Copyright (c) 2017-2018 John Biddiscombe
//  Copyright (c) 2024-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)

#include <hpx/config.hpp>

// make available M_PI
#if !defined(_USE_MATH_DEFINES)
#define _USE_MATH_DEFINES
#endif

#include <hpx/algorithm.hpp>
#include <hpx/execution.hpp>
#include <hpx/init.hpp>
#include <hpx/modules/resource_partitioner.hpp>
#include <hpx/modules/schedulers.hpp>
#include <hpx/runtime.hpp>
#include <hpx/thread.hpp>

#include <cmath>
#include <cstddef>
#include <iostream>
#include <memory>
#include <set>
#include <string>
#include <type_traits>
#include <utility>

#include "system_characteristics.hpp"

static int pool_threads = 0;

#define CUSTOM_POOL_NAME "Custom"

// this is our custom scheduler type
using high_priority_sched =
    hpx::threads::policies::shared_priority_queue_scheduler<>;
using hpx::threads::policies::scheduler_mode;

// Force an instantiation of the pool type templated on our custom scheduler
// we need this to ensure that the pool has the generated member functions needed
// by the linker for this pool type
// template class hpx::threads::detail::scheduled_thread_pool<high_priority_sched>;

// dummy function we will call using async
void async_guided(std::size_t n, bool printout, std::string const& message)
{
    if (printout)
    {
        std::cout << "[async_guided] <std::size_t, bool, const std::string> "
                  << message << " n=" << n << "\n";
    }

    double convert = 2.0 * M_PI / static_cast<double>(n);
    for (std::size_t i(0); i < n; ++i)
    {
        double f = std::sin(convert * static_cast<double>(i));
        if (printout)
        {
            std::cout << "sin(" << i << ") = " << f << ", ";
        }
    }
    if (printout)
    {
        std::cout << "\n";
    }
}
/*
template <typename ... Args>
int a_function(Args...args) {
    std::cout << "A_function double is " << std::endl;
    return 2;
}
*/
std::string a_function(hpx::future<double>&& df)
{
    std::cout << "A_function double is " << df.get() << std::endl;
    return "The number 2";
}

namespace hpx::execution::experimental {

    struct guided_test_tag
    {
    };

    // ------------------------------------------------------------------------
    template <>
    struct pool_numa_hint<guided_test_tag>
    {
        // ------------------------------------------------------------------------
        // specialize the hint operator for params
        int operator()(std::size_t i, bool b, std::string const& msg) const
        {
            std::cout << "<std::size_t, bool, const std::string> hint "
                      << "invoked with : " << i << " " << b << " " << msg
                      << std::endl;
            return 1;
        }

        // ------------------------------------------------------------------------
        // specialize the hint operator for params
        int operator()(int i, double d, std::string const& msg) const
        {
            std::cout << "<int, double, const std::string> hint "
                      << "invoked with : " << i << " " << d << " " << msg
                      << std::endl;
            return 42;
        }

        // ------------------------------------------------------------------------
        // specialize the hint operator for params
        int operator()(double x) const
        {
            std::cout << "double hint invoked with " << x << std::endl;
            return 27;
        }

        // ------------------------------------------------------------------------
        // specialize the hint operator for an arbitrary function/args type
        template <typename... Args>
        int operator()(Args...) const
        {
            std::cout << "Variadic hint invoked " << std::endl;
            return 56;
        }
    };
}    // namespace hpx::execution::experimental

using namespace hpx::execution::experimental;

// this is called on an hpx thread after the runtime starts up
int hpx_main()
{
    std::size_t num_threads = hpx::get_num_worker_threads();
    std::cout << "HPX using threads = " << num_threads << std::endl;

    // ------------------------------------------------------------------------
    // test 1
    // ------------------------------------------------------------------------
    std::cout << std::endl << std::endl;
    std::cout << "----------------------------------------------" << std::endl;
    std::cout << "Testing async guided exec " << std::endl;
    std::cout << "----------------------------------------------" << std::endl;
    // we must specialize the numa callback hint for the function type we are invoking
    using hint_type1 = pool_numa_hint<guided_test_tag>;
    // create an executor using that hint type
    hpx::execution::experimental::guided_pool_executor<hint_type1> guided_exec(
        &hpx::resource::get_thread_pool(CUSTOM_POOL_NAME));
    // invoke an async function using our numa hint executor
    hpx::future<void> gf1 = hpx::async(guided_exec, &async_guided,
        std::size_t(5), true, std::string("Guided function"));
    gf1.get();

    // ------------------------------------------------------------------------
    // test 2
    // ------------------------------------------------------------------------
    std::cout << std::endl << std::endl;
    std::cout << "----------------------------------------------" << std::endl;
    std::cout << "Testing async guided exec lambda" << std::endl;
    std::cout << "----------------------------------------------" << std::endl;
    // specialize the numa hint callback for a lambda type invocation
    // the args of the async lambda must match the args of the hint type
    using hint_type2 = pool_numa_hint<guided_test_tag>;
    // create an executor using the numa hint type
    hpx::execution::experimental::guided_pool_executor<hint_type2>
        guided_lambda_exec(&hpx::resource::get_thread_pool(CUSTOM_POOL_NAME));
    // invoke a lambda asynchronously and use the numa executor
    hpx::future<double> gf2 = hpx::async(
        guided_lambda_exec,
        [](int, double, std::string const& msg) mutable -> double {
            std::cout << "inside <int, double, string> async lambda " << msg
                      << std::endl;
            // return a double as an example
            return 3.1415;
        },
        5, 2.718, "Guided function 2");
    gf2.get();

    // ------------------------------------------------------------------------
    // test 3
    // ------------------------------------------------------------------------
    std::cout << std::endl << std::endl;
    std::cout << "----------------------------------------------" << std::endl;
    std::cout << "Testing async guided exec continuation" << std::endl;
    std::cout << "----------------------------------------------" << std::endl;
    // specialize the numa hint callback for another lambda type invocation
    // the args of the async lambda must match the args of the hint type
    using hint_type3 = pool_numa_hint<guided_test_tag>;
    // create an executor using the numa hint type
    hpx::execution::experimental::guided_pool_executor<hint_type3>
        guided_cont_exec(&hpx::resource::get_thread_pool(CUSTOM_POOL_NAME));
    // invoke the lambda asynchronously and use the numa executor
    auto new_future = hpx::async([]() -> double {
        return 2 * 3.1415;
    }).then(guided_cont_exec, a_function);

    new_future.get();

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

void init_resource_partitioner_handler(
    hpx::resource::partitioner& rp, hpx::program_options::variables_map const&)
{
    // create a thread pool and supply a lambda that returns a new pool with
    // a user supplied scheduler attached
    rp.create_thread_pool(CUSTOM_POOL_NAME,
        [](hpx::threads::thread_pool_init_parameters init,
            hpx::threads::policies::thread_queue_init_parameters const&
                thread_queue_init)
            -> std::unique_ptr<hpx::threads::thread_pool_base> {
            std::cout << "User defined scheduler creation callback "
                      << std::endl;
            high_priority_sched::init_parameter_type scheduler_init(
                init.num_threads_, {1, 1, 64}, init.affinity_data_,
                thread_queue_init, "shared-priority-scheduler");

            auto scheduler =
                std::make_unique<high_priority_sched>(scheduler_init);

            init.mode_ = scheduler_mode::delay_exit;

            auto pool =
                std::make_unique<hpx::threads::detail::scheduled_thread_pool<
                    high_priority_sched>>(std::move(scheduler), init);
            return pool;
        });

    // rp.add_resource(rp.numa_domains()[0].cores()[0].pus(), CUSTOM_POOL_NAME);
    // add N cores to Custom pool
    int count = 0;
    for (hpx::resource::numa_domain const& d : rp.numa_domains())
    {
        for (hpx::resource::core const& c : d.cores())
        {
            for (hpx::resource::pu const& p : c.pus())
            {
                if (count < pool_threads)
                {
                    std::cout << "Added pu " << count++
                              << " to " CUSTOM_POOL_NAME " pool\n";
                    rp.add_resource(p, CUSTOM_POOL_NAME);
                }
            }
        }
    }

    std::cout << "[rp_callback] resources added to thread_pools \n";
}

// the normal int main function that is called at startup and runs on an OS
// thread the user must call hpx::local::init to start the hpx runtime which
// will execute hpx_main on an hpx thread
int main(int argc, char* argv[])
{
    hpx::program_options::options_description desc_cmdline("Test options");
    desc_cmdline.add_options()("pool-threads,m",
        hpx::program_options::value<int>()->default_value(1),
        "Number of threads to assign to custom pool");

    // HPX uses a boost program options variable map, but we need it before
    // hpx-main, so we will create another one here and throw it away after use
    hpx::program_options::variables_map vm;
    hpx::program_options::store(
        hpx::program_options::command_line_parser(argc, argv)
            .allow_unregistered()
            .options(desc_cmdline)
            .run(),
        vm);

    pool_threads = vm["pool-threads"].as<int>();

    // Set up the init parameters
    hpx::local::init_params init_args;
    init_args.desc_cmdline = desc_cmdline;

    // Set the callback to init the thread_pools
    init_args.rp_callback = &init_resource_partitioner_handler;

    return hpx::local::init(hpx_main, argc, argv, init_args);
}

1	// Copyright (c) 2017-2018 John Biddiscombe
2	// Copyright (c) 2024-2025 Hartmut Kaiser
3	//
4	// SPDX-License-Identifier: BSL-1.0
5	// Distributed under the Boost Software License, Version 1.0. (See accompanying
6	// file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
7
8	#include <hpx/config.hpp>
9
10	// make available M_PI
11	#if !defined(_USE_MATH_DEFINES)
12	#define _USE_MATH_DEFINES
13	#endif
14
15	#include <hpx/algorithm.hpp>
16	#include <hpx/execution.hpp>
17	#include <hpx/init.hpp>
18	#include <hpx/modules/resource_partitioner.hpp>
19	#include <hpx/modules/schedulers.hpp>
20	#include <hpx/runtime.hpp>
21	#include <hpx/thread.hpp>
22
23	#include <cmath>
24	#include <cstddef>
25	#include <iostream>
26	#include <memory>
27	#include <set>
28	#include <string>
29	#include <type_traits>
30	#include <utility>
31
32	#include "system_characteristics.hpp"
33
34	static int pool_threads = 0;
35
36	#define CUSTOM_POOL_NAME "Custom"
37
38	// this is our custom scheduler type
39	using high_priority_sched =
40	hpx::threads::policies::shared_priority_queue_scheduler<>;
41	using hpx::threads::policies::scheduler_mode;
42
43	// Force an instantiation of the pool type templated on our custom scheduler
44	// we need this to ensure that the pool has the generated member functions needed	×
45	// by the linker for this pool type
46	// template class hpx::threads::detail::scheduled_thread_pool<high_priority_sched>;	×
47
48	// dummy function we will call using async
49	void async_guided(std::size_t n, bool printout, std::string const& message)	×
50	{
51	if (printout)	×
52	{
53	std::cout << "[async_guided] <std::size_t, bool, const std::string> "	×
54	<< message << " n=" << n << "\n";	×
55	}
56		×
57	double convert = 2.0 * M_PI / static_cast<double>(n);
58	for (std::size_t i(0); i < n; ++i)
59	{	×
60	double f = std::sin(convert * static_cast<double>(i));
61	if (printout)	×
62	{
63	std::cout << "sin(" << i << ") = " << f << ", ";	×
64	}
65	}
66	if (printout)
67	{
68	std::cout << "\n";
69	}
70	}
71	/*	×
72	template <typename ... Args>
73	int a_function(Args...args) {	×
74	std::cout << "A_function double is " << std::endl;	×
75	return 2;
76	}
77	*/
78	std::string a_function(hpx::future<double>&& df)
79	{
80	std::cout << "A_function double is " << df.get() << std::endl;
81	return "The number 2";
82	}
83
84	namespace hpx::execution::experimental {
85
86	struct guided_test_tag
87	{
88	};
89		×
90	// ------------------------------------------------------------------------
91	template <>
92	struct pool_numa_hint<guided_test_tag>	×
93	{
94	// ------------------------------------------------------------------------	×
95	// specialize the hint operator for params
96	int operator()(std::size_t i, bool b, std::string const& msg) const
97	{
98	std::cout << "<std::size_t, bool, const std::string> hint "
99	<< "invoked with : " << i << " " << b << " " << msg
100	<< std::endl;
101	return 1;
102	}
103
104	// ------------------------------------------------------------------------
105	// specialize the hint operator for params
106	int operator()(int i, double d, std::string const& msg) const
107	{
108	std::cout << "<int, double, const std::string> hint "
109	<< "invoked with : " << i << " " << d << " " << msg	×
110	<< std::endl;
111	return 42;
112	}	×
113
114	// ------------------------------------------------------------------------
115	// specialize the hint operator for params
116	int operator()(double x) const
117	{
118	std::cout << "double hint invoked with " << x << std::endl;
119	return 27;
120	}
121
122	// ------------------------------------------------------------------------
123	// specialize the hint operator for an arbitrary function/args type
124	template <typename... Args>
125	int operator()(Args...) const
126	{
127	std::cout << "Variadic hint invoked " << std::endl;
128	return 56;
129	}	×
130	};
131	} // namespace hpx::execution::experimental	×
132
133	using namespace hpx::execution::experimental;
134
135	// this is called on an hpx thread after the runtime starts up
136	int hpx_main()
137	{
138	std::size_t num_threads = hpx::get_num_worker_threads();
139	std::cout << "HPX using threads = " << num_threads << std::endl;
140
141	// ------------------------------------------------------------------------
142	// test 1
143	// ------------------------------------------------------------------------
144	std::cout << std::endl << std::endl;
145	std::cout << "----------------------------------------------" << std::endl;	×
146	std::cout << "Testing async guided exec " << std::endl;
147	std::cout << "----------------------------------------------" << std::endl;	×
148	// we must specialize the numa callback hint for the function type we are invoking	×
149	using hint_type1 = pool_numa_hint<guided_test_tag>;	×
150	// create an executor using that hint type
151	hpx::execution::experimental::guided_pool_executor<hint_type1> guided_exec(
152	&hpx::resource::get_thread_pool(CUSTOM_POOL_NAME));
153	// invoke an async function using our numa hint executor
154	hpx::future<void> gf1 = hpx::async(guided_exec, &async_guided,
155	std::size_t(5), true, std::string("Guided function"));
156	gf1.get();
157
158	// ------------------------------------------------------------------------
159	// test 2
160	// ------------------------------------------------------------------------
161	std::cout << std::endl << std::endl;
162	std::cout << "----------------------------------------------" << std::endl;
163	std::cout << "Testing async guided exec lambda" << std::endl;	×
164	std::cout << "----------------------------------------------" << std::endl;
165	// specialize the numa hint callback for a lambda type invocation
166	// the args of the async lambda must match the args of the hint type
167	using hint_type2 = pool_numa_hint<guided_test_tag>;	×
168	// create an executor using the numa hint type
169	hpx::execution::experimental::guided_pool_executor<hint_type2>
170	guided_lambda_exec(&hpx::resource::get_thread_pool(CUSTOM_POOL_NAME));
171	// invoke a lambda asynchronously and use the numa executor	×
172	hpx::future<double> gf2 = hpx::async(
173	guided_lambda_exec,	×
174	[](int, double, std::string const& msg) mutable -> double {	×
175	std::cout << "inside <int, double, string> async lambda " << msg
176	<< std::endl;
177	// return a double as an example
178	return 3.1415;
179	},
180	5, 2.718, "Guided function 2");
181	gf2.get();
182
183	// ------------------------------------------------------------------------
184	// test 3
185	// ------------------------------------------------------------------------
186	std::cout << std::endl << std::endl;
187	std::cout << "----------------------------------------------" << std::endl;
188	std::cout << "Testing async guided exec continuation" << std::endl;	×
189	std::cout << "----------------------------------------------" << std::endl;
190	// specialize the numa hint callback for another lambda type invocation	×
191	// the args of the async lambda must match the args of the hint type
192	using hint_type3 = pool_numa_hint<guided_test_tag>;	×
193	// create an executor using the numa hint type
194	hpx::execution::experimental::guided_pool_executor<hint_type3>	×
195	guided_cont_exec(&hpx::resource::get_thread_pool(CUSTOM_POOL_NAME));
196	// invoke the lambda asynchronously and use the numa executor	×
197	auto new_future = hpx::async([]() -> double {
198	return 2 * 3.1415;
199	}).then(guided_cont_exec, a_function);	×
200
201	new_future.get();
202
203	return hpx::local::finalize();
204	}	×
205		×
206	void init_resource_partitioner_handler(
207	hpx::resource::partitioner& rp, hpx::program_options::variables_map const&)
208	{
209	// create a thread pool and supply a lambda that returns a new pool with
210	// a user supplied scheduler attached
211	rp.create_thread_pool(CUSTOM_POOL_NAME,
212	[](hpx::threads::thread_pool_init_parameters init,
213	hpx::threads::policies::thread_queue_init_parameters const&	×
214	thread_queue_init)
215	-> std::unique_ptr<hpx::threads::thread_pool_base> {	×
216	std::cout << "User defined scheduler creation callback "
217	<< std::endl;	×
218	high_priority_sched::init_parameter_type scheduler_init(
219	init.num_threads_, {1, 1, 64}, init.affinity_data_,
220	thread_queue_init, "shared-priority-scheduler");
221		×
222	auto scheduler =	×
223	std::make_unique<high_priority_sched>(scheduler_init);	×
224
225	init.mode_ = scheduler_mode::delay_exit;
226
227	auto pool =
228	std::make_unique<hpx::threads::detail::scheduled_thread_pool<	×
229	high_priority_sched>>(std::move(scheduler), init);
230	return pool;	×
231	});
232		×
233	// rp.add_resource(rp.numa_domains()[0].cores()[0].pus(), CUSTOM_POOL_NAME);
234	// add N cores to Custom pool	×
235	int count = 0;
236	for (hpx::resource::numa_domain const& d : rp.numa_domains())
237	{	×
238	for (hpx::resource::core const& c : d.cores())	×
239	{
240	for (hpx::resource::pu const& p : c.pus())
241	{
242	if (count < pool_threads)
243	{
244	std::cout << "Added pu " << count++	×
245	<< " to " CUSTOM_POOL_NAME " pool\n";	×
246	rp.add_resource(p, CUSTOM_POOL_NAME);
247	}
248	}
249	}
250	}	×
251
252	std::cout << "[rp_callback] resources added to thread_pools \n";	×
253	}	×
254		×
255	// the normal int main function that is called at startup and runs on an OS
256	// thread the user must call hpx::local::init to start the hpx runtime which
257	// will execute hpx_main on an hpx thread
258	int main(int argc, char* argv[])
259	{	×
260	hpx::program_options::options_description desc_cmdline("Test options");	×
261	desc_cmdline.add_options()("pool-threads,m",	×
262	hpx::program_options::value<int>()->default_value(1),
263	"Number of threads to assign to custom pool");
264		×
265	// HPX uses a boost program options variable map, but we need it before
266	// hpx-main, so we will create another one here and throw it away after use
267	hpx::program_options::variables_map vm;	×
268	hpx::program_options::store(
269	hpx::program_options::command_line_parser(argc, argv)
270	.allow_unregistered()	×
271	.options(desc_cmdline)	×
272	.run(),
273	vm);
274		×
275	pool_threads = vm["pool-threads"].as<int>();
276		×
277	// Set up the init parameters	×
278	hpx::local::init_params init_args;
279	init_args.desc_cmdline = desc_cmdline;
280
281	// Set the callback to init the thread_pools
282	init_args.rp_callback = &init_resource_partitioner_handler;
283
284	return hpx::local::init(hpx_main, argc, argv, init_args);
285	}

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