cds/intrusive/striped_set/striping_policy.h

   1 //$$CDS-header$$
   2
   3 #ifndef __CDS_INTRUSIVE_STRIPED_SET_STRIPING_POLICY_H
   4 #define __CDS_INTRUSIVE_STRIPED_SET_STRIPING_POLICY_H
   5
   6 #include <memory>
   7 #include <cds/lock/array.h>
   8 #include <cds/os/thread.h>
   9 #include <cds/lock/spinlock.h>
  10
  11 #include <cds/details/std/mutex.h>
  12 //#include <boost/thread/mutex.hpp>
  13 //#include <boost/thread/recursive_mutex.hpp>
  14
  15
  16 namespace cds { namespace intrusive { namespace striped_set {
  17
  18     /// Lock striping concurrent access policy
  19     /**
  20         This is one of available opt::mutex_policy option type for StripedSet
  21
  22         Lock striping is very simple technique.
  23         The set consists of the bucket table and the array of locks.
  24         Initially, the capacity of lock array and bucket table is the same.
  25         When set is resized, bucket table capacity will be doubled but lock array will not.
  26         The lock \p i protects each bucket \p j, where <tt> j = i mod L </tt>,
  27         where \p L - the size of lock array.
  28
  29         The policy contains an internal array of \p Lock locks.
  30
  31         Template arguments:
  32         - \p Lock - the type of mutex. The default is \p cds_std::mutex. The mutex type should be default-constructible.
  33             Note that a spin-lock is not so good suitable for lock striping for performance reason.
  34         - \p Alloc - allocator type used for lock array memory allocation. Default is \p CDS_DEFAULT_ALLOCATOR.
  35     */
  36     template <class Lock = cds_std::mutex, class Alloc = CDS_DEFAULT_ALLOCATOR >
  37     class striping
  38     {
  39     public:
  40         typedef Lock    lock_type       ;   ///< lock type
  41         typedef Alloc   allocator_type  ;   ///< allocator type
  42
  43         typedef cds::lock::array< lock_type, cds::lock::pow2_select_policy, allocator_type >    lock_array_type ;   ///< lock array type
  44
  45     protected:
  46         //@cond
  47         lock_array_type m_Locks;
  48         //@endcond
  49
  50     public:
  51         //@cond
  52         class scoped_cell_lock {
  53             cds::lock::scoped_lock< lock_array_type >   m_guard;
  54
  55         public:
  56             scoped_cell_lock( striping& policy, size_t nHash )
  57                 : m_guard( policy.m_Locks, nHash )
  58             {}
  59         };
  60
  61         class scoped_full_lock {
  62             cds::lock::scoped_lock< lock_array_type >   m_guard;
  63         public:
  64             scoped_full_lock( striping& policy )
  65                 : m_guard( policy.m_Locks )
  66             {}
  67         };
  68
  69         class scoped_resize_lock: public scoped_full_lock {
  70         public:
  71             scoped_resize_lock( striping& policy )
  72                 : scoped_full_lock( policy )
  73             {}
  74
  75             bool success() const
  76             {
  77                 return true;
  78             }
  79         };
  80         //@endcond
  81
  82     public:
  83         /// Constructor
  84         striping(
  85             size_t nLockCount   ///< The size of lock array. Must be power of two.
  86         )
  87             : m_Locks( nLockCount, cds::lock::pow2_select_policy( nLockCount ))
  88         {}
  89
  90         /// Returns lock array size
  91         /**
  92             Lock array size is unchanged during \p striped object lifetime
  93         */
  94         size_t lock_count() const
  95         {
  96             return m_Locks.size();
  97         }
  98
  99         //@cond
 100         void resize( size_t /*nNewCapacity*/ )
 101         {}
 102         //@endcond
 103     };
 104
 105
 106     /// Refinable concurrent access policy
 107     /**
 108         This is one of available opt::mutex_policy option type for StripedSet
 109
 110         Refining is like a striping technique (see striped_set::striping)
 111         but it allows growing the size of lock array when resizing the hash table.
 112         So, the sizes of hash table and lock array are equal.
 113
 114         Template arguments:
 115         - \p RecursiveLock - the type of mutex. Reentrant (recursive) mutex is required.
 116             The default is \p cds_std::recursive_mutex. The mutex type should be default-constructible.
 117         - \p BackOff - back-off strategy. Default is cds::backoff::yield
 118         - \p Alloc - allocator type used for lock array memory allocation. Default is \p CDS_DEFAULT_ALLOCATOR.
 119     */
 120     template <
 121         class RecursiveLock = cds_std::recursive_mutex,
 122         typename BackOff = cds::backoff::yield,
 123         class Alloc = CDS_DEFAULT_ALLOCATOR>
 124     class refinable
 125     {
 126     public:
 127         typedef RecursiveLock   lock_type   ;   ///< lock type
 128         typedef BackOff         back_off    ;   ///< back-off strategy used
 129         typedef Alloc           allocator_type; ///< allocator type
 130
 131     protected:
 132         //@cond
 133         typedef cds::lock::trivial_select_policy  lock_selection_policy;
 134
 135         class lock_array_type
 136             : public cds::lock::array< lock_type, lock_selection_policy, allocator_type >
 137             , public std::enable_shared_from_this< lock_array_type >
 138         {
 139             typedef cds::lock::array< lock_type, lock_selection_policy, allocator_type >    lock_array_base;
 140         public:
 141             lock_array_type( size_t nCapacity )
 142                 : lock_array_base( nCapacity )
 143             {}
 144         };
 145         typedef std::shared_ptr< lock_array_type >  lock_array_ptr;
 146         typedef cds::details::Allocator< lock_array_type, allocator_type >  lock_array_allocator;
 147
 148         typedef unsigned long long  owner_t;
 149         typedef cds::OS::ThreadId   threadId_t;
 150
 151         typedef cds::lock::Spin     spinlock_type;
 152         typedef cds::lock::scoped_lock< spinlock_type > scoped_spinlock;
 153         //@endcond
 154
 155     protected:
 156         //@cond
 157         static owner_t const c_nOwnerMask = (((owner_t) 1) << (sizeof(owner_t) * 8 - 1)) - 1;
 158
 159         lock_array_ptr                  m_arrLocks  ;   ///< Lock array. The capacity of array is specified in constructor.
 160         CDS_ATOMIC::atomic< owner_t >   m_Owner     ;   ///< owner mark (thread id + boolean flag)
 161         CDS_ATOMIC::atomic<size_t>      m_nCapacity ;   ///< Lock array capacity
 162         spinlock_type                   m_access    ;   ///< access to m_arrLocks
 163         //@endcond
 164
 165     protected:
 166         //@cond
 167         struct lock_array_disposer {
 168             void operator()( lock_array_type * pArr )
 169             {
 170                 lock_array_allocator().Delete( pArr );
 171             }
 172         };
 173
 174         lock_array_ptr create_lock_array( size_t nCapacity )
 175         {
 176             m_nCapacity.store( nCapacity, CDS_ATOMIC::memory_order_relaxed );
 177             return lock_array_ptr( lock_array_allocator().New( nCapacity ), lock_array_disposer() );
 178         }
 179
 180         lock_type& acquire( size_t nHash )
 181         {
 182             owner_t me = (owner_t) cds::OS::getCurrentThreadId();
 183             owner_t who;
 184
 185             back_off bkoff;
 186             while ( true ) {
 187                 // wait while resizing
 188                 while ( true ) {
 189                     who = m_Owner.load( CDS_ATOMIC::memory_order_acquire );
 190                     if ( !( who & 1 ) || (who >> 1) == (me & c_nOwnerMask) )
 191                         break;
 192                     bkoff();
 193                 }
 194
 195                 lock_array_ptr pLocks;
 196                 {
 197                     scoped_spinlock sl(m_access);
 198                     pLocks = m_arrLocks;
 199                 }
 200
 201                 lock_type& lock = pLocks->at( nHash & (pLocks->size() - 1));
 202                 lock.lock();
 203
 204                 who = m_Owner.load( CDS_ATOMIC::memory_order_acquire );
 205                 if ( ( !(who & 1) || (who >> 1) == (me & c_nOwnerMask) ) && m_arrLocks == pLocks )
 206                     return lock;
 207                 lock.unlock();
 208             }
 209         }
 210
 211         lock_array_ptr acquire_all()
 212         {
 213             owner_t me = (owner_t) cds::OS::getCurrentThreadId();
 214             owner_t who;
 215
 216             back_off bkoff;
 217             while ( true ) {
 218                 // wait while resizing
 219                 while ( true ) {
 220                     who = m_Owner.load( CDS_ATOMIC::memory_order_acquire );
 221                     if ( !( who & 1 ) || (who >> 1) == (me & c_nOwnerMask) )
 222                         break;
 223                     bkoff();
 224                 }
 225
 226                 lock_array_ptr pLocks;
 227                 {
 228                     scoped_spinlock sl(m_access);
 229                     pLocks = m_arrLocks;
 230                 }
 231
 232                 pLocks->lock_all();
 233
 234                 who = m_Owner.load( CDS_ATOMIC::memory_order_acquire );
 235                 if ( ( !(who & 1) || (who >> 1) == (me & c_nOwnerMask) ) && m_arrLocks == pLocks )
 236                     return pLocks;
 237
 238                 pLocks->unlock_all();
 239             }
 240         }
 241
 242         void release_all( lock_array_ptr p )
 243         {
 244             p->unlock_all();
 245         }
 246
 247         bool acquire_resize()
 248         {
 249             owner_t me = (owner_t) cds::OS::getCurrentThreadId();
 250
 251             back_off bkoff;
 252             for (unsigned int nAttempts = 0; nAttempts < 32; ++nAttempts ) {
 253                 owner_t ownNull = 0;
 254                 if ( m_Owner.compare_exchange_strong( ownNull, (me << 1) | 1, CDS_ATOMIC::memory_order_acquire, CDS_ATOMIC::memory_order_relaxed )) {
 255                     lock_array_ptr pOldLocks = m_arrLocks;
 256                     size_t const nLockCount = pOldLocks->size();
 257                     for ( size_t i = 0; i < nLockCount; ++i ) {
 258                         typename lock_array_type::lock_type& lock = pOldLocks->at(i);
 259                         bkoff.reset();
 260                         while ( !lock.try_lock() )
 261                             bkoff();
 262                         lock.unlock();
 263                     }
 264                     return true;
 265                 }
 266                 else
 267                     bkoff();
 268             }
 269             return false;
 270         }
 271
 272         void release_resize()
 273         {
 274             m_Owner.store( 0, CDS_ATOMIC::memory_order_release );
 275         }
 276         //@endcond
 277     public:
 278         //@cond
 279         class scoped_cell_lock {
 280             cds::lock::scoped_lock< lock_type >   m_guard;
 281
 282         public:
 283             scoped_cell_lock( refinable& policy, size_t nHash )
 284                 : m_guard( policy.acquire( nHash ), true )
 285             {}
 286         };
 287
 288         class scoped_full_lock {
 289             refinable&      m_Policy;
 290             lock_array_ptr  m_Locks;
 291         public:
 292             scoped_full_lock( refinable& policy )
 293                 : m_Policy( policy )
 294             {
 295                 m_Locks = policy.acquire_all();
 296             }
 297             ~scoped_full_lock()
 298             {
 299                 m_Policy.release_all( m_Locks );
 300             }
 301         };
 302
 303         class scoped_resize_lock {
 304             refinable&      m_Policy;
 305             bool            m_bSucceess;
 306
 307         public:
 308             scoped_resize_lock( refinable& policy )
 309                 : m_Policy( policy )
 310             {
 311                 m_bSucceess = policy.acquire_resize();
 312             }
 313
 314             ~scoped_resize_lock()
 315             {
 316                 if ( m_bSucceess )
 317                     m_Policy.release_resize();
 318             }
 319
 320             bool success() const
 321             {
 322                 return m_bSucceess;
 323             }
 324         };
 325         //@endcond
 326
 327     public:
 328         /// Constructor
 329         refinable(
 330             size_t nLockCount   ///< Initial size of lock array. Must be power of two.
 331         )
 332         : m_Owner(0)
 333         , m_nCapacity( nLockCount )
 334         {
 335             assert( cds::beans::is_power2( nLockCount ));
 336             m_arrLocks = create_lock_array( nLockCount );
 337         }
 338
 339         /// Returns lock array size
 340         /**
 341             Lock array size is not a constant for \p refinable policy and can be changed when the set is resized.
 342         */
 343         size_t lock_count() const
 344         {
 345             return m_nCapacity.load( CDS_ATOMIC::memory_order_relaxed );
 346         }
 347
 348         /// Resize for new capacity
 349         void resize( size_t nNewCapacity )
 350         {
 351             // Expect the access is locked by scoped_resize_lock!!!
 352             lock_array_ptr pNewArr = create_lock_array( nNewCapacity );
 353             scoped_spinlock sl(m_access);
 354             m_arrLocks.swap( pNewArr );
 355         }
 356     };
 357
 358 }}} // namespace cds::intrusive::striped_set
 359
 360 #endif