#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

74.63

/libs/core/thread_pools/src/detail/background_thread.cpp

//  Copyright (c) 2023-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>
#include <hpx/assert.hpp>
#include <hpx/modules/execution_base.hpp>
#include <hpx/modules/logging.hpp>
#include <hpx/modules/threading_base.hpp>
#include <hpx/thread_pools/detail/background_thread.hpp>
#include <hpx/thread_pools/detail/scheduling_callbacks.hpp>
#include <hpx/thread_pools/detail/scoped_background_timer.hpp>

#include <cstddef>
#include <cstdint>
#include <memory>

namespace hpx::threads::detail {

    ///////////////////////////////////////////////////////////////////////////
    thread_id_ref_type create_background_thread(
        threads::policies::scheduler_base& scheduler_base,
        std::size_t const num_thread, scheduling_callbacks const& callbacks,
        std::shared_ptr<bool>& background_running,
        std::int64_t& idle_loop_count)
    {
        threads::thread_schedule_hint const schedulehint(
            static_cast<std::int16_t>(num_thread));

        thread_id_ref_type background_thread;
        background_running = std::make_shared<bool>(true);

        thread_init_data background_init(
            [&, background_running](
                thread_restart_state) -> thread_result_type {
                while (*background_running)
                {
                    if (callbacks.background_())
                    {
                        // we only update the idle_loop_count if
                        // background_running is true. If it was false, this
                        // task was given back to the scheduler.
                        if (*background_running)
                        {
                            idle_loop_count = 0;
                        }
                    }

                    // Force yield...
                    hpx::execution_base::this_thread::yield("background_work");
                }

                return {thread_schedule_state::terminated, invalid_thread_id};
            },
            hpx::threads::thread_description("background_work"),
            thread_priority::high_recursive, schedulehint,
            thread_stacksize::large,
            // Create in suspended to prevent the thread from being scheduled
            // directly...
            thread_schedule_state::suspended, true, &scheduler_base);

        scheduler_base.create_thread(
            background_init, &background_thread, hpx::throws);
        HPX_ASSERT(background_thread);

        scheduler_base.increment_background_thread_count();

        LTM_(debug).format("create_background_thread: pool({}), "
                           "scheduler({}), worker_thread({}), thread({})",
            scheduler_base.get_parent_pool(), scheduler_base, num_thread,
            get_thread_id_data(background_thread));

        // We can now set the state to pending
        [[maybe_unused]] auto old_state =
            get_thread_id_data(background_thread)
                ->set_state(thread_schedule_state::pending);
        return background_thread;
    }

    ///////////////////////////////////////////////////////////////////////////
    class switch_status_background
    {
    public:
        switch_status_background(
            thread_id_ref_type const& t, thread_state const prev_state) noexcept
          : thread_(get_thread_id_data(t))
          , prev_state_(prev_state)
          , next_thread_id_(nullptr)
          , need_restore_state_(
                thread_->set_state_tagged(thread_schedule_state::active,
                    prev_state_, orig_state_, std::memory_order_relaxed))
        {
        }

        switch_status_background(switch_status_background const&) = delete;
        switch_status_background(switch_status_background&&) = delete;

        switch_status_background& operator=(
            switch_status_background const&) = delete;
        switch_status_background& operator=(
            switch_status_background&&) = delete;

        ~switch_status_background()
        {
            if (need_restore_state_)
            {
                store_state(prev_state_);
            }
        }

        [[nodiscard]] constexpr bool is_valid() const noexcept
        {
            return need_restore_state_;
        }

        // allow to change the state the thread will be switched to after
        // execution
        switch_status_background& operator=(
            thread_result_type&& new_state) noexcept
        {
            prev_state_ = thread_state(
                new_state.first, prev_state_.state_ex(), prev_state_.tag() + 1);
            if (new_state.second != nullptr)
            {
                next_thread_id_ = HPX_MOVE(new_state.second);
            }
            return *this;
        }

        // Get the state this thread was in before execution (usually pending),
        // this helps to make sure no other worker-thread is started to execute
        // this HPX-thread in the meantime.
        [[nodiscard]] thread_schedule_state get_previous() const noexcept
        {
            return prev_state_.state();
        }

