#845

Committed 05 Dec 2022 08:25PM UTC coverage: 85.664% (-0.1%) from 85.8%

Build # #845

Build Type

push

Committed by StellarBot

Commit Message

Merge #6091

6091: Replace artificial sequencing with fold expressions r=hkaiser a=hkaiser

- flyby: fix left-over problems from namespace change

working towards resolving #5497

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

Run Details

99 of 99 new or added lines in 15 files covered. (100.0%)

171125 of 199764 relevant lines covered (85.66%)

1865254.07 hits per line

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

14.94

/libs/full/parcelset/tests/unit/put_parcels.cpp

//  Copyright (c) 2016-2021 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>
#if !defined(HPX_COMPUTE_DEVICE_CODE)
#include <hpx/hpx.hpp>
#include <hpx/hpx_init.hpp>
#include <hpx/modules/testing.hpp>
#include <hpx/parallel/task_group.hpp>

#include <cstddef>
#include <iostream>
#include <memory>
#include <string>
#include <type_traits>
#include <utility>
#include <vector>

///////////////////////////////////////////////////////////////////////////////
std::size_t const vsize_default = 1024;
std::size_t const numparcels_default = 10;

///////////////////////////////////////////////////////////////////////////////
template <typename Action, typename T>
hpx::parcelset::parcel generate_parcel(
    hpx::id_type const& dest_id, hpx::id_type const& cont, T&& data)
{
    hpx::naming::address addr;
    hpx::naming::gid_type dest = dest_id.get_gid();
    hpx::parcelset::parcel p(hpx::parcelset::detail::create_parcel::call(
        std::move(dest), std::move(addr),
        hpx::actions::typed_continuation<hpx::id_type>(cont), Action(),
        hpx::threads::thread_priority::normal, std::forward<T>(data)));

    p.set_source_id(hpx::find_here());
    p.size() = 4096;
    return p;
}

///////////////////////////////////////////////////////////////////////////////
hpx::id_type test1(std::vector<double> const&)
{
    return hpx::find_here();
}
HPX_PLAIN_ACTION(test1)

void test_plain_argument(hpx::id_type const& id)
{
    std::vector<double> data(vsize_default);
    std::generate(data.begin(), data.end(), std::rand);

    std::vector<hpx::future<hpx::id_type>> results;
    results.reserve(numparcels_default);

    // create parcels
    std::vector<hpx::parcelset::parcel> parcels;
    for (std::size_t i = 0; i != numparcels_default; ++i)
    {
        hpx::distributed::promise<hpx::id_type> p;
        auto f = p.get_future();
        parcels.push_back(generate_parcel<test1_action>(id, p.get_id(), data));
        results.push_back(std::move(f));
    }

    // send parcels
    hpx::get_runtime_distributed().get_parcel_handler().put_parcels(
        std::move(parcels));

    // verify all messages got actually sent to the correct locality
    hpx::wait_all(results);

    for (hpx::future<hpx::id_type>& f : results)
    {
        HPX_TEST_EQ(f.get(), id);
    }
}

///////////////////////////////////////////////////////////////////////////////
hpx::id_type test2(hpx::future<double> const&)
{
    return hpx::find_here();
}
HPX_PLAIN_ACTION(test2)

void test_future_argument(hpx::id_type const& id)
{
    std::vector<hpx::promise<double>> args;
    args.reserve(numparcels_default);

    std::vector<hpx::future<hpx::id_type>> results;
    results.reserve(numparcels_default);

    // create parcels
    std::vector<hpx::parcelset::parcel> parcels;
    for (std::size_t i = 0; i != numparcels_default; ++i)
    {
        hpx::promise<double> p_arg;
        hpx::distributed::promise<hpx::id_type> p_cont;
        auto f_cont = p_cont.get_future();

        parcels.push_back(generate_parcel<test2_action>(
            id, p_cont.get_id(), p_arg.get_future()));

        args.push_back(std::move(p_arg));
        results.push_back(std::move(f_cont));
    }

    // send parcels
    hpx::get_runtime_distributed().get_parcel_handler().put_parcels(
        std::move(parcels));

    // now make the futures ready
    for (hpx::promise<double>& arg : args)
    {
        arg.set_value(42.0);
    }

    // verify all messages got actually sent to the correct locality
    hpx::wait_all(results);

    for (hpx::future<hpx::id_type>& f : results)
    {
        HPX_TEST_EQ(f.get(), id);
    }
}

