EIC code displayed by LXR master/​include/​boost/​log/​sinks/​async_frontend.hpp	Source navigation Diff markup Identifier search General search
	Version: master test Revert   Change

File indexing completed on 2025-01-18 09:39:24

 /*
  *          Copyright Andrey Semashev 2007 - 2015.
  * 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)
  */
 /*!
  * \file   async_frontend.hpp
  * \author Andrey Semashev
  * \date   14.07.2009
  *
  * The header contains implementation of asynchronous sink frontend.
  */
 
 #ifndef BOOST_LOG_SINKS_ASYNC_FRONTEND_HPP_INCLUDED_
 #define BOOST_LOG_SINKS_ASYNC_FRONTEND_HPP_INCLUDED_
 
 #include <exception> // std::terminate
 #include <boost/log/detail/config.hpp>
 
 #ifdef BOOST_HAS_PRAGMA_ONCE
 #pragma once
 #endif
 
 #if defined(BOOST_LOG_NO_THREADS)
 #error Boost.Log: Asynchronous sink frontend is only supported in multithreaded environment
 #endif
 
 #include <boost/memory_order.hpp>
 #include <boost/atomic/atomic.hpp>
 #include <boost/smart_ptr/shared_ptr.hpp>
 #include <boost/smart_ptr/make_shared_object.hpp>
 #include <boost/preprocessor/control/if.hpp>
 #include <boost/preprocessor/comparison/equal.hpp>
 #include <boost/thread/locks.hpp>
 #include <boost/thread/recursive_mutex.hpp>
 #include <boost/thread/thread.hpp>
 #include <boost/thread/condition_variable.hpp>
 #include <boost/log/exceptions.hpp>
 #include <boost/log/detail/locking_ptr.hpp>
 #include <boost/log/detail/parameter_tools.hpp>
 #include <boost/log/core/record_view.hpp>
 #include <boost/log/sinks/basic_sink_frontend.hpp>
 #include <boost/log/sinks/frontend_requirements.hpp>
 #include <boost/log/sinks/unbounded_fifo_queue.hpp>
 #include <boost/log/keywords/start_thread.hpp>
 #include <boost/log/detail/header.hpp>
 
 namespace boost {
 
 BOOST_LOG_OPEN_NAMESPACE
 
 namespace sinks {
 
 #ifndef BOOST_LOG_DOXYGEN_PASS
 
 #define BOOST_LOG_SINK_CTOR_FORWARD_INTERNAL_1(z, n, data)\
     template< typename T0 >\
     explicit asynchronous_sink(T0 const& arg0, typename boost::log::aux::enable_if_named_parameters< T0, boost::log::aux::sfinae_dummy >::type = boost::log::aux::sfinae_dummy()) :\
         base_type(true),\
         queue_base_type(arg0),\
         m_pBackend(boost::make_shared< sink_backend_type >(arg0)),\
         m_ActiveOperation(idle),\
         m_StopRequested(false),\
         m_FlushRequested(false)\
     {\
         if (arg0[keywords::start_thread | true])\
             start_feeding_thread();\
     }\
     template< typename T0 >\
     explicit asynchronous_sink(shared_ptr< sink_backend_type > const& backend, T0 const& arg0) :\
         base_type(true),\
         queue_base_type(arg0),\
         m_pBackend(backend),\
         m_ActiveOperation(idle),\
         m_StopRequested(false),\
         m_FlushRequested(false)\
     {\
         if (arg0[keywords::start_thread | true])\
             start_feeding_thread();\
     }
 
 #define BOOST_LOG_SINK_CTOR_FORWARD_INTERNAL_N(z, n, data)\
     template< BOOST_PP_ENUM_PARAMS_Z(z, n, typename T) >\
     explicit asynchronous_sink(BOOST_PP_ENUM_BINARY_PARAMS_Z(z, n, T, const& arg)) :\
         base_type(true),\
         queue_base_type((BOOST_PP_ENUM_PARAMS_Z(z, n, arg))),\
         m_pBackend(boost::make_shared< sink_backend_type >(BOOST_PP_ENUM_PARAMS_Z(z, n, arg))),\
         m_ActiveOperation(idle),\
         m_StopRequested(false),\
         m_FlushRequested(false)\
     {\
         if ((BOOST_PP_ENUM_PARAMS_Z(z, n, arg))[keywords::start_thread | true])\
             start_feeding_thread();\
     }\
     template< BOOST_PP_ENUM_PARAMS_Z(z, n, typename T) >\
     explicit asynchronous_sink(shared_ptr< sink_backend_type > const& backend, BOOST_PP_ENUM_BINARY_PARAMS_Z(z, n, T, const& arg)) :\
         base_type(true),\
         queue_base_type((BOOST_PP_ENUM_PARAMS_Z(z, n, arg))),\
         m_pBackend(backend),\
         m_ActiveOperation(idle),\
         m_StopRequested(false),\
         m_FlushRequested(false)\
     {\
         if ((BOOST_PP_ENUM_PARAMS_Z(z, n, arg))[keywords::start_thread | true])\
             start_feeding_thread();\
     }
 
 #define BOOST_LOG_SINK_CTOR_FORWARD_INTERNAL(z, n, data)\
     BOOST_PP_IF(BOOST_PP_EQUAL(n, 1), BOOST_LOG_SINK_CTOR_FORWARD_INTERNAL_1, BOOST_LOG_SINK_CTOR_FORWARD_INTERNAL_N)(z, n, data)
 
 #endif // BOOST_LOG_DOXYGEN_PASS
 
 /*!
  * \brief Asynchronous logging sink frontend
  *
  * The frontend starts a separate thread on construction. All logging records are passed
  * to the backend in this dedicated thread.
  *
  * The user can prevent spawning the internal thread by specifying \c start_thread parameter
  * with the value of \c false on construction. In this case log records will be buffered
  * in the internal queue until the user calls \c run, \c feed_records or \c flush in his own
  * thread. Log record queueing strategy is specified in the \c QueueingStrategyT template
  * parameter.
  */
 template< typename SinkBackendT, typename QueueingStrategyT = unbounded_fifo_queue >
 class asynchronous_sink :
     public aux::make_sink_frontend_base< SinkBackendT >::type,
     public QueueingStrategyT
 {
     typedef typename aux::make_sink_frontend_base< SinkBackendT >::type base_type;
     typedef QueueingStrategyT queue_base_type;
 
 private:
     //! Backend synchronization mutex type
     typedef boost::recursive_mutex backend_mutex_type;
     //! Frontend synchronization mutex type
     typedef typename base_type::mutex_type frontend_mutex_type;
 
     //! Operation bit mask
     enum operation
     {
         idle = 0u,
         feeding_records = 1u,
         flushing = 3u
     };
 
     //! Function object to run the log record feeding thread
     class run_func
     {
     public:
         typedef void result_type;
 
     private:
         asynchronous_sink* m_self;
 
     public:
         explicit run_func(asynchronous_sink* self) BOOST_NOEXCEPT : m_self(self)
         {
         }
 
         result_type operator()() const
         {
             m_self->run();
         }
     };
 
     //! A scope guard that implements active operation management
     class scoped_feeding_operation
     {
     private:
         asynchronous_sink& m_self;
 
     public:
         //! Initializing constructor
         explicit scoped_feeding_operation(asynchronous_sink& self) : m_self(self)
         {
         }
         //! Destructor
         ~scoped_feeding_operation()
         {
             m_self.complete_feeding_operation();
         }
 
         BOOST_DELETED_FUNCTION(scoped_feeding_operation(scoped_feeding_operation const&))
         BOOST_DELETED_FUNCTION(scoped_feeding_operation& operator= (scoped_feeding_operation const&))
     };
 
     //! A scope guard that resets a flag on destructor
     class scoped_flag
     {
     private:
         frontend_mutex_type& m_Mutex;
         condition_variable_any& m_Cond;
         boost::atomic< bool >& m_Flag;
 
     public:
         explicit scoped_flag(frontend_mutex_type& mut, condition_variable_any& cond, boost::atomic< bool >& f) :
             m_Mutex(mut), m_Cond(cond), m_Flag(f)
         {
         }
         ~scoped_flag()
         {
             try
             {
                 lock_guard< frontend_mutex_type > lock(m_Mutex);
                 m_Flag.store(false, boost::memory_order_relaxed);
                 m_Cond.notify_all();
             }
             catch (...)
             {
             }
         }
 
         BOOST_DELETED_FUNCTION(scoped_flag(scoped_flag const&))
         BOOST_DELETED_FUNCTION(scoped_flag& operator= (scoped_flag const&))
     };
 
 public:
     //! Sink implementation type
     typedef SinkBackendT sink_backend_type;
     //! \cond
     static_assert(has_requirement< typename sink_backend_type::frontend_requirements, synchronized_feeding >::value, "Asynchronous sink frontend is incompatible with the specified backend: thread synchronization requirements are not met");
     //! \endcond
 
 #ifndef BOOST_LOG_DOXYGEN_PASS
 
     //! A pointer type that locks the backend until it's destroyed
     typedef boost::log::aux::locking_ptr< sink_backend_type, backend_mutex_type > locked_backend_ptr;
 
 #else // BOOST_LOG_DOXYGEN_PASS
 
     //! A pointer type that locks the backend until it's destroyed
     typedef implementation_defined locked_backend_ptr;
 
 #endif // BOOST_LOG_DOXYGEN_PASS
 
 private:
     //! Synchronization mutex
     backend_mutex_type m_BackendMutex;
     //! Pointer to the backend
     const shared_ptr< sink_backend_type > m_pBackend;
 
     //! Dedicated record feeding thread
     thread m_DedicatedFeedingThread;
     //! Condition variable to implement blocking operations
     condition_variable_any m_BlockCond;
 
     //! Currently active operation
     operation m_ActiveOperation;
     //! The flag indicates that the feeding loop has to be stopped
     boost::atomic< bool > m_StopRequested;
     //! The flag indicates that queue flush has been requested
     boost::atomic< bool > m_FlushRequested;
 
 public:
     /*!
      * Default constructor. Constructs the sink backend instance.
      * Requires the backend to be default-constructible.
      *
      * \param start_thread If \c true, the frontend creates a thread to feed
      *                     log records to the backend. Otherwise no thread is
      *                     started and it is assumed that the user will call
      *                     \c run, \c feed_records or \c flush himself.
      */
     explicit asynchronous_sink(bool start_thread = true) :
         base_type(true),
         m_pBackend(boost::make_shared< sink_backend_type >()),
         m_ActiveOperation(idle),
         m_StopRequested(false),
         m_FlushRequested(false)
     {
         if (start_thread)
             start_feeding_thread();
     }
     /*!
      * Constructor attaches user-constructed backend instance
      *
      * \param backend Pointer to the backend instance.
      * \param start_thread If \c true, the frontend creates a thread to feed
      *                     log records to the backend. Otherwise no thread is
      *                     started and it is assumed that the user will call
      *                     \c run, \c feed_records or \c flush himself.
      *
      * \pre \a backend is not \c NULL.
      */
     explicit asynchronous_sink(shared_ptr< sink_backend_type > const& backend, bool start_thread = true) :
         base_type(true),
         m_pBackend(backend),
         m_ActiveOperation(idle),
         m_StopRequested(false),
         m_FlushRequested(false)
     {
         if (start_thread)
             start_feeding_thread();
     }
 
     /*!
      * Constructor that passes arbitrary named parameters to the interprocess sink backend constructor.
      * Refer to the backend documentation for the list of supported parameters.
      *
      * The frontend uses the following named parameters:
      *
      *   \li start_thread - If \c true, the frontend creates a thread to feed
      *                      log records to the backend. Otherwise no thread is
      *                      started and it is assumed that the user will call
      *                      \c run, \c feed_records or \c flush himself.
      */
 #ifndef BOOST_LOG_DOXYGEN_PASS
     BOOST_LOG_PARAMETRIZED_CONSTRUCTORS_GEN(BOOST_LOG_SINK_CTOR_FORWARD_INTERNAL, ~)
 #else
     template< typename... Args >
     explicit asynchronous_sink(Args&&... args);
 #endif
 
     /*!
      * Destructor. Implicitly stops the dedicated feeding thread, if one is running.
      */
     ~asynchronous_sink() BOOST_NOEXCEPT BOOST_OVERRIDE
     {
         try
         {
             boost::this_thread::disable_interruption no_interrupts;
             stop();
         }
         catch (...)
         {
             std::terminate();
         }
     }
 
     /*!
      * Locking accessor to the attached backend
      */
     locked_backend_ptr locked_backend()
     {
         return locked_backend_ptr(m_pBackend, m_BackendMutex);
     }
 
     /*!
      * Enqueues the log record to the backend
      */
     void consume(record_view const& rec) BOOST_OVERRIDE
     {
         if (BOOST_UNLIKELY(m_FlushRequested.load(boost::memory_order_acquire)))
         {
             unique_lock< frontend_mutex_type > lock(base_type::frontend_mutex());
             // Wait until flush is done
             while (m_FlushRequested.load(boost::memory_order_acquire))
                 m_BlockCond.wait(lock);
         }
         queue_base_type::enqueue(rec);
     }
 
     /*!
      * The method attempts to pass logging record to the backend
      */
     bool try_consume(record_view const& rec) BOOST_OVERRIDE
     {
         if (!m_FlushRequested.load(boost::memory_order_acquire))
         {
             return queue_base_type::try_enqueue(rec);
         }
         else
             return false;
     }
 
     /*!
      * The method starts record feeding loop and effectively blocks until either of this happens:
      *
      * \li the thread is interrupted due to either standard thread interruption or a call to \c stop
      * \li an exception is thrown while processing a log record in the backend, and the exception is
      *     not terminated by the exception handler, if one is installed
      *
      * \pre The sink frontend must be constructed without spawning a dedicated thread
      */
     void run()
     {
         // First check that no other thread is running
         {
             unique_lock< frontend_mutex_type > lock(base_type::frontend_mutex());
             if (start_feeding_operation(lock, feeding_records))
                 return;
         }
 
         scoped_feeding_operation guard(*this);
 
         // Now start the feeding loop
         while (true)
         {
             do_feed_records();
             if (!m_StopRequested.load(boost::memory_order_acquire))
             {
                 // Block until new record is available
                 record_view rec;
                 if (queue_base_type::dequeue_ready(rec))
                     base_type::feed_record(rec, m_BackendMutex, *m_pBackend);
             }
             else
                 break;
         }
     }
 
     /*!
      * The method softly interrupts record feeding loop. This method must be called when \c run,
      * \c feed_records or \c flush method execution has to be interrupted. Unlike regular thread
      * interruption, calling \c stop will not interrupt the record processing in the middle.
      * Instead, the sink frontend will attempt to finish its business with the record in progress
      * and return afterwards. This method can be called either if the sink was created with
      * an internal dedicated thread, or if the feeding loop was initiated by user.
      *
      * If no record feeding operation is in progress, calling \c stop marks the sink frontend
      * so that the next feeding operation stops immediately.
      *
      * \note Returning from this method does not guarantee that there are no records left buffered
      *       in the sink frontend. It is possible that log records keep coming during and after this
      *       method is called. At some point of execution of this method log records stop being processed,
      *       and all records that come after this point are put into the queue. These records will be
      *       processed upon further calls to \c run or \c feed_records.
      *
      * \note If the record feeding loop is being run in a user's thread (i.e. \c start_thread was specified
      *       as \c false on frontend construction), this method does not guarantee that upon return the thread
      *       has returned from the record feeding loop or that it won't enter it in the future. The method
      *       only ensures that the record feeding thread will eventually return from the feeding loop. It is
      *       user's responsibility to synchronize with the user's record feeding thread.
      */
     void stop()
     {
         boost::thread feeding_thread;
         {
             lock_guard< frontend_mutex_type > lock(base_type::frontend_mutex());
 
             m_StopRequested.store(true, boost::memory_order_release);
             queue_base_type::interrupt_dequeue();
 
             m_DedicatedFeedingThread.swap(feeding_thread);
         }
 
         if (feeding_thread.joinable())
             feeding_thread.join();
     }
 
     /*!
      * The method feeds log records that may have been buffered to the backend and returns
      *
      * \pre The sink frontend must be constructed without spawning a dedicated thread
      */
     void feed_records()
     {
         // First check that no other thread is running
         {
             unique_lock< frontend_mutex_type > lock(base_type::frontend_mutex());
             if (start_feeding_operation(lock, feeding_records))
                 return;
         }
 
         scoped_feeding_operation guard(*this);
 
         // Now start the feeding loop
         do_feed_records();
     }
 
     /*!
      * The method feeds all log records that may have been buffered to the backend and returns.
      * Unlike \c feed_records, in case of ordering queueing the method also feeds records
      * that were enqueued during the ordering window, attempting to drain the queue completely.
      */
     void flush() BOOST_OVERRIDE
     {
         {
             unique_lock< frontend_mutex_type > lock(base_type::frontend_mutex());
             if (static_cast< unsigned int >(m_ActiveOperation & feeding_records) != 0u)
             {
                 // There is already a thread feeding records, let it do the job
                 m_FlushRequested.store(true, boost::memory_order_release);
                 queue_base_type::interrupt_dequeue();
                 while (!m_StopRequested.load(boost::memory_order_acquire) && m_FlushRequested.load(boost::memory_order_acquire))
                     m_BlockCond.wait(lock);
 
                 // The condition may have been signalled when the feeding operation was finishing.
                 // In that case records may not have been flushed, and we do the flush ourselves.
                 if (m_ActiveOperation != idle)
                     return;
             }
 
             m_ActiveOperation = flushing;
             m_FlushRequested.store(true, boost::memory_order_relaxed);
         }
 
         scoped_feeding_operation guard(*this);
 
         do_feed_records();
     }
 
 private:
 #ifndef BOOST_LOG_DOXYGEN_PASS
     //! The method spawns record feeding thread
     void start_feeding_thread()
     {
         boost::thread(run_func(this)).swap(m_DedicatedFeedingThread);
     }
 
     //! Starts record feeding operation. The method blocks or throws if another feeding operation is in progress.
     bool start_feeding_operation(unique_lock< frontend_mutex_type >& lock, operation op)
     {
         while (m_ActiveOperation != idle)
         {
             if (BOOST_UNLIKELY(op == feeding_records && m_ActiveOperation == feeding_records))
                 BOOST_LOG_THROW_DESCR(unexpected_call, "Asynchronous sink frontend already runs a record feeding thread");
 
             if (BOOST_UNLIKELY(m_StopRequested.load(boost::memory_order_relaxed)))
             {
                 m_StopRequested.store(false, boost::memory_order_relaxed);
                 return true;
             }
 
             m_BlockCond.wait(lock);
         }
 
         m_ActiveOperation = op;
 
         return false;
     }
 
     //! Completes record feeding operation
     void complete_feeding_operation() BOOST_NOEXCEPT
     {
         try
         {
             lock_guard< frontend_mutex_type > lock(base_type::frontend_mutex());
             m_ActiveOperation = idle;
             m_StopRequested.store(false, boost::memory_order_relaxed);
             m_BlockCond.notify_all();
         }
         catch (...)
         {
         }
     }
 
     //! The record feeding loop
     void do_feed_records()
     {
         while (!m_StopRequested.load(boost::memory_order_acquire))
         {
             record_view rec;
             bool dequeued = false;
             if (BOOST_LIKELY(!m_FlushRequested.load(boost::memory_order_acquire)))
                 dequeued = queue_base_type::try_dequeue_ready(rec);
             else
                 dequeued = queue_base_type::try_dequeue(rec);
 
             if (dequeued)
                 base_type::feed_record(rec, m_BackendMutex, *m_pBackend);
             else
                 break;
         }
 
         if (BOOST_UNLIKELY(m_FlushRequested.load(boost::memory_order_acquire)))
         {
             scoped_flag guard(base_type::frontend_mutex(), m_BlockCond, m_FlushRequested);
             base_type::flush_backend(m_BackendMutex, *m_pBackend);
         }
     }
 #endif // BOOST_LOG_DOXYGEN_PASS
 };
 
 #undef BOOST_LOG_SINK_CTOR_FORWARD_INTERNAL_1
 #undef BOOST_LOG_SINK_CTOR_FORWARD_INTERNAL_N
 #undef BOOST_LOG_SINK_CTOR_FORWARD_INTERNAL
 
 } // namespace sinks
 
 BOOST_LOG_CLOSE_NAMESPACE // namespace log
 
 } // namespace boost
 
 #include <boost/log/detail/footer.hpp>
 
 #endif // BOOST_LOG_SINKS_ASYNC_FRONTEND_HPP_INCLUDED_

This page was automatically generated by the 2.3.7 LXR engine. The LXR team