cds/urcu/details/gpb.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_GPB_H
  32 #define CDSLIB_URCU_DETAILS_GPB_H
  33
  34 #include <mutex>
  35 #include <limits>
  36 #include <cds/urcu/details/gp.h>
  37 #include <cds/algo/backoff_strategy.h>
  38 #include <cds/container/vyukov_mpmc_cycle_queue.h>
  39
  40 namespace cds { namespace urcu {
  41
  42     /// User-space general-purpose RCU with deferred (buffered) reclamation
  43     /**
  44         @headerfile cds/urcu/general_buffered.h
  45
  46         This URCU implementation contains an internal buffer where retired objects are
  47         accumulated. When the buffer becomes full, the RCU \p synchronize function is called
  48         that waits until all reader/updater threads end up their read-side critical sections,
  49         i.e. until the RCU quiescent state will come. After that the buffer and all retired objects are freed.
  50         This synchronization cycle may be called in any thread that calls \p retire_ptr function.
  51
  52         The \p Buffer contains items of \ref cds_urcu_retired_ptr "epoch_retired_ptr" type and it should support a queue interface with
  53         three function:
  54         - <tt> bool push( retired_ptr& p ) </tt> - places the retired pointer \p p into queue. If the function
  55             returns \p false it means that the buffer is full and RCU synchronization cycle must be processed.
  56         - <tt>bool pop( retired_ptr& p ) </tt> - pops queue's head item into \p p parameter; if the queue is empty
  57             this function must return \p false
  58         - <tt>size_t size()</tt> - returns queue's item count.
  59
  60         The buffer is considered as full if \p push() returns \p false or the buffer size reaches the RCU threshold.
  61
  62         There is a wrapper \ref cds_urcu_general_buffered_gc "gc<general_buffered>" for \p %general_buffered class
  63         that provides unified RCU interface. You should use this wrapper class instead \p %general_buffered
  64
  65         Template arguments:
  66         - \p Buffer - buffer type. Default is \p cds::container::VyukovMPMCCycleQueue
  67         - \p Lock - mutex type, default is \p std::mutex
  68         - \p Backoff - back-off schema, default is cds::backoff::Default
  69     */
  70     template <
  71         class Buffer = cds::container::VyukovMPMCCycleQueue< epoch_retired_ptr >
  72         ,class Lock = std::mutex
  73         ,class Backoff = cds::backoff::Default
  74     >
  75     class general_buffered: public details::gp_singleton< general_buffered_tag >
  76     {
  77         //@cond
  78         typedef details::gp_singleton< general_buffered_tag > base_class;
  79         //@endcond
  80     public:
  81         typedef general_buffered_tag rcu_tag ;  ///< RCU tag
  82         typedef Buffer  buffer_type ;   ///< Buffer type
  83         typedef Lock    lock_type   ;   ///< Lock type
  84         typedef Backoff back_off    ;   ///< Back-off type
  85
  86         typedef base_class::thread_gc thread_gc ;   ///< Thread-side RCU part
  87         typedef typename thread_gc::scoped_lock scoped_lock ; ///< Access lock class
  88
  89         //@cond
  90         static bool const c_bBuffered = true ; ///< Bufferized RCU
  91         //@endcond
  92
  93     protected:
  94         //@cond
  95         typedef details::gp_singleton_instance< rcu_tag >    singleton_ptr;
  96         //@endcond
  97
  98     protected:
  99         //@cond
 100         buffer_type                m_Buffer;
 101         atomics::atomic<uint64_t>  m_nCurEpoch;
 102         lock_type                  m_Lock;
 103         size_t const               m_nCapacity;
 104         //@endcond
 105
 106     public:
 107         /// Returns singleton instance
 108         static general_buffered * instance()
 109         {
 110             return static_cast<general_buffered *>( base_class::instance());
 111         }
 112         /// Checks if the singleton is created and ready to use
 113         static bool isUsed()
 114         {
 115             return singleton_ptr::s_pRCU != nullptr;
 116         }
 117
 118     protected:
 119         //@cond
 120         general_buffered( size_t nBufferCapacity )
 121             : m_Buffer( nBufferCapacity )
 122             , m_nCurEpoch(0)
 123             , m_nCapacity( nBufferCapacity )
 124         {}
 125
 126         ~general_buffered()
 127         {
 128             clear_buffer( std::numeric_limits< uint64_t >::max());
 129         }
 130
 131         void flip_and_wait()
 132         {
 133             back_off bkoff;
 134             base_class::flip_and_wait( bkoff );
 135         }
 136
 137         void clear_buffer( uint64_t nEpoch )
 138         {
 139             epoch_retired_ptr p;
 140             while ( m_Buffer.pop( p )) {
 141                 if ( p.m_nEpoch <= nEpoch ) {
 142                     p.free();
 143                 }
 144                 else {
 145                     push_buffer( std::move(p));
 146                     break;
 147                 }
 148             }
 149         }
 150
 151         // Return: \p true - synchronize has been called, \p false - otherwise
 152         bool push_buffer( epoch_retired_ptr&& ep )
 153         {
 154             bool bPushed = m_Buffer.push( ep );
 155             if ( !bPushed || m_Buffer.size() >= capacity()) {
 156                 synchronize();
 157                 if ( !bPushed ) {
 158                     ep.free();
 159                 }
 160                 return true;
 161             }
 162             return false;
 163         }
 164         //@endcond
 165
 166     public:
 167         /// Creates singleton object
 168         /**
 169             The \p nBufferCapacity parameter defines RCU threshold.
 170         */
 171         static void Construct( size_t nBufferCapacity = 256 )
 172         {
 173             if ( !singleton_ptr::s_pRCU )
 174                 singleton_ptr::s_pRCU = new general_buffered( nBufferCapacity );
 175         }
 176
 177         /// Destroys singleton object
 178         static void Destruct( bool bDetachAll = false )
 179         {
 180             if ( isUsed()) {
 181                 instance()->clear_buffer( std::numeric_limits< uint64_t >::max());
 182                 if ( bDetachAll )
 183                     instance()->m_ThreadList.detach_all();
 184                 delete instance();
 185                 singleton_ptr::s_pRCU = nullptr;
 186             }
 187         }
 188
 189     public:
 190         /// Retire \p p pointer
 191         /**
 192             The method pushes \p p pointer to internal buffer.
 193             When the buffer becomes full \ref synchronize function is called
 194             to wait for the end of grace period and then to free all pointers from the buffer.
 195         */
 196         virtual void retire_ptr( retired_ptr& p ) override
 197         {
 198             if ( p.m_p )
 199                 push_buffer( epoch_retired_ptr( p, m_nCurEpoch.load( atomics::memory_order_relaxed )));
 200         }
 201
 202         /// Retires the pointer chain [\p itFirst, \p itLast)
 203         template <typename ForwardIterator>
 204         void batch_retire( ForwardIterator itFirst, ForwardIterator itLast )
 205         {
 206             uint64_t nEpoch = m_nCurEpoch.load( atomics::memory_order_relaxed );
 207             while ( itFirst != itLast ) {
 208                 epoch_retired_ptr ep( *itFirst, nEpoch );
 209                 ++itFirst;
 210                 push_buffer( std::move(ep));
 211             }
 212         }
 213
 214         /// Retires the pointer chain until \p Func returns \p nullptr retired pointer
 215         template <typename Func>
 216         void batch_retire( Func e )
 217         {
 218             uint64_t nEpoch = m_nCurEpoch.load( atomics::memory_order_relaxed );
 219             for ( retired_ptr p{ e() }; p.m_p; ) {
 220                 epoch_retired_ptr ep( p, nEpoch );
 221                 p = e();
 222                 push_buffer( std::move(ep));
 223             }
 224         }
 225
 226         /// Wait to finish a grace period and then clear the buffer
 227         void synchronize()
 228         {
 229             epoch_retired_ptr ep( retired_ptr(), m_nCurEpoch.load( atomics::memory_order_relaxed ));
 230             synchronize( ep );
 231         }
 232
 233         //@cond
 234         bool synchronize( epoch_retired_ptr& ep )
 235         {
 236             uint64_t nEpoch;
 237             {
 238                 std::unique_lock<lock_type> sl( m_Lock );
 239                 if ( ep.m_p && m_Buffer.push( ep ))
 240                     return false;
 241                 nEpoch = m_nCurEpoch.fetch_add( 1, atomics::memory_order_relaxed );
 242                 flip_and_wait();
 243                 flip_and_wait();
 244             }
 245             clear_buffer( nEpoch );
 246             return true;
 247         }
 248         //@endcond
 249
 250         /// Returns internal buffer capacity
 251         size_t capacity() const
 252         {
 253             return m_nCapacity;
 254         }
 255     };
 256
 257     /// User-space general-purpose RCU with deferred (buffered) reclamation (stripped version)
 258     /**
 259         @headerfile cds/urcu/general_buffered.h
 260
 261         This short version of \p general_buffered is intended for stripping debug info.
 262         If you use \p %general_buffered with default template arguments you may use
 263         this stripped version. All functionality of both classes are identical.
 264     */
 265     class general_buffered_stripped: public general_buffered<>
 266     {};
 267
 268 }} // namespace cds::urcu
 269
 270 #endif // #ifndef CDSLIB_URCU_DETAILS_GPB_H