#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

50.0

/libs/core/execution_base/src/this_thread.cpp

//  Copyright (c) 2013-2019 Thomas Heller
//  Copyright (c) 2008 Peter Dimov
//  Copyright (c) 2018-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/execution_base/agent_base.hpp>
#include <hpx/execution_base/context_base.hpp>
#include <hpx/execution_base/this_thread.hpp>
#include <hpx/modules/coroutines.hpp>
#include <hpx/modules/errors.hpp>
#include <hpx/modules/format.hpp>
#include <hpx/modules/timing.hpp>

#include <condition_variable>
#include <cstddef>
#include <cstdint>
#include <mutex>
#include <string>
#include <thread>
#include <utility>

#if defined(HPX_WINDOWS)
#include <ctime>
#include <windows.h>
#else
#ifndef _AIX
#include <sched.h>
#else
// AIX's sched.h defines ::var which sometimes conflicts with Lambda's var
extern "C" int sched_yield(void);
#endif
#include <time.h>
#endif

namespace hpx::execution_base {

    namespace {

#if defined(HPX_WINDOWS)
        // Number of performance counter increments per nanosecond, or zero if
        // it could not be determined.
        struct ticks_per_nanosecond
        {
            ticks_per_nanosecond()
            {
                LARGE_INTEGER freq;
                if (QueryPerformanceFrequency(&freq))
                    ticks = static_cast<double>(freq.QuadPart) / 1e9;
            }

            double ticks = 0.0;
        };
        ticks_per_nanosecond ticks;

        // our own (crude) implementation of nanosleep
        int win_nanosleep(std::timespec const& delay)
        {
            if (delay.tv_nsec < 0 || delay.tv_nsec >= 999999999)
            {
                return -1;
            }

            // for small delays we busy-wait
            if (delay.tv_sec == 0 && ticks.ticks != 0.0)
            {
                // compensate for fluctuations introduced by Sleep()
                auto const sleep_for = delay.tv_nsec / 1000000 - 10;

                // overall number of ticks to delay
                auto const ticks_to_delay = static_cast<long>(
                    static_cast<double>(delay.tv_nsec) * ticks.ticks);

                LARGE_INTEGER counter;
                if (QueryPerformanceCounter(&counter))
                {
                    // wait until the performance counter has reached this value
                    auto const wait_until = counter.QuadPart + ticks_to_delay;

                    // use Sleep() if appropriate
                    if (sleep_for > 0)
                    {
                        Sleep(sleep_for);
                    }

                    // simply busy-wait for the remaining amount of time, don't
                    // wait if delay is zero
                    while (counter.QuadPart < wait_until &&
                        QueryPerformanceCounter(&counter))
                    {
                    }
                    return 0;
                }
            }

            // Fallback and longer delays
            Sleep(static_cast<long>(delay.tv_sec) * 1000 +
                delay.tv_nsec / 1000000);

            return 0;
        }

        constexpr std::timespec wait_0ns = {.tv_sec = 0, .tv_nsec = 0};
        constexpr std::timespec wait_1000ns = {.tv_sec = 0, .tv_nsec = 1000};
#endif

        ///////////////////////////////////////////////////////////////////////
        struct default_context final : execution_base::context_base
        {
            [[nodiscard]] resource_base const& resource()
                const noexcept override
            {
                return resource_;
            }

            resource_base resource_;
        };

        struct default_agent : execution_base::agent_base
        {
            default_agent();

            [[nodiscard]] std::string description() const override
            {
                return hpx::util::format("{}", id_);
            }

            [[nodiscard]] default_context const& context()
                const noexcept override
            {
                return context_;
            }

            void yield(char const* desc) override;
            bool yield_k(std::size_t k, char const* desc) override;
            void suspend(char const* desc) override;
            void resume(hpx::threads::thread_priority priority,
                char const* desc) override;
            void abort(char const* desc) override;
            void sleep_for(hpx::chrono::steady_duration const& sleep_duration,
                char const* desc) override;
            void sleep_until(hpx::chrono::steady_time_point const& sleep_time,
                char const* desc) override;

        private:
            bool running_;
            bool aborted_;
            std::thread::id id_;
            std::mutex mtx_;
            std::condition_variable suspend_cv_;
            std::condition_variable resume_cv_;

            default_context context_;
        };

        default_agent::default_agent()
          : running_(true)
          , aborted_(false)
          , id_(std::this_thread::get_id())
        {
        }

        void default_agent::yield(char const* /* desc */)
        {
#if defined(HPX_WINDOWS)
            win_nanosleep(wait_0ns);
#else
            sched_yield();
#endif
        }

