#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

28.77

/libs/core/threading_base/src/set_thread_state.cpp

//  Copyright (c) 2007-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/errors.hpp>
#include <hpx/modules/format.hpp>
#include <hpx/modules/functional.hpp>
#include <hpx/modules/logging.hpp>
#include <hpx/threading_base/create_work.hpp>
#include <hpx/threading_base/register_thread.hpp>
#include <hpx/threading_base/scheduler_base.hpp>
#include <hpx/threading_base/set_thread_state.hpp>
#include <hpx/threading_base/thread_data.hpp>
#include <hpx/threading_base/threading_base_fwd.hpp>

#include <cstddef>
#include <cstdint>
#include <string>
#include <utility>

namespace hpx::threads::detail {

    ///////////////////////////////////////////////////////////////////////////
    namespace {

        thread_result_type set_active_state(thread_id_ref_type const& thrd,
            thread_schedule_state const newstate,
            thread_restart_state const newstate_ex,
            thread_priority const priority, thread_state const previous_state,
            thread_schedule_hint const schedulehint)
        {
            if (HPX_UNLIKELY(!thrd))
            {
                HPX_THROW_EXCEPTION(hpx::error::null_thread_id,
                    "threads::detail::set_active_state",
                    "null thread id encountered");
            }

            // make sure that the thread has not been suspended and set active
            // again in the meantime
            thread_state const current_state =
                get_thread_id_data(thrd)->get_state();

            if (current_state.state() == previous_state.state() &&
                current_state != previous_state)
            {
                LTM_(warning).format(
                    "set_active_state: thread is still active, however it was "
                    "non-active since the original set_state request was "
                    "issued, aborting state change, thread({}), "
                    "description({}), new state({})",
                    thrd, get_thread_id_data(thrd)->get_description(),
                    get_thread_state_name(newstate));
                return {thread_schedule_state::terminated, invalid_thread_id};
            }

            // Just retry, set_state will create new thread if target is still
            // active.
            error_code ec(throwmode::lightweight);    // do not throw
            detail::set_thread_state(thrd.noref(), newstate, newstate_ex,
                priority, schedulehint, true, ec);

            return {thread_schedule_state::terminated, invalid_thread_id};
        }

        thread_state reschedule_active(thread_id_type const& thrd,
            thread_schedule_state new_state, thread_restart_state new_state_ex,
            thread_priority priority, thread_state previous_state,
            thread_schedule_hint schedulehint, error_code& ec)
        {
            // schedule a new thread to set the state
            LTM_(warning).format(
                "set_thread_state: thread is currently active, "
                "scheduling new thread, thread({}), description({}), "
                "new state({})",
                thrd, get_thread_id_data(thrd)->get_description(),
                get_thread_state_name(new_state));

            thread_init_data data(
                hpx::bind(&set_active_state, thread_id_ref_type(thrd),
                    new_state, new_state_ex, priority, previous_state,
                    schedulehint),
                "set state for active thread", priority, schedulehint);

            create_work(
                get_thread_id_data(thrd)->get_scheduler_base(), data, ec);

            if (&ec != &throws)
                ec = make_success_code();

            return previous_state;    // done
        }
    }    // namespace

