cds/urcu/details/gpt.h

   1 //$$CDS-header$$1
   2
   3 #ifndef CDSLIB_URCU_DETAILS_GPT_H
   4 #define CDSLIB_URCU_DETAILS_GPT_H
   5
   6 #include <mutex>    //unique_lock
   7 #include <cds/urcu/details/gp.h>
   8 #include <cds/urcu/dispose_thread.h>
   9 #include <cds/algo/backoff_strategy.h>
  10 #include <cds/container/vyukov_mpmc_cycle_queue.h>
  11
  12 namespace cds { namespace urcu {
  13
  14     /// User-space general-purpose RCU with deferred threaded reclamation
  15     /**
  16         @headerfile cds/urcu/general_threaded.h
  17
  18         This implementation is similar to \ref general_buffered but separate thread is created
  19         for deleting the retired objects. Like \p %general_buffered, the class contains an internal buffer
  20         where retired objects are accumulated. When the buffer becomes full,
  21         the RCU \p synchronize function is called that waits until all reader/updater threads end up their read-side critical sections,
  22         i.e. until the RCU quiescent state will come. After that the "work ready" message is sent to reclamation tread.
  23         The reclamation thread frees the buffer.
  24         This synchronization cycle may be called in any thread that calls \ref retire_ptr function.
  25
  26         There is a wrapper \ref cds_urcu_general_threaded_gc "gc<general_threaded>" for \p %general_threaded class
  27         that provides unified RCU interface. You should use this wrapper class instead \p %general_threaded
  28
  29         Template arguments:
  30         - \p Buffer - buffer type with FIFO semantics. Default is cds::container::VyukovMPMCCycleQueue. See \ref general_buffered
  31             for description of buffer's interface. The buffer contains the objects of \ref epoch_retired_ptr
  32             type that contains additional \p m_nEpoch field. This field specifies an epoch when the object
  33             has been placed into the buffer. The \p %general_threaded object has a global epoch counter
  34             that is incremented on each \p synchronize call. The epoch is used internally to prevent early deletion.
  35         - \p Lock - mutex type, default is \p std::mutex
  36         - \p DisposerThread - the reclamation thread class. Default is \ref cds::urcu::dispose_thread,
  37             see the description of this class for required interface.
  38         - \p Backoff - back-off schema, default is cds::backoff::Default
  39     */
  40     template <
  41         class Buffer = cds::container::VyukovMPMCCycleQueue< epoch_retired_ptr >
  42         ,class Lock = std::mutex
  43         ,class DisposerThread = dispose_thread<Buffer>
  44         ,class Backoff = cds::backoff::Default
  45     >
  46     class general_threaded: public details::gp_singleton< general_threaded_tag >
  47     {
  48         //@cond
  49         typedef details::gp_singleton< general_threaded_tag > base_class;
  50         //@endcond
  51     public:
  52         typedef Buffer          buffer_type ;   ///< Buffer type
  53         typedef Lock            lock_type   ;   ///< Lock type
  54         typedef Backoff         back_off    ;   ///< Back-off scheme
  55         typedef DisposerThread  disposer_thread ;   ///< Disposer thread type
  56
  57         typedef general_threaded_tag    rcu_tag ;       ///< Thread-side RCU part
  58         typedef base_class::thread_gc   thread_gc ;     ///< Access lock class
  59         typedef typename thread_gc::scoped_lock scoped_lock ; ///< Access lock class
  60
  61         static bool const c_bBuffered = true ; ///< This RCU buffers disposed elements
  62
  63     protected:
  64         //@cond
  65         typedef details::gp_singleton_instance< rcu_tag >    singleton_ptr;
  66
  67         struct scoped_disposer {
  68             void operator ()( general_threaded * p )
  69             {
  70                 delete p;
  71             }
  72         };
  73         //@endcond
  74
  75     protected:
  76         //@cond
  77         buffer_type                     m_Buffer;
  78         atomics::atomic<uint64_t>    m_nCurEpoch;
  79         lock_type                       m_Lock;
  80         size_t const                    m_nCapacity;
  81         disposer_thread                 m_DisposerThread;
  82         //@endcond
  83
  84     public:
  85         /// Returns singleton instance
  86         static general_threaded * instance()
  87         {
  88             return static_cast<general_threaded *>( base_class::instance() );
  89         }
  90         /// Checks if the singleton is created and ready to use
  91         static bool isUsed()
  92         {
  93             return singleton_ptr::s_pRCU != nullptr;
  94         }
  95
  96     protected:
  97         //@cond
  98         general_threaded( size_t nBufferCapacity )
  99             : m_Buffer( nBufferCapacity )
 100             , m_nCurEpoch( 1 )
 101             , m_nCapacity( nBufferCapacity )
 102         {}
 103
 104         void flip_and_wait()
 105         {
 106             back_off bkoff;
 107             base_class::flip_and_wait( bkoff );
 108         }
 109
 110         // Return: true - synchronize has been called, false - otherwise
 111         bool push_buffer( epoch_retired_ptr& p )
 112         {
 113             bool bPushed = m_Buffer.push( p );
 114             if ( !bPushed || m_Buffer.size() >= capacity() ) {
 115                 synchronize();
 116                 if ( !bPushed ) {
 117                     CDS_TSAN_ANNOTATE_IGNORE_RW_BEGIN;
 118                     p.free();
 119                     CDS_TSAN_ANNOTATE_IGNORE_RW_END;
 120                 }
 121                 return true;
 122             }
 123             return false;
 124         }
 125
 126         //@endcond
 127
 128     public:
 129         //@cond
 130         ~general_threaded()
 131         {}
 132         //@endcond
 133
 134         /// Creates singleton object and starts reclamation thread
 135         /**
 136             The \p nBufferCapacity parameter defines RCU threshold.
 137         */
 138         static void Construct( size_t nBufferCapacity = 256 )
 139         {
 140             if ( !singleton_ptr::s_pRCU ) {
 141                 std::unique_ptr< general_threaded, scoped_disposer > pRCU( new general_threaded( nBufferCapacity ) );
 142                 pRCU->m_DisposerThread.start();
 143
 144                 singleton_ptr::s_pRCU = pRCU.release();
 145             }
 146         }
 147
 148         /// Destroys singleton object and terminates internal reclamation thread
 149         static void Destruct( bool bDetachAll = false )
 150         {
 151             if ( isUsed() ) {
 152                 general_threaded * pThis = instance();
 153                 if ( bDetachAll )
 154                     pThis->m_ThreadList.detach_all();
 155
 156                 pThis->m_DisposerThread.stop( pThis->m_Buffer, pThis->m_nCurEpoch.load( atomics::memory_order_acquire ));
 157
 158                 delete pThis;
 159                 singleton_ptr::s_pRCU = nullptr;
 160             }
 161         }
 162
 163     public:
 164         /// Retires \p p pointer
 165         /**
 166             The method pushes \p p pointer to internal buffer.
 167             When the buffer becomes full \ref synchronize function is called
 168             to wait for the end of grace period and then
 169             a message is sent to the reclamation thread.
 170         */
 171         virtual void retire_ptr( retired_ptr& p )
 172         {
 173             if ( p.m_p ) {
 174                 epoch_retired_ptr ep( p, m_nCurEpoch.load( atomics::memory_order_acquire ) );
 175                 push_buffer( ep );
 176             }
 177         }
 178
 179         /// Retires the pointer chain [\p itFirst, \p itLast)
 180         template <typename ForwardIterator>
 181         void batch_retire( ForwardIterator itFirst, ForwardIterator itLast )
 182         {
 183             uint64_t nEpoch = m_nCurEpoch.load( atomics::memory_order_relaxed );
 184             while ( itFirst != itLast ) {
 185                 epoch_retired_ptr p( *itFirst, nEpoch );
 186                 ++itFirst;
 187                 push_buffer( p );
 188             }
 189         }
 190
 191         /// Waits to finish a grace period and calls disposing thread
 192         void synchronize()
 193         {
 194             synchronize( false );
 195         }
 196
 197         //@cond
 198         void synchronize( bool bSync )
 199         {
 200             uint64_t nPrevEpoch = m_nCurEpoch.fetch_add( 1, atomics::memory_order_release );
 201
 202             atomics::atomic_thread_fence( atomics::memory_order_acquire );
 203             {
 204                 std::unique_lock<lock_type> sl( m_Lock );
 205                 flip_and_wait();
 206                 flip_and_wait();
 207
 208                 m_DisposerThread.dispose( m_Buffer, nPrevEpoch, bSync );
 209             }
 210             atomics::atomic_thread_fence( atomics::memory_order_release );
 211         }
 212         void force_dispose()
 213         {
 214             synchronize( true );
 215         }
 216         //@endcond
 217
 218         /// Returns the threshold of internal buffer
 219         size_t capacity() const
 220         {
 221             return m_nCapacity;
 222         }
 223     };
 224 }} // namespace cds::urcu
 225
 226 #endif // #ifndef CDSLIB_URCU_DETAILS_GPT_H