#844

Committed 02 Dec 2022 12:37AM UTC coverage: 85.8% (+0.2%) from 85.634%

Build # #844

Build Type

push

Committed by StellarBot

Commit Message

Merge #6084

6084: Renaming hpx::apply and friends to hpx::post r=hkaiser a=hkaiser

- this is needed to be able to rename invoke_fused to apply later

working towards https://github.com/STEllAR-GROUP/hpx/issues/5497



Co-authored-by: Hartmut Kaiser <hartmut.kaiser@gmail.com>

Run Details

468 of 468 new or added lines in 96 files covered. (100.0%)

171389 of 199753 relevant lines covered (85.8%)

1914550.97 hits per line

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

24.82

/libs/full/runtime_components/src/console_logging.cpp

//  Copyright (c) 2007-2022 Hartmut Kaiser
//  Copyright (c)      2011 Bryce Lelbach
//
//  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/agas/addressing_service.hpp>
#include <hpx/assert.hpp>
#include <hpx/async_distributed/continuation.hpp>
#include <hpx/async_distributed/detail/post.hpp>
#include <hpx/components_base/agas_interface.hpp>
#include <hpx/concurrency/spinlock.hpp>
#include <hpx/datastructures/tuple.hpp>
#include <hpx/modules/errors.hpp>
#include <hpx/modules/logging.hpp>
#include <hpx/modules/threadmanager.hpp>
#include <hpx/naming_base/id_type.hpp>
#include <hpx/runtime_components/console_logging.hpp>
#include <hpx/runtime_components/server/console_logging.hpp>
#include <hpx/runtime_local/runtime_local.hpp>
#include <hpx/runtime_local/state.hpp>
#include <hpx/static_reinit/reinitializable_static.hpp>
#include <hpx/synchronization/mutex.hpp>
#include <hpx/thread_support/unlock_guard.hpp>
#include <hpx/type_support/static.hpp>

#include <atomic>
#include <cstddef>
#include <mutex>
#include <string>

///////////////////////////////////////////////////////////////////////////////
namespace hpx { namespace components {

    void fallback_console_logging_locked(
        messages_type const& msgs, std::string fail_msg = "")
    {
        using hpx::get;

        if (!fail_msg.empty())
            fail_msg = "Logging failed due to: " + fail_msg + "\n";

        for (std::size_t i = 0; i != msgs.size(); ++i)
        {
            message_type const& msg = msgs[i];
            switch (get<0>(msg))
            {
            default:
            case destination_hpx:
                LHPX_CONSOLE_(get<1>(msg)) << fail_msg << get<2>(msg);
                break;

            case destination_timing:
                LTIM_CONSOLE_(get<1>(msg)) << fail_msg << get<2>(msg);
                break;

            case destination_agas:
                LAGAS_CONSOLE_(get<1>(msg)) << fail_msg << get<2>(msg);
                break;

            case destination_parcel:
                LPT_CONSOLE_(get<1>(msg)) << fail_msg << get<2>(msg);
                break;

            case destination_app:
                LAPP_CONSOLE_(get<1>(msg)) << fail_msg << get<2>(msg);
                break;

            case destination_debuglog:
                LDEB_CONSOLE_ << fail_msg << get<2>(msg);
                break;
            }
        }
    }

    void console_logging_locked(
        hpx::id_type const& prefix, messages_type const& msgs)
    {
        // If we're not in an HPX thread, we cannot call apply as it may access
        // the AGAS components. We just throw an exception here - there's no
        // thread-manager, so the exception will probably be unhandled. This is
        // desirable in this situation, as we can't trust the logging system to
        // report this error.
        if (HPX_UNLIKELY(!threads::get_self_ptr()))
        {
            // write the message to a local file in any case
            fallback_console_logging_locked(msgs);

            // raise error as this should get called from outside a HPX-thread
            HPX_THROW_EXCEPTION(null_thread_id,
                "components::console_logging_locked",
                "console_logging_locked was not called from a HPX-thread");
        }

        try
        {
            hpx::post<server::console_logging_action<>>(prefix, msgs);
        }
        catch (hpx::exception const& e)
        {
            // if this is not the console locality (or any other error occurs)
            // we might be too late for any logging, write to local file
            fallback_console_logging_locked(msgs, e.what());
        }
    }