    ///////////////////////////////////////////////////////////////////////////
    thread_state set_thread_state(thread_id_type const& thrd,
        thread_schedule_state const new_state,
        thread_restart_state const new_state_ex, thread_priority const priority,
        thread_schedule_hint schedulehint, bool retry_on_active, error_code& ec)
    {
        if (HPX_UNLIKELY(!thrd))
        {
            HPX_THROWS_IF(ec, hpx::error::null_thread_id,
                "threads::detail::set_thread_state",
                "null thread id encountered");
            return thread_state{
                thread_schedule_state::unknown, thread_restart_state::unknown};
        }

        // set_state can't be used to force a thread into active state
        if (new_state == thread_schedule_state::active)
        {
            HPX_THROWS_IF(ec, hpx::error::bad_parameter,
                "threads::detail::set_thread_state", "invalid new state: {}",
                new_state);
            return thread_state{
                thread_schedule_state::unknown, thread_restart_state::unknown};
        }

        thread_state previous_state;
        std::size_t k = 0;
        do
        {
            // action depends on the current state
            previous_state = get_thread_id_data(thrd)->get_state();
            thread_schedule_state previous_state_val = previous_state.state();

            // nothing to do here if the state doesn't change
            if (new_state == previous_state_val)
            {
                LTM_(warning).format(
                    "set_thread_state: old thread state is the same as new "
                    "thread state, aborting state change, thread({}), "
                    "description({}), new state({})",
                    thrd, get_thread_id_data(thrd)->get_description(),
                    get_thread_state_name(new_state));

                if (&ec != &throws)
                    ec = make_success_code();

                return thread_state{new_state, previous_state.state_ex()};
            }

            // the thread to set the state for is currently running, so we
            // schedule another thread to execute the pending set_state
            switch (previous_state_val)
            {
            case thread_schedule_state::active:
            {
                if (retry_on_active)
                {
                    return reschedule_active(thrd, new_state, new_state_ex,
                        priority, previous_state, schedulehint, ec);
                }

                if (hpx::execution_base::this_thread::yield_k(
                        k, "hpx::threads::detail::set_thread_state"))
                {
                    retry_on_active = true;    // don't wait too long
                }
                ++k;

                LTM_(warning).format(
                    "set_thread_state: thread is currently active, but not "
                    "scheduling new thread because retry_on_active = "
                    "false, thread({}), description({}), new state({})",
                    thrd, get_thread_id_data(thrd)->get_description(),
                    get_thread_state_name(new_state));

                continue;
            }
            case thread_schedule_state::deleted:
                [[fallthrough]];
            case thread_schedule_state::terminated:
            {
                LTM_(warning).format(
                    "set_thread_state: thread is terminated, aborting state "
                    "change, thread({}), description({}), new state({})",
                    thrd, get_thread_id_data(thrd)->get_description(),
                    get_thread_state_name(new_state));

                if (&ec != &throws)
                    ec = make_success_code();

                // If the thread has been terminated while this set_state was
                // pending nothing has to be done anymore.
                return previous_state;
            }

            case thread_schedule_state::pending:
                [[fallthrough]];
            case thread_schedule_state::pending_boost:
                if (thread_schedule_state::suspended == new_state)
                {
                    // we do not allow explicit resetting of a state to suspended
                    // without the thread being executed.
                    std::string str = hpx::util::format(
                        "set_thread_state: invalid new state, can't demote a "
                        "pending thread, thread({}), description({}), new "
                        "state({})",
                        thrd, get_thread_id_data(thrd)->get_description(),
                        new_state);

                    LTM_(fatal) << str;

                    HPX_THROWS_IF(ec, hpx::error::bad_parameter,
                        "threads::detail::set_thread_state", str);
                    return thread_state{thread_schedule_state::unknown,
                        thread_restart_state::unknown};
                }
                break;

            case thread_schedule_state::suspended:
                break;    // fine, just set the new state

            case thread_schedule_state::unknown:
            case thread_schedule_state::depleted:
            case thread_schedule_state::staged:
            case thread_schedule_state::pending_do_not_schedule:
                [[fallthrough]];
            default:
            {
                HPX_ASSERT_MSG(false,
                    hpx::util::format("set_thread_state: previous state was {}",
                        previous_state_val));    // should not happen
                break;
            }
            }

            // If the previous state was pending we are supposed to remove the
            // thread from the queue. But in order to avoid linearly looking
            // through the queue we defer this to the thread function, which at
            // some point will ignore this thread by simply skipping it (if it's
            // not pending anymore).

            LTM_(info).format("set_thread_state: thread({}), description({}), "
                              "new state({}), old state({})",
                thrd, get_thread_id_data(thrd)->get_description(),
                get_thread_state_name(new_state),
                get_thread_state_name(previous_state_val));

            // So all what we do here is to set the new state.
            if (get_thread_id_data(thrd)->restore_state(
                    new_state, new_state_ex, previous_state))
            {
                break;
            }

            // state has changed since we fetched it from the thread, retry
            LTM_(warning).format(
                "set_thread_state: state has been changed since it was "
                "fetched, retrying, thread({}), description({}), new "
                "state({}), old state({})",
                get_thread_id_data(thrd),
                get_thread_id_data(thrd)->get_description(),
                get_thread_state_name(new_state),
                get_thread_state_name(previous_state_val));
        } while (true);

        thread_schedule_state const previous_state_val = previous_state.state();
        if (!(previous_state_val == thread_schedule_state::pending ||
                previous_state_val == thread_schedule_state::pending_boost) &&
            (new_state == thread_schedule_state::pending ||
                new_state == thread_schedule_state::pending_boost))
        {
            auto* thrd_data = get_thread_id_data(thrd);
            auto* scheduler = thrd_data->get_scheduler_base();
            if (auto const current_priority = thrd_data->get_priority();
                current_priority == thread_priority::bound)
            {
                schedulehint.mode = thread_schedule_hint_mode::thread;
                schedulehint.hint = static_cast<std::int16_t>(
                    thrd_data->get_last_worker_thread_num());
                scheduler->schedule_thread(
                    thrd, schedulehint, false, thread_priority::bound);
            }
            else if (priority == thread_priority::bound)
            {
                thrd_data->set_priority(thread_priority::bound);
                scheduler->schedule_thread(
                    thrd, schedulehint, false, thread_priority::bound);
            }
            else
            {
                scheduler->schedule_thread(
                    thrd, schedulehint, false, current_priority);
            }

            scheduler->do_some_work(schedulehint.mode ==
                        hpx::threads::thread_schedule_hint_mode::numa ?
                    static_cast<std::size_t>(-1) :
                    schedulehint.hint);
        }

        if (&ec != &throws)
            ec = make_success_code();

        return previous_state;
    }
}    // namespace hpx::threads::detail

