#871

Committed 22 Jan 2023 11:22PM UTC coverage: 86.624% (+0.7%) from 85.97%

Build # #871

Build Type

push

Committed by StellarBot

Commit Message

Merge #6144

6144: General improvements to scheduling and related fixes r=hkaiser a=hkaiser

This is a collection of unrelated improvements applied to different parts of the code

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

Run Details

152 of 152 new or added lines in 23 files covered. (100.0%)

174953 of 201969 relevant lines covered (86.62%)

1838882.76 hits per line

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

79.28

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

//  Copyright (c) 2023 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/execution_base/this_thread.hpp>
#include <hpx/thread_pools/detail/background_thread.hpp>
#include <hpx/thread_pools/detail/scheduling_callbacks.hpp>
#include <hpx/thread_pools/detail/scheduling_counters.hpp>
#include <hpx/thread_pools/detail/scoped_background_timer.hpp>
#include <hpx/threading_base/scheduler_base.hpp>
#include <hpx/threading_base/thread_data.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 num_thread, scheduling_callbacks& callbacks,
        std::shared_ptr<bool>& background_running,
        std::int64_t& idle_loop_count)
    {
        threads::thread_schedule_hint 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_result_type(
                    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();

        // We can now set the state to pending
        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 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()
        {
            if (need_restore_state_)
            {
                store_state(prev_state_);
            }
        }

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

        // allow to change the state the thread will be switched to after
        // execution
        thread_state 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 prev_state_;
        }

        // Get the state this thread was in before execution (usually pending),
        // this helps making sure no other worker-thread is started to execute
        // this HPX-thread in the meantime.
        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;
        }

        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 num_thread,
        [[maybe_unused]] background_work_exec_time& exec_time,
        hpx::execution_base::this_thread::detail::agent_storage*
            context_storage)
    {
        if (HPX_UNLIKELY(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);

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

STEllAR-GROUP / hpx / #871

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