    ///////////////////////////////////////////////////////////////////////////
    struct HPX_EXPORT pending_logs
    {
        using prefix_mutex_type = hpx::mutex;
        using queue_mutex_type = util::spinlock;

        enum
        {
            max_pending = 128
        };

        pending_logs()
          : prefix_mtx_()
          , prefix_(hpx::invalid_id)
          , queue_mtx_()
          , activated_(false)
          , is_sending_(false)
        {
        }

        void add(message_type const& msg);

        void cleanup();

        void activate()
        {
            activated_.store(true);
        }

    private:
        bool ensure_prefix();
        void send();
        bool is_active();

        prefix_mutex_type prefix_mtx_;
        hpx::id_type prefix_;

        queue_mutex_type queue_mtx_;
        messages_type queue_;

        std::atomic<bool> activated_;
        std::atomic<bool> is_sending_;
    };

    bool pending_logs::is_active()
    {
        return threads::get_self_ptr() &&
            threads::threadmanager_is(hpx::state::running) && activated_.load();
    }

    void pending_logs::add(message_type const& msg)
    {
        if (nullptr == hpx::get_runtime_ptr())
        {
            // This branch will be taken if it's too early or too late in the
            // game. We do local logging only. Any queued messages which may be
            // still left in the queue are logged locally as well.

            // queue up the new message and log it with the rest of it
            messages_type msgs;
            {
                std::lock_guard<queue_mutex_type> l(queue_mtx_);
                queue_.push_back(msg);
                queue_.swap(msgs);
            }

            fallback_console_logging_locked(msgs);
        }
        else if (is_active())
        {
            // This branch will be taken under normal circumstances. We queue
            // up the message and either log it immediately (if we are on the
            // console locality) or defer logging until 'max_pending' messages
            // have been queued. Note: we can invoke send only from within a
            // HPX-thread.
            std::size_t size = 0;

            {
                std::lock_guard<queue_mutex_type> l(queue_mtx_);
                queue_.push_back(msg);
                size = queue_.size();
            }

            // Invoke actual logging immediately if we're on the console or
            // if the number of waiting log messages is too large.
            if (agas::is_console() || size > max_pending)
                send();
        }
        else
        {
            // This branch will be taken if the runtime is up and running, but
            // either the thread manager is not active or this is not invoked
            // on a HPX-thread.

            // Note: is_console can be called outside of a HPX-thread
            if (!agas::is_console())
            {
                // queue it for delivery to the console
                std::lock_guard<queue_mutex_type> l(queue_mtx_);
                queue_.push_back(msg);
            }
            else
            {
                // log it locally on the console
                messages_type msgs;
                msgs.push_back(msg);
                fallback_console_logging_locked(msgs);
            }
        }
    }

    void pending_logs::cleanup()
    {
        if (threads::threadmanager_is(hpx::state::running) &&
            threads::get_self_ptr())
        {
            send();
        }
        else
        {
            messages_type msgs;
            {
                std::lock_guard<queue_mutex_type> l(queue_mtx_);
                if (queue_.empty())
                    return;    // some other thread did the deed
                queue_.swap(msgs);
            }

            fallback_console_logging_locked(msgs);
        }
    }

    bool pending_logs::ensure_prefix()
    {
        // Resolve the console prefix if it's still invalid.
        if (HPX_UNLIKELY(hpx::invalid_id == prefix_))
        {
            std::unique_lock<prefix_mutex_type> l(
                prefix_mtx_, std::try_to_lock);

            if (l.owns_lock() && (hpx::invalid_id == prefix_))
            {
                naming::gid_type raw_prefix;
                {
                    unlock_guard<std::unique_lock<prefix_mutex_type>> ul(l);
                    naming::get_agas_client().get_console_locality(raw_prefix);
                }

                HPX_ASSERT(naming::invalid_gid != raw_prefix);
                if (!prefix_)
                {
                    prefix_ = hpx::id_type(
                        raw_prefix, hpx::id_type::management_type::unmanaged);
                }
                else
                {
                    HPX_ASSERT(prefix_.get_gid() == raw_prefix);
                }
            }

            // Someone else started getting the console prefix.
            else
            {
                return false;
            }
        }
        return true;
    }

