#872

Committed 24 Jan 2023 07:29PM UTC coverage: 85.694% (-0.9%) from 86.624%

Build # #872

Build Type

push

Committed by StellarBot

Commit Message

Merge #6148

6148: Investigate the failure of the LCI parcelport. r=hkaiser a=JiakunYan



Co-authored-by: Jiakun Yan <jiakunyan1998@gmail.com>

Run Details

173076 of 201969 relevant lines covered (85.69%)

2110584.0 hits per line

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

17.07

/libs/full/async_distributed/tests/unit/async_cb_remote.cpp

//  Copyright (c) 2007-2015 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_init.hpp>
#include <hpx/include/actions.hpp>
#include <hpx/include/async.hpp>
#include <hpx/include/components.hpp>
#include <hpx/include/lcos.hpp>
#include <hpx/include/runtime.hpp>
#include <hpx/modules/testing.hpp>

#include <atomic>
#include <chrono>
#include <cstdint>
#include <system_error>
#include <vector>

///////////////////////////////////////////////////////////////////////////////
std::int32_t increment(std::int32_t i)
{
    return i + 1;
}
HPX_PLAIN_ACTION(increment)

std::int32_t increment_with_future(hpx::shared_future<std::int32_t> fi)
{
    return fi.get() + 1;
}
HPX_PLAIN_ACTION(increment_with_future)

///////////////////////////////////////////////////////////////////////////////
struct decrement_server
  : hpx::components::managed_component_base<decrement_server>
{
    std::int32_t call(std::int32_t i) const
    {
        return i - 1;
    }

    HPX_DEFINE_COMPONENT_ACTION(decrement_server, call)
};

typedef hpx::components::managed_component<decrement_server> server_type;
HPX_REGISTER_COMPONENT(server_type, decrement_server)

typedef decrement_server::call_action call_action;
HPX_REGISTER_ACTION_DECLARATION(call_action)
HPX_REGISTER_ACTION(call_action)

///////////////////////////////////////////////////////////////////////////////
std::atomic<int> callback_called(0);

#if defined(HPX_HAVE_NETWORKING)
void cb(std::error_code const&, hpx::parcelset::parcel const&)
{
    ++callback_called;
}
#else
void cb()
{
    ++callback_called;
}
#endif

///////////////////////////////////////////////////////////////////////////////
void test_remote_async_cb(hpx::id_type const& target)
{
    {
        increment_action inc;

        callback_called.store(0);
        hpx::future<std::int32_t> f1 = hpx::async_cb(inc, target, &cb, 42);
        HPX_TEST_EQ(f1.get(), 43);

        hpx::future<std::int32_t> f2 =
            hpx::async_cb(hpx::launch::all, inc, target, &cb, 42);
        HPX_TEST_EQ(f2.get(), 43);

        // The callback should have been called 2 times. wait for a short period
        // of time, to allow it for it to be fully executed
        hpx::this_thread::sleep_for(std::chrono::milliseconds(100));
        HPX_TEST_EQ(callback_called.load(), 2);
    }

    {
        increment_with_future_action inc;

        hpx::distributed::promise<std::int32_t> p;
        hpx::shared_future<std::int32_t> f = p.get_future();

        callback_called.store(0);
        hpx::future<std::int32_t> f1 = hpx::async_cb(inc, target, &cb, f);
        hpx::future<std::int32_t> f2 =
            hpx::async_cb(hpx::launch::all, inc, target, &cb, f);

        p.set_value(42);
        HPX_TEST_EQ(f1.get(), 43);
        HPX_TEST_EQ(f2.get(), 43);

        // The callback should have been called 2 times. wait for a short period
        // of time, to allow it for it to be fully executed
        hpx::this_thread::sleep_for(std::chrono::milliseconds(100));
        HPX_TEST_EQ(callback_called.load(), 2);
    }

    {
        callback_called.store(0);
        hpx::future<std::int32_t> f1 =
            hpx::async_cb<increment_action>(target, &cb, 42);
        HPX_TEST_EQ(f1.get(), 43);

        hpx::future<std::int32_t> f2 =
            hpx::async_cb<increment_action>(hpx::launch::all, target, &cb, 42);
        HPX_TEST_EQ(f2.get(), 43);

        // The callback should have been called 2 times. wait for a short period
        // of time, to allow it for it to be fully executed
        hpx::this_thread::sleep_for(std::chrono::milliseconds(100));
        HPX_TEST_EQ(callback_called.load(), 2);
    }

    {
        hpx::future<hpx::id_type> dec_f =
            hpx::components::new_<decrement_server>(target);
        hpx::id_type dec = dec_f.get();

        call_action call;

        callback_called.store(0);
        hpx::future<std::int32_t> f1 = hpx::async_cb(call, dec, &cb, 42);
        HPX_TEST_EQ(f1.get(), 41);

        hpx::future<std::int32_t> f2 =
            hpx::async_cb(hpx::launch::all, call, dec, &cb, 42);
        HPX_TEST_EQ(f2.get(), 41);

        // The callback should have been called 2 times. wait for a short period
        // of time, to allow it for it to be fully executed
        hpx::this_thread::sleep_for(std::chrono::milliseconds(100));
        HPX_TEST_EQ(callback_called.load(), 2);
    }

    {
        hpx::future<hpx::id_type> dec_f =
            hpx::components::new_<decrement_server>(target);
        hpx::id_type dec = dec_f.get();

        callback_called.store(0);
        hpx::future<std::int32_t> f1 = hpx::async_cb<call_action>(dec, &cb, 42);
        HPX_TEST_EQ(f1.get(), 41);

        hpx::future<std::int32_t> f2 =
            hpx::async_cb<call_action>(hpx::launch::all, dec, &cb, 42);
        HPX_TEST_EQ(f2.get(), 41);

        // The callback should have been called 2 times. wait for a short period
        // of time, to allow it for it to be fully executed
        hpx::this_thread::sleep_for(std::chrono::milliseconds(100));
        HPX_TEST_EQ(callback_called.load(), 2);
    }

    {
        increment_with_future_action inc;
        hpx::shared_future<std::int32_t> f =
            hpx::async(hpx::launch::deferred, hpx::bind(&increment, 42));

        callback_called.store(0);
        hpx::future<std::int32_t> f1 = hpx::async_cb(inc, target, &cb, f);
        hpx::future<std::int32_t> f2 =
            hpx::async_cb(hpx::launch::all, inc, target, &cb, f);

        HPX_TEST_EQ(f1.get(), 44);
        HPX_TEST_EQ(f2.get(), 44);

        // The callback should have been called 2 times. wait for a short period
        // of time, to allow it for it to be fully executed
        hpx::this_thread::sleep_for(std::chrono::milliseconds(100));
        HPX_TEST_EQ(callback_called.load(), 2);
    }
}

