src/hp.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/hp.h>
  35 #include <cds/os/thread.h>
  36
  37 namespace cds { namespace gc { namespace hp {
  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         void* ( *s_alloc_memory )( size_t size ) = default_alloc_memory;
  51         void ( *s_free_memory )( void* p ) = default_free_memory;
  52
  53         template <typename T>
  54         class allocator
  55         {
  56         public:
  57             typedef T   value_type;
  58
  59             allocator() {}
  60             allocator( allocator const& ) {}
  61             template <class U>
  62             explicit allocator( allocator<U> const& ) {}
  63
  64             static T* allocate( size_t nCount )
  65             {
  66                 return reinterpret_cast<T*>( s_alloc_memory( sizeof( value_type ) * nCount ) );
  67             }
  68
  69             static void deallocate( T* p, size_t /*nCount*/ )
  70             {
  71                 s_free_memory( reinterpret_cast<void*>( p ) );
  72             }
  73         };
  74
  75         struct defaults {
  76             static const size_t c_nHazardPointerPerThread = 8;
  77             static const size_t c_nMaxThreadCount = 100;
  78         };
  79
  80         size_t calc_retired_size( size_t nSize, size_t nHPCount, size_t nThreadCount )
  81         {
  82             size_t const min_size = nHPCount * nThreadCount;
  83             return nSize < min_size ? min_size * 2 : nSize;
  84         }
  85
  86         stat s_postmortem_stat;
  87     } // namespace
  88
  89     /*static*/ CDS_EXPORT_API smr* smr::instance_ = nullptr;
  90     thread_local thread_data* tls_ = nullptr;
  91
  92     /*static*/ CDS_EXPORT_API thread_data* smr::tls()
  93     {
  94         assert( tls_ != nullptr );
  95         return tls_;
  96     }
  97
  98     struct smr::thread_record: thread_data
  99     {
 100         atomics::atomic<thread_record*>     m_pNextNode; ///< next hazard ptr record in list
 101         atomics::atomic<cds::OS::ThreadId>  m_idOwner;   ///< Owner thread id; 0 - the record is free (not owned)
 102         atomics::atomic<bool>               m_bFree;     ///< true if record is free (not owned)
 103
 104         thread_record( guard* guards, size_t guard_count, retired_ptr* retired_arr, size_t retired_capacity )
 105             : thread_data( guards, guard_count, retired_arr, retired_capacity )
 106             , m_bFree( false )
 107         {}
 108     };
 109
 110     /*static*/ CDS_EXPORT_API void smr::set_memory_allocator(
 111         void* ( *alloc_func )( size_t size ),
 112         void( *free_func )( void * p )
 113     )
 114     {
 115         // The memory allocation functions may be set BEFORE initializing HP SMR!!!
 116         assert( instance_ == nullptr );
 117
 118         s_alloc_memory = alloc_func;
 119         s_free_memory = free_func;
 120     }
 121
 122
 123     /*static*/ CDS_EXPORT_API void smr::construct( size_t nHazardPtrCount, size_t nMaxThreadCount, size_t nMaxRetiredPtrCount, scan_type nScanType )
 124     {
 125         if ( !instance_ ) {
 126             instance_ = new( s_alloc_memory(sizeof(smr))) smr( nHazardPtrCount, nMaxThreadCount, nMaxRetiredPtrCount, nScanType );
 127         }
 128     }
 129
 130     /*static*/ CDS_EXPORT_API void smr::destruct( bool bDetachAll )
 131     {
 132         if ( instance_ ) {
 133             if ( bDetachAll )
 134                 instance_->detach_all_thread();
 135
 136             instance_->~smr();
 137             s_free_memory( instance_ );
 138             instance_ = nullptr;
 139         }
 140     }
 141
 142     CDS_EXPORT_API smr::smr( size_t nHazardPtrCount, size_t nMaxThreadCount, size_t nMaxRetiredPtrCount, scan_type nScanType )
 143         : thread_list_( nullptr )
 144         , hazard_ptr_count_( nHazardPtrCount == 0 ? defaults::c_nHazardPointerPerThread : nHazardPtrCount )
 145         , max_thread_count_( nMaxThreadCount == 0 ? defaults::c_nMaxThreadCount : nMaxThreadCount )
 146         , max_retired_ptr_count_( calc_retired_size( nMaxRetiredPtrCount, hazard_ptr_count_, max_thread_count_ ))
 147         , scan_type_( nScanType )
 148         , scan_func_( nScanType == classic ? &smr::classic_scan : &smr::inplace_scan )
 149     {}
 150
 151     CDS_EXPORT_API smr::~smr()
 152     {
 153         CDS_DEBUG_ONLY( const cds::OS::ThreadId nullThreadId = cds::OS::c_NullThreadId; )
 154         CDS_DEBUG_ONLY( const cds::OS::ThreadId mainThreadId = cds::OS::get_current_thread_id();)
 155
 156         CDS_HPSTAT( statistics( s_postmortem_stat ));
 157
 158         thread_record* pHead = thread_list_.load( atomics::memory_order_relaxed );
 159         thread_list_.store( nullptr, atomics::memory_order_relaxed );
 160
 161         thread_record* pNext = nullptr;
 162         for ( thread_record* hprec = pHead; hprec; hprec = pNext )
 163         {
 164             assert( hprec->m_idOwner.load( atomics::memory_order_relaxed ) == nullThreadId
 165                 || hprec->m_idOwner.load( atomics::memory_order_relaxed ) == mainThreadId );
 166
 167             retired_array& arr = hprec->retired_;
 168             for ( retired_ptr* cur{ arr.first() }, *last{ arr.last() }; cur != last; ++cur ) {
 169                 cur->free();
 170                 CDS_HPSTAT( ++s_postmortem_stat.free_count );
 171             }
 172
 173             arr.reset( 0 );
 174             pNext = hprec->m_pNextNode.load( atomics::memory_order_relaxed );
 175             hprec->m_bFree.store( true, atomics::memory_order_relaxed );
 176             destroy_thread_data( hprec );
 177         }
 178     }
 179
 180
 181     CDS_EXPORT_API smr::thread_record* smr::create_thread_data()
 182     {
 183         size_t const guard_array_size = thread_hp_storage::calc_array_size( get_hazard_ptr_count());
 184         size_t const retired_array_size = retired_array::calc_array_size( get_max_retired_ptr_count());
 185         size_t const nSize = sizeof( thread_record ) + guard_array_size + retired_array_size;
 186
 187         /*
 188             The memory is allocated by contnuous block
 189             Memory layout:
 190             +--------------------------+
 191             |                          |
 192             | thread_record            |
 193             |         hazards_         +---+
 194         +---|         retired_         |   |
 195         |   |                          |   |
 196         |   |--------------------------|   |
 197         |   | hazard_ptr[]             |<--+
 198         |   |                          |
 199         |   |                          |
 200         |   |--------------------------|
 201         +-->| retired_ptr[]            |
 202             |                          |
 203             |                          |
 204             +--------------------------+
 205         */
 206
 207         char* mem = reinterpret_cast<char*>( s_alloc_memory( nSize ));
 208         return new( mem ) thread_record(
 209             reinterpret_cast<guard*>( mem + sizeof( thread_record )), get_hazard_ptr_count(),
 210             reinterpret_cast<retired_ptr*>( mem + sizeof( thread_record ) + guard_array_size ), get_max_retired_ptr_count()
 211         );
 212     }
 213
 214     /*static*/ CDS_EXPORT_API void smr::destroy_thread_data( thread_record* pRec )
 215     {
 216         // all retired pointers must be freed
 217         assert( pRec->retired_.size() == 0 );
 218
 219         pRec->~thread_record();
 220         s_free_memory( pRec );
 221     }
 222
 223
 224     CDS_EXPORT_API smr::thread_record* smr::alloc_thread_data()
 225     {
 226         //CDS_HAZARDPTR_STATISTIC( ++m_Stat.m_AllocHPRec )
 227
 228         thread_record * hprec;
 229         const cds::OS::ThreadId nullThreadId = cds::OS::c_NullThreadId;
 230         const cds::OS::ThreadId curThreadId = cds::OS::get_current_thread_id();
 231
 232         // First try to reuse a free (non-active) HP record
 233         for ( hprec = thread_list_.load( atomics::memory_order_acquire ); hprec; hprec = hprec->m_pNextNode.load( atomics::memory_order_relaxed ) ) {
 234             cds::OS::ThreadId thId = nullThreadId;
 235             if ( !hprec->m_idOwner.compare_exchange_strong( thId, curThreadId, atomics::memory_order_relaxed, atomics::memory_order_relaxed ) )
 236                 continue;
 237             hprec->m_bFree.store( false, atomics::memory_order_release );
 238             return hprec;
 239         }
 240
 241         // No HP records available for reuse
 242         // Allocate and push a new HP record
 243         hprec = create_thread_data();
 244         hprec->m_idOwner.store( curThreadId, atomics::memory_order_relaxed );
 245
 246         thread_record* pOldHead = thread_list_.load( atomics::memory_order_relaxed );
 247         do {
 248             hprec->m_pNextNode.store( pOldHead, atomics::memory_order_relaxed );
 249         } while ( !thread_list_.compare_exchange_weak( pOldHead, hprec, atomics::memory_order_release, atomics::memory_order_acquire ) );
 250
 251         return hprec;
 252     }
 253
 254     CDS_EXPORT_API void smr::free_thread_data( smr::thread_record* pRec )
 255     {
 256         assert( pRec != nullptr );
 257
 258         pRec->hazards_.clear();
 259         scan( pRec );
 260         help_scan( pRec );
 261         pRec->m_idOwner.store( cds::OS::c_NullThreadId, atomics::memory_order_release );
 262     }
 263
 264     CDS_EXPORT_API void smr::detach_all_thread()
 265     {
 266         thread_record * pNext = nullptr;
 267         const cds::OS::ThreadId nullThreadId = cds::OS::c_NullThreadId;
 268
 269         for ( thread_record * hprec = thread_list_.load( atomics::memory_order_relaxed ); hprec; hprec = pNext ) {
 270             pNext = hprec->m_pNextNode.load( atomics::memory_order_relaxed );
 271             if ( hprec->m_idOwner.load( atomics::memory_order_relaxed ) != nullThreadId ) {
 272                 free_thread_data( hprec );
 273             }
 274         }
 275     }
 276
 277     /*static*/ CDS_EXPORT_API void smr::attach_thread()
 278     {
 279         if ( !tls_ )
 280             tls_ = instance().alloc_thread_data();
 281     }
 282
 283     /*static*/ CDS_EXPORT_API void smr::detach_thread()
 284     {
 285         thread_data* rec = tls_;
 286         if ( rec ) {
 287             tls_ = nullptr;
 288             instance().free_thread_data( static_cast<thread_record*>( rec ));
 289         }
 290     }
 291
 292
 293     CDS_EXPORT_API void smr::inplace_scan( thread_data* pThreadRec )
 294     {
 295         thread_record* pRec = static_cast<thread_record*>( pThreadRec );
 296
 297         //CDS_HAZARDPTR_STATISTIC( ++m_Stat.m_ScanCallCount )
 298
 299         // In-place scan algo uses LSB of retired ptr as a mark for internal purposes.
 300         // It is correct if all retired pointers are ar least 2-byte aligned (LSB is zero).
 301         // If it is wrong, we use classic scan algorithm
 302
 303         // Check if all retired pointers has zero LSB
 304         // LSB is used for marking pointers that cannot be deleted yet
 305         retired_ptr* first_retired = pRec->retired_.first();
 306         retired_ptr* last_retired = pRec->retired_.last();
 307         if ( first_retired == last_retired )
 308             return;
 309
 310         for ( auto it = first_retired; it != last_retired; ++it ) {
 311             if ( it->m_n & 1 ) {
 312                 // found a pointer with LSB bit set - use classic_scan
 313                 classic_scan( pRec );
 314                 return;
 315             }
 316         }
 317
 318         CDS_HPSTAT( ++pRec->scan_count_ );
 319
 320         // Sort retired pointer array
 321         std::sort( first_retired, last_retired, retired_ptr::less );
 322
 323         // Check double free
 324 #   ifdef _DEBUG
 325         {
 326             auto it = first_retired;
 327             auto itPrev = it;
 328             while ( ++it != last_retired ) {
 329                 assert( itPrev->m_p < it->m_p );
 330                 itPrev = it;
 331             }
 332         }
 333 #   endif
 334
 335         // Search guarded pointers in retired array
 336         thread_record* pNode = thread_list_.load( atomics::memory_order_acquire );
 337
 338         {
 339             retired_ptr dummy_retired;
 340             while ( pNode ) {
 341                 if ( pNode->m_idOwner.load( atomics::memory_order_relaxed ) != cds::OS::c_NullThreadId ) {
 342                     thread_hp_storage& hpstg = pNode->hazards_;
 343                     for ( size_t i = 0; i < hazard_ptr_count_; ++i ) {
 344                         pRec->sync();
 345                         void * hptr = hpstg[i].get();
 346                         if ( hptr ) {
 347                             dummy_retired.m_p = hptr;
 348                             retired_ptr* it = std::lower_bound( first_retired, last_retired, dummy_retired, retired_ptr::less );
 349                             if ( it != last_retired && it->m_p == hptr ) {
 350                                 // Mark retired pointer as guarded
 351                                 it->m_n |= 1;
 352                             }
 353                         }
 354                     }
 355                 }
 356                 pNode = pNode->m_pNextNode.load( atomics::memory_order_relaxed );
 357             }
 358         }
 359
 360         // Move all marked pointers to head of array
 361         {
 362             retired_ptr* insert_pos = first_retired;
 363             for ( retired_ptr* it = first_retired; it != last_retired; ++it ) {
 364                 if ( it->m_n & 1 ) {
 365                     it->m_n &= ~uintptr_t(1);
 366                     if ( insert_pos != it )
 367                         *insert_pos = *it;
 368                     ++insert_pos;
 369                 }
 370                 else {
 371                     // Retired pointer may be freed
 372                     it->free();
 373                     CDS_HPSTAT( ++pRec->free_count_ );
 374                 }
 375             }
 376             const size_t nDeferred = insert_pos - first_retired;
 377             pRec->retired_.reset( nDeferred );
 378         }
 379     }
 380
 381     // cppcheck-suppress functionConst
 382     CDS_EXPORT_API void smr::classic_scan( thread_data* pThreadRec )
 383     {
 384         thread_record* pRec = static_cast<thread_record*>( pThreadRec );
 385
 386         CDS_HPSTAT( ++pRec->scan_count_ );
 387
 388         std::vector< void*, allocator<void*>>   plist;
 389         plist.reserve( get_max_thread_count() * get_hazard_ptr_count());
 390         assert( plist.size() == 0 );
 391
 392         // Stage 1: Scan HP list and insert non-null values in plist
 393
 394         thread_record* pNode = thread_list_.load( atomics::memory_order_acquire );
 395
 396         while ( pNode ) {
 397             if ( pNode->m_idOwner.load( std::memory_order_relaxed ) != cds::OS::c_NullThreadId ) {
 398                 for ( size_t i = 0; i < get_hazard_ptr_count(); ++i ) {
 399                     pRec->sync();
 400                     void * hptr = pNode->hazards_[i].get();
 401                     if ( hptr )
 402                         plist.push_back( hptr );
 403                 }
 404             }
 405             pNode = pNode->m_pNextNode.load( atomics::memory_order_relaxed );
 406         }
 407
 408         // Sort plist to simplify search in
 409         std::sort( plist.begin(), plist.end() );
 410
 411         // Stage 2: Search plist
 412         retired_array& retired = pRec->retired_;
 413
 414         retired_ptr* first_retired = retired.first();
 415         retired_ptr* last_retired = retired.last();
 416
 417         {
 418             auto itBegin = plist.begin();
 419             auto itEnd = plist.end();
 420             retired_ptr* insert_pos = first_retired;
 421             for ( retired_ptr* it = first_retired; it != last_retired; ++it ) {
 422                 if ( std::binary_search( itBegin, itEnd, first_retired->m_p ) ) {
 423                     if ( insert_pos != it )
 424                         *insert_pos = *it;
 425                     ++insert_pos;
 426                 }
 427                 else {
 428                     it->free();
 429                     CDS_HPSTAT( ++pRec->free_count_ );
 430                 }
 431             }
 432
 433             retired.reset( insert_pos - first_retired );
 434         }
 435     }
 436
 437     CDS_EXPORT_API void smr::help_scan( thread_data* pThis )
 438     {
 439         assert( static_cast<thread_record*>( pThis )->m_idOwner.load( atomics::memory_order_relaxed ) == cds::OS::get_current_thread_id() );
 440
 441         CDS_HPSTAT( ++pThis->help_scan_count_ );
 442
 443         const cds::OS::ThreadId nullThreadId = cds::OS::c_NullThreadId;
 444         const cds::OS::ThreadId curThreadId = cds::OS::get_current_thread_id();
 445         for ( thread_record* hprec = thread_list_.load( atomics::memory_order_acquire ); hprec; hprec = hprec->m_pNextNode.load( atomics::memory_order_relaxed ))
 446         {
 447             if ( hprec == static_cast<thread_record*>( pThis ))
 448                 continue;
 449
 450             // If m_bFree == true then hprec->retired_ is empty - we don't need to see it
 451             if ( hprec->m_bFree.load( atomics::memory_order_acquire ))
 452                 continue;
 453
 454             // Owns hprec if it is empty.
 455             // Several threads may work concurrently so we use atomic technique only.
 456             {
 457                 cds::OS::ThreadId curOwner = hprec->m_idOwner.load( atomics::memory_order_relaxed );
 458                 if ( curOwner == nullThreadId ) {
 459                     if ( !hprec->m_idOwner.compare_exchange_strong( curOwner, curThreadId, atomics::memory_order_acquire, atomics::memory_order_relaxed ) )
 460                         continue;
 461                 }
 462                 else
 463                     continue;
 464             }
 465
 466             // We own the thread record successfully. Now, we can see whether it has retired pointers.
 467             // If it has ones then we move them to pThis that is private for current thread.
 468             retired_array& src = hprec->retired_;
 469             retired_array& dest = pThis->retired_;
 470             assert( !dest.full() );
 471
 472             retired_ptr* src_first = src.first();
 473             retired_ptr* src_last = src.last();
 474
 475             for ( ; src_first != src_last; ++src_first ) {
 476                 if ( !dest.push( std::move( *src_first )))
 477                     scan( pThis );
 478             }
 479
 480             src.reset( 0 );
 481
 482             hprec->m_bFree.store( true, atomics::memory_order_relaxed );
 483             hprec->m_idOwner.store( nullThreadId, atomics::memory_order_release );
 484
 485             scan( pThis );
 486         }
 487     }
 488
 489     CDS_EXPORT_API void smr::statistics( stat& st )
 490     {
 491         st.clear();
 492 #   ifdef CDS_ENABLE_HPSTAT
 493         for ( thread_record* hprec = thread_list_.load( atomics::memory_order_acquire ); hprec; hprec = hprec->m_pNextNode.load( atomics::memory_order_relaxed ) )
 494         {
 495             ++st.thread_rec_count;
 496             st.guard_allocated += hprec->hazards_.alloc_guard_count_;
 497             st.guard_freed     += hprec->hazards_.free_guard_count_;
 498             st.retired_count   += hprec->retired_.retire_call_count_;
 499             st.free_count      += hprec->free_count_;
 500             st.scan_count      += hprec->scan_count_;
 501             st.help_scan_count += hprec->help_scan_count_;
 502         }
 503 #   endif
 504     }
 505
 506 }}} // namespace cds::gc::hp
 507
 508 CDS_EXPORT_API /*static*/ cds::gc::HP::stat const& cds::gc::HP::postmortem_statistics()
 509 {
 510     return cds::gc::hp::s_postmortem_stat;
 511 }
 512