    void pending_logs::send()
    {
        // WARNING: Never, ever call this outside of a HPX-thread.
        HPX_ASSERT(threads::get_self_ptr());

        bool expected = false;
        if (!is_sending_.compare_exchange_strong(expected, true))
            return;

        try
        {
            {
                std::lock_guard<queue_mutex_type> l(queue_mtx_);
                if (queue_.empty())
                    return;    // some other thread did the deed
            }

            if (!ensure_prefix())
                return;    // some other thread tries to do logging

            messages_type msgs;
            {
                std::lock_guard<queue_mutex_type> l(queue_mtx_);
                if (queue_.empty())
                    return;    // some other thread did the deed
                queue_.swap(msgs);
            }

            console_logging_locked(prefix_, msgs);
        }
        catch (...)
        {
            is_sending_ = false;
            throw;
        }

        is_sending_ = false;
    }

    ///////////////////////////////////////////////////////////////////////////
    namespace detail {

        struct pending_logs_tag
        {
        };

        pending_logs& logger()
        {
            util::reinitializable_static<pending_logs, pending_logs_tag> logs;
            return logs.get();
        }
    }    // namespace detail

    void console_logging(
        logging_destination dest, std::size_t level, std::string const& s)
    {
        message_type msg(dest, level, s);
        detail::logger().add(msg);
    }

    void cleanup_logging()
    {
        detail::logger().cleanup();
    }

    void activate_logging()
    {
        detail::logger().activate();
    }
}}    // namespace hpx::components

