#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

16.16

/libs/full/async_colocated/tests/unit/async_cb_colocated.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/include/traits.hpp>
#include <hpx/modules/testing.hpp>

#include <atomic>
#include <chrono>
#include <cstdint>
#include <system_error>
#include <utility>
#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)

///////////////////////////////////////////////////////////////////////////////
struct test_server : hpx::components::component_base<test_server>
{
};

typedef hpx::components::component<test_server> test_server_type;
HPX_REGISTER_COMPONENT(test_server_type, test_server)

struct test_client : hpx::components::client_base<test_client, test_server>
{
    typedef hpx::components::client_base<test_client, test_server> base_type;

    test_client(hpx::id_type const& id)
      : base_type(id)
    {
    }
    test_client(hpx::future<hpx::id_type>&& id)
      : base_type(std::move(id))
    {
    }
};

///////////////////////////////////////////////////////////////////////////////
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_colocated(test_client const& target)
{
    {
        increment_action inc;

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

        hpx::future<std::int32_t> f2 = hpx::async_cb(
            hpx::launch::all, inc, hpx::colocated(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, hpx::colocated(target), &cb, f);
        hpx::future<std::int32_t> f2 = hpx::async_cb(
            hpx::launch::all, inc, hpx::colocated(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>(hpx::colocated(target), &cb, 42);
        HPX_TEST_EQ(f1.get(), 43);

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