3 #ifndef _CDS_URCU_DETAILS_SIG_THREADED_H
4 #define _CDS_URCU_DETAILS_SIG_THREADED_H
6 #include <cds/urcu/details/sh.h>
7 #ifdef CDS_URCU_SIGNAL_HANDLING_ENABLED
9 #include <cds/urcu/dispose_thread.h>
10 #include <cds/algo/backoff_strategy.h>
11 #include <cds/container/vyukov_mpmc_cycle_queue.h>
13 namespace cds { namespace urcu {
15 /// User-space signal-handled RCU with deferred threaded reclamation
17 @headerfile cds/urcu/signal_threaded.h
19 This implementation is similar to \ref signal_buffered but separate thread is created
20 for deleting the retired objects. Like \p %signal_buffered, the class contains an internal buffer
21 where retired objects are accumulated. When the buffer becomes full,
22 the RCU \p synchronize function is called that waits until all reader/updater threads end up their read-side critical sections,
23 i.e. until the RCU quiescent state will come. After that the "work ready" message is sent to reclamation thread.
24 The reclamation thread frees the buffer.
25 This synchronization cycle may be called in any thread that calls \ref retire_ptr function.
27 There is a wrapper \ref cds_urcu_signal_threaded_gc "gc<signal_threaded>" for \p %signal_threaded class
28 that provides unified RCU interface. You should use this wrapper class instead \p %signal_threaded
31 - \p Buffer - buffer type with FIFO semantics. Default is cds::container::VyukovMPMCCycleQueue. See \ref signal_buffered
32 for description of buffer's interface. The buffer contains the objects of \ref epoch_retired_ptr
33 type that contains additional \p m_nEpoch field. This field specifies an epoch when the object
34 has been placed into the buffer. The \p %signal_threaded object has a global epoch counter
35 that is incremented on each \p synchronize call. The epoch is used internally to prevent early deletion.
36 - \p Lock - mutex type, default is \p std::mutex
37 - \p DisposerThread - the reclamation thread class. Default is \ref cds::urcu::dispose_thread,
38 see the description of this class for required interface.
39 - \p Backoff - back-off schema, default is cds::backoff::Default
42 class Buffer = cds::container::VyukovMPMCCycleQueue<
44 ,cds::opt::buffer< cds::opt::v::dynamic_buffer< epoch_retired_ptr > >
46 ,class Lock = std::mutex
47 ,class DisposerThread = dispose_thread<Buffer>
48 ,class Backoff = cds::backoff::Default
50 class signal_threaded: public details::sh_singleton< signal_threaded_tag >
53 typedef details::sh_singleton< signal_threaded_tag > base_class;
56 typedef Buffer buffer_type ; ///< Buffer type
57 typedef Lock lock_type ; ///< Lock type
58 typedef Backoff back_off ; ///< Back-off scheme
59 typedef DisposerThread disposer_thread ; ///< Disposer thread type
61 typedef signal_threaded_tag rcu_tag ; ///< Thread-side RCU part
62 typedef base_class::thread_gc thread_gc ; ///< Access lock class
63 typedef typename thread_gc::scoped_lock scoped_lock ; ///< Access lock class
65 static bool const c_bBuffered = true ; ///< This RCU buffers disposed elements
69 typedef details::sh_singleton_instance< rcu_tag > singleton_ptr;
71 struct scoped_disposer {
72 void operator ()( signal_threaded * p )
82 atomics::atomic<uint64_t> m_nCurEpoch;
84 size_t const m_nCapacity;
85 disposer_thread m_DisposerThread;
89 /// Returns singleton instance
90 static signal_threaded * instance()
92 return static_cast<signal_threaded *>( base_class::instance() );
94 /// Checks if the singleton is created and ready to use
97 return singleton_ptr::s_pRCU != nullptr;
102 signal_threaded( size_t nBufferCapacity, int nSignal = SIGUSR1 )
103 : base_class( nSignal )
104 , m_Buffer( nBufferCapacity )
106 , m_nCapacity( nBufferCapacity )
109 // Return: true - synchronize has been called, false - otherwise
110 bool push_buffer( epoch_retired_ptr& p )
112 bool bPushed = m_Buffer.push( p );
113 if ( !bPushed || m_Buffer.size() >= capacity() ) {
130 /// Creates singleton object and starts reclamation thread
132 The \p nBufferCapacity parameter defines RCU threshold.
134 The \p nSignal parameter defines a signal number stated for RCU, default is \p SIGUSR1
136 static void Construct( size_t nBufferCapacity = 256, int nSignal = SIGUSR1 )
138 if ( !singleton_ptr::s_pRCU ) {
139 std::unique_ptr< signal_threaded, scoped_disposer > pRCU( new signal_threaded( nBufferCapacity, nSignal ) );
140 pRCU->m_DisposerThread.start();
142 singleton_ptr::s_pRCU = pRCU.release();
146 /// Destroys singleton object and terminates internal reclamation thread
147 static void Destruct( bool bDetachAll = false )
150 signal_threaded * pThis = instance();
152 pThis->m_ThreadList.detach_all();
154 pThis->m_DisposerThread.stop( pThis->m_Buffer, pThis->m_nCurEpoch.load( atomics::memory_order_acquire ));
157 singleton_ptr::s_pRCU = nullptr;
162 /// Retires \p p pointer
164 The method pushes \p p pointer to internal buffer.
165 When the buffer becomes full \ref synchronize function is called
166 to wait for the end of grace period and then
167 a message is sent to the reclamation thread.
169 virtual void retire_ptr( retired_ptr& p )
172 epoch_retired_ptr ep( p, m_nCurEpoch.load( atomics::memory_order_acquire ) );
177 /// Retires the pointer chain [\p itFirst, \p itLast)
178 template <typename ForwardIterator>
179 void batch_retire( ForwardIterator itFirst, ForwardIterator itLast )
181 uint64_t nEpoch = m_nCurEpoch.load( atomics::memory_order_relaxed );
182 while ( itFirst != itLast ) {
183 epoch_retired_ptr p( *itFirst, nEpoch );
189 /// Waits to finish a grace period and calls disposing thread
192 synchronize( false );
196 void synchronize( bool bSync )
198 uint64_t nPrevEpoch = m_nCurEpoch.fetch_add( 1, atomics::memory_order_release );
200 atomics::atomic_thread_fence( atomics::memory_order_acquire );
202 cds::lock::scoped_lock<lock_type> sl( m_Lock );
205 base_class::force_membar_all_threads( bkOff );
206 base_class::switch_next_epoch();
208 base_class::wait_for_quiescent_state( bkOff );
209 base_class::switch_next_epoch();
211 base_class::wait_for_quiescent_state( bkOff );
212 base_class::force_membar_all_threads( bkOff );
214 m_DisposerThread.dispose( m_Buffer, nPrevEpoch, bSync );
223 /// Returns the threshold of internal buffer
224 size_t capacity() const
229 /// Returns the signal number stated for RCU
230 int signal_no() const
232 return base_class::signal_no();
235 }} // namespace cds::urcu
237 #endif // #ifdef CDS_URCU_SIGNAL_HANDLING_ENABLED
238 #endif // #ifndef _CDS_URCU_DETAILS_SIG_THREADED_H