#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

83.33

/libs/core/threading_base/src/set_thread_state_timed.cpp

//  Copyright (c) 2007-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/modules/coroutines.hpp>
#include <hpx/modules/errors.hpp>
#include <hpx/modules/functional.hpp>
#include <hpx/threading_base/create_thread.hpp>
#include <hpx/threading_base/detail/get_default_timer_service.hpp>
#include <hpx/threading_base/set_thread_state_timed.hpp>
#include <hpx/threading_base/threading_base_fwd.hpp>

#if defined(WIN32) || defined(_WIN32) || defined(__WIN32__)
#include <winsock2.h>
#endif
#include <asio/basic_waitable_timer.hpp>

#include <atomic>
#include <chrono>
#include <memory>
#include <system_error>
#include <utility>

namespace hpx::threads::detail {

    ///////////////////////////////////////////////////////////////////////////
    /// This thread function is used by the at_timer thread below to trigger
    /// the required action.
    thread_result_type wake_timer_thread(thread_id_ref_type const& thrd,
        thread_schedule_state /*newstate*/,
        thread_restart_state /*newstate_ex*/, thread_priority /*priority*/,
        thread_id_type timer_id,
        std::shared_ptr<std::atomic<bool>> const& triggered,
        bool retry_on_active, thread_restart_state my_statex)
    {
        if (HPX_UNLIKELY(!thrd))
        {
            HPX_THROW_EXCEPTION(hpx::error::null_thread_id,
                "threads::detail::wake_timer_thread",
                "null thread id encountered (id)");
        }

        if (HPX_UNLIKELY(!timer_id))
        {
            HPX_THROW_EXCEPTION(hpx::error::null_thread_id,
                "threads::detail::wake_timer_thread",
                "null thread id encountered (timer_id)");
        }

        HPX_ASSERT(my_statex == thread_restart_state::abort ||
            my_statex == thread_restart_state::timeout);

        if (!triggered->load())
        {
            error_code ec(throwmode::lightweight);    // do not throw
            set_thread_state(timer_id, thread_schedule_state::pending,
                my_statex, thread_priority::boost, thread_schedule_hint(),
                retry_on_active, ec);
        }

        return {thread_schedule_state::terminated, invalid_thread_id};
    }

    // This thread function initiates the required set_state action (on behalf
    // of one of the threads#detail#set_thread_state functions).
    thread_result_type at_timer(policies::scheduler_base* scheduler,
        std::chrono::steady_clock::time_point const& abs_time,
        thread_id_ref_type const& thrd, thread_schedule_state newstate,
        thread_restart_state newstate_ex, thread_priority priority,
        std::atomic<bool>* started, bool retry_on_active)
    {
        if (HPX_UNLIKELY(!thrd))
        {
            HPX_THROW_EXCEPTION(hpx::error::null_thread_id,
                "threads::detail::at_timer", "null thread id encountered");
        }

        // create a new thread in suspended state, which will execute the
        // requested set_state when timer fires and will re-awaken this thread,
        // allowing the deadline_timer to go out of scope gracefully
        thread_id_ref_type const self_id = get_outer_self_id();    // keep alive

        std::shared_ptr<std::atomic<bool>> triggered(
            std::make_shared<std::atomic<bool>>(false));

        thread_init_data data(
            hpx::bind_front(&wake_timer_thread, thrd, newstate, newstate_ex,
                priority, self_id.noref(), triggered, retry_on_active),
            "wake_timer", priority, thread_schedule_hint(),
            thread_stacksize::small_, thread_schedule_state::suspended, true);

        thread_id_ref_type wake_id = invalid_thread_id;
        create_thread(scheduler, data, wake_id);

        // create timer firing in correspondence with given time
        using deadline_timer =
            ::asio::basic_waitable_timer<std::chrono::steady_clock>;

        deadline_timer t(get_default_timer_service(), abs_time);

        // let the timer invoke the set_state on the new (suspended) thread
        t.async_wait([wake_id = HPX_MOVE(wake_id), priority, retry_on_active](
                         std::error_code const& ec) {
            if (ec == std::make_error_code(std::errc::operation_canceled))
            {
                set_thread_state(wake_id.noref(),
                    thread_schedule_state::pending, thread_restart_state::abort,
                    priority, thread_schedule_hint(), retry_on_active, throws);
            }
            else
            {
                set_thread_state(wake_id.noref(),
                    thread_schedule_state::pending,
                    thread_restart_state::timeout, priority,
                    thread_schedule_hint(), retry_on_active, throws);
            }
        });

        if (started != nullptr)
        {
            started->store(true);
        }

        // this waits for the thread to be reactivated when the timer fired
        // if it returns signaled the timer has been canceled, otherwise
        // the timer fired and the wake_timer_thread above has been executed
        thread_restart_state const statex = get_self().yield(thread_result_type(
            thread_schedule_state::suspended, invalid_thread_id));

        HPX_ASSERT(statex == thread_restart_state::abort ||
            statex == thread_restart_state::timeout);

        if (thread_restart_state::timeout != statex)    //-V601
        {
            triggered->store(true);

            // wake_timer_thread has not been executed yet, cancel timer
            t.cancel();
        }
        else
        {
            detail::set_thread_state(
                thrd.noref(), newstate, newstate_ex, priority);
        }

        return {thread_schedule_state::terminated, invalid_thread_id};
    }

