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                 || !cds::OS::is_thread_alive( hprec->m_idOwner.load( atomics::memory_order_relaxed ) )
 167             );
 168
 169             retired_array& arr = hprec->retired_;
 170             for ( retired_ptr* cur{ arr.first() }, *last{ arr.last() }; cur != last; ++cur ) {
 171                 cur->free();
 172                 CDS_HPSTAT( ++s_postmortem_stat.free_count );
 173             }
 174
 175             arr.reset( 0 );
 176             pNext = hprec->m_pNextNode.load( atomics::memory_order_relaxed );
 177             hprec->m_bFree.store( true, atomics::memory_order_relaxed );
 178             destroy_thread_data( hprec );
 179         }
 180     }
 181
 182
 183     CDS_EXPORT_API smr::thread_record* smr::create_thread_data()
 184     {
 185         size_t const guard_array_size = thread_hp_storage::calc_array_size( get_hazard_ptr_count());
 186         size_t const retired_array_size = retired_array::calc_array_size( get_max_retired_ptr_count());
 187         size_t const nSize = sizeof( thread_record ) + guard_array_size + retired_array_size;
 188
 189         /*
 190             The memory is allocated by contnuous block
 191             Memory layout:
 192             +--------------------------+
 193             |                          |
 194             | thread_record            |
 195             |         hazards_         +---+
 196         +---|         retired_         |   |
 197         |   |                          |   |
 198         |   |--------------------------|   |
 199         |   | hazard_ptr[]             |<--+
 200         |   |                          |
 201         |   |                          |
 202         |   |--------------------------|
 203         +-->| retired_ptr[]            |
 204             |                          |
 205             |                          |
 206             +--------------------------+
 207         */
 208
 209         char* mem = reinterpret_cast<char*>( s_alloc_memory( nSize ));
 210         return new( mem ) thread_record(
 211             reinterpret_cast<guard*>( mem + sizeof( thread_record )), get_hazard_ptr_count(),
 212             reinterpret_cast<retired_ptr*>( mem + sizeof( thread_record ) + guard_array_size ), get_max_retired_ptr_count()
 213         );
 214     }
 215
 216     /*static*/ CDS_EXPORT_API void smr::destroy_thread_data( thread_record* pRec )
 217     {
 218         // all retired pointers must be freed
 219         assert( pRec->retired_.size() == 0 );
 220
 221         pRec->~thread_record();
 222         s_free_memory( pRec );
 223     }
 224
 225
 226     CDS_EXPORT_API smr::thread_record* smr::alloc_thread_data()
 227     {
 228         //CDS_HAZARDPTR_STATISTIC( ++m_Stat.m_AllocHPRec )
 229
 230         thread_record * hprec;
 231         const cds::OS::ThreadId nullThreadId = cds::OS::c_NullThreadId;
 232         const cds::OS::ThreadId curThreadId = cds::OS::get_current_thread_id();
 233
 234         // First try to reuse a free (non-active) HP record
 235         for ( hprec = thread_list_.load( atomics::memory_order_acquire ); hprec; hprec = hprec->m_pNextNode.load( atomics::memory_order_relaxed ) ) {
 236             cds::OS::ThreadId thId = nullThreadId;
 237             if ( !hprec->m_idOwner.compare_exchange_strong( thId, curThreadId, atomics::memory_order_relaxed, atomics::memory_order_relaxed ) )
 238                 continue;
 239             hprec->m_bFree.store( false, atomics::memory_order_release );
 240             return hprec;
 241         }
 242
 243         // No HP records available for reuse
 244         // Allocate and push a new HP record
 245         hprec = create_thread_data();
 246         hprec->m_idOwner.store( curThreadId, atomics::memory_order_relaxed );
 247
 248         thread_record* pOldHead = thread_list_.load( atomics::memory_order_relaxed );
 249         do {
 250             hprec->m_pNextNode.store( pOldHead, atomics::memory_order_relaxed );
 251         } while ( !thread_list_.compare_exchange_weak( pOldHead, hprec, atomics::memory_order_release, atomics::memory_order_acquire ) );
 252
 253         return hprec;
 254     }
 255
 256     CDS_EXPORT_API void smr::free_thread_data( smr::thread_record* pRec )
 257     {
 258         assert( pRec != nullptr );
 259         //CDS_HAZARDPTR_STATISTIC( ++m_Stat.m_RetireHPRec )
 260
 261         pRec->hazards_.clear();
 262         scan( pRec );
 263         help_scan( pRec );
 264         pRec->m_idOwner.store( cds::OS::c_NullThreadId, atomics::memory_order_release );
 265     }
 266
 267     CDS_EXPORT_API void smr::detach_all_thread()
 268     {
 269         thread_record * pNext = nullptr;
 270         const cds::OS::ThreadId nullThreadId = cds::OS::c_NullThreadId;
 271
 272         for ( thread_record * hprec = thread_list_.load( atomics::memory_order_relaxed ); hprec; hprec = pNext ) {
 273             pNext = hprec->m_pNextNode.load( atomics::memory_order_relaxed );
 274             if ( hprec->m_idOwner.load( atomics::memory_order_relaxed ) != nullThreadId ) {
 275                 free_thread_data( hprec );
 276             }
 277         }
 278     }
 279
 280     /*static*/ CDS_EXPORT_API void smr::attach_thread()
 281     {
 282         if ( !tls_ )
 283             tls_ = instance().alloc_thread_data();
 284     }
 285
 286     /*static*/ CDS_EXPORT_API void smr::detach_thread()
 287     {
 288         thread_data* rec = tls_;
 289         if ( rec ) {
 290             tls_ = nullptr;
 291             instance().free_thread_data( static_cast<thread_record*>( rec ));
 292         }
 293     }
 294
 295
 296     CDS_EXPORT_API void smr::inplace_scan( thread_data* pThreadRec )
 297     {
 298         thread_record* pRec = static_cast<thread_record*>( pThreadRec );
 299
 300         //CDS_HAZARDPTR_STATISTIC( ++m_Stat.m_ScanCallCount )
 301
 302         // In-place scan algo uses LSB of retired ptr as a mark for internal purposes.
 303         // It is correct if all retired pointers are ar least 2-byte aligned (LSB is zero).
 304         // If it is wrong, we use classic scan algorithm
 305
 306         // Check if all retired pointers has zero LSB
 307         // LSB is used for marking pointers that cannot be deleted yet
 308         retired_ptr* first_retired = pRec->retired_.first();
 309         retired_ptr* last_retired = pRec->retired_.last();
 310         if ( first_retired == last_retired )
 311             return;
 312
 313         for ( auto it = first_retired; it != last_retired; ++it ) {
 314             if ( it->m_n & 1 ) {
 315                 // found a pointer with LSB bit set - use classic_scan
 316                 classic_scan( pRec );
 317                 return;
 318             }
 319         }
 320
 321         CDS_HPSTAT( ++pRec->stat_.scan_count );
 322
 323         // Sort retired pointer array
 324         std::sort( first_retired, last_retired, retired_ptr::less );
 325
 326         // Check double free
 327 #   ifdef _DEBUG
 328         {
 329             auto it = first_retired;
 330             auto itPrev = it;
 331             while ( ++it != last_retired ) {
 332                 assert( itPrev->m_p < it->m_p );
 333                 itPrev = it;
 334             }
 335         }
 336 #   endif
 337
 338         // Search guarded pointers in retired array
 339         thread_record* pNode = thread_list_.load( atomics::memory_order_acquire );
 340
 341         {
 342             retired_ptr dummy_retired;
 343             while ( pNode ) {
 344                 if ( pNode->m_idOwner.load( atomics::memory_order_relaxed ) != cds::OS::c_NullThreadId ) {
 345                     thread_hp_storage& hpstg = pNode->hazards_;
 346                     for ( size_t i = 0; i < hazard_ptr_count_; ++i ) {
 347                         pRec->sync();
 348                         void * hptr = hpstg[i].get();
 349                         if ( hptr ) {
 350                             dummy_retired.m_p = hptr;
 351                             retired_ptr* it = std::lower_bound( first_retired, last_retired, dummy_retired, retired_ptr::less );
 352                             if ( it != last_retired && it->m_p == hptr ) {
 353                                 // Mark retired pointer as guarded
 354                                 it->m_n |= 1;
 355                             }
 356                         }
 357                     }
 358                 }
 359                 pNode = pNode->m_pNextNode.load( atomics::memory_order_relaxed );
 360             }
 361         }
 362
 363         // Move all marked pointers to head of array
 364         {
 365             retired_ptr* insert_pos = first_retired;
 366             for ( retired_ptr* it = first_retired; it != last_retired; ++it ) {
 367                 if ( it->m_n & 1 ) {
 368                     it->m_n &= ~uintptr_t(1);
 369                     if ( insert_pos != it )
 370                         *insert_pos = *it;
 371                     ++insert_pos;
 372                 }
 373                 else {
 374                     // Retired pointer may be freed
 375                     it->free();
 376                     CDS_HPSTAT( ++pRec->stat_.free_count );
 377                 }
 378             }
 379             const size_t nDeferred = insert_pos - first_retired;
 380             pRec->retired_.reset( nDeferred );
 381         }
 382     }
 383
 384     // cppcheck-suppress functionConst
 385     CDS_EXPORT_API void smr::classic_scan( thread_data* pThreadRec )
 386     {
 387         thread_record* pRec = static_cast<thread_record*>( pThreadRec );
 388
 389         CDS_HPSTAT( ++pRec->stat_.scan_count );
 390
 391         std::vector< void*, allocator<void*>>   plist;
 392         plist.reserve( get_max_thread_count() * get_hazard_ptr_count());
 393         assert( plist.size() == 0 );
 394
 395         // Stage 1: Scan HP list and insert non-null values in plist
 396
 397         thread_record* pNode = thread_list_.load( atomics::memory_order_acquire );
 398
 399         while ( pNode ) {
 400             if ( pNode->m_idOwner.load( std::memory_order_relaxed ) != cds::OS::c_NullThreadId ) {
 401                 for ( size_t i = 0; i < get_hazard_ptr_count(); ++i ) {
 402                     pRec->sync();
 403                     void * hptr = pNode->hazards_[i].get();
 404                     if ( hptr )
 405                         plist.push_back( hptr );
 406                 }
 407             }
 408             pNode = pNode->m_pNextNode.load( atomics::memory_order_relaxed );
 409         }
 410
 411         // Sort plist to simplify search in
 412         std::sort( plist.begin(), plist.end() );
 413
 414         // Stage 2: Search plist
 415         retired_array& retired = pRec->retired_;
 416
 417         retired_ptr* first_retired = retired.first();
 418         retired_ptr* last_retired = retired.last();
 419
 420         {
 421             auto itBegin = plist.begin();
 422             auto itEnd = plist.end();
 423             retired_ptr* insert_pos = first_retired;
 424             for ( retired_ptr* it = first_retired; it != last_retired; ++it ) {
 425                 if ( std::binary_search( itBegin, itEnd, first_retired->m_p ) ) {
 426                     if ( insert_pos != it )
 427                         *insert_pos = *it;
 428                     ++insert_pos;
 429                 }
 430                 else {
 431                     it->free();
 432                     CDS_HPSTAT( ++pRec->stat_.free_count );
 433                 }
 434             }
 435
 436             retired.reset( insert_pos - first_retired );
 437         }
 438     }
 439
 440     CDS_EXPORT_API void smr::help_scan( thread_data* pThis )
 441     {
 442         assert( static_cast<thread_record*>( pThis )->m_idOwner.load( atomics::memory_order_relaxed ) == cds::OS::get_current_thread_id() );
 443
 444         CDS_HPSTAT( ++pThis->stat_.help_scan_count );
 445
 446         const cds::OS::ThreadId nullThreadId = cds::OS::c_NullThreadId;
 447         const cds::OS::ThreadId curThreadId = cds::OS::get_current_thread_id();
 448         for ( thread_record* hprec = thread_list_.load( atomics::memory_order_acquire ); hprec; hprec = hprec->m_pNextNode.load( atomics::memory_order_relaxed ))
 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 || !cds::OS::is_thread_alive( curOwner ) ) {
 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 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     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->stat_.free_count;
 500             st.scan_count      += hprec->stat_.scan_count;
 501             st.help_scan_count += hprec->stat_.help_scan_count;
 502         }
 503 #   endif
 504     }
 505
 506 }}} // namespace cds::gc::hp
 507
 508 /*static*/ cds::gc::HP::stat const& cds::gc::HP::postmortem_statistics()
 509 {
 510     return cds::gc::hp::s_postmortem_stat;
 511 }
 512