#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

41.3

/libs/core/runtime_local/src/interval_timer.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/functional.hpp>
#include <hpx/modules/thread_support.hpp>
#include <hpx/modules/threading_base.hpp>
#include <hpx/modules/type_support.hpp>
#include <hpx/runtime_local/interval_timer.hpp>
#include <hpx/runtime_local/shutdown_function.hpp>

#include <chrono>
#include <cstddef>
#include <cstdint>
#include <memory>
#include <mutex>
#include <string>

namespace hpx::util::detail {

    ///////////////////////////////////////////////////////////////////////////
    interval_timer::interval_timer() = default;

    interval_timer::interval_timer(hpx::function<bool()> const& f,
        std::int64_t microsecs, std::string const& description,
        bool pre_shutdown)
      : f_(f)
      , microsecs_(microsecs)
      , description_(description)
      , pre_shutdown_(pre_shutdown)
    {
    }

    interval_timer::interval_timer(hpx::function<bool()> const& f,
        hpx::function<void()> const& on_term, std::int64_t microsecs,
        std::string const& description, bool pre_shutdown)
      : f_(f)
      , on_term_(on_term)
      , microsecs_(microsecs)
      , description_(description)
      , pre_shutdown_(pre_shutdown)
    {
    }

    bool interval_timer::start(bool evaluate_)
    {
        std::unique_lock<mutex_type> l(mtx_);
        if (is_terminated_)
            return false;

        if (!is_started_)
        {
            if (first_start_)
            {
                first_start_ = false;

                hpx::unlock_guard<std::unique_lock<mutex_type>> ul(l);
                if (pre_shutdown_)
                {
                    register_pre_shutdown_function(util::deferred_call(
                        &interval_timer::terminate, this->shared_from_this()));
                }
                else
                {
                    register_shutdown_function(util::deferred_call(
                        &interval_timer::terminate, this->shared_from_this()));
                }
            }

            is_stopped_ = false;

            if (evaluate_)
            {
                l.unlock();
                evaluate(threads::thread_restart_state::signaled);
            }
            else
            {
                schedule_thread(l);
            }

            return true;
        }
        return false;
    }

    bool interval_timer::restart(bool evaluate_)
    {
        if (!is_started_)
            return start(evaluate_);

        std::unique_lock<mutex_type> l(mtx_);

        if (is_terminated_)
            return false;

        // interrupt timer thread, if needed
        stop_locked();

        // reschedule evaluation thread
        if (evaluate_)
        {
            l.unlock();
            evaluate(threads::thread_restart_state::signaled);
        }
        else
        {
            schedule_thread(l);
        }
        return true;
    }

    bool interval_timer::stop(bool terminate_timer)
    {
        if (terminate_timer)
        {
            terminate();
            return true;
        }

        std::lock_guard<mutex_type> l(mtx_);
        is_stopped_ = true;
        return stop_locked();
    }

    bool interval_timer::stop_locked()
    {
        if (is_started_)
        {
            is_started_ = false;

            if (timerid_)
            {
                error_code ec(
                    throwmode::lightweight);    // avoid throwing on error
                threads::set_thread_state(timerid_.noref(),
                    threads::thread_schedule_state::pending,
                    threads::thread_restart_state::abort,
                    threads::thread_priority::boost, true, ec);
                timerid_.reset();
            }
            if (id_)
            {
                error_code ec(
                    throwmode::lightweight);    // avoid throwing on error
                threads::set_thread_state(id_.noref(),
                    threads::thread_schedule_state::pending,
                    threads::thread_restart_state::abort,
                    threads::thread_priority::boost, true, ec);
                id_.reset();
            }
            return true;
        }

        HPX_ASSERT(id_ == nullptr);
        HPX_ASSERT(timerid_ == nullptr);
        return false;
    }

    void interval_timer::terminate()
    {
        std::unique_lock<mutex_type> l(mtx_);
        if (!is_terminated_)
        {
            is_terminated_ = true;
            stop_locked();

            if (on_term_)
            {
                l.unlock();
                on_term_();
            }
        }
    }

    interval_timer::~interval_timer()
    {
        try
        {
            terminate();
        }
        // NOLINTNEXTLINE(bugprone-empty-catch)
        catch (...)
        {
        }
    }

