#853

Committed 19 Dec 2022 01:01AM UTC coverage: 86.287% (+0.4%) from 85.912%

Build # #853

Build Type

push

Committed by StellarBot

Commit Message

Merge #6109

6109: Modernize serialization module r=hkaiser a=hkaiser

- flyby separate serialization of Boost types

working towards https://github.com/STEllAR-GROUP/hpx/issues/5497

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

Run Details

53 of 53 new or added lines in 6 files covered. (100.0%)

173939 of 201582 relevant lines covered (86.29%)

1931657.12 hits per line

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

55.56

/libs/core/thread_pool_util/src/thread_pool_suspension_helpers.cpp

//  Copyright (c) 2019 Mikael Simberg
//
//  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/async_local/async.hpp>
#include <hpx/async_local/post.hpp>
#include <hpx/functional/function.hpp>
#include <hpx/futures/future.hpp>
#include <hpx/thread_pool_util/thread_pool_suspension_helpers.hpp>
#include <hpx/threading_base/scheduler_base.hpp>
#include <hpx/threading_base/thread_data.hpp>
#include <hpx/threading_base/thread_pool_base.hpp>

#include <cstddef>
#include <utility>

namespace hpx { namespace threads {
    hpx::future<void> resume_processing_unit(
        thread_pool_base& pool, std::size_t virt_core)
    {
        if (!threads::get_self_ptr())
        {
            HPX_THROW_EXCEPTION(hpx::error::invalid_status,
                "resume_processing_unit",
                "cannot call resume_processing_unit from outside HPX, use"
                "resume_processing_unit_cb instead");
        }
        else if (!pool.get_scheduler()->has_scheduler_mode(
                     policies::scheduler_mode::enable_elasticity))
        {
            return hpx::make_exceptional_future<void>(HPX_GET_EXCEPTION(
                hpx::error::invalid_status, "resume_processing_unit",
                "this thread pool does not support suspending "
                "processing units"));
        }

        return hpx::async([&pool, virt_core]() -> void {
            return pool.resume_processing_unit_direct(virt_core, throws);
        });
    }

    void resume_processing_unit_cb(thread_pool_base& pool,
        hpx::function<void(void)> callback, std::size_t virt_core,
        error_code& ec)
    {
        if (!pool.get_scheduler()->has_scheduler_mode(
                policies::scheduler_mode::enable_elasticity))
        {
            HPX_THROWS_IF(ec, hpx::error::invalid_status,
                "resume_processing_unit_cb",
                "this thread pool does not support suspending "
                "processing units");
            return;
        }

        auto resume_direct_wrapper = [&pool, virt_core,
                                         callback = HPX_MOVE(callback)]() {
            pool.resume_processing_unit_direct(virt_core, throws);
            callback();
        };

        if (threads::get_self_ptr())
        {
            hpx::post(HPX_MOVE(resume_direct_wrapper));
        }
        else
        {
            std::thread(HPX_MOVE(resume_direct_wrapper)).detach();
        }
    }

    hpx::future<void> suspend_processing_unit(
        thread_pool_base& pool, std::size_t virt_core)
    {
        if (!threads::get_self_ptr())
        {
            HPX_THROW_EXCEPTION(hpx::error::invalid_status,
                "suspend_processing_unit",
                "cannot call suspend_processing_unit from outside HPX, use"
                "suspend_processing_unit_cb instead");
        }
        if (!pool.get_scheduler()->has_scheduler_mode(
                policies::scheduler_mode::enable_elasticity))
        {
            return hpx::make_exceptional_future<void>(HPX_GET_EXCEPTION(
                hpx::error::invalid_status, "suspend_processing_unit",
                "this thread pool does not support suspending "
                "processing units"));
        }
        if (!pool.get_scheduler()->has_scheduler_mode(
                policies::scheduler_mode::enable_stealing) &&
            hpx::this_thread::get_pool() == &pool)
        {
            return hpx::make_exceptional_future<void>(HPX_GET_EXCEPTION(
                hpx::error::invalid_status, "suspend_processing_unit",
                "this thread pool does not support suspending "
                "processing units from itself (no thread stealing)"));
        }

        return hpx::async([&pool, virt_core]() -> void {
            return pool.suspend_processing_unit_direct(virt_core, throws);
        });
    }

