#862

Committed 10 Jan 2023 05:30PM UTC coverage: 86.582% (-0.05%) from 86.634%

Build # #862

Build Type

push

Committed by StellarBot

Commit Message

Merge #6130

6130: Remove the mutex lock in the critical path of get_partitioner. r=hkaiser a=JiakunYan

Remove the mutex lock in the critical path of hpx::resource::detail::get_partitioner.

The protected variable `partitioner_ref` is only set once during initialization.

Co-authored-by: Jiakun Yan <jiakunyan1998@gmail.com>

Run Details

6 of 6 new or added lines in 1 file covered. (100.0%)

174767 of 201851 relevant lines covered (86.58%)

2069816.07 hits per line

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

70.0

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

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

namespace hpx::util::detail {

    ///////////////////////////////////////////////////////////////////////////
    interval_timer::interval_timer()
      : microsecs_(0)
      , id_()
      , timerid_()
    {
    }

    interval_timer::interval_timer(hpx::function<bool()> const& f,
        std::int64_t microsecs, std::string const& description,
        bool pre_shutdown)
      : f_(f)
      , on_term_()
      , microsecs_(microsecs)
      , id_()
      , timerid_()
      , 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)
      , id_()
      , timerid_()
      , 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();
        }
        catch (...)
        {
            ;    // there is nothing we can do here
        }
    }

    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_result_type(
                    threads::thread_schedule_state::terminated,
                    threads::invalid_thread_id);
            }

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

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

            bool result = false;

            {
                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_result_type(
            threads::thread_schedule_state::terminated,
            threads::invalid_thread_id);
    }

    // schedule a high priority task after a given time interval
    void interval_timer::schedule_thread(std::unique_lock<mutex_type>& l)
    {
        HPX_ASSERT(l.owns_lock());
        HPX_UNUSED(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 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)
    {
        return timer_->change_interval(new_interval);
    }

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

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

STEllAR-GROUP / hpx / #862

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