        bool default_agent::yield_k(std::size_t k, char const* /* desc */)
        {
            if (k < 4)    //-V112
            {
                return false;
            }
            else if (k < 16)
            {
                HPX_SMT_PAUSE;
                return false;
            }
            else if (k < 32 || k & 1)    //-V112
            {
#if defined(HPX_WINDOWS)
                win_nanosleep(wait_0ns);
                return true;
#else
                sched_yield();
                return true;
#endif
            }
            else
            {
#if defined(HPX_WINDOWS)
                win_nanosleep(wait_1000ns);
                return true;
#else
                // g++ -Wextra warns on {} or {0}
                struct timespec rqtp = {0, 0};

                // POSIX says that timespec has tv_sec and tv_nsec
                // But it doesn't guarantee order or placement

                rqtp.tv_sec = 0;
                rqtp.tv_nsec = 1000;

                nanosleep(&rqtp, nullptr);
                return true;
#endif
            }
        }

        void default_agent::suspend(char const* /* desc */)
        {
            std::unique_lock<std::mutex> l(mtx_);
            HPX_ASSERT(running_);

            running_ = false;
            resume_cv_.notify_all();

            while (!running_)
            {
                suspend_cv_.wait(l);    //-V1089
            }

            if (aborted_)
            {
                HPX_THROW_EXCEPTION(hpx::error::yield_aborted, "suspend",
                    "std::thread({}) aborted (yield returned wait_abort)", id_);
            }
        }

        void default_agent::resume(hpx::threads::thread_priority, char const*)
        {
            {
                std::unique_lock<std::mutex> l(mtx_);
                while (running_)
                {
                    resume_cv_.wait(l);    //-V1089
                }
                running_ = true;
            }
            suspend_cv_.notify_one();
        }

        void default_agent::abort(char const* /* desc */)
        {
            {
                std::unique_lock<std::mutex> l(mtx_);
                while (running_)
                {
                    resume_cv_.wait(l);    //-V1089
                }
                running_ = true;
                aborted_ = true;
            }
            suspend_cv_.notify_one();
        }

        void default_agent::sleep_for(
            hpx::chrono::steady_duration const& sleep_duration,
            char const* /* desc */)
        {
            std::this_thread::sleep_for(sleep_duration.value());
        }

        void default_agent::sleep_until(
            hpx::chrono::steady_time_point const& sleep_time,
            char const* /* desc */)
        {
            std::this_thread::sleep_until(sleep_time.value());
        }
    }    // namespace

    namespace detail {

        [[nodiscard]] agent_base& get_default_agent()
        {
            static thread_local default_agent agent;
            return agent;
        }
    }    // namespace detail

    namespace this_thread {

        namespace detail {

            struct agent_storage
            {
                agent_storage()
                  : impl_(&hpx::execution_base::detail::get_default_agent())
                {
                }

                agent_base* set(agent_base* context) noexcept
                {
                    std::swap(context, impl_);
                    return context;
                }

                agent_base* impl_;
            };

            [[nodiscard]] agent_storage* get_agent_storage()
            {
                static thread_local agent_storage storage;
                return &storage;
            }
        }    // namespace detail

        reset_agent::reset_agent(
            detail::agent_storage* storage, agent_base& impl)
          : storage_(storage)
          , old_(storage_->set(&impl))
        {
        }

        reset_agent::reset_agent(agent_base& impl)
          : reset_agent(detail::get_agent_storage(), impl)
        {
        }

        reset_agent::~reset_agent()
        {
            storage_->set(old_);
        }

        [[nodiscard]] agent_ref agent()
        {
            return agent_ref(detail::get_agent_storage()->impl_);
        }

        void yield(char const* desc)
        {
            agent().yield(desc);
        }

        bool yield_k(std::size_t k, char const* desc)
        {
            return agent().yield_k(k, desc);
        }

        void suspend(char const* desc)
        {
            agent().suspend(desc);
        }
    }    // namespace this_thread
}    // namespace hpx::execution_base