void test_mixed_arguments(hpx::id_type const& id)
{
    std::vector<double> data(vsize_default);
    std::generate(data.begin(), data.end(), std::rand);

    std::vector<hpx::promise<double>> args;
    args.reserve(numparcels_default);

    std::vector<hpx::future<hpx::id_type>> results;
    results.reserve(numparcels_default);

    // create parcels
    std::vector<hpx::parcelset::parcel> parcels;
    for (std::size_t i = 0; i != numparcels_default; ++i)
    {
        hpx::distributed::promise<hpx::id_type> p_cont;
        auto f_cont = p_cont.get_future();

        if (std::rand() % 2)
        {
            parcels.push_back(
                generate_parcel<test1_action>(id, p_cont.get_id(), data));
        }
        else
        {
            hpx::promise<double> p_arg;

            parcels.push_back(generate_parcel<test2_action>(
                id, p_cont.get_id(), p_arg.get_future()));

            args.push_back(std::move(p_arg));
        }

        results.push_back(std::move(f_cont));
    }

    // send parcels
    hpx::get_runtime_distributed().get_parcel_handler().put_parcels(
        std::move(parcels));

    // now make the futures ready
    for (hpx::promise<double>& arg : args)
    {
        arg.set_value(42.0);
    }

    // verify all messages got actually sent to the correct locality
    hpx::wait_all(results);

    for (hpx::future<hpx::id_type>& f : results)
    {
        HPX_TEST_EQ(f.get(), id);
    }
}

///////////////////////////////////////////////////////////////////////////////
hpx::id_type test3(std::shared_ptr<hpx::execution::experimental::task_group> tg)
{
    tg->wait();
    return hpx::find_here();
}
HPX_PLAIN_ACTION(test3)

void wait_a_while()
{
    hpx::this_thread::sleep_for(std::chrono::milliseconds(100));
}

void test_task_group_argument(hpx::id_type const& id)
{
    std::vector<std::shared_ptr<hpx::execution::experimental::task_group>> args;
    args.reserve(numparcels_default);

    std::vector<hpx::future<hpx::id_type>> results;
    results.reserve(numparcels_default);

    // create parcels
    std::vector<hpx::parcelset::parcel> parcels;
    for (std::size_t i = 0; i != numparcels_default; ++i)
    {
        auto tg = std::make_shared<hpx::execution::experimental::task_group>();
        tg->run(wait_a_while);
        tg->run(wait_a_while);
        tg->run(wait_a_while);

        hpx::distributed::promise<hpx::id_type> p_cont;
        auto f_cont = p_cont.get_future();

        parcels.push_back(
            generate_parcel<test3_action>(id, p_cont.get_id(), tg));

        args.push_back(std::move(tg));
        results.push_back(std::move(f_cont));
    }

    // send parcels
    hpx::get_runtime_distributed().get_parcel_handler().put_parcels(
        std::move(parcels));

    // now wait for task groups
    for (auto& arg : args)
    {
        arg->wait();
    }

    // verify all messages got actually sent to the correct locality
    hpx::wait_all(results);

    for (hpx::future<hpx::id_type>& f : results)
    {
        HPX_TEST_EQ(f.get(), id);
    }
}

///////////////////////////////////////////////////////////////////////////////
void print_counters(char const* name)
{
    using namespace hpx::performance_counters;

    std::vector<performance_counter> counters = discover_counters(name);

    for (performance_counter const& c : counters)
    {
        counter_value value = c.get_counter_value(hpx::launch::sync);
        std::cout << "counter: " << c.get_name(hpx::launch::sync)
                  << ", value: " << value.get_value<double>() << std::endl;
    }
}

///////////////////////////////////////////////////////////////////////////////
int hpx_main(hpx::program_options::variables_map& vm)
{
    unsigned int seed = (unsigned int) std::time(nullptr);
    if (vm.count("seed"))
        seed = vm["seed"].as<unsigned int>();

    std::cout << "using seed: " << seed << std::endl;
    std::srand(seed);

    for (hpx::id_type const& id : hpx::find_remote_localities())
    {
        test_plain_argument(id);
        test_future_argument(id);
        test_mixed_arguments(id);
        test_task_group_argument(id);
    }

#if defined(HPX_HAVE_NETWORKING) && defined(PX_HAVE_PARCELPORT_COUNTERS)
    if (hpx::is_networking_enabled())
    {
        // compare number of parcels with number of messages generated
        print_counters("/parcels/count/*/sent");
        print_counters("/messages/count/*/sent");
    }
#endif

    return hpx::finalize();
}