        // This restores the previous state, while making sure that the original
        // state has not been changed since we started executing this thread.
        // The function returns true if the state has been set, false otherwise.
        bool store_state(thread_state& newstate) noexcept
        {
            disable_restore();
            if (thread_->restore_state(prev_state_, orig_state_))
            {
                newstate = prev_state_;
                return true;
            }
            return false;
        }

        // disable default handling in destructor
        void disable_restore() noexcept
        {
            need_restore_state_ = false;
        }

        [[nodiscard]] constexpr thread_id_ref_type const& get_next_thread()
            const noexcept
        {
            return next_thread_id_;
        }

        thread_id_ref_type move_next_thread() noexcept
        {
            return HPX_MOVE(next_thread_id_);
        }

    private:
        thread_data* thread_;
        thread_state prev_state_;
        thread_state orig_state_;
        thread_id_ref_type next_thread_id_;
        bool need_restore_state_;
    };

    // This function tries to invoke the background work thread. It returns
    // false when we need to give the background thread back to scheduler and
    // create a new one that is supposed to be executed inside the
    // scheduling_loop, true otherwise
    bool call_background_thread(thread_id_ref_type& background_thread,
        thread_id_ref_type& next_thrd,
        threads::policies::scheduler_base& scheduler_base,
        std::size_t const num_thread,
        [[maybe_unused]] background_work_exec_time& exec_time,
        hpx::execution_base::this_thread::detail::agent_storage*
            context_storage)
    {
        LTM_(debug).format("call_background_thread: pool({}), "
                           "scheduler({}), worker_thread({}), thread({})",
            scheduler_base.get_parent_pool(), scheduler_base, num_thread,
            get_thread_id_data(background_thread));

        if (HPX_LIKELY(background_thread))
        {
            auto* thrdptr = get_thread_id_data(background_thread);
            thread_state state = thrdptr->get_state();
            thread_schedule_state state_val = state.state();

            // we should only deal with pending here.
            HPX_ASSERT(thread_schedule_state::pending == state_val);

            // tries to set state to active (only if state is still
            // the same as 'state')
            detail::switch_status_background thrd_stat(
                background_thread, state);

            if (HPX_LIKELY(thrd_stat.is_valid() &&
                    thrd_stat.get_previous() == thread_schedule_state::pending))
            {
#if defined(HPX_HAVE_BACKGROUND_THREAD_COUNTERS) &&                            \
    defined(HPX_HAVE_THREAD_IDLE_RATES)
                // measure background work duration
                background_work_duration_counter bg_work_duration(
                    exec_time.timer);
                background_exec_time_wrapper bg_exec_time(bg_work_duration);
#endif    // HPX_HAVE_BACKGROUND_THREAD_COUNTERS

                // invoke background thread
                thrd_stat = (*thrdptr)(context_storage);

                if (thread_id_ref_type next = thrd_stat.move_next_thread();
                    next && next != background_thread)
                {
                    if (!next_thrd)
                    {
                        next_thrd = HPX_MOVE(next);
                    }
                    else
                    {
                        auto* scheduler =
                            get_thread_id_data(next)->get_scheduler_base();
                        scheduler->schedule_thread(HPX_MOVE(next),
                            threads::thread_schedule_hint(
                                static_cast<std::int16_t>(num_thread)),
                            true);
                        scheduler->do_some_work(num_thread);
                    }
                }
            }
            thrd_stat.store_state(state);
            state_val = state.state();

            if (HPX_LIKELY(state_val == thread_schedule_state::pending_boost))
            {
                [[maybe_unused]] auto old_state =
                    thrdptr->set_state(thread_schedule_state::pending);
            }
            else if (thread_schedule_state::terminated == state_val)
            {
                LTM_(debug).format(
                    "call_background_thread terminated: pool({}), "
                    "scheduler({}), worker_thread({}), thread({})",
                    scheduler_base.get_parent_pool(), scheduler_base,
                    num_thread, get_thread_id_data(background_thread));

                scheduler_base.decrement_background_thread_count();
                background_thread = thread_id_type();
            }
            else if (thread_schedule_state::suspended == state_val)
            {
                LTM_(debug).format(
                    "call_background_thread suspended: pool({}), "
                    "scheduler({}), worker_thread({}), thread({})",
                    scheduler_base.get_parent_pool(), scheduler_base,
                    num_thread, get_thread_id_data(background_thread));

                return false;
            }
        }
        return true;
    }

