Removed redundant comment
[libcds.git] / src / dhp.cpp
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 #include <algorithm>
32 #include <vector>
33
34 #include <cds/gc/dhp.h>
35 #include <cds/os/thread.h>
36
37 namespace cds { namespace gc { namespace dhp {
38
39     namespace {
40         void * default_alloc_memory( size_t size )
41         {
42             return new uintptr_t[( size + sizeof( uintptr_t ) - 1 ) / sizeof( uintptr_t )];
43         }
44
45         void default_free_memory( void* p )
46         {
47             delete[] reinterpret_cast<uintptr_t*>( p );
48         }
49
50         struct defaults {
51             static size_t const c_extended_guard_block_size = 16;
52         };
53
54         void* ( *s_alloc_memory )( size_t size ) = default_alloc_memory;
55         void( *s_free_memory )( void* p ) = default_free_memory;
56
57         template <typename T>
58         class allocator
59         {
60         public:
61             typedef T   value_type;
62
63             allocator() {}
64             allocator( allocator const& ) {}
65             template <class U>
66             explicit allocator( allocator<U> const& ) {}
67
68             static T* allocate( size_t nCount )
69             {
70                 return reinterpret_cast<T*>( s_alloc_memory( sizeof( value_type ) * nCount ));
71             }
72
73             static void deallocate( T* p, size_t /*nCount*/ )
74             {
75                 s_free_memory( reinterpret_cast<void*>( p ));
76             }
77         };
78
79         stat s_postmortem_stat;
80     } // namespace
81
82     /*static*/ CDS_EXPORT_API smr* smr::instance_ = nullptr;
83     thread_local thread_data* tls_ = nullptr;
84
85     CDS_EXPORT_API hp_allocator::~hp_allocator()
86     {
87         while ( guard_block* gp = static_cast<guard_block*>( free_list_.get())) {
88             gp->~guard_block();
89             s_free_memory( gp );
90         }
91     }
92
93     CDS_EXPORT_API guard_block* hp_allocator::alloc()
94     {
95         guard_block* gb;
96         auto block = free_list_.get();
97         if ( block )
98             gb = static_cast< guard_block* >( block );
99         else {
100             // allocate new block
101             gb = new( s_alloc_memory( sizeof( guard_block ) + sizeof( guard ) * defaults::c_extended_guard_block_size )) guard_block;
102             new ( gb->first() ) guard[defaults::c_extended_guard_block_size];
103
104             CDS_HPSTAT( block_allocated_.fetch_add( 1, atomics::memory_order_relaxed ));
105         }
106
107         // links guards in the block
108         guard* p = gb->first();
109         for ( guard* last = p + defaults::c_extended_guard_block_size - 1; p != last; ++p ) {
110             p->clear( atomics::memory_order_relaxed );
111             p->next_ = p + 1;
112         }
113         p->next_ = nullptr;
114         p->clear();
115
116         return gb;
117     }
118
119     CDS_EXPORT_API retired_allocator::~retired_allocator()
120     {
121         while ( retired_block* rb = static_cast<retired_block*>( free_list_.get() ) ) {
122             rb->~retired_block();
123             s_free_memory( rb );
124         }
125     }
126
127     CDS_EXPORT_API retired_block* retired_allocator::alloc()
128     {
129         retired_block* rb;
130         auto block = free_list_.get();
131         if ( block )
132             rb = static_cast< retired_block* >( block );
133         else {
134             // allocate new block
135             rb = new( s_alloc_memory( sizeof( retired_block ) + sizeof( retired_ptr ) * retired_block::c_capacity )) retired_block;
136             new ( rb->first()) retired_ptr[retired_block::c_capacity];
137             CDS_HPSTAT( block_allocated_.fetch_add( 1, atomics::memory_order_relaxed ) );
138         }
139
140         rb->next_ = nullptr;
141         return rb;
142     }
143
144     struct smr::thread_record: thread_data
145     {
146         atomics::atomic<thread_record*>     m_pNextNode; ///< next hazard ptr record in list
147         atomics::atomic<cds::OS::ThreadId>  m_idOwner;   ///< Owner thread id; 0 - the record is free (not owned)
148         atomics::atomic<bool>               m_bFree;     ///< true if record is free (not owned)
149
150         thread_record( guard* guards, size_t guard_count )
151             : thread_data( guards, guard_count )
152             , m_pNextNode( nullptr )
153             , m_idOwner( cds::OS::c_NullThreadId )
154             , m_bFree( false )
155         {}
156     };
157
158     /*static*/ CDS_EXPORT_API thread_data* smr::tls()
159     {
160         assert( tls_ != nullptr );
161         return tls_;
162     }
163
164     /*static*/ CDS_EXPORT_API void smr::set_memory_allocator(
165         void* ( *alloc_func )( size_t size ),
166         void( *free_func )( void * p )
167     )
168     {
169         // The memory allocation functions may be set BEFORE initializing DHP SMR!!!
170         assert( instance_ == nullptr );
171
172         s_alloc_memory = alloc_func;
173         s_free_memory = free_func;
174     }
175
176     /*static*/ CDS_EXPORT_API void smr::construct( size_t nInitialHazardPtrCount )
177     {
178         if ( !instance_ ) {
179             instance_ = new( s_alloc_memory( sizeof( smr ))) smr( nInitialHazardPtrCount );
180         }
181     }
182
183     /*static*/ CDS_EXPORT_API void smr::destruct( bool bDetachAll )
184     {
185         if ( instance_ ) {
186             if ( bDetachAll )
187                 instance_->detach_all_thread();
188
189             instance_->~smr();
190             s_free_memory( instance_ );
191             instance_ = nullptr;
192         }
193     }
194
195     CDS_EXPORT_API smr::smr( size_t nInitialHazardPtrCount )
196         : initial_hazard_count_( nInitialHazardPtrCount < 4 ? 16 : nInitialHazardPtrCount )
197         , last_plist_size_( initial_hazard_count_ * 64 )
198     {
199         thread_list_.store( nullptr, atomics::memory_order_release );
200     }
201
202     CDS_EXPORT_API smr::~smr()
203     {
204         CDS_DEBUG_ONLY( const cds::OS::ThreadId nullThreadId = cds::OS::c_NullThreadId; )
205         CDS_DEBUG_ONLY( const cds::OS::ThreadId mainThreadId = cds::OS::get_current_thread_id(); )
206
207         CDS_HPSTAT( statistics( s_postmortem_stat ) );
208
209         thread_record* pHead = thread_list_.load( atomics::memory_order_relaxed );
210         thread_list_.store( nullptr, atomics::memory_order_release );
211
212         thread_record* pNext = nullptr;
213         for ( thread_record* hprec = pHead; hprec; hprec = pNext )
214         {
215             assert( hprec->m_idOwner.load( atomics::memory_order_relaxed ) == nullThreadId
216                 || hprec->m_idOwner.load( atomics::memory_order_relaxed ) == mainThreadId );
217
218             retired_array& retired = hprec->retired_;
219
220             // delete retired data
221             for ( retired_block* block = retired.list_head_; block && block != retired.current_block_; block = block->next_ ) {
222                 for ( retired_ptr* p = block->first(); p != block->last(); ++p ) {
223                     p->free();
224                     CDS_HPSTAT( ++s_postmortem_stat.free_count );
225                 }
226             }
227             if ( retired.current_block_ ) {
228                 for ( retired_ptr* p = retired.current_block_->first(); p != retired.current_cell_; ++p ) {
229                     p->free();
230                     CDS_HPSTAT( ++s_postmortem_stat.free_count );
231                 }
232             }
233             hprec->retired_.fini();
234             hprec->hazards_.clear();
235
236             pNext = hprec->m_pNextNode.load( atomics::memory_order_relaxed );
237             hprec->m_bFree.store( true, atomics::memory_order_relaxed );
238             destroy_thread_data( hprec );
239         }
240     }
241
242     /*static*/ CDS_EXPORT_API void smr::attach_thread()
243     {
244         if ( !tls_ )
245             tls_ = instance().alloc_thread_data();
246     }
247
248     /*static*/ CDS_EXPORT_API void smr::detach_thread()
249     {
250         thread_data* rec = tls_;
251         if ( rec ) {
252             tls_ = nullptr;
253             instance().free_thread_data( static_cast<thread_record*>( rec ) );
254         }
255     }
256
257     CDS_EXPORT_API void smr::detach_all_thread()
258     {
259         thread_record * pNext = nullptr;
260         const cds::OS::ThreadId nullThreadId = cds::OS::c_NullThreadId;
261
262         for ( thread_record * hprec = thread_list_.load( atomics::memory_order_relaxed ); hprec; hprec = pNext ) {
263             pNext = hprec->m_pNextNode.load( atomics::memory_order_relaxed );
264             if ( hprec->m_idOwner.load( atomics::memory_order_relaxed ) != nullThreadId ) {
265                 free_thread_data( hprec );
266             }
267         }
268     }
269
270     CDS_EXPORT_API smr::thread_record* smr::create_thread_data()
271     {
272         size_t const guard_array_size = sizeof( guard ) * initial_hazard_count_;
273
274         /*
275             The memory is allocated by contnuous block
276             Memory layout:
277             +--------------------------+
278             |                          |
279             | thread_record            |
280             |         hazards_         +---+
281             |         retired_         |   |
282             |                          |   |
283             |--------------------------|   |
284             | hazard_ptr[]             |<--+
285             |  initial HP array        |
286             |                          |
287             +--------------------------+
288         */
289
290         char* mem = reinterpret_cast<char*>( s_alloc_memory( sizeof( thread_record ) + guard_array_size ));
291         return new( mem ) thread_record(
292             reinterpret_cast<guard*>( mem + sizeof( thread_record ) ), initial_hazard_count_
293         );
294     }
295
296     /*static*/ CDS_EXPORT_API void smr::destroy_thread_data( thread_record* pRec )
297     {
298         // all retired pointers must be freed
299         pRec->~thread_record();
300         s_free_memory( pRec );
301     }
302
303     CDS_EXPORT_API smr::thread_record* smr::alloc_thread_data()
304     {
305         thread_record * hprec = nullptr;
306         const cds::OS::ThreadId nullThreadId = cds::OS::c_NullThreadId;
307         const cds::OS::ThreadId curThreadId = cds::OS::get_current_thread_id();
308
309         // First try to reuse a free (non-active) DHP record
310         for ( hprec = thread_list_.load( atomics::memory_order_acquire ); hprec; hprec = hprec->m_pNextNode.load( atomics::memory_order_acquire ) ) {
311             cds::OS::ThreadId thId = nullThreadId;
312             if ( !hprec->m_idOwner.compare_exchange_strong( thId, curThreadId, atomics::memory_order_relaxed, atomics::memory_order_relaxed ) )
313                 continue;
314             hprec->m_bFree.store( false, atomics::memory_order_release );
315             break;
316         }
317         
318         if ( !hprec ) {
319             // No HP records available for reuse
320             // Allocate and push a new HP record
321             hprec = create_thread_data();
322             hprec->m_idOwner.store( curThreadId, atomics::memory_order_relaxed );
323
324             thread_record* pOldHead = thread_list_.load( atomics::memory_order_acquire );
325             do {
326                 hprec->m_pNextNode.store( pOldHead, atomics::memory_order_release );
327             } while ( !thread_list_.compare_exchange_weak( pOldHead, hprec, atomics::memory_order_release, atomics::memory_order_acquire ) );
328         }
329
330         hprec->hazards_.init();
331         hprec->retired_.init();
332
333         return hprec;
334     }
335
336     CDS_EXPORT_API void smr::free_thread_data( thread_record* pRec )
337     {
338         assert( pRec != nullptr );
339         //CDS_HAZARDPTR_STATISTIC( ++m_Stat.m_RetireHPRec )
340
341         pRec->hazards_.clear();
342         scan( pRec );
343         help_scan( pRec );
344
345         if ( pRec->retired_.empty() ) {
346             pRec->retired_.fini();
347             pRec->m_bFree.store( true, std::memory_order_release );
348         }
349         else {
350             // Free all empty blocks
351             retired_block* free_block = pRec->retired_.current_block_->next_;
352             if ( free_block ) {
353                 pRec->retired_.current_block_->next_ = nullptr;
354                 while ( free_block ) {
355                     retired_block* next = free_block->next_;
356                     retired_allocator_.free( free_block );
357                     free_block = next;
358                     --pRec->retired_.block_count_;
359                 }
360             }
361         }
362
363         pRec->m_idOwner.store( cds::OS::c_NullThreadId, atomics::memory_order_release );
364     }
365
366     namespace {
367         typedef std::vector<void*, allocator<void*>> hp_vector;
368
369         inline void copy_hazards( hp_vector& vect, guard const* arr, size_t size )
370         {
371             for ( guard const* end = arr + size; arr != end; ++arr ) {
372                 void* hp = arr->get();
373                 if ( hp )
374                     vect.push_back( hp );
375             }
376         }
377
378         inline size_t retire_data( hp_vector const& plist, retired_array& stg, retired_block* block, size_t block_size )
379         {
380             auto hp_begin = plist.begin();
381             auto hp_end = plist.end();
382             size_t count = 0;
383
384             for ( retired_ptr* p = block->first(), *end = p + block_size; p != end; ++p ) {
385                 if ( cds_unlikely( std::binary_search( hp_begin, hp_end, p->m_p )))
386                     stg.repush( p );
387                 else {
388                     p->free();
389                     ++count;
390                 }
391             }
392
393             return count;
394         }
395
396     } // namespace
397
398     CDS_EXPORT_API void smr::scan( thread_data* pThreadRec )
399     {
400         thread_record* pRec = static_cast<thread_record*>( pThreadRec );
401
402         CDS_HPSTAT( ++pRec->scan_call_count_ );
403
404         hp_vector plist;
405         size_t plist_size = last_plist_size_.load( std::memory_order_relaxed );
406         plist.reserve( plist_size );
407
408         // Stage 1: Scan HP list and insert non-null values in plist
409         thread_record* pNode = thread_list_.load( atomics::memory_order_acquire );
410         while ( pNode ) {
411             if ( pNode->m_idOwner.load( std::memory_order_relaxed ) != cds::OS::c_NullThreadId ) {
412                 copy_hazards( plist, pNode->hazards_.array_, pNode->hazards_.initial_capacity_ );
413
414                 for ( guard_block* block = pNode->hazards_.extended_list_.load( atomics::memory_order_acquire ); 
415                     block;
416                     block = block->next_block_.load( atomics::memory_order_acquire ) )
417                 {
418                     copy_hazards( plist, block->first(), defaults::c_extended_guard_block_size );
419                 }
420             }
421
422             pNode = pNode->m_pNextNode.load( atomics::memory_order_relaxed );
423         }
424
425         // Store plist size for next scan() call (vector reallocation optimization)
426         if ( plist.size() > plist_size )
427             last_plist_size_.compare_exchange_weak( plist_size, plist.size(), std::memory_order_relaxed, std::memory_order_relaxed );
428
429         // Sort plist to simplify search in
430         std::sort( plist.begin(), plist.end() );
431
432         // Stage 2: Search plist
433         size_t free_count = 0;
434         size_t retired_count = 0;
435         retired_block* last_block = pRec->retired_.current_block_;
436         retired_ptr*   last_block_cell = pRec->retired_.current_cell_;
437
438         pRec->retired_.current_block_ = pRec->retired_.list_head_;
439         pRec->retired_.current_cell_ = pRec->retired_.current_block_->first();
440
441         for ( retired_block* block = pRec->retired_.list_head_; block; block = block->next_ ) {
442             bool const end_block = block == last_block;
443             size_t const size = end_block ? last_block_cell - block->first() : retired_block::c_capacity;
444
445             retired_count += retired_block::c_capacity;
446             free_count += retire_data( plist, pRec->retired_, block, size );
447
448             if ( end_block )
449                 break;
450         }
451         CDS_HPSTAT( pRec->free_call_count_ += free_count );
452
453         // If the count of freed elements is too small, increase retired array
454         if ( free_count < retired_count / 4 && last_block == pRec->retired_.list_tail_ && last_block_cell == last_block->last() )
455             pRec->retired_.extend();
456     }
457
458     CDS_EXPORT_API void smr::help_scan( thread_data* pThis )
459     {
460         assert( static_cast<thread_record*>( pThis )->m_idOwner.load( atomics::memory_order_relaxed ) == cds::OS::get_current_thread_id() );
461         CDS_HPSTAT( ++pThis->help_scan_call_count_ );
462
463         const cds::OS::ThreadId nullThreadId = cds::OS::c_NullThreadId;
464         const cds::OS::ThreadId curThreadId = cds::OS::get_current_thread_id();
465         for ( thread_record* hprec = thread_list_.load( atomics::memory_order_acquire ); hprec; hprec = hprec->m_pNextNode.load( atomics::memory_order_relaxed ) )
466         {
467             if ( hprec == static_cast<thread_record*>( pThis ))
468                 continue;
469
470             // If m_bFree == true then hprec->retired_ is empty - we don't need to see it
471             if ( hprec->m_bFree.load( atomics::memory_order_acquire ) ) {
472                 CDS_TSAN_ANNOTATE_IGNORE_READS_BEGIN;
473                 assert( hprec->retired_.empty() );
474                 CDS_TSAN_ANNOTATE_IGNORE_READS_END;
475                 continue;
476             }
477
478             // Owns hprec
479             // Several threads may work concurrently so we use atomic technique
480             {
481                 cds::OS::ThreadId curOwner = hprec->m_idOwner.load( atomics::memory_order_relaxed );
482                 if ( curOwner == nullThreadId ) {
483                     if ( !hprec->m_idOwner.compare_exchange_strong( curOwner, curThreadId, atomics::memory_order_acquire, atomics::memory_order_relaxed ) )
484                         continue;
485                 }
486                 else
487                     continue;
488             }
489
490             // We own the thread record successfully. Now, we can see whether it has retired pointers.
491             // If it has ones then we move them to pThis that is private for current thread.
492             retired_array& src = hprec->retired_;
493             retired_array& dest = pThis->retired_;
494
495             for ( retired_block* block = src.list_head_; block; block = block->next_ ) {
496                 retired_ptr* last = block == src.current_block_ ? src.current_cell_ : block->last();
497                 for ( retired_ptr* p = block->first(); p != last; ++p ) {
498                     if ( !dest.push( *p ) )
499                         scan( pThis );
500                 }
501
502                 if ( block == src.current_block_ )
503                     break;
504             }
505
506             src.fini();
507             hprec->m_bFree.store( true, atomics::memory_order_relaxed );
508             hprec->m_idOwner.store( nullThreadId, atomics::memory_order_release );
509         }
510
511         scan( pThis );
512     }
513
514     CDS_EXPORT_API void smr::statistics( stat& st )
515     {
516         st.clear();
517 #   ifdef CDS_ENABLE_HPSTAT
518         for ( thread_record* hprec = thread_list_.load( atomics::memory_order_acquire ); hprec; hprec = hprec->m_pNextNode.load( atomics::memory_order_relaxed ) )
519         {
520             CDS_TSAN_ANNOTATE_IGNORE_READS_BEGIN;
521             ++st.thread_rec_count;
522             st.guard_allocated      += hprec->hazards_.alloc_guard_count_;
523             st.guard_freed          += hprec->hazards_.free_guard_count_;
524             st.hp_extend_count      += hprec->hazards_.extend_call_count_;
525             st.retired_count        += hprec->retired_.retire_call_count_;
526             st.retired_extend_count += hprec->retired_.extend_call_count_;
527             st.free_count           += hprec->free_call_count_;
528             st.scan_count           += hprec->scan_call_count_;
529             st.help_scan_count      += hprec->help_scan_call_count_;
530             CDS_TSAN_ANNOTATE_IGNORE_READS_END;
531         }
532
533         CDS_TSAN_ANNOTATE_IGNORE_READS_BEGIN;
534         st.hp_block_count = hp_allocator_.block_allocated_.load( atomics::memory_order_relaxed );
535         st.retired_block_count = retired_allocator_.block_allocated_.load( atomics::memory_order_relaxed );
536         CDS_TSAN_ANNOTATE_IGNORE_READS_END;
537 #   endif
538     }
539
540
541 }}} // namespace cds::gc::dhp
542
543 CDS_EXPORT_API /*static*/ cds::gc::DHP::stat const& cds::gc::DHP::postmortem_statistics()
544 {
545     return cds::gc::dhp::s_postmortem_stat;
546 }
547