    void suspend_processing_unit_cb(thread_pool_base& pool,
        hpx::function<void(void)> callback, std::size_t virt_core,
        error_code& ec)
    {
        if (!pool.get_scheduler()->has_scheduler_mode(
                policies::scheduler_mode::enable_elasticity))
        {
            HPX_THROWS_IF(ec, hpx::error::invalid_status,
                "suspend_processing_unit_cb",
                "this thread pool does not support suspending processing "
                "units");
            return;
        }

        auto suspend_direct_wrapper = [&pool, virt_core,
                                          callback = HPX_MOVE(callback)]() {
            pool.suspend_processing_unit_direct(virt_core, throws);
            callback();
        };

        if (threads::get_self_ptr())
        {
            if (!pool.get_scheduler()->has_scheduler_mode(
                    policies::scheduler_mode::enable_stealing) &&
                hpx::this_thread::get_pool() == &pool)
            {
                HPX_THROW_EXCEPTION(hpx::error::invalid_status,
                    "suspend_processing_unit_cb",
                    "this thread pool does not support suspending "
                    "processing units from itself (no thread stealing)");
            }

            hpx::post(HPX_MOVE(suspend_direct_wrapper));
        }
        else
        {
            std::thread(HPX_MOVE(suspend_direct_wrapper)).detach();
        }
    }

    future<void> resume_pool(thread_pool_base& pool)
    {
        if (!threads::get_self_ptr())
        {
            HPX_THROW_EXCEPTION(hpx::error::invalid_status, "resume_pool",
                "cannot call resume_pool from outside HPX, use resume_pool_cb "
                "or the member function resume_direct instead");
            return hpx::make_ready_future();
        }

        return hpx::async(
            [&pool]() -> void { return pool.resume_direct(throws); });
    }

    void resume_pool_cb(thread_pool_base& pool,
        hpx::function<void(void)> callback, error_code& /* ec */)
    {
        auto resume_direct_wrapper =
            [&pool, callback = HPX_MOVE(callback)]() -> void {
            pool.resume_direct(throws);
            callback();
        };

        if (threads::get_self_ptr())
        {
            hpx::post(HPX_MOVE(resume_direct_wrapper));
        }
        else
        {
            std::thread(HPX_MOVE(resume_direct_wrapper)).detach();
        }
    }

    future<void> suspend_pool(thread_pool_base& pool)
    {
        if (!threads::get_self_ptr())
        {
            HPX_THROW_EXCEPTION(hpx::error::invalid_status, "suspend_pool",
                "cannot call suspend_pool from outside HPX, use "
                "suspend_pool_cb or the member function suspend_direct "
                "instead");
            return hpx::make_ready_future();
        }
        if (threads::get_self_ptr() && hpx::this_thread::get_pool() == &pool)
        {
            return hpx::make_exceptional_future<void>(
                HPX_GET_EXCEPTION(hpx::error::bad_parameter, "suspend_pool",
                    "cannot suspend a pool from itself"));
        }

        return hpx::async(
            [&pool]() -> void { return pool.suspend_direct(throws); });
    }

    void suspend_pool_cb(thread_pool_base& pool,
        hpx::function<void(void)> callback, error_code& ec)
    {
        if (threads::get_self_ptr() && hpx::this_thread::get_pool() == &pool)
        {
            HPX_THROWS_IF(ec, hpx::error::bad_parameter, "suspend_pool_cb",
                "cannot suspend a pool from itself");
            return;
        }

        auto suspend_direct_wrapper = [&pool, callback = HPX_MOVE(callback)]() {
            pool.suspend_direct(throws);
            callback();
        };

        if (threads::get_self_ptr())
        {
            hpx::post(HPX_MOVE(suspend_direct_wrapper));
        }
        else
        {
            std::thread(HPX_MOVE(suspend_direct_wrapper)).detach();
        }
    }

}}    // namespace hpx::threads