    std::int64_t interval_timer::get_interval() const
    {
        std::lock_guard<mutex_type> l(mtx_);
        return microsecs_;
    }

    void interval_timer::change_interval(std::int64_t new_interval)
    {
        HPX_ASSERT(new_interval > 0);

        std::lock_guard<mutex_type> l(mtx_);
        microsecs_ = new_interval;
    }

    threads::thread_result_type interval_timer::evaluate(
        threads::thread_restart_state statex)
    {
        try
        {
            std::unique_lock<mutex_type> l(mtx_);

            if (is_stopped_ || is_terminated_ ||
                statex == threads::thread_restart_state::abort ||
                0 == microsecs_)
            {
                // object has been finalized, exit
                return {threads::thread_schedule_state::terminated,
                    threads::invalid_thread_id};
            }

            if (id_ != nullptr && id_ != threads::get_self_id())
            {
                // obsolete timer thread
                return {threads::thread_schedule_state::terminated,
                    threads::invalid_thread_id};
            }

            id_.reset();
            timerid_.reset();
            is_started_ = false;

            bool result;

            {
                hpx::unlock_guard<std::unique_lock<mutex_type>> ul(l);
                result = f_();    // invoke the supplied function
            }

            // some other thread might already have started the timer
            if (nullptr == id_ && result)
            {
                HPX_ASSERT(!is_started_);
                schedule_thread(l);    // wait and repeat
            }

            if (!result)
                is_terminated_ = true;
        }
        catch (hpx::exception const& e)
        {
            // the lock above might throw yield_aborted
            if (e.get_error() != hpx::error::yield_aborted)
                throw;
        }

        // do not re-schedule this thread
        return {threads::thread_schedule_state::terminated,
            threads::invalid_thread_id};
    }

    // schedule a high priority task after a given time interval
    void interval_timer::schedule_thread(
        [[maybe_unused]] std::unique_lock<mutex_type>& l)
    {
        HPX_ASSERT_OWNS_LOCK(l);

        using namespace hpx::threads;

        error_code ec;

        // create a new suspended thread
        threads::thread_id_ref_type id;
        {
            // FIXME: registering threads might lead to thread suspension since
            // the allocators use hpx::spinlock. Unlocking the lock here would
            // be the right thing but leads to crashes and hangs at shutdown.
            // unlock_guard<std::unique_lock<mutex_type> > ul(l);
            hpx::threads::thread_init_data data(
                hpx::threads::make_thread_function(hpx::bind_front(
                    &interval_timer::evaluate, this->shared_from_this())),
                description_.c_str(), threads::thread_priority::boost,
                threads::thread_schedule_hint(),
                threads::thread_stacksize::default_,
                threads::thread_schedule_state::suspended, true);
            id = hpx::threads::register_thread(data, ec);
        }

        if (ec)
        {
            is_terminated_ = true;
            is_started_ = false;
            return;
        }

        // schedule this thread to be run after the given amount of seconds
        threads::thread_id_ref_type const timerid = threads::set_thread_state(
            id.noref(), std::chrono::microseconds(microsecs_),
            threads::thread_schedule_state::pending,
            threads::thread_restart_state::signaled,
            threads::thread_priority::boost, true, ec);

        if (ec)
        {
            is_terminated_ = true;
            is_started_ = false;

            // abort the newly created thread
            threads::set_thread_state(id.noref(),
                threads::thread_schedule_state::pending,
                threads::thread_restart_state::abort,
                threads::thread_priority::boost, true, ec);

            return;
        }

        id_ = id;
        timerid_ = timerid;
        is_started_ = true;
    }
}    // namespace hpx::util::detail

namespace hpx::util {

    interval_timer::interval_timer() = default;

    interval_timer::interval_timer(    // -V730
        hpx::function<bool()> const& f, std::int64_t microsecs,
        std::string const& description, bool pre_shutdown)
      : timer_(std::make_shared<detail::interval_timer>(
            f, microsecs, description, pre_shutdown))
    {
    }

    interval_timer::interval_timer(    // -V730
        hpx::function<bool()> const& f, hpx::function<void()> const& on_term,
        std::int64_t microsecs, std::string const& description,
        bool pre_shutdown)
      : timer_(std::make_shared<detail::interval_timer>(
            f, on_term, microsecs, description, pre_shutdown))
    {
    }

