cds/intrusive/split_list_rcu.h

   1 //$$CDS-header$$
   2
   3 #ifndef CDSLIB_INTRUSIVE_SPLIT_LIST_RCU_H
   4 #define CDSLIB_INTRUSIVE_SPLIT_LIST_RCU_H
   5
   6 #include <cds/intrusive/details/split_list_base.h>
   7 #include <cds/details/binary_functor_wrapper.h>
   8
   9 namespace cds { namespace intrusive {
  10
  11     /// Split-ordered list RCU specialization
  12     /** @ingroup cds_intrusive_map
  13         \anchor cds_intrusive_SplitListSet_rcu
  14
  15         Hash table implementation based on split-ordered list algorithm discovered by Ori Shalev and Nir Shavit, see
  16         - [2003] Ori Shalev, Nir Shavit "Split-Ordered Lists - Lock-free Resizable Hash Tables"
  17         - [2008] Nir Shavit "The Art of Multiprocessor Programming"
  18
  19         The split-ordered list is a lock-free implementation of an extensible unbounded hash table. It uses original
  20         recursive split-ordering algorithm discovered by Ori Shalev and Nir Shavit that allows to split buckets
  21         without moving an item on resizing, see \ref cds_SplitList_algo_desc "short algo description".
  22
  23         <b>Implementation</b>
  24
  25         Template parameters are:
  26         - \p RCU - one of \ref cds_urcu_gc "RCU type"
  27         - \p OrderedList - ordered list implementation used as bucket for hash set, for example, MichaelList, LazyList.
  28             The intrusive ordered list implementation specifies the type \p T stored in the hash-set,
  29             the comparing functor for the type \p T and other features specific for the ordered list.
  30         - \p Traits - set traits, default isd \p split_list::traits.
  31             Instead of defining \p Traits struct you may use option-based syntax with \p split_list::make_traits metafunction.
  32
  33         @note About reqired features of hash functor see \ref cds_SplitList_hash_functor "SplitList general description".
  34
  35         \par How to use
  36         Before including <tt><cds/intrusive/split_list_rcu.h></tt> you should include appropriate RCU header file,
  37         see \ref cds_urcu_gc "RCU type" for list of existing RCU class and corresponding header files.
  38         For example, for \ref cds_urcu_general_buffered_gc "general-purpose buffered RCU" and
  39         MichaelList-based split-list you should include:
  40         \code
  41         #include <cds/urcu/general_buffered.h>
  42         #include <cds/intrusive/michael_list_rcu.h>
  43         #include <cds/intrusive/split_list_rcu.h>
  44
  45         // Declare Michael's list for type Foo and default traits:
  46         typedef cds::intrusive::MichaelList< cds::urcu::gc< cds::urcu::general_buffered<> >, Foo > rcu_michael_list;
  47
  48         // Declare split-list based on rcu_michael_list
  49         typedef cds::intrusive::SplitListSet< cds::urcu::gc< cds::urcu::general_buffered<> >, rcu_michael_list > rcu_split_list;
  50         \endcode
  51
  52     */
  53     template <
  54         class RCU,
  55         class OrderedList,
  56 #   ifdef CDS_DOXYGEN_INVOKED
  57         class Traits = split_list::traits
  58 #   else
  59         class Traits
  60 #   endif
  61     >
  62     class SplitListSet< cds::urcu::gc< RCU >, OrderedList, Traits >
  63     {
  64     public:
  65         typedef cds::urcu::gc< RCU > gc;   ///< RCU garbage collector
  66         typedef Traits           traits;   ///< Traits template parameters
  67
  68         /// Hash functor for \ref value_type and all its derivatives that you use
  69         typedef typename cds::opt::v::hash_selector< typename traits::hash >::type   hash;
  70
  71         //@cond
  72         typedef cds::intrusive::split_list::implementation_tag implementation_tag;
  73         //@endcond
  74
  75     protected:
  76         //@cond
  77         typedef split_list::details::rebind_list_traits<OrderedList, traits> wrapped_ordered_list;
  78         //@endcond
  79
  80     public:
  81 #   ifdef CDS_DOXYGEN_INVOKED
  82         typedef OrderedList         ordered_list;   ///< type of ordered list used as base for split-list
  83 #   else
  84         typedef typename wrapped_ordered_list::result    ordered_list;
  85 #   endif
  86         typedef typename ordered_list::value_type     value_type;     ///< type of value stored in the split-list
  87         typedef typename ordered_list::key_comparator key_comparator; ///< key compare functor
  88         typedef typename ordered_list::disposer       disposer;       ///< Node disposer functor
  89         typedef typename ordered_list::rcu_lock       rcu_lock;       ///< RCU scoped lock
  90         typedef typename ordered_list::exempt_ptr     exempt_ptr;     ///< pointer to extracted node
  91         /// Group of \p extract_xxx functions require external locking if underlying ordered list requires that
  92         static CDS_CONSTEXPR const bool c_bExtractLockExternal = ordered_list::c_bExtractLockExternal;
  93
  94         typedef typename traits::item_counter item_counter; ///< Item counter type
  95         typedef typename traits::back_off     back_off;     ///< back-off strategy for spinning
  96         typedef typename traits::memory_model memory_model; ///< Memory ordering. See cds::opt::memory_model option
  97         typedef typename traits::stat         stat;         ///< Internal statistics
  98
  99     protected:
 100         typedef typename ordered_list::node_type    list_node_type;  ///< Node type as declared in ordered list
 101         typedef split_list::node<list_node_type>    node_type;       ///< split-list node type
 102         typedef node_type                           dummy_node_type; ///< dummy node type
 103
 104         /// Split-list node traits
 105         /**
 106             This traits is intended for converting between underlying ordered list node type \ref list_node_type
 107             and split-list node type \ref node_type
 108         */
 109         typedef split_list::node_traits<typename ordered_list::node_traits>  node_traits;
 110
 111         //@cond
 112         /// Bucket table implementation
 113         typedef typename split_list::details::bucket_table_selector<
 114             traits::dynamic_bucket_table
 115             , gc
 116             , dummy_node_type
 117             , opt::allocator< typename traits::allocator >
 118             , opt::memory_model< memory_model >
 119         >::type bucket_table;
 120
 121         //@endcond
 122
 123     protected:
 124         //@cond
 125         /// Ordered list wrapper to access protected members of OrderedList
 126         class ordered_list_wrapper: public ordered_list
 127         {
 128             typedef ordered_list base_class;
 129             typedef typename base_class::auxiliary_head       bucket_head_type;
 130
 131         public:
 132             bool insert_at( dummy_node_type * pHead, value_type& val )
 133             {
 134                 assert( pHead != nullptr );
 135                 bucket_head_type h(pHead);
 136                 return base_class::insert_at( h, val );
 137             }
 138
 139             template <typename Func>
 140             bool insert_at( dummy_node_type * pHead, value_type& val, Func f )
 141             {
 142                 assert( pHead != nullptr );
 143                 bucket_head_type h(pHead);
 144                 return base_class::insert_at( h, val, f );
 145             }
 146
 147             template <typename Func>
 148             std::pair<bool, bool> ensure_at( dummy_node_type * pHead, value_type& val, Func func )
 149             {
 150                 assert( pHead != nullptr );
 151                 bucket_head_type h(pHead);
 152                 return base_class::ensure_at( h, val, func );
 153             }
 154
 155             bool unlink_at( dummy_node_type * pHead, value_type& val )
 156             {
 157                 assert( pHead != nullptr );
 158                 bucket_head_type h(pHead);
 159                 return base_class::unlink_at( h, val );
 160             }
 161
 162             template <typename Q, typename Compare, typename Func>
 163             bool erase_at( dummy_node_type * pHead, split_list::details::search_value_type<Q> const& val, Compare cmp, Func f )
 164             {
 165                 assert( pHead != nullptr );
 166                 bucket_head_type h(pHead);
 167                 return base_class::erase_at( h, val, cmp, f );
 168             }
 169
 170             template <typename Q, typename Compare>
 171             bool erase_at( dummy_node_type * pHead, split_list::details::search_value_type<Q> const& val, Compare cmp )
 172             {
 173                 assert( pHead != nullptr );
 174                 bucket_head_type h(pHead);
 175                 return base_class::erase_at( h, val, cmp );
 176             }
 177
 178             template <typename Q, typename Compare>
 179             value_type * extract_at( dummy_node_type * pHead, split_list::details::search_value_type<Q>& val, Compare cmp )
 180             {
 181                 assert( pHead != nullptr );
 182                 bucket_head_type h(pHead);
 183                 return base_class::extract_at( h, val, cmp );
 184             }
 185
 186             template <typename Q, typename Compare, typename Func>
 187             bool find_at( dummy_node_type * pHead, split_list::details::search_value_type<Q>& val, Compare cmp, Func f ) const
 188             {
 189                 assert( pHead != nullptr );
 190                 bucket_head_type h(pHead);
 191                 return base_class::find_at( h, val, cmp, f );
 192             }
 193
 194             template <typename Q, typename Compare>
 195             bool find_at( dummy_node_type * pHead, split_list::details::search_value_type<Q> const & val, Compare cmp ) const
 196             {
 197                 assert( pHead != nullptr );
 198                 bucket_head_type h(pHead);
 199                 return base_class::find_at( h, val, cmp );
 200             }
 201
 202             template <typename Q, typename Compare>
 203             value_type * get_at( dummy_node_type * pHead, split_list::details::search_value_type<Q>& val, Compare cmp ) const
 204             {
 205                 assert( pHead != nullptr );
 206                 bucket_head_type h(pHead);
 207                 return base_class::get_at( h, val, cmp );
 208             }
 209
 210             bool insert_aux_node( dummy_node_type * pNode )
 211             {
 212                 return base_class::insert_aux_node( pNode );
 213             }
 214             bool insert_aux_node( dummy_node_type * pHead, dummy_node_type * pNode )
 215             {
 216                 bucket_head_type h(pHead);
 217                 return base_class::insert_aux_node( h, pNode );
 218             }
 219         };
 220
 221         template <typename Less>
 222         struct less_wrapper: public cds::opt::details::make_comparator_from_less<Less>
 223         {
 224             typedef cds::opt::details::make_comparator_from_less<Less> base_wrapper;
 225
 226             template <typename Q1, typename Q2>
 227             int operator()( split_list::details::search_value_type<Q1> const& v1, Q2 const& v2 ) const
 228             {
 229                 return base_wrapper::operator()( v1.val, v2 );
 230             }
 231
 232             template <typename Q1, typename Q2>
 233             int operator()( Q1 const& v1, split_list::details::search_value_type<Q2> const& v2 ) const
 234             {
 235                 return base_wrapper::operator()( v1, v2.val );
 236             }
 237         };
 238         //@endcond
 239
 240     protected:
 241         ordered_list_wrapper    m_List;             ///< Ordered list containing split-list items
 242         bucket_table            m_Buckets;          ///< bucket table
 243         atomics::atomic<size_t> m_nBucketCountLog2; ///< log2( current bucket count )
 244         item_counter            m_ItemCounter;      ///< Item counter
 245         hash                    m_HashFunctor;      ///< Hash functor
 246         stat                    m_Stat;             ///< Internal stattistics accumulator
 247
 248     protected:
 249         //@cond
 250         typedef cds::details::Allocator< dummy_node_type, typename traits::allocator >   dummy_node_allocator;
 251
 252         dummy_node_type * alloc_dummy_node( size_t nHash )
 253         {
 254             m_Stat.onHeadNodeAllocated();
 255             return dummy_node_allocator().New( nHash );
 256         }
 257         void free_dummy_node( dummy_node_type * p )
 258         {
 259             dummy_node_allocator().Delete( p );
 260             m_Stat.onHeadNodeFreed();
 261         }
 262
 263         /// Calculates hash value of \p key
 264         template <typename Q>
 265         size_t hash_value( Q const& key ) const
 266         {
 267             return m_HashFunctor( key );
 268         }
 269
 270         size_t bucket_no( size_t nHash ) const
 271         {
 272             return nHash & ( (1 << m_nBucketCountLog2.load(atomics::memory_order_relaxed)) - 1 );
 273         }
 274
 275         static size_t parent_bucket( size_t nBucket )
 276         {
 277             assert( nBucket > 0 );
 278             return nBucket & ~( 1 << bitop::MSBnz( nBucket ) );
 279         }
 280
 281         dummy_node_type * init_bucket( size_t nBucket )
 282         {
 283             assert( nBucket > 0 );
 284             size_t nParent = parent_bucket( nBucket );
 285
 286             dummy_node_type * pParentBucket = m_Buckets.bucket( nParent );
 287             if ( pParentBucket == nullptr ) {
 288                 pParentBucket = init_bucket( nParent );
 289                 m_Stat.onRecursiveInitBucket();
 290             }
 291
 292             assert( pParentBucket != nullptr );
 293
 294             // Allocate a dummy node for new bucket
 295             {
 296                 dummy_node_type * pBucket = alloc_dummy_node( split_list::dummy_hash( nBucket ) );
 297                 if ( m_List.insert_aux_node( pParentBucket, pBucket ) ) {
 298                     m_Buckets.bucket( nBucket, pBucket );
 299                     m_Stat.onNewBucket();
 300                     return pBucket;
 301                 }
 302                 free_dummy_node( pBucket );
 303             }
 304
 305             // Another thread set the bucket. Wait while it done
 306
 307             // In this point, we must wait while nBucket is empty.
 308             // The compiler can decide that waiting loop can be "optimized" (stripped)
 309             // To prevent this situation, we use waiting on volatile bucket_head_ptr pointer.
 310             //
 311             m_Stat.onBucketInitContenton();
 312             back_off bkoff;
 313             while ( true ) {
 314                 dummy_node_type volatile * p = m_Buckets.bucket( nBucket );
 315                 if ( p != nullptr )
 316                     return const_cast<dummy_node_type *>( p );
 317                 bkoff();
 318                 m_Stat.onBusyWaitBucketInit();
 319             }
 320         }
 321
 322         dummy_node_type * get_bucket( size_t nHash )
 323         {
 324             size_t nBucket = bucket_no( nHash );
 325
 326             dummy_node_type * pHead = m_Buckets.bucket( nBucket );
 327             if ( pHead == nullptr )
 328                 pHead = init_bucket( nBucket );
 329
 330             assert( pHead->is_dummy() );
 331
 332             return pHead;
 333         }
 334
 335         void init()
 336         {
 337             // GC and OrderedList::gc must be the same
 338             static_assert( std::is_same<gc, typename ordered_list::gc>::value, "GC and OrderedList::gc must be the same");
 339
 340             // atomicity::empty_item_counter is not allowed as a item counter
 341             static_assert( !std::is_same<item_counter, cds::atomicity::empty_item_counter>::value,
 342                            "cds::atomicity::empty_item_counter is not allowed as a item counter");
 343
 344             // Initialize bucket 0
 345             dummy_node_type * pNode = alloc_dummy_node( 0 /*split_list::dummy_hash(0)*/ );
 346
 347             // insert_aux_node cannot return false for empty list
 348             CDS_VERIFY( m_List.insert_aux_node( pNode ));
 349
 350             m_Buckets.bucket( 0, pNode );
 351         }
 352
 353         void    inc_item_count()
 354         {
 355             size_t sz = m_nBucketCountLog2.load(atomics::memory_order_relaxed);
 356             if ( ( ++m_ItemCounter >> sz ) > m_Buckets.load_factor() && ((size_t)(1 << sz )) < m_Buckets.capacity() )
 357             {
 358                 m_nBucketCountLog2.compare_exchange_strong( sz, sz + 1, atomics::memory_order_seq_cst, atomics::memory_order_relaxed );
 359             }
 360         }
 361
 362         template <typename Q, typename Compare, typename Func>
 363         bool find_( Q& val, Compare cmp, Func f )
 364         {
 365             size_t nHash = hash_value( val );
 366             split_list::details::search_value_type<Q>  sv( val, split_list::regular_hash( nHash ));
 367             dummy_node_type * pHead = get_bucket( nHash );
 368             assert( pHead != nullptr );
 369
 370             return m_Stat.onFind( m_List.find_at( pHead, sv, cmp,
 371                 [&f](value_type& item, split_list::details::search_value_type<Q>& val){ f(item, val.val ); }));
 372         }
 373
 374         template <typename Q, typename Compare>
 375         bool find_value( Q const& val, Compare cmp )
 376         {
 377             size_t nHash = hash_value( val );
 378             split_list::details::search_value_type<Q const>  sv( val, split_list::regular_hash( nHash ));
 379             dummy_node_type * pHead = get_bucket( nHash );
 380             assert( pHead != nullptr );
 381
 382             return m_Stat.onFind( m_List.find_at( pHead, sv, cmp ));
 383         }
 384
 385         template <typename Q, typename Compare>
 386         value_type * get_( Q const& val, Compare cmp )
 387         {
 388             size_t nHash = hash_value( val );
 389             split_list::details::search_value_type<Q const>  sv( val, split_list::regular_hash( nHash ));
 390             dummy_node_type * pHead = get_bucket( nHash );
 391             assert( pHead != nullptr );
 392
 393             value_type * p = m_List.get_at( pHead, sv, cmp );
 394             m_Stat.onFind( p != nullptr );
 395             return p;
 396         }
 397
 398         template <typename Q, typename Compare>
 399         value_type * extract_( Q const& val, Compare cmp )
 400         {
 401             size_t nHash = hash_value( val );
 402             split_list::details::search_value_type<Q const>  sv( val, split_list::regular_hash( nHash ));
 403             dummy_node_type * pHead = get_bucket( nHash );
 404             assert( pHead != nullptr );
 405
 406             value_type * pNode = m_List.extract_at( pHead, sv, cmp );
 407             if ( pNode ) {
 408                 --m_ItemCounter;
 409                 m_Stat.onExtractSuccess();
 410             }
 411             else
 412                 m_Stat.onExtractFailed();
 413             return pNode;
 414         }
 415
 416         template <typename Q, typename Less>
 417         value_type * extract_with_( Q const& val, Less pred )
 418         {
 419             CDS_UNUSED( pred );
 420             return extract_( val, typename wrapped_ordered_list::template make_compare_from_less<Less>());
 421         }
 422
 423         template <typename Q, typename Compare>
 424         bool erase_( const Q& val, Compare cmp )
 425         {
 426             size_t nHash = hash_value( val );
 427             split_list::details::search_value_type<Q const>  sv( val, split_list::regular_hash( nHash ));
 428             dummy_node_type * pHead = get_bucket( nHash );
 429             assert( pHead != nullptr );
 430
 431             if ( m_List.erase_at( pHead, sv, cmp ) ) {
 432                 --m_ItemCounter;
 433                 m_Stat.onEraseSuccess();
 434                 return true;
 435             }
 436             m_Stat.onEraseFailed();
 437             return false;
 438         }
 439
 440         template <typename Q, typename Compare, typename Func>
 441         bool erase_( Q const& val, Compare cmp, Func f )
 442         {
 443             size_t nHash = hash_value( val );
 444             split_list::details::search_value_type<Q const>  sv( val, split_list::regular_hash( nHash ));
 445             dummy_node_type * pHead = get_bucket( nHash );
 446             assert( pHead != nullptr );
 447
 448             if ( m_List.erase_at( pHead, sv, cmp, f )) {
 449                 --m_ItemCounter;
 450                 m_Stat.onEraseSuccess();
 451                 return true;
 452             }
 453             m_Stat.onEraseFailed();
 454             return false;
 455         }
 456
 457         //@endcond
 458
 459     public:
 460         /// Initialize split-ordered list of default capacity
 461         /**
 462             The default capacity is defined in bucket table constructor.
 463             See split_list::expandable_bucket_table, split_list::static_ducket_table
 464             which selects by split_list::dynamic_bucket_table option.
 465         */
 466         SplitListSet()
 467             : m_nBucketCountLog2(1)
 468         {
 469             init();
 470         }
 471
 472         /// Initialize split-ordered list
 473         SplitListSet(
 474             size_t nItemCount           ///< estimate average of item count
 475             , size_t nLoadFactor = 1    ///< load factor - average item count per bucket. Small integer up to 8, default is 1.
 476             )
 477             : m_Buckets( nItemCount, nLoadFactor )
 478             , m_nBucketCountLog2(1)
 479         {
 480             init();
 481         }
 482
 483     public:
 484         /// Inserts new node
 485         /**
 486             The function inserts \p val in the set if it does not contain
 487             an item with key equal to \p val.
 488
 489             The function makes RCU lock internally.
 490
 491             Returns \p true if \p val is placed into the set, \p false otherwise.
 492         */
 493         bool insert( value_type& val )
 494         {
 495             size_t nHash = hash_value( val );
 496             dummy_node_type * pHead = get_bucket( nHash );
 497             assert( pHead != nullptr );
 498
 499             node_traits::to_node_ptr( val )->m_nHash = split_list::regular_hash( nHash );
 500
 501             if ( m_List.insert_at( pHead, val )) {
 502                 inc_item_count();
 503                 m_Stat.onInsertSuccess();
 504                 return true;
 505             }
 506             m_Stat.onInsertFailed();
 507             return false;
 508         }
 509
 510         /// Inserts new node
 511         /**
 512             This function is intended for derived non-intrusive containers.
 513
 514             The function allows to split creating of new item into two part:
 515             - create item with key only
 516             - insert new item into the set
 517             - if inserting is success, calls  \p f functor to initialize value-field of \p val.
 518
 519             The functor signature is:
 520             \code
 521                 void func( value_type& val );
 522             \endcode
 523             where \p val is the item inserted.
 524
 525             The function makes RCU lock internally.
 526
 527             @warning For \ref cds_intrusive_MichaelList_rcu "MichaelList" as the bucket see \ref cds_intrusive_item_creating "insert item troubleshooting".
 528             \ref cds_intrusive_LazyList_rcu "LazyList" provides exclusive access to inserted item and does not require any node-level
 529             synchronization.
 530             */
 531         template <typename Func>
 532         bool insert( value_type& val, Func f )
 533         {
 534             size_t nHash = hash_value( val );
 535             dummy_node_type * pHead = get_bucket( nHash );
 536             assert( pHead != nullptr );
 537
 538             node_traits::to_node_ptr( val )->m_nHash = split_list::regular_hash( nHash );
 539
 540             if ( m_List.insert_at( pHead, val, f )) {
 541                 inc_item_count();
 542                 m_Stat.onInsertSuccess();
 543                 return true;
 544             }
 545             m_Stat.onInsertFailed();
 546             return false;
 547         }
 548
 549         /// Ensures that the \p val exists in the set
 550         /**
 551             The operation performs inserting or changing data with lock-free manner.
 552
 553             If the item \p val is not found in the set, then \p val is inserted into the set.
 554             Otherwise, the functor \p func is called with item found.
 555             The functor signature is:
 556             \code
 557                 void func( bool bNew, value_type& item, value_type& val );
 558             \endcode
 559             with arguments:
 560             - \p bNew - \p true if the item has been inserted, \p false otherwise
 561             - \p item - item of the set
 562             - \p val - argument \p val passed into the \p ensure function
 563             If new item has been inserted (i.e. \p bNew is \p true) then \p item and \p val arguments
 564             refers to the same thing.
 565
 566             The function makes RCU lock internally.
 567
 568             Returns std::pair<bool, bool> where \p first is \p true if operation is successfull,
 569             \p second is \p true if new item has been added or \p false if the item with \p key
 570             already is in the set.
 571
 572             @warning For \ref cds_intrusive_MichaelList_rcu "MichaelList" as the bucket see \ref cds_intrusive_item_creating "insert item troubleshooting".
 573             \ref cds_intrusive_LazyList_rcu "LazyList" provides exclusive access to inserted item and does not require any node-level
 574             synchronization.
 575             */
 576         template <typename Func>
 577         std::pair<bool, bool> ensure( value_type& val, Func func )
 578         {
 579             size_t nHash = hash_value( val );
 580             dummy_node_type * pHead = get_bucket( nHash );
 581             assert( pHead != nullptr );
 582
 583             node_traits::to_node_ptr( val )->m_nHash = split_list::regular_hash( nHash );
 584
 585             std::pair<bool, bool> bRet = m_List.ensure_at( pHead, val, func );
 586             if ( bRet.first && bRet.second ) {
 587                 inc_item_count();
 588                 m_Stat.onEnsureNew();
 589             }
 590             else
 591                 m_Stat.onEnsureExist();
 592             return bRet;
 593         }
 594
 595         /// Unlinks the item \p val from the set
 596         /**
 597             The function searches the item \p val in the set and unlinks it from the set
 598             if it is found and is equal to \p val.
 599
 600             Difference between \ref erase and \p unlink functions: \p erase finds <i>a key</i>
 601             and deletes the item found. \p unlink finds an item by key and deletes it
 602             only if \p val is an item of that set, i.e. the pointer to item found
 603             is equal to <tt> &val </tt>.
 604
 605             RCU \p synchronize method can be called, therefore, RCU should not be locked.
 606
 607             The function returns \p true if success and \p false otherwise.
 608         */
 609         bool unlink( value_type& val )
 610         {
 611             size_t nHash = hash_value( val );
 612             dummy_node_type * pHead = get_bucket( nHash );
 613             assert( pHead != nullptr );
 614
 615             if ( m_List.unlink_at( pHead, val ) ) {
 616                 --m_ItemCounter;
 617                 m_Stat.onEraseSuccess();
 618                 return true;
 619             }
 620             m_Stat.onEraseFailed();
 621             return false;
 622         }
 623
 624         /// Deletes the item from the set
 625         /** \anchor cds_intrusive_SplitListSet_rcu_erase
 626             The function searches an item with key equal to \p key in the set,
 627             unlinks it from the set, and returns \p true.
 628             If the item with key equal to \p key is not found the function return \p false.
 629
 630             Difference between \ref erase and \p unlink functions: \p erase finds <i>a key</i>
 631             and deletes the item found. \p unlink finds an item by key and deletes it
 632             only if \p key is an item of that set, i.e. the pointer to item found
 633             is equal to <tt> &key </tt>.
 634
 635             RCU \p synchronize method can be called, therefore, RCU should not be locked.
 636
 637             Note the hash functor should accept a parameter of type \p Q that can be not the same as \p value_type.
 638         */
 639         template <typename Q>
 640         bool erase( Q const& key )
 641         {
 642             return erase_( key, key_comparator() );
 643         }
 644
 645         /// Deletes the item from the set using \p pred for searching
 646         /**
 647             The function is an analog of \ref cds_intrusive_SplitListSet_rcu_erase "erase(Q const&)"
 648             but \p cmp is used for key compare.
 649             \p Less has the interface like \p std::less.
 650             \p pred must imply the same element order as the comparator used for building the set.
 651         */
 652         template <typename Q, typename Less>
 653         bool erase_with( Q const& key, Less pred )
 654         {
 655             CDS_UNUSED( pred );
 656             return erase_( key, typename wrapped_ordered_list::template make_compare_from_less<Less>() );
 657         }
 658
 659         /// Deletes the item from the set
 660         /** \anchor cds_intrusive_SplitListSet_rcu_erase_func
 661             The function searches an item with key equal to \p key in the set,
 662             call \p f functor with item found, unlinks it from the set, and returns \p true.
 663             The \ref disposer specified by \p OrderedList class template parameter is called
 664             by garbage collector \p GC asynchronously.
 665
 666             The \p Func interface is
 667             \code
 668             struct functor {
 669                 void operator()( value_type const& item );
 670             };
 671             \endcode
 672
 673             If the item with key equal to \p key is not found the function return \p false.
 674
 675             RCU \p synchronize method can be called, therefore, RCU should not be locked.
 676
 677             Note the hash functor should accept a parameter of type \p Q that can be not the same as \p value_type.
 678         */
 679         template <typename Q, typename Func>
 680         bool erase( Q const& key, Func f )
 681         {
 682             return erase_( key, key_comparator(), f );
 683         }
 684
 685         /// Deletes the item from the set using \p pred for searching
 686         /**
 687             The function is an analog of \ref cds_intrusive_SplitListSet_rcu_erase_func "erase(Q const&, Func)"
 688             but \p cmp is used for key compare.
 689             \p Less has the interface like \p std::less.
 690             \p pred must imply the same element order as the comparator used for building the set.
 691         */
 692         template <typename Q, typename Less, typename Func>
 693         bool erase_with( Q const& key, Less pred, Func f )
 694         {
 695             CDS_UNUSED( pred );
 696             return erase_( key, typename wrapped_ordered_list::template make_compare_from_less<Less>(), f );
 697         }
 698
 699         /// Extracts an item from the set
 700         /** \anchor cds_intrusive_SplitListSet_rcu_extract
 701             The function searches an item with key equal to \p key in the set,
 702             unlinks it, and returns \ref cds::urcu::exempt_ptr "exempt_ptr" pointer to the item found.
 703             If the item with the key equal to \p key is not found the function returns an empty \p exempt_ptr.
 704
 705             @note The function does NOT call RCU read-side lock or synchronization,
 706             and does NOT dispose the item found. It just excludes the item from the set
 707             and returns a pointer to item found.
 708             You should lock RCU before calling of the function, and you should synchronize RCU
 709             outside the RCU lock before reusing returned pointer.
 710
 711             \code
 712             typedef cds::urcu::gc< general_buffered<> > rcu;
 713             typedef cds::intrusive::MichaelList< rcu, Foo > rcu_michael_list;
 714             typedef cds::intrusive::SplitListSet< rcu, rcu_michael_list, foo_traits > rcu_splitlist_set;
 715
 716             rcu_splitlist_set theSet;
 717             // ...
 718
 719             rcu_splitlist_set::exempt_ptr p;
 720             {
 721                 // first, we should lock RCU
 722                 rcu_splitlist_set::rcu_lock lock;
 723
 724                 // Now, you can apply extract function
 725                 // Note that you must not delete the item found inside the RCU lock
 726                 p = theList.extract( 10 );
 727                 if ( p ) {
 728                     // do something with p
 729                     ...
 730                 }
 731             }
 732
 733             // We may safely release p here
 734             // release() passes the pointer to RCU reclamation cycle:
 735             // it invokes RCU retire_ptr function with the disposer you provided for rcu_michael_list.
 736             p.release();
 737             \endcode
 738         */
 739         template <typename Q>
 740         exempt_ptr extract( Q const& key )
 741         {
 742             return exempt_ptr(extract_( key, key_comparator() ));
 743         }
 744
 745         /// Extracts an item from the set using \p pred for searching
 746         /**
 747             The function is an analog of \p extract(Q const&) but \p pred is used for key compare.
 748             \p Less functor has the interface like \p std::less.
 749             \p pred must imply the same element order as the comparator used for building the set.
 750         */
 751         template <typename Q, typename Less>
 752         exempt_ptr extract_with( Q const& key, Less pred )
 753         {
 754             return exempt_ptr( extract_with_( key, pred ));
 755         }
 756
 757         /// Finds the key \p key
 758         /** \anchor cds_intrusive_SplitListSet_rcu_find_func
 759             The function searches the item with key equal to \p key and calls the functor \p f for item found.
 760             The interface of \p Func functor is:
 761             \code
 762             struct functor {
 763                 void operator()( value_type& item, Q& key );
 764             };
 765             \endcode
 766             where \p item is the item found, \p key is the <tt>find</tt> function argument.
 767
 768             The functor can change non-key fields of \p item. Note that the functor is only guarantee
 769             that \p item cannot be disposed during functor is executing.
 770             The functor does not serialize simultaneous access to the set \p item. If such access is
 771             possible you must provide your own synchronization schema on item level to exclude unsafe item modifications.
 772
 773             The \p key argument is non-const since it can be used as \p f functor destination i.e., the functor
 774             can modify both arguments.
 775
 776             Note the hash functor specified for class \p Traits template parameter
 777             should accept a parameter of type \p Q that can be not the same as \p value_type.
 778
 779             The function applies RCU lock internally.
 780
 781             The function returns \p true if \p key is found, \p false otherwise.
 782         */
 783         template <typename Q, typename Func>
 784         bool find( Q& key, Func f )
 785         {
 786             return find_( key, key_comparator(), f );
 787         }
 788         //@cond
 789         template <typename Q, typename Func>
 790         bool find( Q const& key, Func f )
 791         {
 792             return find_( key, key_comparator(), f );
 793         }
 794         //@endcond
 795
 796         /// Finds the key \p key with \p pred predicate for comparing
 797         /**
 798             The function is an analog of \ref cds_intrusive_SplitListSet_rcu_find_func "find(Q&, Func)"
 799             but \p cmp is used for key compare.
 800             \p Less has the interface like \p std::less.
 801             \p cmp must imply the same element order as the comparator used for building the set.
 802         */
 803         template <typename Q, typename Less, typename Func>
 804         bool find_with( Q& key, Less pred, Func f )
 805         {
 806             CDS_UNUSED( pred );
 807             return find_( key, typename wrapped_ordered_list::template make_compare_from_less<Less>(), f );
 808         }
 809         //@cond
 810         template <typename Q, typename Less, typename Func>
 811         bool find_with( Q const& key, Less pred, Func f )
 812         {
 813             CDS_UNUSED( pred );
 814             return find_( key, typename wrapped_ordered_list::template make_compare_from_less<Less>(), f );
 815         }
 816         //@endcond
 817
 818         /// Finds the key \p key
 819         /** \anchor cds_intrusive_SplitListSet_rcu_find_val
 820             The function searches the item with key equal to \p key
 821             and returns \p true if \p key found or \p false otherwise.
 822         */
 823         template <typename Q>
 824         bool find( Q const& key )
 825         {
 826             return find_value( key, key_comparator() );
 827         }
 828
 829         /// Finds the key \p key with \p pred predicate for comparing
 830         /**
 831             The function is an analog of \ref cds_intrusive_SplitListSet_rcu_find_val "find(Q const&)"
 832             but \p cmp is used for key compare.
 833             \p Less has the interface like \p std::less.
 834             \p pred must imply the same element order as the comparator used for building the set.
 835         */
 836         template <typename Q, typename Less>
 837         bool find_with( Q const& key, Less pred )
 838         {
 839             CDS_UNUSED( pred );
 840             return find_value( key, typename wrapped_ordered_list::template make_compare_from_less<Less>() );
 841         }
 842
 843         /// Finds the key \p key and return the item found
 844         /** \anchor cds_intrusive_SplitListSet_rcu_get
 845             The function searches the item with key equal to \p key and returns the pointer to item found.
 846             If \p key is not found it returns \p nullptr.
 847
 848             Note the compare functor should accept a parameter of type \p Q that can be not the same as \p value_type.
 849
 850             RCU should be locked before call of this function.
 851             Returned item is valid only while RCU is locked:
 852             \code
 853             cds::intrusive::SplitListSet< your_template_parameters > theSet;
 854             // ...
 855             {
 856                 // Lock RCU
 857                 hash_set::rcu_lock lock;
 858
 859                 foo * pVal = theSet.get( 5 );
 860                 if ( pVal ) {
 861                     // Deal with pVal
 862                     //...
 863                 }
 864                 // Unlock RCU by rcu_lock destructor
 865                 // pVal can be retired by disposer at any time after RCU has been unlocked
 866             }
 867             \endcode
 868         */
 869         template <typename Q>
 870         value_type * get( Q const& key )
 871         {
 872             return get_( key, key_comparator() );
 873         }
 874
 875         /// Finds the key \p key and return the item found
 876         /**
 877             The function is an analog of \ref cds_intrusive_SplitListSet_rcu_get "get(Q const&)"
 878             but \p pred is used for comparing the keys.
 879
 880             \p Less functor has the semantics like \p std::less but should take arguments of type \ref value_type and \p Q
 881             in any order.
 882             \p pred must imply the same element order as the comparator used for building the set.
 883         */
 884         template <typename Q, typename Less>
 885         value_type * get_with( Q const& key, Less pred )
 886         {
 887             CDS_UNUSED( pred );
 888             return get_( key, typename wrapped_ordered_list::template make_compare_from_less<Less>());
 889         }
 890
 891
 892         /// Returns item count in the set
 893         size_t size() const
 894         {
 895             return m_ItemCounter;
 896         }
 897
 898         /// Checks if the set is empty
 899         /**
 900             Emptiness is checked by item counting: if item count is zero then the set is empty.
 901             Thus, the correct item counting feature is an important part of split-list set implementation.
 902         */
 903         bool empty() const
 904         {
 905             return size() == 0;
 906         }
 907
 908         /// Clears the set (not atomic)
 909         void clear()
 910         {
 911             iterator it = begin();
 912             while ( it != end() ) {
 913                 iterator i(it);
 914                 ++i;
 915                 unlink( *it );
 916                 it = i;
 917             }
 918         }
 919
 920         /// Returns internal statistics
 921         stat const& statistics() const
 922         {
 923             return m_Stat;
 924         }
 925
 926     protected:
 927         //@cond
 928         template <bool IsConst>
 929         class iterator_type
 930             :public split_list::details::iterator_type<node_traits, ordered_list, IsConst>
 931         {
 932             typedef split_list::details::iterator_type<node_traits, ordered_list, IsConst> iterator_base_class;
 933             typedef typename iterator_base_class::list_iterator list_iterator;
 934         public:
 935             iterator_type()
 936                 : iterator_base_class()
 937             {}
 938
 939             iterator_type( iterator_type const& src )
 940                 : iterator_base_class( src )
 941             {}
 942
 943             // This ctor should be protected...
 944             iterator_type( list_iterator itCur, list_iterator itEnd )
 945                 : iterator_base_class( itCur, itEnd )
 946             {}
 947         };
 948         //@endcond
 949     public:
 950         /// Forward iterator
 951         /**
 952             The forward iterator for a split-list has some features:
 953             - it has no post-increment operator
 954             - it depends on iterator of underlying \p OrderedList
 955             - The iterator cannot be moved across thread boundary since it may contain GC's guard that is thread-private GC data.
 956             - Iterator ensures thread-safety even if you delete the item that iterator points to. However, in case of concurrent
 957               deleting operations it is no guarantee that you iterate all item in the split-list.
 958
 959             Therefore, the use of iterators in concurrent environment is not good idea. Use the iterator on the concurrent container
 960             for debug purpose only.
 961         */
 962         typedef iterator_type<false>    iterator;
 963         /// Const forward iterator
 964         /**
 965             For iterator's features and requirements see \ref iterator
 966         */
 967         typedef iterator_type<true>     const_iterator;
 968
 969         /// Returns a forward iterator addressing the first element in a split-list
 970         /**
 971             For empty list \code begin() == end() \endcode
 972         */
 973         iterator begin()
 974         {
 975             return iterator( m_List.begin(), m_List.end() );
 976         }
 977
 978         /// Returns an iterator that addresses the location succeeding the last element in a split-list
 979         /**
 980             Do not use the value returned by <tt>end</tt> function to access any item.
 981
 982             The returned value can be used only to control reaching the end of the split-list.
 983             For empty list \code begin() == end() \endcode
 984         */
 985         iterator end()
 986         {
 987             return iterator( m_List.end(), m_List.end() );
 988         }
 989
 990         /// Returns a forward const iterator addressing the first element in a split-list
 991         const_iterator begin() const
 992         {
 993             return cbegin();
 994         }
 995         /// Returns a forward const iterator addressing the first element in a split-list
 996         const_iterator cbegin() const
 997         {
 998             return const_iterator( m_List.cbegin(), m_List.cend() );
 999         }
1000
1001         /// Returns an const iterator that addresses the location succeeding the last element in a split-list
1002         const_iterator end() const
1003         {
1004             return cend();
1005         }
1006         /// Returns an const iterator that addresses the location succeeding the last element in a split-list
1007         const_iterator cend() const
1008         {
1009             return const_iterator( m_List.cend(), m_List.cend() );
1010         }
1011
1012     };
1013
1014 }}  // namespace cds::intrusive
1015
1016 #endif // #ifndef CDSLIB_INTRUSIVE_SPLIT_LIST_RCU_H