1	// Copyright (c) 2007-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/errors.hpp>
10	#include <hpx/modules/format.hpp>
11	#include <hpx/modules/functional.hpp>
12	#include <hpx/modules/logging.hpp>
13	#include <hpx/threading_base/create_work.hpp>
14	#include <hpx/threading_base/register_thread.hpp>
15	#include <hpx/threading_base/scheduler_base.hpp>
16	#include <hpx/threading_base/set_thread_state.hpp>
17	#include <hpx/threading_base/thread_data.hpp>
18	#include <hpx/threading_base/threading_base_fwd.hpp>
19
20	#include <cstddef>
21	#include <cstdint>
22	#include <string>
23	#include <utility>
24
25	namespace hpx::threads::detail {
26
27	///////////////////////////////////////////////////////////////////////////	×
28	namespace {
29
30	thread_result_type set_active_state(thread_id_ref_type const& thrd,
31	thread_schedule_state const newstate,	×
32	thread_restart_state const newstate_ex,
33	thread_priority const priority, thread_state const previous_state,	×
34	thread_schedule_hint const schedulehint)
35	{
36	if (HPX_UNLIKELY(!thrd))
37	{
38	HPX_THROW_EXCEPTION(hpx::error::null_thread_id,
39	"threads::detail::set_active_state",
40	"null thread id encountered");
41	}
42
43	// make sure that the thread has not been suspended and set active	×
44	// again in the meantime
45	thread_state const current_state =
46	get_thread_id_data(thrd)->get_state();	×
47
48	if (current_state.state() == previous_state.state() &&
49	current_state != previous_state)
50	{
51	LTM_(warning).format(	×
52	"set_active_state: thread is still active, however it was "	×
53	"non-active since the original set_state request was "
54	"issued, aborting state change, thread({}), "
55	"description({}), new state({})",
56	thrd, get_thread_id_data(thrd)->get_description(),
57	get_thread_state_name(newstate));
58	return {thread_schedule_state::terminated, invalid_thread_id};	×
59	}
60
61	// Just retry, set_state will create new thread if target is still
62	// active.
63	error_code ec(throwmode::lightweight); // do not throw
64	detail::set_thread_state(thrd.noref(), newstate, newstate_ex,
65	priority, schedulehint, true, ec);	28,247✔
66
67	return {thread_schedule_state::terminated, invalid_thread_id};
68	}
69
70	thread_state reschedule_active(thread_id_type const& thrd,	28,247✔
71	thread_schedule_state new_state, thread_restart_state new_state_ex,
72	thread_priority priority, thread_state previous_state,	×
73	thread_schedule_hint schedulehint, error_code& ec)
74	{
75	// schedule a new thread to set the state
76	LTM_(warning).format(	×
77	"set_thread_state: thread is currently active, "
78	"scheduling new thread, thread({}), description({}), "
79	"new state({})",
80	thrd, get_thread_id_data(thrd)->get_description(),	28,247✔
81	get_thread_state_name(new_state));
82		×
83	thread_init_data data(
84	hpx::bind(&set_active_state, thread_id_ref_type(thrd),
85	new_state, new_state_ex, priority, previous_state,
86	schedulehint),	×
87	"set state for active thread", priority, schedulehint);
88
89	create_work(
90	get_thread_id_data(thrd)->get_scheduler_base(), data, ec);
91
92	if (&ec != &throws)
93	ec = make_success_code();
94
95	return previous_state; // done
96	}
97	} // namespace
98		28,263✔
99	///////////////////////////////////////////////////////////////////////////
100	thread_state set_thread_state(thread_id_type const& thrd,	×
101	thread_schedule_state const new_state,
102	thread_restart_state const new_state_ex, thread_priority const priority,
103	thread_schedule_hint schedulehint, bool retry_on_active, error_code& ec)
104	{	×
105	if (HPX_UNLIKELY(!thrd))	×
106	{
107	HPX_THROWS_IF(ec, hpx::error::null_thread_id,	×
108	"threads::detail::set_thread_state",	×
109	"null thread id encountered");
110	return thread_state{	×
111	thread_schedule_state::unknown, thread_restart_state::unknown};
112	}
113
114	// set_state can't be used to force a thread into active state
115	if (new_state == thread_schedule_state::active)	28,263✔
116	{
117	HPX_THROWS_IF(ec, hpx::error::bad_parameter,	16✔
118	"threads::detail::set_thread_state", "invalid new state: {}",
119	new_state);	16✔
120	return thread_state{
121	thread_schedule_state::unknown, thread_restart_state::unknown};
122	}	×
123
124	thread_state previous_state;
125	std::size_t k = 0;
126	do	×
127	{	×
128	// action depends on the current state
129	previous_state = get_thread_id_data(thrd)->get_state();
130	thread_schedule_state previous_state_val = previous_state.state();	×
131
132	// nothing to do here if the state doesn't change
133	if (new_state == previous_state_val)
134	{	×
135	LTM_(warning).format(
136	"set_thread_state: old thread state is the same as new "
137	"thread state, aborting state change, thread({}), "	×
138	"description({}), new state({})",	×
139	thrd, get_thread_id_data(thrd)->get_description(),
140	get_thread_state_name(new_state));
141
142	if (&ec != &throws)	16✔
143	ec = make_success_code();
144		16✔
145	return thread_state{new_state, previous_state.state_ex()};
146	}	16✔
147
148	// the thread to set the state for is currently running, so we
149	// schedule another thread to execute the pending set_state
150	switch (previous_state_val)	×
151	{	×
152	case thread_schedule_state::active:
153	{	16✔
154	if (retry_on_active)
155	{
156	return reschedule_active(thrd, new_state, new_state_ex,	×
157	priority, previous_state, schedulehint, ec);	×
158	}
159		×
160	if (hpx::execution_base::this_thread::yield_k(
161	k, "hpx::threads::detail::set_thread_state"))	×
162	{
163	retry_on_active = true; // don't wait too long
164	}
165	++k;	×
166
167	LTM_(warning).format(
168	"set_thread_state: thread is currently active, but not "	×
169	"scheduling new thread because retry_on_active = "	×
170	"false, thread({}), description({}), new state({})",
171	thrd, get_thread_id_data(thrd)->get_description(),	×
172	get_thread_state_name(new_state));	×
173
174	continue;
175	}
176	case thread_schedule_state::deleted:	×
177	[[fallthrough]];
178	case thread_schedule_state::terminated:
179	{	×
180	LTM_(warning).format(
181	"set_thread_state: thread is terminated, aborting state "
182	"change, thread({}), description({}), new state({})",	×
183	thrd, get_thread_id_data(thrd)->get_description(),
184	get_thread_state_name(new_state));
185
186	if (&ec != &throws)
187	ec = make_success_code();
188
189	// If the thread has been terminated while this set_state was
190	// pending nothing has to be done anymore.	×
191	return previous_state;	×
192	}
193		×
194	case thread_schedule_state::pending:
195	[[fallthrough]];	×
196	case thread_schedule_state::pending_boost:
197	if (thread_schedule_state::suspended == new_state)
198	{	×
199	// we do not allow explicit resetting of a state to suspended
200	// without the thread being executed.
201	std::string str = hpx::util::format(
202	"set_thread_state: invalid new state, can't demote a "
203	"pending thread, thread({}), description({}), new "
204	"state({})",
205	thrd, get_thread_id_data(thrd)->get_description(),
206	new_state);
207
208	LTM_(fatal) << str;
209
210	HPX_THROWS_IF(ec, hpx::error::bad_parameter,
211	"threads::detail::set_thread_state", str);
212	return thread_state{thread_schedule_state::unknown,
213	thread_restart_state::unknown};
214	}
215	break;
216
217	case thread_schedule_state::suspended:
218	break; // fine, just set the new state
219
220	case thread_schedule_state::unknown:
221	case thread_schedule_state::depleted:
222	case thread_schedule_state::staged:
223	case thread_schedule_state::pending_do_not_schedule:
224	[[fallthrough]];
225	default:	28,247✔
226	{
227	HPX_ASSERT_MSG(false,	×
228	hpx::util::format("set_thread_state: previous state was {}",	×
229	previous_state_val)); // should not happen	×
230	break;
231	}
232	}	28,247✔
233
234	// If the previous state was pending we are supposed to remove the
235	// thread from the queue. But in order to avoid linearly looking
236	// through the queue we defer this to the thread function, which at
237	// some point will ignore this thread by simply skipping it (if it's
238	// not pending anymore).
239		×
240	LTM_(info).format("set_thread_state: thread({}), description({}), "
241	"new state({}), old state({})",
242	thrd, get_thread_id_data(thrd)->get_description(),
243	get_thread_state_name(new_state),	×
244	get_thread_state_name(previous_state_val));	×
245		×
246	// So all what we do here is to set the new state.	×
247	if (get_thread_id_data(thrd)->restore_state(
248	new_state, new_state_ex, previous_state))
249	{
250	break;	28,247✔
251	}	28,247✔
252		28,247✔
253	// state has changed since we fetched it from the thread, retry
254	LTM_(warning).format(
255	"set_thread_state: state has been changed since it was "
256	"fetched, retrying, thread({}), description({}), new "
257	"state({}), old state({})",
258	get_thread_id_data(thrd),
259	get_thread_id_data(thrd)->get_description(),
260	get_thread_state_name(new_state),	28,247✔
261	get_thread_state_name(previous_state_val));
262	} while (true);
263
264	thread_schedule_state const previous_state_val = previous_state.state();
265	if (!(previous_state_val == thread_schedule_state::pending \|\|	28,247✔
266	previous_state_val == thread_schedule_state::pending_boost) &&
267	(new_state == thread_schedule_state::pending \|\|
268	new_state == thread_schedule_state::pending_boost))	28,247✔
269	{	282✔
270	auto* thrd_data = get_thread_id_data(thrd);
271	auto* scheduler = thrd_data->get_scheduler_base();	28,247✔
272	if (auto const current_priority = thrd_data->get_priority();
273	current_priority == thread_priority::bound)
274	{
275	schedulehint.mode = thread_schedule_hint_mode::thread;
276	schedulehint.hint = static_cast<std::int16_t>(
277	thrd_data->get_last_worker_thread_num());
278	scheduler->schedule_thread(
279	thrd, schedulehint, false, thread_priority::bound);
280	}
281	else if (priority == thread_priority::bound)
282	{
283	thrd_data->set_priority(thread_priority::bound);
284	scheduler->schedule_thread(
285	thrd, schedulehint, false, thread_priority::bound);
286	}
287	else
288	{
289	scheduler->schedule_thread(
290	thrd, schedulehint, false, current_priority);
291	}
292
293	scheduler->do_some_work(schedulehint.mode ==
294	hpx::threads::thread_schedule_hint_mode::numa ?
295	static_cast<std::size_t>(-1) :
296	schedulehint.hint);
297	}
298
299	if (&ec != &throws)
300	ec = make_success_code();
301
302	return previous_state;
303	}
304	} // 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