int hpx_main()
{
    std::vector<hpx::id_type> localities = hpx::find_all_localities();
    for (hpx::id_type const& id : localities)
    {
        test_remote_async_cb(id);
    }
    return hpx::finalize();
}

int main(int argc, char* argv[])
{
    // Initialize and run HPX
    HPX_TEST_EQ_MSG(
        hpx::init(argc, argv), 0, "HPX main exited with non-zero status");

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

1	// Copyright (c) 2007-2015 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_init.hpp>
10	#include <hpx/include/actions.hpp>
11	#include <hpx/include/async.hpp>
12	#include <hpx/include/components.hpp>
13	#include <hpx/include/lcos.hpp>
14	#include <hpx/include/runtime.hpp>
15	#include <hpx/modules/testing.hpp>
16
17	#include <atomic>
18	#include <chrono>
19	#include <cstdint>
20	#include <system_error>
21	#include <vector>
22
23	///////////////////////////////////////////////////////////////////////////////
24	std::int32_t increment(std::int32_t i)	4✔
25	{
26	return i + 1;	4✔
27	}
28	HPX_PLAIN_ACTION(increment)	7✔
29
30	std::int32_t increment_with_future(hpx::shared_future<std::int32_t> fi)	4✔
31	{
32	return fi.get() + 1;	4✔
33	}
34	HPX_PLAIN_ACTION(increment_with_future)	7✔
35
36	///////////////////////////////////////////////////////////////////////////////
37	struct decrement_server	4✔
38	: hpx::components::managed_component_base<decrement_server>
39	{
40	std::int32_t call(std::int32_t i) const	4✔
41	{
42	return i - 1;	4✔
43	}
44
45	HPX_DEFINE_COMPONENT_ACTION(decrement_server, call)
46	};
47
48	typedef hpx::components::managed_component<decrement_server> server_type;
49	HPX_REGISTER_COMPONENT(server_type, decrement_server)	46✔
50
51	typedef decrement_server::call_action call_action;
52	HPX_REGISTER_ACTION_DECLARATION(call_action)
53	HPX_REGISTER_ACTION(call_action)	7✔
54
55	///////////////////////////////////////////////////////////////////////////////
56	std::atomic<int> callback_called(0);
57
58	#if defined(HPX_HAVE_NETWORKING)
59	void cb(std::error_code const&, hpx::parcelset::parcel const&)	×
60	{
61	++callback_called;	×
62	}	×
63	#else
64	void cb()
65	{
66	++callback_called;
67	}
68	#endif
69
70	///////////////////////////////////////////////////////////////////////////////
71	void test_remote_async_cb(hpx::id_type const& target)	×
72	{
73	{
74	increment_action inc;
75
76	callback_called.store(0);	×
77	hpx::future<std::int32_t> f1 = hpx::async_cb(inc, target, &cb, 42);	×
78	HPX_TEST_EQ(f1.get(), 43);	×
79
80	hpx::future<std::int32_t> f2 =
81	hpx::async_cb(hpx::launch::all, inc, target, &cb, 42);	×
82	HPX_TEST_EQ(f2.get(), 43);	×
83
84	// The callback should have been called 2 times. wait for a short period
85	// of time, to allow it for it to be fully executed
86	hpx::this_thread::sleep_for(std::chrono::milliseconds(100));	×
87	HPX_TEST_EQ(callback_called.load(), 2);	×
88	}	×
89
90	{
91	increment_with_future_action inc;
92
93	hpx::distributed::promise<std::int32_t> p;	×
94	hpx::shared_future<std::int32_t> f = p.get_future();	×
95
96	callback_called.store(0);	×
97	hpx::future<std::int32_t> f1 = hpx::async_cb(inc, target, &cb, f);	×
98	hpx::future<std::int32_t> f2 =
99	hpx::async_cb(hpx::launch::all, inc, target, &cb, f);	×
100
101	p.set_value(42);	×
102	HPX_TEST_EQ(f1.get(), 43);	×
103	HPX_TEST_EQ(f2.get(), 43);	×
104
105	// The callback should have been called 2 times. wait for a short period
106	// of time, to allow it for it to be fully executed
107	hpx::this_thread::sleep_for(std::chrono::milliseconds(100));	×
108	HPX_TEST_EQ(callback_called.load(), 2);	×
109	}	×
110
111	{
112	callback_called.store(0);	×
113	hpx::future<std::int32_t> f1 =
114	hpx::async_cb<increment_action>(target, &cb, 42);	×
115	HPX_TEST_EQ(f1.get(), 43);	×
116
117	hpx::future<std::int32_t> f2 =
118	hpx::async_cb<increment_action>(hpx::launch::all, target, &cb, 42);	×
119	HPX_TEST_EQ(f2.get(), 43);	×
120
121	// The callback should have been called 2 times. wait for a short period
122	// of time, to allow it for it to be fully executed
123	hpx::this_thread::sleep_for(std::chrono::milliseconds(100));	×
124	HPX_TEST_EQ(callback_called.load(), 2);	×
125	}	×
126
127	{
128	hpx::future<hpx::id_type> dec_f =
129	hpx::components::new_<decrement_server>(target);	×
130	hpx::id_type dec = dec_f.get();	×
131
132	call_action call;
133
134	callback_called.store(0);	×
135	hpx::future<std::int32_t> f1 = hpx::async_cb(call, dec, &cb, 42);	×
136	HPX_TEST_EQ(f1.get(), 41);	×
137
138	hpx::future<std::int32_t> f2 =
139	hpx::async_cb(hpx::launch::all, call, dec, &cb, 42);	×
140	HPX_TEST_EQ(f2.get(), 41);	×
141
142	// The callback should have been called 2 times. wait for a short period
143	// of time, to allow it for it to be fully executed
144	hpx::this_thread::sleep_for(std::chrono::milliseconds(100));	×
145	HPX_TEST_EQ(callback_called.load(), 2);	×
146	}	×
147
148	{
149	hpx::future<hpx::id_type> dec_f =
150	hpx::components::new_<decrement_server>(target);	×
151	hpx::id_type dec = dec_f.get();	×
152
153	callback_called.store(0);	×
154	hpx::future<std::int32_t> f1 = hpx::async_cb<call_action>(dec, &cb, 42);	×
155	HPX_TEST_EQ(f1.get(), 41);	×
156
157	hpx::future<std::int32_t> f2 =
158	hpx::async_cb<call_action>(hpx::launch::all, dec, &cb, 42);	×
159	HPX_TEST_EQ(f2.get(), 41);	×
160
161	// The callback should have been called 2 times. wait for a short period
162	// of time, to allow it for it to be fully executed
163	hpx::this_thread::sleep_for(std::chrono::milliseconds(100));	×
164	HPX_TEST_EQ(callback_called.load(), 2);	×
165	}	×
166
167	{
168	increment_with_future_action inc;
169	hpx::shared_future<std::int32_t> f =
170	hpx::async(hpx::launch::deferred, hpx::bind(&increment, 42));	×
171
172	callback_called.store(0);	×
173	hpx::future<std::int32_t> f1 = hpx::async_cb(inc, target, &cb, f);	×
174	hpx::future<std::int32_t> f2 =
175	hpx::async_cb(hpx::launch::all, inc, target, &cb, f);	×
176
177	HPX_TEST_EQ(f1.get(), 44);	×
178	HPX_TEST_EQ(f2.get(), 44);	×
179
180	// The callback should have been called 2 times. wait for a short period
181	// of time, to allow it for it to be fully executed
182	hpx::this_thread::sleep_for(std::chrono::milliseconds(100));	×
183	HPX_TEST_EQ(callback_called.load(), 2);	×
184	}	×
185	}	×
186
187	int hpx_main()	×
188	{
189	std::vector<hpx::id_type> localities = hpx::find_all_localities();	×
190	for (hpx::id_type const& id : localities)	×
191	{
192	test_remote_async_cb(id);	×
193	}
194	return hpx::finalize();	×
195	}	×
196
197	int main(int argc, char* argv[])	1✔
198	{
199	// Initialize and run HPX
200	HPX_TEST_EQ_MSG(	1✔
201	hpx::init(argc, argv), 0, "HPX main exited with non-zero status");
202
203	return hpx::util::report_errors();	1✔
204	}	×
205	#endif

STEllAR-GROUP / hpx / #872

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