cds/urcu/details/gpt.h

   1 /*
   2     This file is a part of libcds - Concurrent Data Structures library
   3
   4     (C) Copyright Maxim Khizhinsky (libcds.dev@gmail.com) 2006-2017
   5
   6     Source code repo: http://github.com/khizmax/libcds/
   7     Download: http://sourceforge.net/projects/libcds/files/
   8
   9     Redistribution and use in source and binary forms, with or without
  10     modification, are permitted provided that the following conditions are met:
  11
  12     * Redistributions of source code must retain the above copyright notice, this
  13       list of conditions and the following disclaimer.
  14
  15     * Redistributions in binary form must reproduce the above copyright notice,
  16       this list of conditions and the following disclaimer in the documentation
  17       and/or other materials provided with the distribution.
  18
  19     THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
  20     AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
  21     IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
  22     DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE
  23     FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
  24     DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
  25     SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
  26     CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
  27     OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
  28     OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
  29 */
  30
  31 #ifndef CDSLIB_URCU_DETAILS_GPT_H
  32 #define CDSLIB_URCU_DETAILS_GPT_H
  33
  34 #include <mutex>    //unique_lock
  35 #include <limits>
  36 #include <cds/urcu/details/gp.h>
  37 #include <cds/urcu/dispose_thread.h>
  38 #include <cds/algo/backoff_strategy.h>
  39 #include <cds/container/vyukov_mpmc_cycle_queue.h>
  40
  41 namespace cds { namespace urcu {
  42
  43     /// User-space general-purpose RCU with deferred threaded reclamation
  44     /**
  45         @headerfile cds/urcu/general_threaded.h
  46
  47         This implementation is similar to \ref general_buffered but separate thread is created
  48         for deleting the retired objects. Like \p %general_buffered, the class contains an internal buffer
  49         where retired objects are accumulated. When the buffer becomes full,
  50         the RCU \p synchronize() function is called that waits until all reader/updater threads end up their read-side critical sections,
  51         i.e. until the RCU quiescent state will come. After that the "work ready" message is sent to reclamation thread.
  52         The reclamation thread frees the buffer.
  53         This synchronization cycle may be called in any thread that calls \p retire_ptr() function.
  54
  55         There is a wrapper \ref cds_urcu_general_threaded_gc "gc<general_threaded>" for \p %general_threaded class
  56         that provides unified RCU interface. You should use this wrapper class instead \p %general_threaded
  57
  58         The \p Buffer contains items of \ref cds_urcu_retired_ptr "epoch_retired_ptr" type
  59
  60         and it should support a multiple producer/single consumer queue with the following interface:
  61         - <tt> bool push( epoch_retired_ptr& p ) </tt> - places the retired pointer \p p into queue. If the function
  62         returns \p false it means that the buffer is full and RCU synchronization cycle must be processed.
  63         - <tt>epoch_retired_ptr * front() </tt> - returns a pointer to the top element or \p nullptr if the buffer is empty.
  64         - <tt>bool pop_front() </tt> - pops the top element; returns \p false if the buffer is empty.
  65         - <tt>size_t size()</tt> - returns queue's item count.
  66
  67         The buffer is considered as full if \p push() returns \p false or the buffer size reaches the RCU threshold.
  68
  69         Template arguments:
  70         - \p Buffer - MPSC (muliple producer/single consumer) buffer type with FIFO semantics.
  71
  72             Default is \p cds::container::VyukovMPSCCycleQueue. The buffer contains the objects of \ref epoch_retired_ptr
  73             type that contains additional \p m_nEpoch field. This field specifies an epoch when the object
  74             has been placed into the buffer. The \p %general_threaded object has a global epoch counter
  75             that is incremented on each \p synchronize() call. The epoch is used internally to prevent early deletion.
  76         - \p Lock - mutex type, default is \p std::mutex
  77         - \p DisposerThread - the reclamation thread class. Default is \ref cds::urcu::dispose_thread,
  78             see the description of this class for required interface.
  79         - \p Backoff - back-off schema, default is cds::backoff::Default
  80     */
  81     template <
  82         class Buffer = cds::container::VyukovMPSCCycleQueue< epoch_retired_ptr >
  83         ,class Lock = std::mutex
  84         ,class DisposerThread = dispose_thread<Buffer>
  85         ,class Backoff = cds::backoff::Default
  86     >
  87     class general_threaded: public details::gp_singleton< general_threaded_tag >
  88     {
  89         //@cond
  90         typedef details::gp_singleton< general_threaded_tag > base_class;
  91         //@endcond
  92     public:
  93         typedef Buffer          buffer_type ;   ///< Buffer type
  94         typedef Lock            lock_type   ;   ///< Lock type
  95         typedef Backoff         back_off    ;   ///< Back-off scheme
  96         typedef DisposerThread  disposer_thread ;   ///< Disposer thread type
  97
  98         typedef general_threaded_tag    rcu_tag ;       ///< Thread-side RCU part
  99         typedef base_class::thread_gc   thread_gc ;     ///< Access lock class
 100         typedef typename thread_gc::scoped_lock scoped_lock ; ///< Access lock class
 101
 102         static bool const c_bBuffered = true ; ///< This RCU buffers disposed elements
 103
 104     protected:
 105         //@cond
 106         typedef details::gp_singleton_instance< rcu_tag >    singleton_ptr;
 107
 108         struct scoped_disposer {
 109             void operator ()( general_threaded * p )
 110             {
 111                 delete p;
 112             }
 113         };
 114         //@endcond
 115
 116     protected:
 117         //@cond
 118         buffer_type               m_Buffer;
 119         atomics::atomic<uint64_t> m_nCurEpoch;
 120         lock_type                 m_Lock;
 121         size_t const              m_nCapacity;
 122         disposer_thread           m_DisposerThread;
 123         //@endcond
 124
 125     public:
 126         /// Returns singleton instance
 127         static general_threaded * instance()
 128         {
 129             return static_cast<general_threaded *>( base_class::instance());
 130         }
 131         /// Checks if the singleton is created and ready to use
 132         static bool isUsed()
 133         {
 134             return singleton_ptr::s_pRCU != nullptr;
 135         }
 136
 137     protected:
 138         //@cond
 139         general_threaded( size_t nBufferCapacity )
 140             : m_Buffer( nBufferCapacity )
 141             , m_nCurEpoch( 1 )
 142             , m_nCapacity( nBufferCapacity )
 143         {}
 144
 145         void flip_and_wait()
 146         {
 147             back_off bkoff;
 148             base_class::flip_and_wait( bkoff );
 149         }
 150
 151         // Return: true - synchronize has been called, false - otherwise
 152         bool push_buffer( epoch_retired_ptr&& p )
 153         {
 154             bool bPushed = m_Buffer.push( p );
 155             if ( !bPushed || m_Buffer.size() >= capacity()) {
 156                 synchronize();
 157                 if ( !bPushed )
 158                     p.free();
 159                 return true;
 160             }
 161             return false;
 162         }
 163
 164         //@endcond
 165
 166     public:
 167         //@cond
 168         ~general_threaded()
 169         {}
 170         //@endcond
 171
 172         /// Creates singleton object and starts reclamation thread
 173         /**
 174             The \p nBufferCapacity parameter defines RCU threshold.
 175         */
 176         static void Construct( size_t nBufferCapacity = 256 )
 177         {
 178             if ( !singleton_ptr::s_pRCU ) {
 179                 std::unique_ptr< general_threaded, scoped_disposer > pRCU( new general_threaded( nBufferCapacity ));
 180                 pRCU->m_DisposerThread.start();
 181
 182                 singleton_ptr::s_pRCU = pRCU.release();
 183             }
 184         }
 185
 186         /// Destroys singleton object and terminates internal reclamation thread
 187         static void Destruct( bool bDetachAll = false )
 188         {
 189             if ( isUsed()) {
 190                 general_threaded * pThis = instance();
 191                 if ( bDetachAll )
 192                     pThis->m_ThreadList.detach_all();
 193
 194                 pThis->m_DisposerThread.stop( pThis->m_Buffer, std::numeric_limits< uint64_t >::max());
 195
 196                 delete pThis;
 197                 singleton_ptr::s_pRCU = nullptr;
 198             }
 199         }
 200
 201     public:
 202         /// Retires \p p pointer
 203         /**
 204             The method pushes \p p pointer to internal buffer.
 205             When the buffer becomes full \ref synchronize function is called
 206             to wait for the end of grace period and then
 207             a message is sent to the reclamation thread.
 208         */
 209         virtual void retire_ptr( retired_ptr& p )
 210         {
 211             if ( p.m_p )
 212                 push_buffer( epoch_retired_ptr( p, m_nCurEpoch.load( atomics::memory_order_acquire )));
 213         }
 214
 215         /// Retires the pointer chain [\p itFirst, \p itLast)
 216         template <typename ForwardIterator>
 217         void batch_retire( ForwardIterator itFirst, ForwardIterator itLast )
 218         {
 219             uint64_t nEpoch = m_nCurEpoch.load( atomics::memory_order_acquire );
 220             while ( itFirst != itLast ) {
 221                 epoch_retired_ptr ep( *itFirst, nEpoch );
 222                 ++itFirst;
 223                 push_buffer( std::move(ep));
 224             }
 225         }
 226
 227         /// Retires the pointer chain until \p Func returns \p nullptr retired pointer
 228         template <typename Func>
 229         void batch_retire( Func e )
 230         {
 231             uint64_t nEpoch = m_nCurEpoch.load( atomics::memory_order_acquire );
 232             for ( retired_ptr p{ e() }; p.m_p; ) {
 233                 epoch_retired_ptr ep( p, nEpoch );
 234                 p = e();
 235                 push_buffer( std::move(ep));
 236             }
 237         }
 238
 239         /// Waits to finish a grace period and calls disposing thread
 240         void synchronize()
 241         {
 242             synchronize( false );
 243         }
 244
 245         //@cond
 246         void synchronize( bool bSync )
 247         {
 248             uint64_t nPrevEpoch = m_nCurEpoch.fetch_add( 1, atomics::memory_order_release );
 249             {
 250                 std::unique_lock<lock_type> sl( m_Lock );
 251                 flip_and_wait();
 252                 flip_and_wait();
 253             }
 254             m_DisposerThread.dispose( m_Buffer, nPrevEpoch, bSync );
 255         }
 256         void force_dispose()
 257         {
 258             synchronize( true );
 259         }
 260         //@endcond
 261
 262         /// Returns the threshold of internal buffer
 263         size_t capacity() const
 264         {
 265             return m_nCapacity;
 266         }
 267     };
 268
 269     /// User-space general-purpose RCU with deferred threaded reclamation (stripped version)
 270     /**
 271         @headerfile cds/urcu/general_threaded.h
 272
 273         This short version of \p general_threaded is intended for stripping debug info.
 274         If you use \p %general_threaded with default template arguments you may use
 275         this stripped version. All functionality of both classes are identical.
 276     */
 277     class general_threaded_stripped: public general_threaded<>
 278     {};
 279
 280 }} // namespace cds::urcu
 281
 282 #endif // #ifndef CDSLIB_URCU_DETAILS_GPT_H