log/detail/threadsafe_queue.hpp

0001 /*
0002  *          Copyright Andrey Semashev 2007 - 2021.
0003  * Distributed under the Boost Software License, Version 1.0.
0004  *    (See accompanying file LICENSE_1_0.txt or copy at
0005  *          http://www.boost.org/LICENSE_1_0.txt)
0006  */
0007 /*!
0008  * \file   threadsafe_queue.hpp
0009  * \author Andrey Semashev
0010  * \date   05.11.2010
0011  *
0012  * \brief  This header is the Boost.Log library implementation, see the library documentation
0013  *         at http://www.boost.org/doc/libs/release/libs/log/doc/html/index.html.
0014  */
0015
0016 #ifndef BOOST_LOG_DETAIL_THREADSAFE_QUEUE_HPP_INCLUDED_
0017 #define BOOST_LOG_DETAIL_THREADSAFE_QUEUE_HPP_INCLUDED_
0018
0019 #include <boost/log/detail/config.hpp>
0020
0021 #ifdef BOOST_HAS_PRAGMA_ONCE
0022 #pragma once
0023 #endif
0024
0025 #ifndef BOOST_LOG_NO_THREADS
0026
0027 #include <new>
0028 #include <memory>
0029 #include <cstddef>
0030 #include <boost/atomic/atomic.hpp>
0031 #include <boost/move/core.hpp>
0032 #include <boost/move/utility_core.hpp>
0033 #include <boost/type_traits/alignment_of.hpp>
0034 #include <boost/type_traits/aligned_storage.hpp>
0035 #include <boost/log/utility/use_std_allocator.hpp>
0036 #include <boost/log/detail/allocator_traits.hpp>
0037 #include <boost/log/detail/header.hpp>
0038
0039 namespace boost {
0040
0041 BOOST_LOG_OPEN_NAMESPACE
0042
0043 namespace aux {
0044
0045 //! Base class for the thread-safe queue implementation
0046 class threadsafe_queue_impl
0047 {
0048 public:
0049     struct node_base
0050     {
0051         boost::atomic< node_base* > next;
0052     };
0053
0054 protected:
0055     threadsafe_queue_impl();
0056     ~threadsafe_queue_impl();
0057
0058 public:
0059     static BOOST_LOG_API threadsafe_queue_impl* create(node_base* first_node);
0060     static BOOST_LOG_API void destroy(threadsafe_queue_impl* impl) BOOST_NOEXCEPT;
0061
0062     static BOOST_LOG_API node_base* reset_last_node(threadsafe_queue_impl* impl) BOOST_NOEXCEPT;
0063     static BOOST_LOG_API bool unsafe_empty(const threadsafe_queue_impl* impl) BOOST_NOEXCEPT;
0064     static BOOST_LOG_API void push(threadsafe_queue_impl* impl, node_base* p);
0065     static BOOST_LOG_API bool try_pop(threadsafe_queue_impl* impl, node_base*& node_to_free, node_base*& node_with_value);
0066
0067     // Copying and assignment is prohibited
0068     BOOST_DELETED_FUNCTION(threadsafe_queue_impl(threadsafe_queue_impl const&))
0069     BOOST_DELETED_FUNCTION(threadsafe_queue_impl& operator= (threadsafe_queue_impl const&))
0070 };
0071
0072 //! Thread-safe queue node type
0073 template< typename T >
0074 struct threadsafe_queue_node :
0075     public threadsafe_queue_impl::node_base
0076 {
0077     typedef typename aligned_storage< sizeof(T), alignment_of< T >::value >::type storage_type;
0078     storage_type storage;
0079
0080     BOOST_DEFAULTED_FUNCTION(threadsafe_queue_node(), {})
0081     explicit threadsafe_queue_node(T const& val) { new (storage.address()) T(val); }
0082     T& value() BOOST_NOEXCEPT { return *static_cast< T* >(storage.address()); }
0083     void destroy() BOOST_NOEXCEPT { static_cast< T* >(storage.address())->~T(); }
0084
0085     // Copying and assignment is prohibited
0086     BOOST_DELETED_FUNCTION(threadsafe_queue_node(threadsafe_queue_node const&))
0087     BOOST_DELETED_FUNCTION(threadsafe_queue_node& operator= (threadsafe_queue_node const&))
0088 };
0089
0090 /*!
0091  * \brief An unbounded thread-safe queue
0092  *
0093  * The implementation is based on algorithms published in the "Simple, Fast,
0094  * and Practical Non-Blocking and Blocking Concurrent Queue Algorithms" article
0095  * in PODC96 by Maged M. Michael and Michael L. Scott. Pseudocode is available here:
0096  * http://www.cs.rochester.edu/research/synchronization/pseudocode/queues.html
0097  *
0098  * The implementation provides thread-safe \c push and \c try_pop operations, as well as
0099  * a thread-unsafe \c empty operation. The queue imposes the following requirements
0100  * on the element type:
0101  *
0102  * \li Default constructible, the default constructor must not throw.
0103  * \li Copy constructible.
0104  * \li Movable (i.e. there should be an efficient move assignment for this type).
0105  *
0106  * The last requirement is not mandatory but is crucial for decent performance.
0107  */
0108 template< typename T, typename AllocatorT = use_std_allocator >
0109 class threadsafe_queue :
0110     private boost::log::aux::rebind_alloc< AllocatorT, threadsafe_queue_node< T > >::type
0111 {
0112 private:
0113     typedef threadsafe_queue_node< T > node;
0114
0115 public:
0116     typedef typename boost::log::aux::rebind_alloc< AllocatorT, node >::type allocator_type;
0117     typedef T value_type;
0118     typedef T& reference;
0119     typedef T const& const_reference;
0120     typedef T* pointer;
0121     typedef T const* const_pointer;
0122     typedef std::ptrdiff_t difference_type;
0123     typedef std::size_t size_type;
0124
0125 private:
0126     typedef boost::log::aux::allocator_traits< allocator_type > alloc_traits;
0127
0128     //! A simple scope guard to automate memory reclaiming
0129     struct auto_deallocate;
0130     friend struct auto_deallocate;
0131     struct auto_deallocate
0132     {
0133         auto_deallocate(allocator_type* alloc, node* dealloc, node* destr) BOOST_NOEXCEPT :
0134             m_pAllocator(alloc),
0135             m_pDeallocate(dealloc),
0136             m_pDestroy(destr)
0137         {
0138         }
0139         ~auto_deallocate() BOOST_NOEXCEPT
0140         {
0141             alloc_traits::destroy(*m_pAllocator, m_pDeallocate);
0142             alloc_traits::deallocate(*m_pAllocator, m_pDeallocate, 1);
0143             m_pDestroy->destroy();
0144         }
0145
0146     private:
0147         allocator_type* m_pAllocator;
0148         node* m_pDeallocate;
0149         node* m_pDestroy;
0150     };
0151
0152 public:
0153     /*!
0154      * Default constructor, creates an empty queue. Unlike most containers,
0155      * the constructor requires memory allocation.
0156      *
0157      * \throw std::bad_alloc if there is not sufficient memory
0158      */
0159     threadsafe_queue(allocator_type const& alloc = allocator_type()) :
0160         allocator_type(alloc)
0161     {
0162         node* p = alloc_traits::allocate(get_allocator(), 1);
0163         if (BOOST_LIKELY(!!p))
0164         {
0165             try
0166             {
0167                 alloc_traits::construct(get_allocator(), p);
0168                 try
0169                 {
0170                     m_pImpl = threadsafe_queue_impl::create(p);
0171                 }
0172                 catch (...)
0173                 {
0174                     alloc_traits::destroy(get_allocator(), p);
0175                     throw;
0176                 }
0177             }
0178             catch (...)
0179             {
0180                 alloc_traits::deallocate(get_allocator(), p, 1);
0181                 throw;
0182             }
0183         }
0184         else
0185             throw std::bad_alloc();
0186     }
0187     /*!
0188      * Destructor
0189      */
0190     ~threadsafe_queue() BOOST_NOEXCEPT
0191     {
0192         // Clear the queue
0193         if (!unsafe_empty())
0194         {
0195             value_type value;
0196             while (try_pop(value)) {}
0197         }
0198
0199         // Remove the last dummy node
0200         node* p = static_cast< node* >(threadsafe_queue_impl::reset_last_node(m_pImpl));
0201         alloc_traits::destroy(get_allocator(), p);
0202         alloc_traits::deallocate(get_allocator(), p, 1);
0203
0204         threadsafe_queue_impl::destroy(m_pImpl);
0205     }
0206
0207     /*!
0208      * Checks if the queue is empty. Not thread-safe, the returned result may not be actual.
0209      */
0210     bool unsafe_empty() const BOOST_NOEXCEPT { return threadsafe_queue_impl::unsafe_empty(m_pImpl); }
0211
0212     /*!
0213      * Puts a new element to the end of the queue. Thread-safe, can be called
0214      * concurrently by several threads, and concurrently with the \c pop operation.
0215      */
0216     void push(const_reference value)
0217     {
0218         node* p = alloc_traits::allocate(get_allocator(), 1);
0219         if (BOOST_LIKELY(!!p))
0220         {
0221             try
0222             {
0223                 alloc_traits::construct(get_allocator(), p, value);
0224             }
0225             catch (...)
0226             {
0227                 alloc_traits::deallocate(get_allocator(), p, 1);
0228                 throw;
0229             }
0230             threadsafe_queue_impl::push(m_pImpl, p);
0231         }
0232         else
0233             throw std::bad_alloc();
0234     }
0235
0236     /*!
0237      * Attempts to pop an element from the beginning of the queue. Thread-safe, can
0238      * be called concurrently with the \c push operation. Should not be called by
0239      * several threads concurrently.
0240      */
0241     bool try_pop(reference value)
0242     {
0243         threadsafe_queue_impl::node_base *dealloc, *destr;
0244         if (threadsafe_queue_impl::try_pop(m_pImpl, dealloc, destr))
0245         {
0246             node* p = static_cast< node* >(destr);
0247             auto_deallocate guard(static_cast< allocator_type* >(this), static_cast< node* >(dealloc), p);
0248             value = boost::move(p->value());
0249             return true;
0250         }
0251         else
0252             return false;
0253     }
0254
0255     // Copying and assignment is prohibited
0256     BOOST_DELETED_FUNCTION(threadsafe_queue(threadsafe_queue const&))
0257     BOOST_DELETED_FUNCTION(threadsafe_queue& operator= (threadsafe_queue const&))
0258
0259 private:
0260     //! Returns the allocator instance
0261     allocator_type& get_allocator() BOOST_NOEXCEPT { return *static_cast< allocator_type* >(this); }
0262
0263 private:
0264     //! Pointer to the implementation
0265     threadsafe_queue_impl* m_pImpl;
0266 };
0267
0268 } // namespace aux
0269
0270 BOOST_LOG_CLOSE_NAMESPACE // namespace log
0271
0272 } // namespace boost
0273
0274 #include <boost/log/detail/footer.hpp>
0275
0276 #endif // BOOST_LOG_NO_THREADS
0277
0278 #endif // BOOST_LOG_DETAIL_THREADSAFE_QUEUE_HPP_INCLUDED_