    interval_timer::interval_timer(    // -V730
        hpx::function<bool()> const& f,
        hpx::chrono::steady_duration const& rel_time, char const* description,
        bool pre_shutdown)
      : timer_(std::make_shared<detail::interval_timer>(
            f, rel_time.value().count() / 1000, description, pre_shutdown))
    {
    }

    interval_timer::interval_timer(    // -V730
        hpx::function<bool()> const& f, hpx::function<void()> const& on_term,
        hpx::chrono::steady_duration const& rel_time, char const* description,
        bool pre_shutdown)
      : timer_(std::make_shared<detail::interval_timer>(f, on_term,
            rel_time.value().count() / 1000, description, pre_shutdown))
    {
    }

    interval_timer::~interval_timer()
    {
        timer_->terminate();
    }

    std::int64_t interval_timer::get_interval() const
    {
        return timer_->get_interval();
    }

    void interval_timer::change_interval(std::int64_t new_interval) const
    {
        return timer_->change_interval(new_interval);
    }

    void interval_timer::change_interval(
        hpx::chrono::steady_duration const& new_interval) const
    {
        return timer_->change_interval(new_interval.value().count() / 1000);
    }
}    // namespace hpx::util

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/functional.hpp>
11	#include <hpx/modules/thread_support.hpp>
12	#include <hpx/modules/threading_base.hpp>
13	#include <hpx/modules/type_support.hpp>
14	#include <hpx/runtime_local/interval_timer.hpp>
15	#include <hpx/runtime_local/shutdown_function.hpp>
16
17	#include <chrono>
18	#include <cstddef>
19	#include <cstdint>
20	#include <memory>
21	#include <mutex>
22	#include <string>
23
24	namespace hpx::util::detail {
25
26	///////////////////////////////////////////////////////////////////////////
27	interval_timer::interval_timer() = default;	×
28		×
29	interval_timer::interval_timer(hpx::function<bool()> const& f,	×
30	std::int64_t microsecs, std::string const& description,	×
31	bool pre_shutdown)
32	: f_(f)	×
33	, microsecs_(microsecs)
34	, description_(description)	1✔
35	, pre_shutdown_(pre_shutdown)
36	{	1✔
37	}
38
39	interval_timer::interval_timer(hpx::function<bool()> const& f,	1✔
40	hpx::function<void()> const& on_term, std::int64_t microsecs,	1✔
41	std::string const& description, bool pre_shutdown)	1✔
42	: f_(f)	1✔
43	, on_term_(on_term)	1✔
44	, microsecs_(microsecs)
45	, description_(description)	1✔
46	, pre_shutdown_(pre_shutdown)
47	{	×
48	}
49		×
50	bool interval_timer::start(bool evaluate_)
51	{
52	std::unique_lock<mutex_type> l(mtx_);	×
53	if (is_terminated_)	×
54	return false;	×
55		×
56	if (!is_started_)	×
57	{
58	if (first_start_)	×
59	{
60	first_start_ = false;	1✔
61
62	hpx::unlock_guard<std::unique_lock<mutex_type>> ul(l);	1✔
63	if (pre_shutdown_)	1✔
64	{
65	register_pre_shutdown_function(util::deferred_call(
66	&interval_timer::terminate, this->shared_from_this()));	1✔
67	}
68	else	1✔
69	{
70	register_shutdown_function(util::deferred_call(	1✔
71	&interval_timer::terminate, this->shared_from_this()));
72	}
73	}	1✔
74
75	is_stopped_ = false;	×
76		×
77	if (evaluate_)
78	{
79	l.unlock();
80	evaluate(threads::thread_restart_state::signaled);	1✔
81	}	1✔
82	else
83	{
84	schedule_thread(l);
85	}	1✔
86
87	return true;	1✔
88	}
89	return false;	1✔
90	}	1✔
91
92	bool interval_timer::restart(bool evaluate_)
93	{
94	if (!is_started_)	×
95	return start(evaluate_);
96
97	std::unique_lock<mutex_type> l(mtx_);	1✔
98
99	if (is_terminated_)
100	return false;
101
102	// interrupt timer thread, if needed	×
103	stop_locked();
104		×
105	// reschedule evaluation thread	×
106	if (evaluate_)
107	{	×
108	l.unlock();
109	evaluate(threads::thread_restart_state::signaled);	×
110	}
111	else
112	{
113	schedule_thread(l);	×
114	}
115	return true;
116	}	×
117
118	bool interval_timer::stop(bool terminate_timer)	×
119	{	×
120	if (terminate_timer)
121	{
122	terminate();
123	return true;	×
124	}
125
126	std::lock_guard<mutex_type> l(mtx_);
127	is_stopped_ = true;
128	return stop_locked();	×
129	}
130		×
131	bool interval_timer::stop_locked()
132	{	×
133	if (is_started_)	×
134	{
135	is_started_ = false;
136		×
137	if (timerid_)	×
138	{	×
139	error_code ec(
140	throwmode::lightweight); // avoid throwing on error
141	threads::set_thread_state(timerid_.noref(),	×
142	threads::thread_schedule_state::pending,
143	threads::thread_restart_state::abort,	×
144	threads::thread_priority::boost, true, ec);
145	timerid_.reset();	×
146	}
147	if (id_)	×
148	{
149	error_code ec(
150	throwmode::lightweight); // avoid throwing on error
151	threads::set_thread_state(id_.noref(),	×
152	threads::thread_schedule_state::pending,
153	threads::thread_restart_state::abort,
154	threads::thread_priority::boost, true, ec);
155	id_.reset();
156	}
157	return true;	×
158	}
159
160	HPX_ASSERT(id_ == nullptr);
161	HPX_ASSERT(timerid_ == nullptr);	×
162	return false;
163	}
164
165	void interval_timer::terminate()
166	{
167	std::unique_lock<mutex_type> l(mtx_);	×
168	if (!is_terminated_)
169	{
170	is_terminated_ = true;
171	stop_locked();
172
173	if (on_term_)
174	{
175	l.unlock();	2✔
176	on_term_();
177	}	2✔
178	}	2✔
179	}
180		×
181	interval_timer::~interval_timer()	×
182	{
183	try	×
184	{
185	terminate();	×
186	}
187	// NOLINTNEXTLINE(bugprone-empty-catch)
188	catch (...)
189	{	2✔
190	}
191	}	1✔
192
193	std::int64_t interval_timer::get_interval() const
194	{
195	std::lock_guard<mutex_type> l(mtx_);	1✔
196	return microsecs_;
197	}	×
198
199	void interval_timer::change_interval(std::int64_t new_interval)
200	{	×
201	HPX_ASSERT(new_interval > 0);	2✔
202
203	std::lock_guard<mutex_type> l(mtx_);	×
204	microsecs_ = new_interval;
205	}	×
206		×
207	threads::thread_result_type interval_timer::evaluate(
208	threads::thread_restart_state statex)
209	{	×
210	try
211	{
212	std::unique_lock<mutex_type> l(mtx_);
213		×
214	if (is_stopped_ \|\| is_terminated_ \|\|	×
215	statex == threads::thread_restart_state::abort \|\|	×
216	0 == microsecs_)
217	{	10✔
218	// object has been finalized, exit
219	return {threads::thread_schedule_state::terminated,
220	threads::invalid_thread_id};
221	}
222		10✔
223	if (id_ != nullptr && id_ != threads::get_self_id())
224	{	10✔
225	// obsolete timer thread	10✔
226	return {threads::thread_schedule_state::terminated,	10✔
227	threads::invalid_thread_id};
228	}
229
230	id_.reset();
231	timerid_.reset();
232	is_started_ = false;
233
234	bool result;	28✔
235
236	{
237	hpx::unlock_guard<std::unique_lock<mutex_type>> ul(l);
238	result = f_(); // invoke the supplied function
239	}
240
241	// some other thread might already have started the timer
242	if (nullptr == id_ && result)
243	{
244	HPX_ASSERT(!is_started_);	10✔
245	schedule_thread(l); // wait and repeat
246	}
247
248	if (!result)
249	is_terminated_ = true;
250	}
251	catch (hpx::exception const& e)
252	{
253	// the lock above might throw yield_aborted
254	if (e.get_error() != hpx::error::yield_aborted)	10✔
255	throw;
256	}
257		9✔
258	// do not re-schedule this thread
259	return {threads::thread_schedule_state::terminated,
260	threads::invalid_thread_id};	10✔
261	}	1✔
262
263	// schedule a high priority task after a given time interval	×
264	void interval_timer::schedule_thread(
265	[[maybe_unused]] std::unique_lock<mutex_type>& l)
266	{	×
267	HPX_ASSERT_OWNS_LOCK(l);	×
268		×
269	using namespace hpx::threads;
270
271	error_code ec;
272
273	// create a new suspended thread
274	threads::thread_id_ref_type id;
275	{
276	// FIXME: registering threads might lead to thread suspension since
277	// the allocators use hpx::spinlock. Unlocking the lock here would	9✔
278	// be the right thing but leads to crashes and hangs at shutdown.
279	// unlock_guard<std::unique_lock<mutex_type> > ul(l);
280	hpx::threads::thread_init_data data(
281	hpx::threads::make_thread_function(hpx::bind_front(
282	&interval_timer::evaluate, this->shared_from_this())),
283	description_.c_str(), threads::thread_priority::boost,
284	threads::thread_schedule_hint(),
285	threads::thread_stacksize::default_,
286	threads::thread_schedule_state::suspended, true);
287	id = hpx::threads::register_thread(data, ec);	9✔
288	}
289
290	if (ec)
291	{
292	is_terminated_ = true;
293	is_started_ = false;
294	return;	9✔
295	}	9✔
296
297	// schedule this thread to be run after the given amount of seconds
298	threads::thread_id_ref_type const timerid = threads::set_thread_state(
299	id.noref(), std::chrono::microseconds(microsecs_),
300	threads::thread_schedule_state::pending,	9✔
301	threads::thread_restart_state::signaled,
302	threads::thread_priority::boost, true, ec);
303		9✔
304	if (ec)
305	{	×
306	is_terminated_ = true;	×
307	is_started_ = false;	×
308
309	// abort the newly created thread
310	threads::set_thread_state(id.noref(),
311	threads::thread_schedule_state::pending,
312	threads::thread_restart_state::abort,	9✔
313	threads::thread_priority::boost, true, ec);
314
315	return;	9✔
316	}
317		9✔
318	id_ = id;
319	timerid_ = timerid;	×
320	is_started_ = true;	×
321	}
322	} // namespace hpx::util::detail
323		×
324	namespace hpx::util {
325
326	interval_timer::interval_timer() = default;
327
328	interval_timer::interval_timer( // -V730
329	hpx::function<bool()> const& f, std::int64_t microsecs,
330	std::string const& description, bool pre_shutdown)
331	: timer_(std::make_shared<detail::interval_timer>(
332	f, microsecs, description, pre_shutdown))
333	{	9✔
334	}
335
336	interval_timer::interval_timer( // -V730
337	hpx::function<bool()> const& f, hpx::function<void()> const& on_term,
338	std::int64_t microsecs, std::string const& description,
339	bool pre_shutdown)	×
340	: timer_(std::make_shared<detail::interval_timer>(
341	f, on_term, microsecs, description, pre_shutdown))	×
342	{
343	}	×
344
345	interval_timer::interval_timer( // -V730
346	hpx::function<bool()> const& f,
347	hpx::chrono::steady_duration const& rel_time, char const* description,	×
348	bool pre_shutdown)
349	: timer_(std::make_shared<detail::interval_timer>(	×
350	f, rel_time.value().count() / 1000, description, pre_shutdown))
351	{
352	}	×
353
354	interval_timer::interval_timer( // -V730
355	hpx::function<bool()> const& f, hpx::function<void()> const& on_term,
356	hpx::chrono::steady_duration const& rel_time, char const* description,	×
357	bool pre_shutdown)
358	: timer_(std::make_shared<detail::interval_timer>(f, on_term,	1✔
359	rel_time.value().count() / 1000, description, pre_shutdown))
360	{
361	}	1✔
362
363	interval_timer::~interval_timer()	1✔
364	{
365	timer_->terminate();	1✔
366	}
367		×
368	std::int64_t interval_timer::get_interval() const
369	{
370	return timer_->get_interval();	×
371	}
372		×
373	void interval_timer::change_interval(std::int64_t new_interval) const
374	{	×
375	return timer_->change_interval(new_interval);
376	}	1✔
377
378	void interval_timer::change_interval(	1✔
379	hpx::chrono::steady_duration const& new_interval) const	1✔
380	{
381	return timer_->change_interval(new_interval.value().count() / 1000);	×
382	}
383	} // namespace hpx::util	×

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