///////////////////////////////////////////////////////////////////////////////
int main(int argc, char* argv[])
{
    // add command line option which controls the random number generator seed
    using namespace hpx::program_options;
    options_description desc_commandline(
        "Usage: " HPX_APPLICATION_STRING " [options]");

    // clang-format off
    desc_commandline.add_options()
        ("seed,s", value<unsigned int>(),
         "the random number generator seed to use for this run")
        ;
    // clang-format on

    // explicitly disable message handlers (parcel coalescing)
    std::vector<std::string> const cfg = {
#if defined(HPX_HAVE_NETWORKING)
        "hpx.parcel.message_handlers=0"
#endif
    };

    // Initialize and run HPX
    hpx::init_params init_args;
    init_args.desc_cmdline = desc_commandline;
    init_args.cfg = cfg;

    HPX_TEST_EQ_MSG(hpx::init(argc, argv, init_args), 0,
        "HPX main exited with non-zero status");

    return hpx::util::report_errors();
}
#endif

1	// Copyright (c) 2016-2021 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	#include <hpx/config.hpp>
8	#if !defined(HPX_COMPUTE_DEVICE_CODE)
9	#include <hpx/hpx.hpp>
10	#include <hpx/hpx_init.hpp>
11	#include <hpx/modules/testing.hpp>
12	#include <hpx/parallel/task_group.hpp>
13
14	#include <cstddef>
15	#include <iostream>
16	#include <memory>
17	#include <string>
18	#include <type_traits>
19	#include <utility>
20	#include <vector>
21
22	///////////////////////////////////////////////////////////////////////////////
23	std::size_t const vsize_default = 1024;
24	std::size_t const numparcels_default = 10;
25
26	///////////////////////////////////////////////////////////////////////////////
27	template <typename Action, typename T>
28	hpx::parcelset::parcel generate_parcel(	×
29	hpx::id_type const& dest_id, hpx::id_type const& cont, T&& data)
30	{
31	hpx::naming::address addr;	×
32	hpx::naming::gid_type dest = dest_id.get_gid();	×
33	hpx::parcelset::parcel p(hpx::parcelset::detail::create_parcel::call(	×
34	std::move(dest), std::move(addr),	×
35	hpx::actions::typed_continuation<hpx::id_type>(cont), Action(),	×
36	hpx::threads::thread_priority::normal, std::forward<T>(data)));	×
37
38	p.set_source_id(hpx::find_here());	×
39	p.size() = 4096;	×
40	return p;	×
41	}	×
42
43	///////////////////////////////////////////////////////////////////////////////
44	hpx::id_type test1(std::vector<double> const&)	17✔
45	{
46	return hpx::find_here();	17✔
47	}
48	HPX_PLAIN_ACTION(test1)	20✔
49
50	void test_plain_argument(hpx::id_type const& id)	×
51	{
52	std::vector<double> data(vsize_default);	×
53	std::generate(data.begin(), data.end(), std::rand);	×
54
55	std::vector<hpx::future<hpx::id_type>> results;	×
56	results.reserve(numparcels_default);	×
57
58	// create parcels
59	std::vector<hpx::parcelset::parcel> parcels;	×
60	for (std::size_t i = 0; i != numparcels_default; ++i)	×
61	{
62	hpx::distributed::promise<hpx::id_type> p;	×
63	auto f = p.get_future();	×
64	parcels.push_back(generate_parcel<test1_action>(id, p.get_id(), data));	×
65	results.push_back(std::move(f));	×
66	}	×
67
68	// send parcels
69	hpx::get_runtime_distributed().get_parcel_handler().put_parcels(	×
70	std::move(parcels));	×
71
72	// verify all messages got actually sent to the correct locality
73	hpx::wait_all(results);	×
74
75	for (hpx::future<hpx::id_type>& f : results)	×
76	{
77	HPX_TEST_EQ(f.get(), id);	×
78	}
79	}	×
80
81	///////////////////////////////////////////////////////////////////////////////
82	hpx::id_type test2(hpx::future<double> const&)	13✔
83	{
84	return hpx::find_here();	13✔
85	}
86	HPX_PLAIN_ACTION(test2)	16✔
87
88	void test_future_argument(hpx::id_type const& id)	×
89	{
90	std::vector<hpx::promise<double>> args;	×
91	args.reserve(numparcels_default);	×
92
93	std::vector<hpx::future<hpx::id_type>> results;	×
94	results.reserve(numparcels_default);	×
95
96	// create parcels
97	std::vector<hpx::parcelset::parcel> parcels;	×
98	for (std::size_t i = 0; i != numparcels_default; ++i)	×
99	{
100	hpx::promise<double> p_arg;	×
101	hpx::distributed::promise<hpx::id_type> p_cont;	×
102	auto f_cont = p_cont.get_future();	×
103
104	parcels.push_back(generate_parcel<test2_action>(	×
105	id, p_cont.get_id(), p_arg.get_future()));	×
106
107	args.push_back(std::move(p_arg));	×
108	results.push_back(std::move(f_cont));	×
109	}	×
110
111	// send parcels
112	hpx::get_runtime_distributed().get_parcel_handler().put_parcels(	×
113	std::move(parcels));	×
114
115	// now make the futures ready
116	for (hpx::promise<double>& arg : args)	×
117	{
118	arg.set_value(42.0);	×
119	}
120
121	// verify all messages got actually sent to the correct locality
122	hpx::wait_all(results);	×
123
124	for (hpx::future<hpx::id_type>& f : results)	×
125	{
126	HPX_TEST_EQ(f.get(), id);	×
127	}
128	}	×
129
130	void test_mixed_arguments(hpx::id_type const& id)	×
131	{
132	std::vector<double> data(vsize_default);	×
133	std::generate(data.begin(), data.end(), std::rand);	×
134
135	std::vector<hpx::promise<double>> args;	×
136	args.reserve(numparcels_default);	×
137
138	std::vector<hpx::future<hpx::id_type>> results;	×
139	results.reserve(numparcels_default);	×
140
141	// create parcels
142	std::vector<hpx::parcelset::parcel> parcels;	×
143	for (std::size_t i = 0; i != numparcels_default; ++i)	×
144	{
145	hpx::distributed::promise<hpx::id_type> p_cont;	×
146	auto f_cont = p_cont.get_future();	×
147
148	if (std::rand() % 2)	×
149	{
150	parcels.push_back(	×
151	generate_parcel<test1_action>(id, p_cont.get_id(), data));	×
152	}	×
153	else
154	{
155	hpx::promise<double> p_arg;	×
156
157	parcels.push_back(generate_parcel<test2_action>(	×
158	id, p_cont.get_id(), p_arg.get_future()));	×
159
160	args.push_back(std::move(p_arg));	×
161	}	×
162
163	results.push_back(std::move(f_cont));	×
164	}	×
165
166	// send parcels
167	hpx::get_runtime_distributed().get_parcel_handler().put_parcels(	×
168	std::move(parcels));	×
169
170	// now make the futures ready
171	for (hpx::promise<double>& arg : args)	×
172	{
173	arg.set_value(42.0);	×
174	}
175
176	// verify all messages got actually sent to the correct locality
177	hpx::wait_all(results);	×
178
179	for (hpx::future<hpx::id_type>& f : results)	×
180	{
181	HPX_TEST_EQ(f.get(), id);	×
182	}
183	}	×
184
185	///////////////////////////////////////////////////////////////////////////////
186	hpx::id_type test3(std::shared_ptr<hpx::execution::experimental::task_group> tg)	10✔
187	{
188	tg->wait();	10✔
189	return hpx::find_here();	10✔
190	}
191	HPX_PLAIN_ACTION(test3)	13✔
192
193	void wait_a_while()	×
194	{
195	hpx::this_thread::sleep_for(std::chrono::milliseconds(100));	×
196	}	×
197
198	void test_task_group_argument(hpx::id_type const& id)	×
199	{
200	std::vector<std::shared_ptr<hpx::execution::experimental::task_group>> args;	×
201	args.reserve(numparcels_default);	×
202
203	std::vector<hpx::future<hpx::id_type>> results;	×
204	results.reserve(numparcels_default);	×
205
206	// create parcels
207	std::vector<hpx::parcelset::parcel> parcels;	×
208	for (std::size_t i = 0; i != numparcels_default; ++i)	×
209	{
210	auto tg = std::make_shared<hpx::execution::experimental::task_group>();	×
211	tg->run(wait_a_while);	×
212	tg->run(wait_a_while);	×
213	tg->run(wait_a_while);	×
214
215	hpx::distributed::promise<hpx::id_type> p_cont;	×
216	auto f_cont = p_cont.get_future();	×
217
218	parcels.push_back(	×
219	generate_parcel<test3_action>(id, p_cont.get_id(), tg));	×
220
221	args.push_back(std::move(tg));	×
222	results.push_back(std::move(f_cont));	×
223	}	×
224
225	// send parcels
226	hpx::get_runtime_distributed().get_parcel_handler().put_parcels(	×
227	std::move(parcels));	×
228
229	// now wait for task groups
230	for (auto& arg : args)	×
231	{
232	arg->wait();	×
233	}
234
235	// verify all messages got actually sent to the correct locality
236	hpx::wait_all(results);	×
237
238	for (hpx::future<hpx::id_type>& f : results)	×
239	{
240	HPX_TEST_EQ(f.get(), id);	×
241	}
242	}	×
243
244	///////////////////////////////////////////////////////////////////////////////
245	void print_counters(char const* name)	×
246	{
247	using namespace hpx::performance_counters;
248
249	std::vector<performance_counter> counters = discover_counters(name);	×
250
251	for (performance_counter const& c : counters)	×
252	{
253	counter_value value = c.get_counter_value(hpx::launch::sync);	×
254	std::cout << "counter: " << c.get_name(hpx::launch::sync)	×
255	<< ", value: " << value.get_value<double>() << std::endl;	×
256	}
257	}	×
258
259	///////////////////////////////////////////////////////////////////////////////
260	int hpx_main(hpx::program_options::variables_map& vm)	×
261	{
262	unsigned int seed = (unsigned int) std::time(nullptr);	×
263	if (vm.count("seed"))	×
264	seed = vm["seed"].as<unsigned int>();	×
265
266	std::cout << "using seed: " << seed << std::endl;	×
267	std::srand(seed);	×
268
269	for (hpx::id_type const& id : hpx::find_remote_localities())	×
270	{
271	test_plain_argument(id);	×
272	test_future_argument(id);	×
273	test_mixed_arguments(id);	×
274	test_task_group_argument(id);	×
275	}
276
277	#if defined(HPX_HAVE_NETWORKING) && defined(PX_HAVE_PARCELPORT_COUNTERS)
278	if (hpx::is_networking_enabled())
279	{
280	// compare number of parcels with number of messages generated
281	print_counters("/parcels/count/*/sent");
282	print_counters("/messages/count/*/sent");
283	}
284	#endif
285
286	return hpx::finalize();	×
287	}	×
288
289	///////////////////////////////////////////////////////////////////////////////
290	int main(int argc, char* argv[])	1✔
291	{
292	// add command line option which controls the random number generator seed
293	using namespace hpx::program_options;
294	options_description desc_commandline(	1✔
295	"Usage: " HPX_APPLICATION_STRING " [options]");	1✔
296
297	// clang-format off
298	desc_commandline.add_options()	1✔
299	("seed,s", value<unsigned int>(),	1✔
300	"the random number generator seed to use for this run")
301	;
302	// clang-format on
303
304	// explicitly disable message handlers (parcel coalescing)
305	std::vector<std::string> const cfg = {	1✔
306	#if defined(HPX_HAVE_NETWORKING)
307	"hpx.parcel.message_handlers=0"	1✔
308	#endif
309	};
310
311	// Initialize and run HPX
312	hpx::init_params init_args;	1✔
313	init_args.desc_cmdline = desc_commandline;	1✔
314	init_args.cfg = cfg;	1✔
315
316	HPX_TEST_EQ_MSG(hpx::init(argc, argv, init_args), 0,	1✔
317	"HPX main exited with non-zero status");
318
319	return hpx::util::report_errors();	1✔
320	}	1✔
321	#endif

STEllAR-GROUP / hpx / #845

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