1	// Copyright (c) 2013-2019 Thomas Heller
2	// Copyright (c) 2008 Peter Dimov
3	// Copyright (c) 2018-2025 Hartmut Kaiser
4	//
5	// SPDX-License-Identifier: BSL-1.0
6	// Distributed under the Boost Software License, Version 1.0. (See accompanying
7	// file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
8
9	#include <hpx/config.hpp>
10	#include <hpx/assert.hpp>
11	#include <hpx/execution_base/agent_base.hpp>
12	#include <hpx/execution_base/context_base.hpp>
13	#include <hpx/execution_base/this_thread.hpp>
14	#include <hpx/modules/coroutines.hpp>
15	#include <hpx/modules/errors.hpp>
16	#include <hpx/modules/format.hpp>
17	#include <hpx/modules/timing.hpp>
18
19	#include <condition_variable>
20	#include <cstddef>
21	#include <cstdint>
22	#include <mutex>
23	#include <string>
24	#include <thread>
25	#include <utility>
26
27	#if defined(HPX_WINDOWS)
28	#include <ctime>
29	#include <windows.h>
30	#else
31	#ifndef _AIX
32	#include <sched.h>
33	#else
34	// AIX's sched.h defines ::var which sometimes conflicts with Lambda's var
35	extern "C" int sched_yield(void);
36	#endif
37	#include <time.h>
38	#endif
39
40	namespace hpx::execution_base {
41
42	namespace {	248✔
43
44	#if defined(HPX_WINDOWS)	×
45	// Number of performance counter increments per nanosecond, or zero if
46	// it could not be determined.	×
47	struct ticks_per_nanosecond
48	{
49	ticks_per_nanosecond()
50	{
51	LARGE_INTEGER freq;
52	if (QueryPerformanceFrequency(&freq))
53	ticks = static_cast<double>(freq.QuadPart) / 1e9;
54	}
55		×
56	double ticks = 0.0;
57	};	×
58	ticks_per_nanosecond ticks;
59
60	// our own (crude) implementation of nanosleep	×
61	int win_nanosleep(std::timespec const& delay)
62	{
63	if (delay.tv_nsec < 0 \|\| delay.tv_nsec >= 999999999)	×
64	{
65	return -1;
66	}
67
68	// for small delays we busy-wait
69	if (delay.tv_sec == 0 && ticks.ticks != 0.0)
70	{
71	// compensate for fluctuations introduced by Sleep()
72	auto const sleep_for = delay.tv_nsec / 1000000 - 10;
73
74	// overall number of ticks to delay
75	auto const ticks_to_delay = static_cast<long>(
76	static_cast<double>(delay.tv_nsec) * ticks.ticks);
77
78	LARGE_INTEGER counter;
79	if (QueryPerformanceCounter(&counter))
80	{
81	// wait until the performance counter has reached this value
82	auto const wait_until = counter.QuadPart + ticks_to_delay;
83
84	// use Sleep() if appropriate
85	if (sleep_for > 0)
86	{
87	Sleep(sleep_for);	248✔
88	}	248✔
89		248✔
90	// simply busy-wait for the remaining amount of time, don't	248✔
91	// wait if delay is zero
92	while (counter.QuadPart < wait_until &&	248✔
93	QueryPerformanceCounter(&counter))
94	{	33✔
95	}
96	return 0;
97	}
98	}
99		33✔
100	// Fallback and longer delays
101	Sleep(static_cast<long>(delay.tv_sec) * 1000 +	33✔
102	delay.tv_nsec / 1000000);
103		33,246✔
104	return 0;
105	}	33,246✔
106
107	constexpr std::timespec wait_0ns = {.tv_sec = 0, .tv_nsec = 0};
108	constexpr std::timespec wait_1000ns = {.tv_sec = 0, .tv_nsec = 1000};	32,600✔
109	#endif
110		1,837✔
111	///////////////////////////////////////////////////////////////////////
112	struct default_context final : execution_base::context_base	30,763✔
113	{
114	[[nodiscard]] resource_base const& resource()
115	const noexcept override
116	{
117	return resource_;	16,201✔
118	}
119
120	resource_base resource_;
121	};
122
123	struct default_agent : execution_base::agent_base
124	{
125	default_agent();
126		14,562✔
127	[[nodiscard]] std::string description() const override
128	{
129	return hpx::util::format("{}", id_);
130	}
131
132	[[nodiscard]] default_context const& context()	14,562✔
133	const noexcept override
134	{	14,562✔
135	return context_;
136	}
137		33,246✔
138	void yield(char const* desc) override;
139	bool yield_k(std::size_t k, char const* desc) override;	×
140	void suspend(char const* desc) override;
141	void resume(hpx::threads::thread_priority priority,	×
142	char const* desc) override;
143	void abort(char const* desc) override;
144	void sleep_for(hpx::chrono::steady_duration const& sleep_duration,	×
145	char const* desc) override;	×
146	void sleep_until(hpx::chrono::steady_time_point const& sleep_time,
147	char const* desc) override;	×
148
149	private:	×
150	bool running_;
151	bool aborted_;
152	std::thread::id id_;	×
153	std::mutex mtx_;
154	std::condition_variable suspend_cv_;	×
155	std::condition_variable resume_cv_;
156
157	default_context context_;	×
158	};
159		×
160	default_agent::default_agent()
161	: running_(true)
162	, aborted_(false)	×
163	, id_(std::this_thread::get_id())	×
164	{
165	}	×
166
167	void default_agent::yield(char const* /* desc */)	×
168	{
169	#if defined(HPX_WINDOWS)	×
170	win_nanosleep(wait_0ns);	×
171	#else
172	sched_yield();	×
173	#endif
174	}
175		×
176	bool default_agent::yield_k(std::size_t k, char const* /* desc */)	×
177	{
178	if (k < 4) //-V112	×
179	{
180	return false;	×
181	}	×
182	else if (k < 16)
183	{	×
184	HPX_SMT_PAUSE;	×
185	return false;
186	}	×
187	else if (k < 32 \|\| k & 1) //-V112
188	{
189	#if defined(HPX_WINDOWS)
190	win_nanosleep(wait_0ns);	×
191	return true;	×
192	#else
193	sched_yield();	×
194	return true;
195	#endif
196	}
197	else	×
198	{	×
199	#if defined(HPX_WINDOWS)
200	win_nanosleep(wait_1000ns);
201	return true;
202	#else
203	// g++ -Wextra warns on {} or {0}	249✔
204	struct timespec rqtp = {0, 0};
205		249✔
206	// POSIX says that timespec has tv_sec and tv_nsec	249✔
207	// But it doesn't guarantee order or placement
208
209	rqtp.tv_sec = 0;
210	rqtp.tv_nsec = 1000;
211
212	nanosleep(&rqtp, nullptr);
213	return true;
214	#endif
215	}
216	}
217		248✔
218	void default_agent::suspend(char const* /* desc */)
219	{
220	std::unique_lock<std::mutex> l(mtx_);
221	HPX_ASSERT(running_);
222
223	running_ = false;
224	resume_cv_.notify_all();
225
226	while (!running_)
227	{
228	suspend_cv_.wait(l); //-V1089
229	}
230		774,486✔
231	if (aborted_)
232	{	774,734✔
233	HPX_THROW_EXCEPTION(hpx::error::yield_aborted, "suspend",	774,486✔
234	"std::thread({}) aborted (yield returned wait_abort)", id_);
235	}
236	}
237		1,831,158✔
238	void default_agent::resume(hpx::threads::thread_priority, char const*)	1,831,158✔
239	{	1,831,158✔
240	{	1,831,158✔
241	std::unique_lock<std::mutex> l(mtx_);
242	while (running_)	1,831,158✔
243	{
244	resume_cv_.wait(l); //-V1089	×
245	}	×
246	running_ = true;
247	}	×
248	suspend_cv_.notify_one();
249	}	1,831,158✔
250
251	void default_agent::abort(char const* /* desc */)	1,831,158✔
252	{	1,831,158✔
253	{
254	std::unique_lock<std::mutex> l(mtx_);	774,024✔
255	while (running_)
256	{	774,024✔
257	resume_cv_.wait(l); //-V1089
258	}
259	running_ = true;	686,365✔
260	aborted_ = true;
261	}	686,365✔
262	suspend_cv_.notify_one();	686,365✔
263	}
264		47,453✔
265	void default_agent::sleep_for(
266	hpx::chrono::steady_duration const& sleep_duration,	47,453✔
267	char const* /* desc */)	47,453✔
268	{
269	std::this_thread::sleep_for(sleep_duration.value());	×
270	}
271		×
272	void default_agent::sleep_until(	×
273	hpx::chrono::steady_time_point const& sleep_time,
274	char const* /* desc */)
275	{
276	std::this_thread::sleep_until(sleep_time.value());
277	}
278	} // namespace
279
280	namespace detail {
281
282	[[nodiscard]] agent_base& get_default_agent()
283	{
284	static thread_local default_agent agent;
285	return agent;
286	}
287	} // namespace detail
288
289	namespace this_thread {
290
291	namespace detail {
292
293	struct agent_storage
294	{
295	agent_storage()
296	: impl_(&hpx::execution_base::detail::get_default_agent())
297	{
298	}
299
300	agent_base* set(agent_base* context) noexcept
301	{
302	std::swap(context, impl_);
303	return context;
304	}
305
306	agent_base* impl_;
307	};
308
309	[[nodiscard]] agent_storage* get_agent_storage()
310	{
311	static thread_local agent_storage storage;
312	return &storage;
313	}
314	} // namespace detail
315
316	reset_agent::reset_agent(
317	detail::agent_storage* storage, agent_base& impl)
318	: storage_(storage)
319	, old_(storage_->set(&impl))
320	{
321	}
322
323	reset_agent::reset_agent(agent_base& impl)
324	: reset_agent(detail::get_agent_storage(), impl)
325	{
326	}
327
328	reset_agent::~reset_agent()
329	{
330	storage_->set(old_);
331	}
332
333	[[nodiscard]] agent_ref agent()
334	{
335	return agent_ref(detail::get_agent_storage()->impl_);
336	}
337
338	void yield(char const* desc)
339	{
340	agent().yield(desc);
341	}
342
343	bool yield_k(std::size_t k, char const* desc)
344	{
345	return agent().yield_k(k, desc);
346	}
347
348	void suspend(char const* desc)
349	{
350	agent().suspend(desc);
351	}
352	} // namespace this_thread
353	} // namespace hpx::execution_base

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