    ///////////////////////////////////////////////////////////////////////////
    bool call_and_create_background_thread(
        thread_id_ref_type& background_thread, thread_id_ref_type& next_thrd,
        threads::policies::scheduler_base& scheduler_base,
        std::size_t const num_thread, background_work_exec_time& exec_time,
        hpx::execution_base::this_thread::detail::agent_storage*
            context_storage,
        scheduling_callbacks const& callbacks, std::shared_ptr<bool>& running,
        std::int64_t& idle_loop_count)
    {
        if (!call_background_thread(background_thread, next_thrd,
                scheduler_base, num_thread, exec_time, context_storage))
        {
            // Let the current background thread terminate as soon as possible.
            // No need to reschedule, as another thread will set it to pending
            // and schedule it back eventually
            HPX_ASSERT(background_thread);
            HPX_ASSERT(running);

            *running = false;
            scheduler_base.decrement_background_thread_count();

            // Create a new one that will replace the current such we avoid
            // deadlock situations, if all background threads are blocked.
            background_thread = create_background_thread(scheduler_base,
                num_thread, callbacks, running, idle_loop_count);

            return true;
        }
        return false;
    }
}    // namespace hpx::threads::detail

1	// Copyright (c) 2023-2025 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	#include <hpx/assert.hpp>
9	#include <hpx/modules/execution_base.hpp>
10	#include <hpx/modules/logging.hpp>
11	#include <hpx/modules/threading_base.hpp>
12	#include <hpx/thread_pools/detail/background_thread.hpp>
13	#include <hpx/thread_pools/detail/scheduling_callbacks.hpp>
14	#include <hpx/thread_pools/detail/scoped_background_timer.hpp>
15
16	#include <cstddef>
17	#include <cstdint>
18	#include <memory>
19
20	namespace hpx::threads::detail {
21
22	///////////////////////////////////////////////////////////////////////////
23	thread_id_ref_type create_background_thread(
24	threads::policies::scheduler_base& scheduler_base,	6✔
25	std::size_t const num_thread, scheduling_callbacks const& callbacks,
26	std::shared_ptr<bool>& background_running,
27	std::int64_t& idle_loop_count)
28	{
29	threads::thread_schedule_hint const schedulehint(
30	static_cast<std::int16_t>(num_thread));
31		6✔
32	thread_id_ref_type background_thread;
33	background_running = std::make_shared<bool>(true);	6✔
34		12✔
35	thread_init_data background_init(
36	[&, background_running](
37	thread_restart_state) -> thread_result_type {	12✔
38	while (*background_running)
39	{	686,025✔
40	if (callbacks.background_())
41	{	686,019✔
42	// we only update the idle_loop_count if
43	// background_running is true. If it was false, this
44	// task was given back to the scheduler.
45	if (*background_running)
46	{	1,768✔
47	idle_loop_count = 0;
48	}	1,768✔
49	}
50
51	// Force yield...
52	hpx::execution_base::this_thread::yield("background_work");
53	}	686,019✔
54
55	return {thread_schedule_state::terminated, invalid_thread_id};
56	},	6✔
57	hpx::threads::thread_description("background_work"),
58	thread_priority::high_recursive, schedulehint,
59	thread_stacksize::large,
60	// Create in suspended to prevent the thread from being scheduled
61	// directly...
62	thread_schedule_state::suspended, true, &scheduler_base);
63		6✔
64	scheduler_base.create_thread(
65	background_init, &background_thread, hpx::throws);	6✔
66	HPX_ASSERT(background_thread);
67
68	scheduler_base.increment_background_thread_count();
69		6✔
70	LTM_(debug).format("create_background_thread: pool({}), "
71	"scheduler({}), worker_thread({}), thread({})",	6✔
72	scheduler_base.get_parent_pool(), scheduler_base, num_thread,
73	get_thread_id_data(background_thread));	×
74		×
75	// We can now set the state to pending
76	[[maybe_unused]] auto old_state =
77	get_thread_id_data(background_thread)
78	->set_state(thread_schedule_state::pending);	6✔
79	return background_thread;	6✔
80	}
81
82	///////////////////////////////////////////////////////////////////////////
83	class switch_status_background
84	{
85	public:
86	switch_status_background(
87	thread_id_ref_type const& t, thread_state const prev_state) noexcept
88	: thread_(get_thread_id_data(t))	686,035✔
89	, prev_state_(prev_state)	686,035✔
90	, next_thread_id_(nullptr)
91	, need_restore_state_(	686,035✔
92	thread_->set_state_tagged(thread_schedule_state::active,	686,035✔
93	prev_state_, orig_state_, std::memory_order_relaxed))
94	{
95	}
96
97	switch_status_background(switch_status_background const&) = delete;
98	switch_status_background(switch_status_background&&) = delete;
99
100	switch_status_background& operator=(
101	switch_status_background const&) = delete;
102	switch_status_background& operator=(
103	switch_status_background&&) = delete;
104
105	~switch_status_background()	686,035✔
106	{
107	if (need_restore_state_)	686,035✔
108	{
109	store_state(prev_state_);	×
110	}
111	}	686,035✔
112
113	[[nodiscard]] constexpr bool is_valid() const noexcept
114	{
115	return need_restore_state_;
116	}
117
118	// allow to change the state the thread will be switched to after
119	// execution
120	switch_status_background& operator=(	686,035✔
121	thread_result_type&& new_state) noexcept
122	{	686,035✔
123	prev_state_ = thread_state(
124	new_state.first, prev_state_.state_ex(), prev_state_.tag() + 1);	686,035✔
125	if (new_state.second != nullptr)
126	{
127	next_thread_id_ = HPX_MOVE(new_state.second);
128	}	686,035✔
129	return *this;
130	}
131
132	// Get the state this thread was in before execution (usually pending),
133	// this helps to make sure no other worker-thread is started to execute
134	// this HPX-thread in the meantime.
135	[[nodiscard]] thread_schedule_state get_previous() const noexcept
136	{
137	return prev_state_.state();
138	}
139
140	// This restores the previous state, while making sure that the original
141	// state has not been changed since we started executing this thread.
142	// The function returns true if the state has been set, false otherwise.	686,035✔
143	bool store_state(thread_state& newstate) noexcept
144	{
145	disable_restore();	686,035✔
146	if (thread_->restore_state(prev_state_, orig_state_))
147	{	686,035✔
148	newstate = prev_state_;	686,035✔
149	return true;
150	}
151	return false;
152	}
153
154	// disable default handling in destructor
155	void disable_restore() noexcept
156	{	686,035✔
157	need_restore_state_ = false;
158	}
159
160	[[nodiscard]] constexpr thread_id_ref_type const& get_next_thread()
161	const noexcept
162	{
163	return next_thread_id_;
164	}
165
166	thread_id_ref_type move_next_thread() noexcept
167	{
168	return HPX_MOVE(next_thread_id_);
169	}
170
171	private:
172	thread_data* thread_;
173	thread_state prev_state_;
174	thread_state orig_state_;
175	thread_id_ref_type next_thread_id_;
176	bool need_restore_state_;
177	};
178
179	// This function tries to invoke the background work thread. It returns
180	// false when we need to give the background thread back to scheduler and
181	// create a new one that is supposed to be executed inside the	686,041✔
182	// scheduling_loop, true otherwise
183	bool call_background_thread(thread_id_ref_type& background_thread,
184	thread_id_ref_type& next_thrd,
185	threads::policies::scheduler_base& scheduler_base,
186	std::size_t const num_thread,
187	[[maybe_unused]] background_work_exec_time& exec_time,
188	hpx::execution_base::this_thread::detail::agent_storage*
189	context_storage)	686,041✔
190	{
191	LTM_(debug).format("call_background_thread: pool({}), "	×
192	"scheduler({}), worker_thread({}), thread({})",	×
193	scheduler_base.get_parent_pool(), scheduler_base, num_thread,
194	get_thread_id_data(background_thread));	686,041✔
195
196	if (HPX_LIKELY(background_thread))
197	{	686,035✔
198	auto* thrdptr = get_thread_id_data(background_thread);
199	thread_state state = thrdptr->get_state();
200	thread_schedule_state state_val = state.state();
201
202	// we should only deal with pending here.
203	HPX_ASSERT(thread_schedule_state::pending == state_val);
204
205	// tries to set state to active (only if state is still
206	// the same as 'state')
207	detail::switch_status_background thrd_stat(
208	background_thread, state);	686,035✔
209
210	if (HPX_LIKELY(thrd_stat.is_valid() &&
211	thrd_stat.get_previous() == thread_schedule_state::pending))
212	{
213	#if defined(HPX_HAVE_BACKGROUND_THREAD_COUNTERS) && \
214	defined(HPX_HAVE_THREAD_IDLE_RATES)
215	// measure background work duration
216	background_work_duration_counter bg_work_duration(
217	exec_time.timer);
218	background_exec_time_wrapper bg_exec_time(bg_work_duration);
219	#endif // HPX_HAVE_BACKGROUND_THREAD_COUNTERS
220		686,035✔
221	// invoke background thread
222	thrd_stat = (*thrdptr)(context_storage);
223		686,035✔
224	if (thread_id_ref_type next = thrd_stat.move_next_thread();
225	next && next != background_thread)	×
226	{
227	if (!next_thrd)
228	{
229	next_thrd = HPX_MOVE(next);
230	}
231	else
232	{
233	auto* scheduler =	×
234	get_thread_id_data(next)->get_scheduler_base();
235	scheduler->schedule_thread(HPX_MOVE(next),
236	threads::thread_schedule_hint(
237	static_cast<std::int16_t>(num_thread)),	×
238	true);
239	scheduler->do_some_work(num_thread);
240	}
241	}	686,035✔
242	}
243	thrd_stat.store_state(state);
244	state_val = state.state();	686,035✔
245
246	if (HPX_LIKELY(state_val == thread_schedule_state::pending_boost))	1✔
247	{
248	[[maybe_unused]] auto old_state =	686,034✔
249	thrdptr->set_state(thread_schedule_state::pending);
250	}	6✔
251	else if (thread_schedule_state::terminated == state_val)
252	{
253	LTM_(debug).format(	×
254	"call_background_thread terminated: pool({}), "	×
255	"scheduler({}), worker_thread({}), thread({})",
256	scheduler_base.get_parent_pool(), scheduler_base,	6✔
257	num_thread, get_thread_id_data(background_thread));
258
259	scheduler_base.decrement_background_thread_count();	686,028✔
260	background_thread = thread_id_type();
261	}	×
262	else if (thread_schedule_state::suspended == state_val)
263	{
264	LTM_(debug).format(	×
265	"call_background_thread suspended: pool({}), "	×
266	"scheduler({}), worker_thread({}), thread({})",
267	scheduler_base.get_parent_pool(), scheduler_base,
268	num_thread, get_thread_id_data(background_thread));
269		686,035✔
270	return false;
271	}
272	}
273	return true;
274	}	686,035✔
275
276	///////////////////////////////////////////////////////////////////////////
277	bool call_and_create_background_thread(
278	thread_id_ref_type& background_thread, thread_id_ref_type& next_thrd,
279	threads::policies::scheduler_base& scheduler_base,
280	std::size_t const num_thread, background_work_exec_time& exec_time,
281	hpx::execution_base::this_thread::detail::agent_storage*
282	context_storage,
283	scheduling_callbacks const& callbacks, std::shared_ptr<bool>& running,	686,035✔
284	std::int64_t& idle_loop_count)
285	{
286	if (!call_background_thread(background_thread, next_thrd,
287	scheduler_base, num_thread, exec_time, context_storage))
288	{
289	// Let the current background thread terminate as soon as possible.
290	// No need to reschedule, as another thread will set it to pending
291	// and schedule it back eventually
292	HPX_ASSERT(background_thread);	×
293	HPX_ASSERT(running);	×
294
295	*running = false;
296	scheduler_base.decrement_background_thread_count();
297		×
298	// Create a new one that will replace the current such we avoid
299	// deadlock situations, if all background threads are blocked.
300	background_thread = create_background_thread(scheduler_base,	×
301	num_thread, callbacks, running, idle_loop_count);
302
303	return true;
304	}
305	return false;
306	}
307	} // namespace hpx::threads::detail

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