1	// Copyright (c) 2019 Mikael Simberg
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/async_local/async.hpp>
8	#include <hpx/async_local/post.hpp>
9	#include <hpx/functional/function.hpp>
10	#include <hpx/futures/future.hpp>
11	#include <hpx/thread_pool_util/thread_pool_suspension_helpers.hpp>
12	#include <hpx/threading_base/scheduler_base.hpp>
13	#include <hpx/threading_base/thread_data.hpp>
14	#include <hpx/threading_base/thread_pool_base.hpp>
15
16	#include <cstddef>
17	#include <utility>
18
19	namespace hpx { namespace threads {
20	hpx::future<void> resume_processing_unit(	758✔
21	thread_pool_base& pool, std::size_t virt_core)
22	{
23	if (!threads::get_self_ptr())	758✔
24	{
25	HPX_THROW_EXCEPTION(hpx::error::invalid_status,	×
26	"resume_processing_unit",
27	"cannot call resume_processing_unit from outside HPX, use"
28	"resume_processing_unit_cb instead");
29	}
30	else if (!pool.get_scheduler()->has_scheduler_mode(	758✔
31	policies::scheduler_mode::enable_elasticity))
32	{
33	return hpx::make_exceptional_future<void>(HPX_GET_EXCEPTION(	×
34	hpx::error::invalid_status, "resume_processing_unit",
35	"this thread pool does not support suspending "
36	"processing units"));
37	}
38
39	return hpx::async([&pool, virt_core]() -> void {	1,516✔
40	return pool.resume_processing_unit_direct(virt_core, throws);	758✔
41	});
42	}	758✔
43
44	void resume_processing_unit_cb(thread_pool_base& pool,	×
45	hpx::function<void(void)> callback, std::size_t virt_core,
46	error_code& ec)
47	{
48	if (!pool.get_scheduler()->has_scheduler_mode(	×
49	policies::scheduler_mode::enable_elasticity))
50	{
51	HPX_THROWS_IF(ec, hpx::error::invalid_status,	×
52	"resume_processing_unit_cb",
53	"this thread pool does not support suspending "
54	"processing units");
55	return;	×
56	}
57
58	auto resume_direct_wrapper = [&pool, virt_core,	×
59	callback = HPX_MOVE(callback)]() {	×
60	pool.resume_processing_unit_direct(virt_core, throws);	×
61	callback();	×
62	};	×
63
64	if (threads::get_self_ptr())	×
65	{
66	hpx::post(HPX_MOVE(resume_direct_wrapper));	×
67	}	×
68	else
69	{
70	std::thread(HPX_MOVE(resume_direct_wrapper)).detach();	×
71	}
72	}	×
73
74	hpx::future<void> suspend_processing_unit(	707✔
75	thread_pool_base& pool, std::size_t virt_core)
76	{
77	if (!threads::get_self_ptr())	707✔
78	{
79	HPX_THROW_EXCEPTION(hpx::error::invalid_status,	×
80	"suspend_processing_unit",
81	"cannot call suspend_processing_unit from outside HPX, use"
82	"suspend_processing_unit_cb instead");
83	}
84	if (!pool.get_scheduler()->has_scheduler_mode(	707✔
85	policies::scheduler_mode::enable_elasticity))
86	{
87	return hpx::make_exceptional_future<void>(HPX_GET_EXCEPTION(	17✔
88	hpx::error::invalid_status, "suspend_processing_unit",
89	"this thread pool does not support suspending "
90	"processing units"));
91	}
92	if (!pool.get_scheduler()->has_scheduler_mode(	690✔
93	policies::scheduler_mode::enable_stealing) &&	690✔
94	hpx::this_thread::get_pool() == &pool)	89✔
95	{
96	return hpx::make_exceptional_future<void>(HPX_GET_EXCEPTION(	4✔
97	hpx::error::invalid_status, "suspend_processing_unit",
98	"this thread pool does not support suspending "
99	"processing units from itself (no thread stealing)"));
100	}
101
102	return hpx::async([&pool, virt_core]() -> void {	1,372✔
103	return pool.suspend_processing_unit_direct(virt_core, throws);	686✔
104	});
105	}	707✔
106
107	void suspend_processing_unit_cb(thread_pool_base& pool,	×
108	hpx::function<void(void)> callback, std::size_t virt_core,
109	error_code& ec)
110	{
111	if (!pool.get_scheduler()->has_scheduler_mode(	×
112	policies::scheduler_mode::enable_elasticity))
113	{
114	HPX_THROWS_IF(ec, hpx::error::invalid_status,	×
115	"suspend_processing_unit_cb",
116	"this thread pool does not support suspending processing "
117	"units");
118	return;	×
119	}
120
121	auto suspend_direct_wrapper = [&pool, virt_core,	×
122	callback = HPX_MOVE(callback)]() {	×
123	pool.suspend_processing_unit_direct(virt_core, throws);	×
124	callback();	×
125	};	×
126
127	if (threads::get_self_ptr())	×
128	{
129	if (!pool.get_scheduler()->has_scheduler_mode(	×
130	policies::scheduler_mode::enable_stealing) &&	×
131	hpx::this_thread::get_pool() == &pool)	×
132	{
133	HPX_THROW_EXCEPTION(hpx::error::invalid_status,	×
134	"suspend_processing_unit_cb",
135	"this thread pool does not support suspending "
136	"processing units from itself (no thread stealing)");
137	}
138
139	hpx::post(HPX_MOVE(suspend_direct_wrapper));	×
140	}	×
141	else
142	{
143	std::thread(HPX_MOVE(suspend_direct_wrapper)).detach();	×
144	}
145	}	×
146
147	future<void> resume_pool(thread_pool_base& pool)	467✔
148	{
149	if (!threads::get_self_ptr())	467✔
150	{
151	HPX_THROW_EXCEPTION(hpx::error::invalid_status, "resume_pool",	×
152	"cannot call resume_pool from outside HPX, use resume_pool_cb "
153	"or the member function resume_direct instead");
154	return hpx::make_ready_future();
155	}
156
157	return hpx::async(	467✔
158	[&pool]() -> void { return pool.resume_direct(throws); });	934✔
159	}	×
160
161	void resume_pool_cb(thread_pool_base& pool,	16✔
162	hpx::function<void(void)> callback, error_code& /* ec */)
163	{
164	auto resume_direct_wrapper =
165	[&pool, callback = HPX_MOVE(callback)]() -> void {	128✔
166	pool.resume_direct(throws);	16✔
167	callback();	16✔
168	};	16✔
169
170	if (threads::get_self_ptr())	16✔
171	{
172	hpx::post(HPX_MOVE(resume_direct_wrapper));	8✔
173	}	8✔
174	else
175	{
176	std::thread(HPX_MOVE(resume_direct_wrapper)).detach();	8✔
177	}
178	}	16✔
179
180	future<void> suspend_pool(thread_pool_base& pool)	24✔
181	{
182	if (!threads::get_self_ptr())	24✔
183	{
184	HPX_THROW_EXCEPTION(hpx::error::invalid_status, "suspend_pool",	×
185	"cannot call suspend_pool from outside HPX, use "
186	"suspend_pool_cb or the member function suspend_direct "
187	"instead");
188	return hpx::make_ready_future();
189	}
190	if (threads::get_self_ptr() && hpx::this_thread::get_pool() == &pool)	24✔
191	{
192	return hpx::make_exceptional_future<void>(	8✔
193	HPX_GET_EXCEPTION(hpx::error::bad_parameter, "suspend_pool",	8✔
194	"cannot suspend a pool from itself"));
195	}
196
197	return hpx::async(	16✔
198	[&pool]() -> void { return pool.suspend_direct(throws); });	32✔
199	}	24✔
200
201	void suspend_pool_cb(thread_pool_base& pool,	16✔
202	hpx::function<void(void)> callback, error_code& ec)
203	{
204	if (threads::get_self_ptr() && hpx::this_thread::get_pool() == &pool)	16✔
205	{
206	HPX_THROWS_IF(ec, hpx::error::bad_parameter, "suspend_pool_cb",	×
207	"cannot suspend a pool from itself");
208	return;	×
209	}
210
211	auto suspend_direct_wrapper = [&pool, callback = HPX_MOVE(callback)]() {	128✔
212	pool.suspend_direct(throws);	16✔
213	callback();	16✔
214	};	16✔
215
216	if (threads::get_self_ptr())	16✔
217	{
218	hpx::post(HPX_MOVE(suspend_direct_wrapper));	8✔
219	}	8✔
220	else
221	{
222	std::thread(HPX_MOVE(suspend_direct_wrapper)).detach();	8✔
223	}
224	}	16✔
225
226	}} // namespace hpx::threads

STEllAR-GROUP / hpx / #853

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