    // Set a timer to set the state of the given \a thread to the given new
    // value after it expired (at the given time)
    thread_id_ref_type set_thread_state_timed(
        policies::scheduler_base* scheduler,
        hpx::chrono::steady_time_point const& abs_time,
        thread_id_type const& thrd, thread_schedule_state newstate,
        thread_restart_state newstate_ex, thread_priority priority,
        thread_schedule_hint schedulehint, std::atomic<bool>* started,
        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 invalid_thread_id;
        }

        // this creates a new thread that creates the timer and handles the
        // requested actions
        thread_init_data data(
            hpx::bind(&at_timer, scheduler, abs_time.value(),
                thread_id_ref_type(thrd), newstate, newstate_ex, priority,
                started, retry_on_active),
            "at_timer (expire at)", priority, schedulehint,
            thread_stacksize::small_, thread_schedule_state::pending, true);

        thread_id_ref_type newid = invalid_thread_id;
        create_thread(scheduler, data, newid, ec);    //-V601
        return newid;
    }
}    // namespace hpx::threads::detail

1	// Copyright (c) 2007-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/modules/coroutines.hpp>
10	#include <hpx/modules/errors.hpp>
11	#include <hpx/modules/functional.hpp>
12	#include <hpx/threading_base/create_thread.hpp>
13	#include <hpx/threading_base/detail/get_default_timer_service.hpp>
14	#include <hpx/threading_base/set_thread_state_timed.hpp>
15	#include <hpx/threading_base/threading_base_fwd.hpp>
16
17	#if defined(WIN32) \|\| defined(_WIN32) \|\| defined(__WIN32__)
18	#include <winsock2.h>
19	#endif
20	#include <asio/basic_waitable_timer.hpp>
21
22	#include <atomic>
23	#include <chrono>
24	#include <memory>
25	#include <system_error>
26	#include <utility>
27
28	namespace hpx::threads::detail {
29
30	///////////////////////////////////////////////////////////////////////////
31	/// This thread function is used by the at_timer thread below to trigger
32	/// the required action.
33	thread_result_type wake_timer_thread(thread_id_ref_type const& thrd,
34	thread_schedule_state /newstate/,	141✔
35	thread_restart_state /newstate_ex/, thread_priority /priority/,
36	thread_id_type timer_id,
37	std::shared_ptr<std::atomic<bool>> const& triggered,
38	bool retry_on_active, thread_restart_state my_statex)
39	{
40	if (HPX_UNLIKELY(!thrd))
41	{	141✔
42	HPX_THROW_EXCEPTION(hpx::error::null_thread_id,
43	"threads::detail::wake_timer_thread",	×
44	"null thread id encountered (id)");
45	}
46
47	if (HPX_UNLIKELY(!timer_id))
48	{	141✔
49	HPX_THROW_EXCEPTION(hpx::error::null_thread_id,
50	"threads::detail::wake_timer_thread",	×
51	"null thread id encountered (timer_id)");
52	}
53
54	HPX_ASSERT(my_statex == thread_restart_state::abort \|\|
55	my_statex == thread_restart_state::timeout);
56
57	if (!triggered->load())
58	{	141✔
59	error_code ec(throwmode::lightweight); // do not throw
60	set_thread_state(timer_id, thread_schedule_state::pending,
61	my_statex, thread_priority::boost, thread_schedule_hint(),	141✔
62	retry_on_active, ec);
63	}
64
65	return {thread_schedule_state::terminated, invalid_thread_id};
66	}	141✔
67
68	// This thread function initiates the required set_state action (on behalf
69	// of one of the threads#detail#set_thread_state functions).
70	thread_result_type at_timer(policies::scheduler_base* scheduler,
71	std::chrono::steady_clock::time_point const& abs_time,	141✔
72	thread_id_ref_type const& thrd, thread_schedule_state newstate,
73	thread_restart_state newstate_ex, thread_priority priority,
74	std::atomic<bool>* started, bool retry_on_active)
75	{
76	if (HPX_UNLIKELY(!thrd))
77	{	141✔
78	HPX_THROW_EXCEPTION(hpx::error::null_thread_id,
79	"threads::detail::at_timer", "null thread id encountered");	×
80	}
81
82	// create a new thread in suspended state, which will execute the
83	// requested set_state when timer fires and will re-awaken this thread,
84	// allowing the deadline_timer to go out of scope gracefully
85	thread_id_ref_type const self_id = get_outer_self_id(); // keep alive
86		141✔
87	std::shared_ptr<std::atomic<bool>> triggered(
88	std::make_shared<std::atomic<bool>>(false));
89		141✔
90	thread_init_data data(
91	hpx::bind_front(&wake_timer_thread, thrd, newstate, newstate_ex,
92	priority, self_id.noref(), triggered, retry_on_active),	141✔
93	"wake_timer", priority, thread_schedule_hint(),	141✔
94	thread_stacksize::small_, thread_schedule_state::suspended, true);
95		141✔
96	thread_id_ref_type wake_id = invalid_thread_id;
97	create_thread(scheduler, data, wake_id);
98		141✔
99	// create timer firing in correspondence with given time
100	using deadline_timer =
101	::asio::basic_waitable_timer<std::chrono::steady_clock>;
102
103	deadline_timer t(get_default_timer_service(), abs_time);
104		141✔
105	// let the timer invoke the set_state on the new (suspended) thread
106	t.async_wait([wake_id = HPX_MOVE(wake_id), priority, retry_on_active](
107	std::error_code const& ec) {	564✔
108	if (ec == std::make_error_code(std::errc::operation_canceled))	141✔
109	{
110	set_thread_state(wake_id.noref(),
111	thread_schedule_state::pending, thread_restart_state::abort,	×
112	priority, thread_schedule_hint(), retry_on_active, throws);
113	}
114	else
115	{
116	set_thread_state(wake_id.noref(),
117	thread_schedule_state::pending,	141✔
118	thread_restart_state::timeout, priority,
119	thread_schedule_hint(), retry_on_active, throws);
120	}
121	});
122		141✔
123	if (started != nullptr)
124	{	141✔
125	started->store(true);
126	}
127
128	// this waits for the thread to be reactivated when the timer fired
129	// if it returns signaled the timer has been canceled, otherwise
130	// the timer fired and the wake_timer_thread above has been executed
131	thread_restart_state const statex = get_self().yield(thread_result_type(
132	thread_schedule_state::suspended, invalid_thread_id));	141✔
133
134	HPX_ASSERT(statex == thread_restart_state::abort \|\|
135	statex == thread_restart_state::timeout);
136
137	if (thread_restart_state::timeout != statex) //-V601
138	{	141✔
139	triggered->store(true);
140
141	// wake_timer_thread has not been executed yet, cancel timer
142	t.cancel();
143	}	×
144	else
145	{
146	detail::set_thread_state(
147	thrd.noref(), newstate, newstate_ex, priority);	141✔
148	}	141✔
149
150	return {thread_schedule_state::terminated, invalid_thread_id};
151	}	141✔
152
153	// Set a timer to set the state of the given \a thread to the given new
154	// value after it expired (at the given time)
155	thread_id_ref_type set_thread_state_timed(
156	policies::scheduler_base* scheduler,	141✔
157	hpx::chrono::steady_time_point const& abs_time,
158	thread_id_type const& thrd, thread_schedule_state newstate,
159	thread_restart_state newstate_ex, thread_priority priority,
160	thread_schedule_hint schedulehint, std::atomic<bool>* started,
161	bool retry_on_active, error_code& ec)
162	{
163	if (HPX_UNLIKELY(!thrd))
164	{	141✔
165	HPX_THROWS_IF(ec, hpx::error::null_thread_id,
166	"threads::detail::set_thread_state",	×
167	"null thread id encountered");
168	return invalid_thread_id;
169	}
170
171	// this creates a new thread that creates the timer and handles the
172	// requested actions
173	thread_init_data data(
174	hpx::bind(&at_timer, scheduler, abs_time.value(),
175	thread_id_ref_type(thrd), newstate, newstate_ex, priority,	141✔
176	started, retry_on_active),
177	"at_timer (expire at)", priority, schedulehint,
178	thread_stacksize::small_, thread_schedule_state::pending, true);
179		141✔
180	thread_id_ref_type newid = invalid_thread_id;
181	create_thread(scheduler, data, newid, ec); //-V601
182	return newid;	141✔
183	}
184	} // 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