2 * Copyright Andrey Semashev 2007 - 2015.
3 * Distributed under the Boost Software License, Version 1.0.
4 * (See accompanying file LICENSE_1_0.txt or copy at
5 * http://www.boost.org/LICENSE_1_0.txt)
8 * \file bounded_ordering_queue.hpp
9 * \author Andrey Semashev
12 * The header contains implementation of bounded ordering queueing strategy for
13 * the asynchronous sink frontend.
16 #ifndef BOOST_LOG_SINKS_BOUNDED_ORDERING_QUEUE_HPP_INCLUDED_
17 #define BOOST_LOG_SINKS_BOUNDED_ORDERING_QUEUE_HPP_INCLUDED_
19 #include <boost/log/detail/config.hpp>
21 #ifdef BOOST_HAS_PRAGMA_ONCE
25 #if defined(BOOST_LOG_NO_THREADS)
26 #error Boost.Log: This header content is only supported in multithreaded environment
32 #include <boost/cstdint.hpp>
33 #include <boost/thread/locks.hpp>
34 #include <boost/thread/mutex.hpp>
35 #include <boost/thread/condition_variable.hpp>
36 #include <boost/thread/thread_time.hpp>
37 #include <boost/date_time/posix_time/posix_time_types.hpp>
38 #include <boost/log/detail/timestamp.hpp>
39 #include <boost/log/detail/enqueued_record.hpp>
40 #include <boost/log/keywords/order.hpp>
41 #include <boost/log/keywords/ordering_window.hpp>
42 #include <boost/log/core/record_view.hpp>
43 #include <boost/log/detail/header.hpp>
47 BOOST_LOG_OPEN_NAMESPACE
52 * \brief Bounded ordering log record queueing strategy
54 * The \c bounded_ordering_queue class is intended to be used with
55 * the \c asynchronous_sink frontend as a log record queueing strategy.
57 * This strategy provides the following properties to the record queueing mechanism:
59 * \li The queue has limited capacity specified by the \c MaxQueueSizeV template parameter.
60 * \li Upon reaching the size limit, the queue invokes the overflow handling strategy
61 * specified in the \c OverflowStrategyT template parameter to handle the situation.
62 * The library provides overflow handling strategies for most common cases:
63 * \c drop_on_overflow will silently discard the log record, and \c block_on_overflow
64 * will put the enqueueing thread to wait until there is space in the queue.
65 * \li The queue has a fixed latency window. This means that each log record put
66 * into the queue will normally not be dequeued for a certain period of time.
67 * \li The queue performs stable record ordering within the latency window.
68 * The ordering predicate can be specified in the \c OrderT template parameter.
70 template< typename OrderT, std::size_t MaxQueueSizeV, typename OverflowStrategyT >
71 class bounded_ordering_queue :
72 private OverflowStrategyT
75 typedef OverflowStrategyT overflow_strategy;
76 typedef boost::mutex mutex_type;
77 typedef sinks::aux::enqueued_record enqueued_record;
79 typedef std::priority_queue<
81 std::vector< enqueued_record >,
82 enqueued_record::order< OrderT >
86 //! Ordering window duration, in milliseconds
87 const uint64_t m_ordering_window;
88 //! Synchronization primitive
90 //! Condition to block the consuming thread on
91 condition_variable m_cond;
95 bool m_interruption_requested;
99 * Returns ordering window size specified during initialization
101 posix_time::time_duration get_ordering_window() const
103 return posix_time::milliseconds(m_ordering_window);
107 * Returns default ordering window size.
108 * The default window size is specific to the operating system thread scheduling mechanism.
110 static posix_time::time_duration get_default_ordering_window()
112 // The main idea behind this parameter is that the ordering window should be large enough
113 // to allow the frontend to order records from different threads on an attribute
114 // that contains system time. Thus this value should be:
115 // * No less than the minimum time resolution quant that Boost.DateTime provides on the current OS.
116 // For instance, on Windows it defaults to around 15-16 ms.
117 // * No less than thread switching quant on the current OS. For now 30 ms is large enough window size to
118 // switch threads on any known OS. It can be tuned for other platforms as needed.
119 return posix_time::milliseconds(30);
123 //! Initializing constructor
124 template< typename ArgsT >
125 explicit bounded_ordering_queue(ArgsT const& args) :
126 m_ordering_window(args[keywords::ordering_window || &bounded_ordering_queue::get_default_ordering_window].total_milliseconds()),
127 m_queue(args[keywords::order]),
128 m_interruption_requested(false)
132 //! Enqueues log record to the queue
133 void enqueue(record_view const& rec)
135 unique_lock< mutex_type > lock(m_mutex);
136 std::size_t size = m_queue.size();
137 for (; size >= MaxQueueSizeV; size = m_queue.size())
139 if (!overflow_strategy::on_overflow(rec, lock))
143 m_queue.push(enqueued_record(rec));
148 //! Attempts to enqueue log record to the queue
149 bool try_enqueue(record_view const& rec)
151 unique_lock< mutex_type > lock(m_mutex, try_to_lock);
152 if (lock.owns_lock())
154 const std::size_t size = m_queue.size();
156 // Do not invoke the bounding strategy in case of overflow as it may block
157 if (size < MaxQueueSizeV)
159 m_queue.push(enqueued_record(rec));
169 //! Attempts to dequeue a log record ready for processing from the queue, does not block if the queue is empty
170 bool try_dequeue_ready(record_view& rec)
172 lock_guard< mutex_type > lock(m_mutex);
173 const std::size_t size = m_queue.size();
176 const boost::log::aux::timestamp now = boost::log::aux::get_timestamp();
177 enqueued_record const& elem = m_queue.top();
178 if (static_cast< uint64_t >((now - elem.m_timestamp).milliseconds()) >= m_ordering_window)
180 // We got a new element
183 overflow_strategy::on_queue_space_available();
191 //! Attempts to dequeue log record from the queue, does not block if the queue is empty
192 bool try_dequeue(record_view& rec)
194 lock_guard< mutex_type > lock(m_mutex);
195 const std::size_t size = m_queue.size();
198 enqueued_record const& elem = m_queue.top();
201 overflow_strategy::on_queue_space_available();
208 //! Dequeues log record from the queue, blocks if the queue is empty
209 bool dequeue_ready(record_view& rec)
211 unique_lock< mutex_type > lock(m_mutex);
213 while (!m_interruption_requested)
215 const std::size_t size = m_queue.size();
218 const boost::log::aux::timestamp now = boost::log::aux::get_timestamp();
219 enqueued_record const& elem = m_queue.top();
220 const uint64_t difference = (now - elem.m_timestamp).milliseconds();
221 if (difference >= m_ordering_window)
225 overflow_strategy::on_queue_space_available();
230 // Wait until the element becomes ready to be processed
231 m_cond.timed_wait(lock, posix_time::milliseconds(m_ordering_window - difference));
239 m_interruption_requested = false;
244 //! Wakes a thread possibly blocked in the \c dequeue method
245 void interrupt_dequeue()
247 lock_guard< mutex_type > lock(m_mutex);
248 m_interruption_requested = true;
249 overflow_strategy::interrupt();
256 BOOST_LOG_CLOSE_NAMESPACE // namespace log
260 #include <boost/log/detail/footer.hpp>
262 #endif // BOOST_LOG_SINKS_BOUNDED_ORDERING_QUEUE_HPP_INCLUDED_