1	// Copyright (c) 2007-2022 Hartmut Kaiser
2	// Copyright (c) 2011 Bryce Lelbach
3	//
4	// SPDX-License-Identifier: BSL-1.0
5	// Distributed under the Boost Software License, Version 1.0. (See accompanying
6	// file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
7
8	#include <hpx/config.hpp>
9	#include <hpx/agas/addressing_service.hpp>
10	#include <hpx/assert.hpp>
11	#include <hpx/async_distributed/continuation.hpp>
12	#include <hpx/async_distributed/detail/post.hpp>
13	#include <hpx/components_base/agas_interface.hpp>
14	#include <hpx/concurrency/spinlock.hpp>
15	#include <hpx/datastructures/tuple.hpp>
16	#include <hpx/modules/errors.hpp>
17	#include <hpx/modules/logging.hpp>
18	#include <hpx/modules/threadmanager.hpp>
19	#include <hpx/naming_base/id_type.hpp>
20	#include <hpx/runtime_components/console_logging.hpp>
21	#include <hpx/runtime_components/server/console_logging.hpp>
22	#include <hpx/runtime_local/runtime_local.hpp>
23	#include <hpx/runtime_local/state.hpp>
24	#include <hpx/static_reinit/reinitializable_static.hpp>
25	#include <hpx/synchronization/mutex.hpp>
26	#include <hpx/thread_support/unlock_guard.hpp>
27	#include <hpx/type_support/static.hpp>
28
29	#include <atomic>
30	#include <cstddef>
31	#include <mutex>
32	#include <string>
33
34	///////////////////////////////////////////////////////////////////////////////
35	namespace hpx { namespace components {
36
37	void fallback_console_logging_locked(	×
38	messages_type const& msgs, std::string fail_msg = "")
39	{
40	using hpx::get;
41
42	if (!fail_msg.empty())	×
43	fail_msg = "Logging failed due to: " + fail_msg + "\n";	×
44
45	for (std::size_t i = 0; i != msgs.size(); ++i)	×
46	{
47	message_type const& msg = msgs[i];	×
48	switch (get<0>(msg))	×
49	{	×
50	default:
51	case destination_hpx:
52	LHPX_CONSOLE_(get<1>(msg)) << fail_msg << get<2>(msg);	×
53	break;	×
54
55	case destination_timing:
56	LTIM_CONSOLE_(get<1>(msg)) << fail_msg << get<2>(msg);	×
57	break;	×
58
59	case destination_agas:
60	LAGAS_CONSOLE_(get<1>(msg)) << fail_msg << get<2>(msg);	×
61	break;	×
62
63	case destination_parcel:
64	LPT_CONSOLE_(get<1>(msg)) << fail_msg << get<2>(msg);	×
65	break;	×
66
67	case destination_app:
68	LAPP_CONSOLE_(get<1>(msg)) << fail_msg << get<2>(msg);	×
69	break;	×
70
71	case destination_debuglog:
72	LDEB_CONSOLE_ << fail_msg << get<2>(msg);	×
73	break;	×
74	}
75	}	×
76	}	×
77
78	void console_logging_locked(	×
79	hpx::id_type const& prefix, messages_type const& msgs)
80	{
81	// If we're not in an HPX thread, we cannot call apply as it may access
82	// the AGAS components. We just throw an exception here - there's no
83	// thread-manager, so the exception will probably be unhandled. This is
84	// desirable in this situation, as we can't trust the logging system to
85	// report this error.
86	if (HPX_UNLIKELY(!threads::get_self_ptr()))	×
87	{
88	// write the message to a local file in any case
89	fallback_console_logging_locked(msgs);	×
90
91	// raise error as this should get called from outside a HPX-thread
92	HPX_THROW_EXCEPTION(null_thread_id,	×
93	"components::console_logging_locked",
94	"console_logging_locked was not called from a HPX-thread");
95	}
96
97	try
98	{
99	hpx::post<server::console_logging_action<>>(prefix, msgs);	×
100	}	×
101	catch (hpx::exception const& e)
102	{
103	// if this is not the console locality (or any other error occurs)
104	// we might be too late for any logging, write to local file
105	fallback_console_logging_locked(msgs, e.what());	×
106	}	×
107	}	×
108
109	///////////////////////////////////////////////////////////////////////////
110	struct HPX_EXPORT pending_logs	597✔
111	{
112	using prefix_mutex_type = hpx::mutex;
113	using queue_mutex_type = util::spinlock;
114
115	enum
116	{
117	max_pending = 128
118	};
119
120	pending_logs()	1,406✔
121	: prefix_mtx_()	703✔
122	, prefix_(hpx::invalid_id)	703✔
123	, queue_mtx_()	703✔
124	, activated_(false)	703✔
125	, is_sending_(false)	703✔
126	{
127	}	703✔
128
129	void add(message_type const& msg);
130
131	void cleanup();
132
133	void activate()	594✔
134	{
135	activated_.store(true);	594✔
136	}	594✔
137
138	private:
139	bool ensure_prefix();
140	void send();
141	bool is_active();
142
143	prefix_mutex_type prefix_mtx_;
144	hpx::id_type prefix_;
145
146	queue_mutex_type queue_mtx_;
147	messages_type queue_;
148
149	std::atomic<bool> activated_;
150	std::atomic<bool> is_sending_;
151	};
152
153	bool pending_logs::is_active()	×
154	{
155	return threads::get_self_ptr() &&	×
156	threads::threadmanager_is(hpx::state::running) && activated_.load();	×
157	}
158
159	void pending_logs::add(message_type const& msg)	×
160	{
161	if (nullptr == hpx::get_runtime_ptr())	×
162	{
163	// This branch will be taken if it's too early or too late in the
164	// game. We do local logging only. Any queued messages which may be
165	// still left in the queue are logged locally as well.
166
167	// queue up the new message and log it with the rest of it
168	messages_type msgs;	×
169	{
170	std::lock_guard<queue_mutex_type> l(queue_mtx_);	×
171	queue_.push_back(msg);	×
172	queue_.swap(msgs);	×
173	}	×
174
175	fallback_console_logging_locked(msgs);	×
176	}	×
177	else if (is_active())	×
178	{
179	// This branch will be taken under normal circumstances. We queue
180	// up the message and either log it immediately (if we are on the
181	// console locality) or defer logging until 'max_pending' messages
182	// have been queued. Note: we can invoke send only from within a
183	// HPX-thread.
184	std::size_t size = 0;	×
185
186	{
187	std::lock_guard<queue_mutex_type> l(queue_mtx_);	×
188	queue_.push_back(msg);	×
189	size = queue_.size();	×
190	}	×
191
192	// Invoke actual logging immediately if we're on the console or
193	// if the number of waiting log messages is too large.
194	if (agas::is_console() \|\| size > max_pending)	×
195	send();	×
196	}	×
197	else
198	{
199	// This branch will be taken if the runtime is up and running, but
200	// either the thread manager is not active or this is not invoked
201	// on a HPX-thread.
202
203	// Note: is_console can be called outside of a HPX-thread
204	if (!agas::is_console())	×
205	{
206	// queue it for delivery to the console
207	std::lock_guard<queue_mutex_type> l(queue_mtx_);	×
208	queue_.push_back(msg);	×
209	}	×
210	else
211	{
212	// log it locally on the console
213	messages_type msgs;	×
214	msgs.push_back(msg);	×
215	fallback_console_logging_locked(msgs);	×
216	}	×
217	}
218	}	×
219
220	void pending_logs::cleanup()	594✔
221	{
222	if (threads::threadmanager_is(hpx::state::running) &&	594✔
223	threads::get_self_ptr())	594✔
224	{
225	send();	594✔
226	}	594✔
227	else
228	{
229	messages_type msgs;	×
230	{
231	std::lock_guard<queue_mutex_type> l(queue_mtx_);	×
232	if (queue_.empty())	×
233	return; // some other thread did the deed	×
234	queue_.swap(msgs);	×
235	}	×
236
237	fallback_console_logging_locked(msgs);	×
238	}	×
239	}	594✔
240
241	bool pending_logs::ensure_prefix()	×
242	{
243	// Resolve the console prefix if it's still invalid.
244	if (HPX_UNLIKELY(hpx::invalid_id == prefix_))	×
245	{
246	std::unique_lock<prefix_mutex_type> l(	×
247	prefix_mtx_, std::try_to_lock);	×
248
249	if (l.owns_lock() && (hpx::invalid_id == prefix_))	×
250	{
251	naming::gid_type raw_prefix;	×
252	{
253	unlock_guard<std::unique_lock<prefix_mutex_type>> ul(l);	×
254	naming::get_agas_client().get_console_locality(raw_prefix);	×
255	}	×
256
257	HPX_ASSERT(naming::invalid_gid != raw_prefix);	×
258	if (!prefix_)	×
259	{
260	prefix_ = hpx::id_type(	×
261	raw_prefix, hpx::id_type::management_type::unmanaged);
262	}	×
263	else
264	{
265	HPX_ASSERT(prefix_.get_gid() == raw_prefix);	×
266	}
267	}	×
268
269	// Someone else started getting the console prefix.
270	else
271	{
272	return false;	×
273	}
274	}	×
275	return true;	×
276	}	×
277
278	void pending_logs::send()	594✔
279	{
280	// WARNING: Never, ever call this outside of a HPX-thread.
281	HPX_ASSERT(threads::get_self_ptr());	594✔
282
283	bool expected = false;	594✔
284	if (!is_sending_.compare_exchange_strong(expected, true))	594✔
285	return;	×
286
287	try
288	{
289	{
290	std::lock_guard<queue_mutex_type> l(queue_mtx_);	594✔
291	if (queue_.empty())	594✔
292	return; // some other thread did the deed	594✔
293	}	594✔
294
295	if (!ensure_prefix())	×
296	return; // some other thread tries to do logging	×
297
298	messages_type msgs;	×
299	{
300	std::lock_guard<queue_mutex_type> l(queue_mtx_);	×
301	if (queue_.empty())	×
302	return; // some other thread did the deed	×
303	queue_.swap(msgs);	×
304	}	×
305
306	console_logging_locked(prefix_, msgs);	×
307	}	×
308	catch (...)
309	{
310	is_sending_ = false;	×
311	throw;	×
312	}	×
313
314	is_sending_ = false;	×
315	}	594✔
316
317	///////////////////////////////////////////////////////////////////////////
318	namespace detail {
319
320	struct pending_logs_tag
321	{
322	};
323
324	pending_logs& logger()	1,188✔
325	{
326	util::reinitializable_static<pending_logs, pending_logs_tag> logs;	1,188✔
327	return logs.get();	1,188✔
328	}
329	} // namespace detail
330
331	void console_logging(	×
332	logging_destination dest, std::size_t level, std::string const& s)
333	{
334	message_type msg(dest, level, s);	×
335	detail::logger().add(msg);	×
336	}	×
337
338	void cleanup_logging()	594✔
339	{
340	detail::logger().cleanup();	594✔
341	}	594✔
342
343	void activate_logging()	594✔
344	{
345	detail::logger().activate();	594✔
346	}	594✔
347	}} // namespace hpx::components

STEllAR-GROUP / hpx / #844

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