cds/intrusive/skip_list_rcu.h

   1 //$$CDS-header$$
   2
   3 #ifndef CDSLIB_INTRUSIVE_SKIP_LIST_RCU_H
   4 #define CDSLIB_INTRUSIVE_SKIP_LIST_RCU_H
   5
   6 #include <type_traits>
   7 #include <memory>
   8 #include <cds/intrusive/details/skip_list_base.h>
   9 #include <cds/opt/compare.h>
  10 #include <cds/urcu/details/check_deadlock.h>
  11 #include <cds/details/binary_functor_wrapper.h>
  12 #include <cds/urcu/exempt_ptr.h>
  13 #include <cds/urcu/raw_ptr.h>
  14 #include <cds/intrusive/details/raw_ptr_disposer.h>
  15
  16 namespace cds { namespace intrusive {
  17
  18     //@cond
  19     namespace skip_list {
  20
  21         template <class RCU, typename Tag>
  22         class node< cds::urcu::gc< RCU >, Tag >
  23         {
  24         public:
  25             typedef cds::urcu::gc< RCU >    gc; ///< Garbage collector
  26             typedef Tag     tag               ; ///< tag
  27
  28             // Mark bits:
  29             //  bit 0 - the item is logically deleted
  30             //  bit 1 - the item is extracted (only for level 0)
  31             typedef cds::details::marked_ptr<node, 3> marked_ptr;        ///< marked pointer
  32             typedef atomics::atomic< marked_ptr >     atomic_marked_ptr; ///< atomic marked pointer
  33             typedef atomic_marked_ptr                 tower_item_type;
  34
  35         protected:
  36             atomic_marked_ptr       m_pNext;     ///< Next item in bottom-list (list at level 0)
  37         public:
  38             node *                  m_pDelChain; ///< Deleted node chain (local for a thread)
  39 #       ifdef _DEBUG
  40             bool volatile           m_bLinked;
  41             bool volatile           m_bUnlinked;
  42 #       endif
  43         protected:
  44             unsigned int            m_nHeight;   ///< Node height (size of m_arrNext array). For node at level 0 the height is 1.
  45             atomic_marked_ptr *     m_arrNext;   ///< Array of next items for levels 1 .. m_nHeight - 1. For node at level 0 \p m_arrNext is \p nullptr
  46
  47         public:
  48             /// Constructs a node of height 1 (a bottom-list node)
  49             CDS_CONSTEXPR node()
  50                 : m_pNext( nullptr )
  51                 , m_pDelChain( nullptr )
  52 #       ifdef _DEBUG
  53                 , m_bLinked( false )
  54                 , m_bUnlinked( false )
  55 #       endif
  56                 , m_nHeight(1)
  57                 , m_arrNext( nullptr )
  58             {}
  59
  60 #       ifdef _DEBUG
  61             ~node()
  62             {
  63                 assert( !m_bLinked || m_bUnlinked );
  64             }
  65 #       endif
  66
  67             /// Constructs a node of height \p nHeight
  68             void make_tower( unsigned int nHeight, atomic_marked_ptr * nextTower )
  69             {
  70                 assert( nHeight > 0 );
  71                 assert( (nHeight == 1 && nextTower == nullptr)  // bottom-list node
  72                         || (nHeight > 1 && nextTower != nullptr)   // node at level of more than 0
  73                     );
  74
  75                 m_arrNext = nextTower;
  76                 m_nHeight = nHeight;
  77             }
  78
  79             atomic_marked_ptr * release_tower()
  80             {
  81                 atomic_marked_ptr * pTower = m_arrNext;
  82                 m_arrNext = nullptr;
  83                 m_nHeight = 1;
  84                 return pTower;
  85             }
  86
  87             atomic_marked_ptr * get_tower() const
  88             {
  89                 return m_arrNext;
  90             }
  91
  92             void clear_tower()
  93             {
  94                 for ( unsigned int nLevel = 1; nLevel < m_nHeight; ++nLevel )
  95                     next(nLevel).store( marked_ptr(), atomics::memory_order_relaxed );
  96             }
  97
  98             /// Access to element of next pointer array
  99             atomic_marked_ptr& next( unsigned int nLevel )
 100             {
 101                 assert( nLevel < height() );
 102                 assert( nLevel == 0 || (nLevel > 0 && m_arrNext != nullptr) );
 103
 104                 return nLevel ? m_arrNext[nLevel - 1] : m_pNext;
 105             }
 106
 107             /// Access to element of next pointer array (const version)
 108             atomic_marked_ptr const& next( unsigned int nLevel ) const
 109             {
 110                 assert( nLevel < height() );
 111                 assert( nLevel == 0 || nLevel > 0 && m_arrNext != nullptr );
 112
 113                 return nLevel ? m_arrNext[nLevel - 1] : m_pNext;
 114             }
 115
 116             /// Access to element of next pointer array (same as \ref next function)
 117             atomic_marked_ptr& operator[]( unsigned int nLevel )
 118             {
 119                 return next( nLevel );
 120             }
 121
 122             /// Access to element of next pointer array (same as \ref next function)
 123             atomic_marked_ptr const& operator[]( unsigned int nLevel ) const
 124             {
 125                 return next( nLevel );
 126             }
 127
 128             /// Height of the node
 129             unsigned int height() const
 130             {
 131                 return m_nHeight;
 132             }
 133
 134             /// Clears internal links
 135             void clear()
 136             {
 137                 assert( m_arrNext == nullptr );
 138                 m_pNext.store( marked_ptr(), atomics::memory_order_release );
 139                 m_pDelChain = nullptr;
 140             }
 141
 142             bool is_cleared() const
 143             {
 144                 return m_pNext == atomic_marked_ptr()
 145                     && m_arrNext == nullptr
 146                     && m_nHeight <= 1;
 147             }
 148         };
 149     } // namespace skip_list
 150     //@endcond
 151
 152     //@cond
 153     namespace skip_list { namespace details {
 154
 155         template <class RCU, typename NodeTraits, typename BackOff, bool IsConst>
 156         class iterator< cds::urcu::gc< RCU >, NodeTraits, BackOff, IsConst >
 157         {
 158         public:
 159             typedef cds::urcu::gc< RCU >                gc;
 160             typedef NodeTraits                          node_traits;
 161             typedef BackOff                             back_off;
 162             typedef typename node_traits::node_type     node_type;
 163             typedef typename node_traits::value_type    value_type;
 164             static bool const c_isConst = IsConst;
 165
 166             typedef typename std::conditional< c_isConst, value_type const &, value_type &>::type   value_ref;
 167
 168         protected:
 169             typedef typename node_type::marked_ptr          marked_ptr;
 170             typedef typename node_type::atomic_marked_ptr   atomic_marked_ptr;
 171
 172             node_type *             m_pNode;
 173
 174         protected:
 175             void next()
 176             {
 177                 // RCU should be locked before iterating!!!
 178                 assert( gc::is_locked() );
 179
 180                 back_off bkoff;
 181
 182                 for (;;) {
 183                     if ( m_pNode->next( m_pNode->height() - 1 ).load( atomics::memory_order_acquire ).bits() ) {
 184                         // Current node is marked as deleted. So, its next pointer can point to anything
 185                         // In this case we interrupt our iteration and returns end() iterator.
 186                         *this = iterator();
 187                         return;
 188                     }
 189
 190                     marked_ptr p = m_pNode->next(0).load( atomics::memory_order_relaxed );
 191                     node_type * pp = p.ptr();
 192                     if ( p.bits() ) {
 193                         // p is marked as deleted. Spin waiting for physical removal
 194                         bkoff();
 195                         continue;
 196                     }
 197                     else if ( pp && pp->next( pp->height() - 1 ).load( atomics::memory_order_relaxed ).bits() ) {
 198                         // p is marked as deleted. Spin waiting for physical removal
 199                         bkoff();
 200                         continue;
 201                     }
 202
 203                     m_pNode = pp;
 204                     break;
 205                 }
 206             }
 207
 208         public: // for internal use only!!!
 209             iterator( node_type& refHead )
 210                 : m_pNode( nullptr )
 211             {
 212                 // RCU should be locked before iterating!!!
 213                 assert( gc::is_locked() );
 214
 215                 back_off bkoff;
 216
 217                 for (;;) {
 218                     marked_ptr p = refHead.next(0).load( atomics::memory_order_relaxed );
 219                     if ( !p.ptr() ) {
 220                         // empty skip-list
 221                         break;
 222                     }
 223
 224                     node_type * pp = p.ptr();
 225                     // Logically deleted node is marked from highest level
 226                     if ( !pp->next( pp->height() - 1 ).load( atomics::memory_order_acquire ).bits() ) {
 227                         m_pNode = pp;
 228                         break;
 229                     }
 230
 231                     bkoff();
 232                 }
 233             }
 234
 235         public:
 236             iterator()
 237                 : m_pNode( nullptr )
 238             {
 239                 // RCU should be locked before iterating!!!
 240                 assert( gc::is_locked() );
 241             }
 242
 243             iterator( iterator const& s)
 244                 : m_pNode( s.m_pNode )
 245             {
 246                 // RCU should be locked before iterating!!!
 247                 assert( gc::is_locked() );
 248             }
 249
 250             value_type * operator ->() const
 251             {
 252                 assert( m_pNode != nullptr );
 253                 assert( node_traits::to_value_ptr( m_pNode ) != nullptr );
 254
 255                 return node_traits::to_value_ptr( m_pNode );
 256             }
 257
 258             value_ref operator *() const
 259             {
 260                 assert( m_pNode != nullptr );
 261                 assert( node_traits::to_value_ptr( m_pNode ) != nullptr );
 262
 263                 return *node_traits::to_value_ptr( m_pNode );
 264             }
 265
 266             /// Pre-increment
 267             iterator& operator ++()
 268             {
 269                 next();
 270                 return *this;
 271             }
 272
 273             iterator& operator = (const iterator& src)
 274             {
 275                 m_pNode = src.m_pNode;
 276                 return *this;
 277             }
 278
 279             template <typename Bkoff, bool C>
 280             bool operator ==(iterator<gc, node_traits, Bkoff, C> const& i ) const
 281             {
 282                 return m_pNode == i.m_pNode;
 283             }
 284             template <typename Bkoff, bool C>
 285             bool operator !=(iterator<gc, node_traits, Bkoff, C> const& i ) const
 286             {
 287                 return !( *this == i );
 288             }
 289         };
 290     }}  // namespace skip_list::details
 291     //@endcond
 292
 293     /// Lock-free skip-list set (template specialization for \ref cds_urcu_desc "RCU")
 294     /** @ingroup cds_intrusive_map
 295         @anchor cds_intrusive_SkipListSet_rcu
 296
 297         The implementation of well-known probabilistic data structure called skip-list
 298         invented by W.Pugh in his papers:
 299             - [1989] W.Pugh Skip Lists: A Probabilistic Alternative to Balanced Trees
 300             - [1990] W.Pugh A Skip List Cookbook
 301
 302         A skip-list is a probabilistic data structure that provides expected logarithmic
 303         time search without the need of rebalance. The skip-list is a collection of sorted
 304         linked list. Nodes are ordered by key. Each node is linked into a subset of the lists.
 305         Each list has a level, ranging from 0 to 32. The bottom-level list contains
 306         all the nodes, and each higher-level list is a sublist of the lower-level lists.
 307         Each node is created with a random top level (with a random height), and belongs
 308         to all lists up to that level. The probability that a node has the height 1 is 1/2.
 309         The probability that a node has the height N is 1/2 ** N (more precisely,
 310         the distribution depends on an random generator provided, but our generators
 311         have this property).
 312
 313         The lock-free variant of skip-list is implemented according to book
 314             - [2008] M.Herlihy, N.Shavit "The Art of Multiprocessor Programming",
 315                 chapter 14.4 "A Lock-Free Concurrent Skiplist".
 316
 317         <b>Template arguments</b>:
 318             - \p RCU - one of \ref cds_urcu_gc "RCU type"
 319             - \p T - type to be stored in the list. The type must be based on \p skip_list::node (for \p skip_list::base_hook)
 320                 or it must have a member of type \p skip_list::node (for \p skip_list::member_hook).
 321             - \p Traits - set traits, default is \p skip_list::traits
 322                 It is possible to declare option-based list with \p cds::intrusive::skip_list::make_traits metafunction
 323                 instead of \p Traits template argument.
 324
 325         @note Before including <tt><cds/intrusive/skip_list_rcu.h></tt> you should include appropriate RCU header file,
 326         see \ref cds_urcu_gc "RCU type" for list of existing RCU class and corresponding header files.
 327
 328         <b>Iterators</b>
 329
 330         The class supports a forward iterator (\ref iterator and \ref const_iterator).
 331         The iteration is ordered.
 332
 333         You may iterate over skip-list set items only under RCU lock.
 334         Only in this case the iterator is thread-safe since
 335         while RCU is locked any set's item cannot be reclaimed.
 336
 337         @note The requirement of RCU lock during iterating means that any type of modification of the skip list
 338         (i.e. inserting, erasing and so on) is not possible.
 339
 340         @warning The iterator object cannot be passed between threads.
 341
 342         Example how to use skip-list set iterators:
 343         \code
 344         // First, you should include the header for RCU type you have chosen
 345         #include <cds/urcu/general_buffered.h>
 346         #include <cds/intrusive/skip_list_rcu.h>
 347
 348         typedef cds::urcu::gc< cds::urcu::general_buffered<> > rcu_type;
 349
 350         struct Foo {
 351             // ...
 352         };
 353
 354         // Traits for your skip-list.
 355         // At least, you should define cds::opt::less or cds::opt::compare for Foo struct
 356         struct my_traits: public cds::intrusive::skip_list::traits
 357         {
 358             // ...
 359         };
 360         typedef cds::intrusive::SkipListSet< rcu_type, Foo, my_traits > my_skiplist_set;
 361
 362         my_skiplist_set theSet;
 363
 364         // ...
 365
 366         // Begin iteration
 367         {
 368             // Apply RCU locking manually
 369             typename rcu_type::scoped_lock sl;
 370
 371             for ( auto it = theList.begin(); it != theList.end(); ++it ) {
 372                 // ...
 373             }
 374
 375             // rcu_type::scoped_lock destructor releases RCU lock implicitly
 376         }
 377         \endcode
 378
 379         The iterator class supports the following minimalistic interface:
 380         \code
 381         struct iterator {
 382             // Default ctor
 383             iterator();
 384
 385             // Copy ctor
 386             iterator( iterator const& s);
 387
 388             value_type * operator ->() const;
 389             value_type& operator *() const;
 390
 391             // Pre-increment
 392             iterator& operator ++();
 393
 394             // Copy assignment
 395             iterator& operator = (const iterator& src);
 396
 397             bool operator ==(iterator const& i ) const;
 398             bool operator !=(iterator const& i ) const;
 399         };
 400         \endcode
 401         Note, the iterator object returned by \ref end, \p cend member functions points to \p nullptr and should not be dereferenced.
 402
 403         <b>How to use</b>
 404
 405         You should incorporate skip_list::node into your struct \p T and provide
 406         appropriate skip_list::traits::hook in your \p Traits template parameters. Usually, for \p Traits you
 407         define a struct based on \p skip_list::traits.
 408
 409         Example for <tt>cds::urcu::general_buffered<></tt> RCU and base hook:
 410         \code
 411         // First, you should include the header for RCU type you have chosen
 412         #include <cds/urcu/general_buffered.h>
 413
 414         // Include RCU skip-list specialization
 415         #include <cds/intrusive/skip_list_rcu.h>
 416
 417         // RCU type typedef
 418         typedef cds::urcu::gc< cds::urcu::general_buffered<> > rcu_type;
 419
 420         // Data stored in skip list
 421         struct my_data: public cds::intrusive::skip_list::node< rcu_type >
 422         {
 423             // key field
 424             std::string     strKey;
 425
 426             // other data
 427             // ...
 428         };
 429
 430         // my_data compare functor
 431         struct my_data_cmp {
 432             int operator()( const my_data& d1, const my_data& d2 )
 433             {
 434                 return d1.strKey.compare( d2.strKey );
 435             }
 436
 437             int operator()( const my_data& d, const std::string& s )
 438             {
 439                 return d.strKey.compare(s);
 440             }
 441
 442             int operator()( const std::string& s, const my_data& d )
 443             {
 444                 return s.compare( d.strKey );
 445             }
 446         };
 447
 448         // Declare traits
 449         struct my_traits: public cds::intrusive::skip_list::traits
 450         {
 451             typedef cds::intrusive::skip_list::base_hook< cds::opt::gc< rcu_type > >   hook;
 452             typedef my_data_cmp compare;
 453         };
 454
 455         // Declare skip-list set type
 456         typedef cds::intrusive::SkipListSet< rcu_type, my_data, my_traits >     traits_based_set;
 457         \endcode
 458
 459         Equivalent option-based code:
 460         \code
 461         #include <cds/urcu/general_buffered.h>
 462         #include <cds/intrusive/skip_list_rcu.h>
 463
 464         typedef cds::urcu::gc< cds::urcu::general_buffered<> > rcu_type;
 465
 466         struct my_data {
 467             // see above
 468         };
 469         struct compare {
 470             // see above
 471         };
 472
 473         // Declare option-based skip-list set
 474         typedef cds::intrusive::SkipListSet< rcu_type
 475             ,my_data
 476             , typename cds::intrusive::skip_list::make_traits<
 477                 cds::intrusive::opt::hook< cds::intrusive::skip_list::base_hook< cds::opt::gc< rcu_type > > >
 478                 ,cds::intrusive::opt::compare< my_data_cmp >
 479             >::type
 480         > option_based_set;
 481
 482         \endcode
 483     */
 484     template <
 485         class RCU
 486        ,typename T
 487 #ifdef CDS_DOXYGEN_INVOKED
 488        ,typename Traits = skip_list::traits
 489 #else
 490        ,typename Traits
 491 #endif
 492     >
 493     class SkipListSet< cds::urcu::gc< RCU >, T, Traits >
 494     {
 495     public:
 496         typedef cds::urcu::gc< RCU > gc; ///< Garbage collector
 497         typedef T       value_type;      ///< type of value stored in the skip-list
 498         typedef Traits  traits;          ///< Traits template parameter
 499
 500         typedef typename traits::hook    hook;      ///< hook type
 501         typedef typename hook::node_type node_type; ///< node type
 502
 503 #   ifdef CDS_DOXYGEN_INVOKED
 504         typedef implementation_defined key_comparator  ;    ///< key comparison functor based on \p Traits::compare and \p Traits::less
 505 #   else
 506         typedef typename opt::details::make_comparator< value_type, traits >::type key_comparator;
 507 #   endif
 508
 509         typedef typename traits::disposer  disposer;   ///< disposer
 510         typedef typename get_node_traits< value_type, node_type, hook>::type node_traits;    ///< node traits
 511
 512         typedef typename traits::item_counter  item_counter;   ///< Item counting policy used
 513         typedef typename traits::memory_model  memory_model;   ///< Memory ordering, see \p cds::opt::memory_model option
 514         typedef typename traits::random_level_generator    random_level_generator;   ///< random level generator
 515         typedef typename traits::allocator     allocator_type; ///< allocator for maintaining array of next pointers of the node
 516         typedef typename traits::back_off      back_off;       ///< Back-off strategy
 517         typedef typename traits::stat          stat;           ///< internal statistics type
 518         typedef typename traits::rcu_check_deadlock rcu_check_deadlock; ///< Deadlock checking policy
 519         typedef typename gc::scoped_lock       rcu_lock;      ///< RCU scoped lock
 520         static CDS_CONSTEXPR const bool c_bExtractLockExternal = false; ///< Group of \p extract_xxx functions does not require external locking
 521
 522
 523         /// Max node height. The actual node height should be in range <tt>[0 .. c_nMaxHeight)</tt>
 524         /**
 525             The max height is specified by \ref skip_list::random_level_generator "random level generator" constant \p m_nUpperBound
 526             but it should be no more than 32 (\ref skip_list::c_nHeightLimit).
 527         */
 528         static unsigned int const c_nMaxHeight = std::conditional<
 529             (random_level_generator::c_nUpperBound <= skip_list::c_nHeightLimit),
 530             std::integral_constant< unsigned int, random_level_generator::c_nUpperBound >,
 531             std::integral_constant< unsigned int, skip_list::c_nHeightLimit >
 532         >::type::value;
 533
 534         //@cond
 535         static unsigned int const c_nMinHeight = 5;
 536         typedef cds::intrusive::skip_list::implementation_tag implementation_tag;
 537         //@endcond
 538
 539     protected:
 540         typedef typename node_type::atomic_marked_ptr   atomic_node_ptr ;   ///< Atomic marked node pointer
 541         typedef typename node_type::marked_ptr          marked_node_ptr ;   ///< Node marked pointer
 542
 543     protected:
 544         //@cond
 545         typedef skip_list::details::intrusive_node_builder< node_type, atomic_node_ptr, allocator_type > intrusive_node_builder;
 546
 547         typedef typename std::conditional<
 548             std::is_same< typename traits::internal_node_builder, cds::opt::none >::value
 549             ,intrusive_node_builder
 550             ,typename traits::internal_node_builder
 551         >::type node_builder;
 552
 553         typedef std::unique_ptr< node_type, typename node_builder::node_disposer >    scoped_node_ptr;
 554
 555         static void dispose_node( value_type * pVal )
 556         {
 557             assert( pVal );
 558
 559             typename node_builder::node_disposer()( node_traits::to_node_ptr(pVal) );
 560             disposer()( pVal );
 561         }
 562
 563         struct node_disposer
 564         {
 565             void operator()( value_type * pVal )
 566             {
 567                 dispose_node( pVal );
 568             }
 569         };
 570
 571         static void dispose_chain( node_type * pChain )
 572         {
 573             if ( pChain ) {
 574                 assert( !gc::is_locked() );
 575
 576                 auto f = [&pChain]() -> cds::urcu::retired_ptr {
 577                     node_type * p = pChain;
 578                     if ( p ) {
 579                         pChain = p->m_pDelChain;
 580                         return cds::urcu::make_retired_ptr<node_disposer>( node_traits::to_value_ptr( p ));
 581                     }
 582                     return cds::urcu::make_retired_ptr<node_disposer>( static_cast<value_type *>(nullptr));
 583                 };
 584                 gc::batch_retire(std::ref(f));
 585             }
 586         }
 587
 588         struct position {
 589             node_type *   pPrev[ c_nMaxHeight ];
 590             node_type *   pSucc[ c_nMaxHeight ];
 591             node_type *   pNext[ c_nMaxHeight ];
 592
 593             node_type *   pCur;
 594             node_type *   pDelChain;
 595
 596             position()
 597                 : pDelChain( nullptr )
 598             {}
 599
 600             ~position()
 601             {
 602                 dispose_chain( pDelChain );
 603             }
 604
 605             void dispose( node_type * p )
 606             {
 607                 assert( p != nullptr );
 608                 assert( p->m_pDelChain == nullptr );
 609
 610                 p->m_pDelChain = pDelChain;
 611                 pDelChain = p;
 612             }
 613         };
 614
 615         typedef cds::urcu::details::check_deadlock_policy< gc, rcu_check_deadlock>   check_deadlock_policy;
 616         //@endcond
 617
 618     protected:
 619         skip_list::details::head_node< node_type > m_Head;   ///< head tower (max height)
 620
 621         item_counter                m_ItemCounter;      ///< item counter
 622         random_level_generator      m_RandomLevelGen;   ///< random level generator instance
 623         atomics::atomic<unsigned int>    m_nHeight;     ///< estimated high level
 624         atomics::atomic<node_type *>     m_pDeferredDelChain ;   ///< Deferred deleted node chain
 625         mutable stat                m_Stat;             ///< internal statistics
 626
 627     protected:
 628         //@cond
 629         unsigned int random_level()
 630         {
 631             // Random generator produces a number from range [0..31]
 632             // We need a number from range [1..32]
 633             return m_RandomLevelGen() + 1;
 634         }
 635
 636         template <typename Q>
 637         node_type * build_node( Q v )
 638         {
 639             return node_builder::make_tower( v, m_RandomLevelGen );
 640         }
 641         //@endcond
 642
 643     public:
 644         using exempt_ptr = cds::urcu::exempt_ptr< gc, value_type, value_type, node_disposer, void >; ///< pointer to extracted node
 645
 646     private:
 647         //@cond
 648         struct chain_disposer {
 649             void operator()( node_type * pChain ) const
 650             {
 651                 dispose_chain( pChain );
 652             }
 653         };
 654         typedef cds::intrusive::details::raw_ptr_disposer< gc, node_type, chain_disposer> raw_ptr_disposer;
 655         //@endcond
 656
 657     public:
 658         /// Result of \p get(), \p get_with() functions - pointer to the node found
 659         typedef cds::urcu::raw_ptr< gc, value_type, raw_ptr_disposer > raw_ptr;
 660
 661     protected:
 662         //@cond
 663
 664         bool is_extracted( marked_node_ptr const p ) const
 665         {
 666             return (p.bits() & 2) != 0;
 667         }
 668
 669         template <typename Q, typename Compare >
 670         bool find_position( Q const& val, position& pos, Compare cmp, bool bStopIfFound )
 671         {
 672             assert( gc::is_locked() );
 673
 674             node_type * pPred;
 675             marked_node_ptr pSucc;
 676             marked_node_ptr pCur;
 677             int nCmp = 1;
 678
 679         retry:
 680             pPred = m_Head.head();
 681
 682             for ( int nLevel = static_cast<int>(c_nMaxHeight - 1); nLevel >= 0; --nLevel ) {
 683
 684                 while ( true ) {
 685                     pCur = pPred->next( nLevel ).load( memory_model::memory_order_relaxed );
 686                     if ( pCur.bits() ) {
 687                         // pCur.bits() means that pPred is logically deleted
 688                         goto retry;
 689                     }
 690
 691                     if ( pCur.ptr() == nullptr ) {
 692                         // end of the list at level nLevel - goto next level
 693                         break;
 694                     }
 695
 696                     // pSucc contains deletion mark for pCur
 697                     pSucc = pCur->next( nLevel ).load( memory_model::memory_order_relaxed );
 698
 699                     if ( pPred->next( nLevel ).load( memory_model::memory_order_relaxed ).all() != pCur.ptr() )
 700                         goto retry;
 701
 702                     if ( pSucc.bits() ) {
 703                         // pCur is marked, i.e. logically deleted.
 704                         marked_node_ptr p( pCur.ptr() );
 705                         if ( pPred->next( nLevel ).compare_exchange_strong( p, marked_node_ptr( pSucc.ptr() ),
 706                              memory_model::memory_order_release, atomics::memory_order_relaxed ))
 707                         {
 708                             if ( nLevel == 0 ) {
 709 #                       ifdef _DEBUG
 710                                 pCur->m_bUnlinked = true;
 711 #                       endif
 712
 713                                 if ( !is_extracted( pSucc )) {
 714                                     // We cannot free the node at this moment since RCU is locked
 715                                     // Link deleted nodes to a chain to free later
 716                                     pos.dispose( pCur.ptr() );
 717                                     m_Stat.onEraseWhileFind();
 718                                 }
 719                                 else {
 720                                     m_Stat.onExtractWhileFind();
 721                                 }
 722                             }
 723                         }
 724                         goto retry;
 725                     }
 726                     else {
 727                         nCmp = cmp( *node_traits::to_value_ptr( pCur.ptr()), val );
 728                         if ( nCmp < 0 )
 729                             pPred = pCur.ptr();
 730                         else if ( nCmp == 0 && bStopIfFound )
 731                             goto found;
 732                         else
 733                             break;
 734                     }
 735                 }
 736
 737                 // Next level
 738                 pos.pPrev[ nLevel ] = pPred;
 739                 pos.pSucc[ nLevel ] = pCur.ptr();
 740             }
 741
 742             if ( nCmp != 0 )
 743                 return false;
 744
 745         found:
 746             pos.pCur = pCur.ptr();
 747             return pCur.ptr() && nCmp == 0;
 748         }
 749
 750         bool find_min_position( position& pos )
 751         {
 752             assert( gc::is_locked() );
 753
 754             node_type * pPred;
 755             marked_node_ptr pSucc;
 756             marked_node_ptr pCur;
 757
 758         retry:
 759             pPred = m_Head.head();
 760
 761             for ( int nLevel = static_cast<int>(c_nMaxHeight - 1); nLevel >= 0; --nLevel ) {
 762
 763                 pCur = pPred->next( nLevel ).load( memory_model::memory_order_relaxed );
 764                 // pCur.bits() means that pPred is logically deleted
 765                 // head cannot be deleted
 766                 assert( pCur.bits() == 0 );
 767
 768                 if ( pCur.ptr() ) {
 769
 770                     // pSucc contains deletion mark for pCur
 771                     pSucc = pCur->next( nLevel ).load( memory_model::memory_order_relaxed );
 772
 773                     if ( pPred->next( nLevel ).load( memory_model::memory_order_relaxed ).all() != pCur.ptr() )
 774                         goto retry;
 775
 776                     if ( pSucc.bits() ) {
 777                         // pCur is marked, i.e. logically deleted.
 778                         marked_node_ptr p( pCur.ptr() );
 779                         if ( pPred->next( nLevel ).compare_exchange_strong( p, marked_node_ptr( pSucc.ptr() ),
 780                             memory_model::memory_order_release, atomics::memory_order_relaxed ))
 781                         {
 782                             if ( nLevel == 0 ) {
 783 #                       ifdef _DEBUG
 784                                 pCur->m_bUnlinked = true;
 785 #                       endif
 786
 787                                 if ( !is_extracted( pSucc )) {
 788                                     // We cannot free the node at this moment since RCU is locked
 789                                     // Link deleted nodes to a chain to free later
 790                                     pos.dispose( pCur.ptr() );
 791                                     m_Stat.onEraseWhileFind();
 792                                 }
 793                                 else {
 794                                     m_Stat.onExtractWhileFind();
 795                                 }
 796                             }
 797                         }
 798                         goto retry;
 799                     }
 800                 }
 801
 802                 // Next level
 803                 pos.pPrev[ nLevel ] = pPred;
 804                 pos.pSucc[ nLevel ] = pCur.ptr();
 805             }
 806             return (pos.pCur = pCur.ptr()) != nullptr;
 807         }
 808
 809         bool find_max_position( position& pos )
 810         {
 811             assert( gc::is_locked() );
 812
 813             node_type * pPred;
 814             marked_node_ptr pSucc;
 815             marked_node_ptr pCur;
 816
 817         retry:
 818             pPred = m_Head.head();
 819
 820             for ( int nLevel = static_cast<int>(c_nMaxHeight - 1); nLevel >= 0; --nLevel ) {
 821
 822                 while ( true ) {
 823                     pCur = pPred->next( nLevel ).load( memory_model::memory_order_relaxed );
 824                     if ( pCur.bits() ) {
 825                         // pCur.bits() means that pPred is logically deleted
 826                         goto retry;
 827                     }
 828
 829                     if ( pCur.ptr() == nullptr ) {
 830                         // end of the list at level nLevel - goto next level
 831                         break;
 832                     }
 833
 834                     // pSucc contains deletion mark for pCur
 835                     pSucc = pCur->next( nLevel ).load( memory_model::memory_order_relaxed );
 836
 837                     if ( pPred->next( nLevel ).load( memory_model::memory_order_relaxed ).all() != pCur.ptr() )
 838                         goto retry;
 839
 840                     if ( pSucc.bits() ) {
 841                         // pCur is marked, i.e. logically deleted.
 842                         marked_node_ptr p( pCur.ptr() );
 843                         if ( pPred->next( nLevel ).compare_exchange_strong( p, marked_node_ptr( pSucc.ptr() ),
 844                             memory_model::memory_order_release, atomics::memory_order_relaxed ))
 845                         {
 846                             if ( nLevel == 0 ) {
 847 #                       ifdef _DEBUG
 848                                 pCur->m_bUnlinked = true;
 849 #                       endif
 850
 851                                 if ( !is_extracted( pSucc )) {
 852                                     // We cannot free the node at this moment since RCU is locked
 853                                     // Link deleted nodes to a chain to free later
 854                                     pos.dispose( pCur.ptr() );
 855                                     m_Stat.onEraseWhileFind();
 856                                 }
 857                                 else {
 858                                     m_Stat.onExtractWhileFind();
 859                                 }
 860                             }
 861                         }
 862                         goto retry;
 863                     }
 864                     else {
 865                         if ( !pSucc.ptr() )
 866                             break;
 867
 868                         pPred = pCur.ptr();
 869                     }
 870                 }
 871
 872                 // Next level
 873                 pos.pPrev[ nLevel ] = pPred;
 874                 pos.pSucc[ nLevel ] = pCur.ptr();
 875             }
 876
 877             return (pos.pCur = pCur.ptr()) != nullptr;
 878         }
 879
 880         template <typename Func>
 881         bool insert_at_position( value_type& val, node_type * pNode, position& pos, Func f )
 882         {
 883             assert( gc::is_locked() );
 884
 885             unsigned int nHeight = pNode->height();
 886             pNode->clear_tower();
 887
 888             {
 889                 marked_node_ptr p( pos.pSucc[0] );
 890                 pNode->next( 0 ).store( p, memory_model::memory_order_release );
 891                 if ( !pos.pPrev[0]->next(0).compare_exchange_strong( p, marked_node_ptr(pNode), memory_model::memory_order_release, atomics::memory_order_relaxed )) {
 892                     return false;
 893                 }
 894 #       ifdef _DEBUG
 895                 pNode->m_bLinked = true;
 896 #       endif
 897                 f( val );
 898             }
 899
 900             for ( unsigned int nLevel = 1; nLevel < nHeight; ++nLevel ) {
 901                 marked_node_ptr p;
 902                 while ( true ) {
 903                     marked_node_ptr q( pos.pSucc[ nLevel ]);
 904                     if ( !pNode->next( nLevel ).compare_exchange_strong( p, q, memory_model::memory_order_acquire, atomics::memory_order_relaxed )) {
 905                         // pNode has been marked as removed while we are inserting it
 906                         // Stop inserting
 907                         assert( p.bits() );
 908                         m_Stat.onLogicDeleteWhileInsert();
 909                         return true;
 910                     }
 911                     p = q;
 912                     if ( pos.pPrev[nLevel]->next(nLevel).compare_exchange_strong( q, marked_node_ptr( pNode ), memory_model::memory_order_release, atomics::memory_order_relaxed ) )
 913                         break;
 914
 915                     // Renew insert position
 916                     m_Stat.onRenewInsertPosition();
 917                     if ( !find_position( val, pos, key_comparator(), false )) {
 918                         // The node has been deleted while we are inserting it
 919                         m_Stat.onNotFoundWhileInsert();
 920                         return true;
 921                     }
 922                 }
 923             }
 924             return true;
 925         }
 926
 927         template <typename Func>
 928         bool try_remove_at( node_type * pDel, position& pos, Func f, bool bExtract )
 929         {
 930             assert( pDel != nullptr );
 931             assert( gc::is_locked() );
 932
 933             marked_node_ptr pSucc;
 934
 935             // logical deletion (marking)
 936             for ( unsigned int nLevel = pDel->height() - 1; nLevel > 0; --nLevel ) {
 937                 pSucc = pDel->next(nLevel).load( memory_model::memory_order_relaxed );
 938                 while ( true ) {
 939                     if ( pSucc.bits()
 940                       || pDel->next(nLevel).compare_exchange_weak( pSucc, pSucc | 1, memory_model::memory_order_acquire, atomics::memory_order_relaxed ))
 941                     {
 942                         break;
 943                     }
 944                 }
 945             }
 946
 947             pSucc = pDel->next(0).load( memory_model::memory_order_relaxed );
 948             while ( true ) {
 949                 if ( pSucc.bits() )
 950                     return false;
 951
 952                 int const nMask = bExtract ? 3 : 1;
 953                 if ( pDel->next(0).compare_exchange_strong( pSucc, pSucc | nMask, memory_model::memory_order_acquire, atomics::memory_order_relaxed ))
 954                 {
 955                     f( *node_traits::to_value_ptr( pDel ));
 956
 957                     // physical deletion
 958                     // try fast erase
 959                     pSucc = pDel;
 960                     for ( int nLevel = static_cast<int>( pDel->height() - 1 ); nLevel >= 0; --nLevel ) {
 961                         if ( !pos.pPrev[nLevel]->next(nLevel).compare_exchange_strong( pSucc,
 962                             marked_node_ptr( pDel->next(nLevel).load(memory_model::memory_order_relaxed).ptr() ),
 963                             memory_model::memory_order_release, atomics::memory_order_relaxed) )
 964                         {
 965                             // Do slow erase
 966                             find_position( *node_traits::to_value_ptr(pDel), pos, key_comparator(), false );
 967                             if ( bExtract )
 968                                 m_Stat.onSlowExtract();
 969                             else
 970                                 m_Stat.onSlowErase();
 971 #                        ifdef _DEBUG
 972                             assert( pDel->m_bUnlinked );
 973 #                        endif
 974                             return true;
 975                         }
 976                     }
 977
 978 #               ifdef _DEBUG
 979                     pDel->m_bUnlinked = true;
 980 #               endif
 981                     if ( !bExtract ) {
 982                         // We cannot free the node at this moment since RCU is locked
 983                         // Link deleted nodes to a chain to free later
 984                         pos.dispose( pDel );
 985                         m_Stat.onFastErase();
 986                     }
 987                     else
 988                         m_Stat.onFastExtract();
 989
 990                     return true;
 991                 }
 992             }
 993         }
 994
 995         enum finsd_fastpath_result {
 996             find_fastpath_found,
 997             find_fastpath_not_found,
 998             find_fastpath_abort
 999         };
1000         template <typename Q, typename Compare, typename Func>
1001         finsd_fastpath_result find_fastpath( Q& val, Compare cmp, Func f ) const
1002         {
1003             node_type * pPred;
1004             marked_node_ptr pCur;
1005             marked_node_ptr pSucc;
1006             marked_node_ptr pNull;
1007
1008             back_off bkoff;
1009
1010             pPred = m_Head.head();
1011             for ( int nLevel = static_cast<int>(m_nHeight.load(memory_model::memory_order_relaxed) - 1); nLevel >= 0; --nLevel ) {
1012                 pCur = pPred->next(nLevel).load( memory_model::memory_order_acquire );
1013                 if ( pCur == pNull )
1014                     continue;
1015
1016                 while ( pCur != pNull ) {
1017                     if ( pCur.bits() ) {
1018                         // Wait until pCur is removed
1019                         unsigned int nAttempt = 0;
1020                         while ( pCur.bits() && nAttempt++ < 16 ) {
1021                             bkoff();
1022                             pCur = pPred->next(nLevel).load( memory_model::memory_order_acquire );
1023                         }
1024                         bkoff.reset();
1025
1026                         if ( pCur.bits() ) {
1027                             // Maybe, we are on deleted node sequence
1028                             // Abort searching, try slow-path
1029                             return find_fastpath_abort;
1030                         }
1031                     }
1032
1033                     if ( pCur.ptr() ) {
1034                         int nCmp = cmp( *node_traits::to_value_ptr( pCur.ptr() ), val );
1035                         if ( nCmp < 0 ) {
1036                             pPred = pCur.ptr();
1037                             pCur = pCur->next(nLevel).load( memory_model::memory_order_acquire );
1038                         }
1039                         else if ( nCmp == 0 ) {
1040                             // found
1041                             f( *node_traits::to_value_ptr( pCur.ptr() ), val );
1042                             return find_fastpath_found;
1043                         }
1044                         else // pCur > val - go down
1045                             break;
1046                     }
1047                 }
1048             }
1049
1050             return find_fastpath_not_found;
1051         }
1052
1053         template <typename Q, typename Compare, typename Func>
1054         bool find_slowpath( Q& val, Compare cmp, Func f, position& pos )
1055         {
1056             if ( find_position( val, pos, cmp, true )) {
1057                 assert( cmp( *node_traits::to_value_ptr( pos.pCur ), val ) == 0 );
1058
1059                 f( *node_traits::to_value_ptr( pos.pCur ), val );
1060                 return true;
1061             }
1062             else
1063                 return false;
1064         }
1065
1066         template <typename Q, typename Compare, typename Func>
1067         bool do_find_with( Q& val, Compare cmp, Func f )
1068         {
1069             position pos;
1070             return do_find_with( val, cmp, f, pos );
1071         }
1072
1073         template <typename Q, typename Compare, typename Func>
1074         bool do_find_with( Q& val, Compare cmp, Func f, position& pos )
1075         {
1076             bool bRet;
1077
1078             {
1079                 rcu_lock l;
1080
1081                 switch ( find_fastpath( val, cmp, f )) {
1082                 case find_fastpath_found:
1083                     m_Stat.onFindFastSuccess();
1084                     return true;
1085                 case find_fastpath_not_found:
1086                     m_Stat.onFindFastFailed();
1087                     return false;
1088                 default:
1089                     break;
1090                 }
1091
1092                 if ( find_slowpath( val, cmp, f, pos )) {
1093                     m_Stat.onFindSlowSuccess();
1094                     bRet = true;
1095                 }
1096                 else {
1097                     m_Stat.onFindSlowFailed();
1098                     bRet = false;
1099                 }
1100             }
1101             return bRet;
1102         }
1103
1104         template <typename Q, typename Compare, typename Func>
1105         bool do_erase( Q const& val, Compare cmp, Func f )
1106         {
1107             check_deadlock_policy::check();
1108
1109             position pos;
1110             bool bRet;
1111
1112             {
1113                 rcu_lock rcuLock;
1114
1115                 if ( !find_position( val, pos, cmp, false ) ) {
1116                     m_Stat.onEraseFailed();
1117                     bRet = false;
1118                 }
1119                 else {
1120                     node_type * pDel = pos.pCur;
1121                     assert( cmp( *node_traits::to_value_ptr( pDel ), val ) == 0 );
1122
1123                     unsigned int nHeight = pDel->height();
1124                     if ( try_remove_at( pDel, pos, f, false )) {
1125                         --m_ItemCounter;
1126                         m_Stat.onRemoveNode( nHeight );
1127                         m_Stat.onEraseSuccess();
1128                         bRet = true;
1129                     }
1130                     else {
1131                         m_Stat.onEraseFailed();
1132                         bRet = false;
1133                     }
1134                 }
1135             }
1136
1137             return bRet;
1138         }
1139
1140         template <typename Q, typename Compare>
1141         value_type * do_extract_key( Q const& key, Compare cmp, position& pos )
1142         {
1143             // RCU should be locked!!!
1144             assert( gc::is_locked() );
1145
1146             node_type * pDel;
1147
1148             if ( !find_position( key, pos, cmp, false ) ) {
1149                 m_Stat.onExtractFailed();
1150                 pDel = nullptr;
1151             }
1152             else {
1153                 pDel = pos.pCur;
1154                 assert( cmp( *node_traits::to_value_ptr( pDel ), key ) == 0 );
1155
1156                 unsigned int const nHeight = pDel->height();
1157
1158                 if ( try_remove_at( pDel, pos, [](value_type const&) {}, true )) {
1159                     --m_ItemCounter;
1160                     m_Stat.onRemoveNode( nHeight );
1161                     m_Stat.onExtractSuccess();
1162                 }
1163                 else {
1164                     m_Stat.onExtractFailed();
1165                     pDel = nullptr;
1166                 }
1167             }
1168
1169             return pDel ? node_traits::to_value_ptr( pDel ) : nullptr;
1170         }
1171
1172         template <typename Q>
1173         value_type * do_extract( Q const& key )
1174         {
1175             check_deadlock_policy::check();
1176             value_type * pDel = nullptr;
1177             position pos;
1178             {
1179                 rcu_lock l;
1180                 pDel = do_extract_key( key, key_comparator(), pos );
1181             }
1182
1183             return pDel;
1184         }
1185
1186         template <typename Q, typename Less>
1187         value_type * do_extract_with( Q const& key, Less pred )
1188         {
1189             CDS_UNUSED(pred);
1190             check_deadlock_policy::check();
1191             value_type * pDel = nullptr;
1192             position pos;
1193             {
1194                 rcu_lock l;
1195                 pDel = do_extract_key( key, cds::opt::details::make_comparator_from_less<Less>(), pos );
1196             }
1197
1198             return pDel;
1199         }
1200
1201         value_type * do_extract_min()
1202         {
1203             assert( !gc::is_locked() );
1204
1205             position pos;
1206             node_type * pDel;
1207
1208             {
1209                 rcu_lock l;
1210
1211                 if ( !find_min_position( pos ) ) {
1212                     m_Stat.onExtractMinFailed();
1213                     pDel = nullptr;
1214                 }
1215                 else {
1216                     pDel = pos.pCur;
1217                     unsigned int const nHeight = pDel->height();
1218
1219                     if ( try_remove_at( pDel, pos, []( value_type const& ) {}, true ) ) {
1220                         --m_ItemCounter;
1221                         m_Stat.onRemoveNode( nHeight );
1222                         m_Stat.onExtractMinSuccess();
1223                     }
1224                     else {
1225                         m_Stat.onExtractMinFailed();
1226                         pDel = nullptr;
1227                     }
1228                 }
1229             }
1230
1231             return pDel ? node_traits::to_value_ptr( pDel ) : nullptr;
1232         }
1233
1234         value_type * do_extract_max()
1235         {
1236             assert( !gc::is_locked() );
1237
1238             position pos;
1239             node_type * pDel;
1240
1241             {
1242                 rcu_lock l;
1243
1244                 if ( !find_max_position( pos ) ) {
1245                     m_Stat.onExtractMaxFailed();
1246                     pDel = nullptr;
1247                 }
1248                 else {
1249                     pDel = pos.pCur;
1250                     unsigned int const nHeight = pDel->height();
1251
1252                     if ( try_remove_at( pDel, pos, []( value_type const& ) {}, true ) ) {
1253                         --m_ItemCounter;
1254                         m_Stat.onRemoveNode( nHeight );
1255                         m_Stat.onExtractMaxSuccess();
1256                     }
1257                     else {
1258                         m_Stat.onExtractMaxFailed();
1259                         pDel = nullptr;
1260                     }
1261                 }
1262             }
1263
1264             return pDel ? node_traits::to_value_ptr( pDel ) : nullptr;
1265         }
1266
1267         void increase_height( unsigned int nHeight )
1268         {
1269             unsigned int nCur = m_nHeight.load( memory_model::memory_order_relaxed );
1270             if ( nCur < nHeight )
1271                 m_nHeight.compare_exchange_strong( nCur, nHeight, memory_model::memory_order_release, atomics::memory_order_relaxed );
1272         }
1273         //@endcond
1274
1275     public:
1276         /// Default constructor
1277         SkipListSet()
1278             : m_Head( c_nMaxHeight )
1279             , m_nHeight( c_nMinHeight )
1280             , m_pDeferredDelChain( nullptr )
1281         {
1282             static_assert( (std::is_same< gc, typename node_type::gc >::value), "GC and node_type::gc must be the same type" );
1283
1284             // Barrier for head node
1285             atomics::atomic_thread_fence( memory_model::memory_order_release );
1286         }
1287
1288         /// Clears and destructs the skip-list
1289         ~SkipListSet()
1290         {
1291             clear();
1292         }
1293
1294     public:
1295         /// Iterator type
1296         typedef skip_list::details::iterator< gc, node_traits, back_off, false >  iterator;
1297
1298         /// Const iterator type
1299         typedef skip_list::details::iterator< gc, node_traits, back_off, true >   const_iterator;
1300
1301         /// Returns a forward iterator addressing the first element in a set
1302         iterator begin()
1303         {
1304             return iterator( *m_Head.head() );
1305         }
1306
1307         /// Returns a forward const iterator addressing the first element in a set
1308         const_iterator begin() const
1309         {
1310             return const_iterator( *m_Head.head() );
1311         }
1312
1313         /// Returns a forward const iterator addressing the first element in a set
1314         const_iterator cbegin() const
1315         {
1316             return const_iterator( *m_Head.head() );
1317         }
1318
1319         /// Returns a forward iterator that addresses the location succeeding the last element in a set.
1320         iterator end()
1321         {
1322             return iterator();
1323         }
1324
1325         /// Returns a forward const iterator that addresses the location succeeding the last element in a set.
1326         const_iterator end() const
1327         {
1328             return const_iterator();
1329         }
1330
1331         /// Returns a forward const iterator that addresses the location succeeding the last element in a set.
1332         const_iterator cend() const
1333         {
1334             return const_iterator();
1335         }
1336
1337     public:
1338         /// Inserts new node
1339         /**
1340             The function inserts \p val in the set if it does not contain
1341             an item with key equal to \p val.
1342
1343             The function applies RCU lock internally.
1344
1345             Returns \p true if \p val is placed into the set, \p false otherwise.
1346         */
1347         bool insert( value_type& val )
1348         {
1349             return insert( val, []( value_type& ) {} );
1350         }
1351
1352         /// Inserts new node
1353         /**
1354             This function is intended for derived non-intrusive containers.
1355
1356             The function allows to split creating of new item into two part:
1357             - create item with key only
1358             - insert new item into the set
1359             - if inserting is success, calls  \p f functor to initialize value-field of \p val.
1360
1361             The functor signature is:
1362             \code
1363                 void func( value_type& val );
1364             \endcode
1365             where \p val is the item inserted. User-defined functor \p f should guarantee that during changing
1366             \p val no any other changes could be made on this set's item by concurrent threads.
1367             The user-defined functor is called only if the inserting is success.
1368
1369             RCU \p synchronize method can be called. RCU should not be locked.
1370         */
1371         template <typename Func>
1372         bool insert( value_type& val, Func f )
1373         {
1374             check_deadlock_policy::check();
1375
1376             position pos;
1377             bool bRet;
1378
1379             {
1380                 node_type * pNode = node_traits::to_node_ptr( val );
1381                 scoped_node_ptr scp( pNode );
1382                 unsigned int nHeight = pNode->height();
1383                 bool bTowerOk = nHeight > 1 && pNode->get_tower() != nullptr;
1384                 bool bTowerMade = false;
1385
1386                 rcu_lock rcuLock;
1387
1388                 while ( true )
1389                 {
1390                     bool bFound = find_position( val, pos, key_comparator(), true );
1391                     if ( bFound ) {
1392                         // scoped_node_ptr deletes the node tower if we create it
1393                         if ( !bTowerMade )
1394                             scp.release();
1395
1396                         m_Stat.onInsertFailed();
1397                         bRet = false;
1398                         break;
1399                     }
1400
1401                     if ( !bTowerOk ) {
1402                         build_node( pNode );
1403                         nHeight = pNode->height();
1404                         bTowerMade =
1405                             bTowerOk = true;
1406                     }
1407
1408                     if ( !insert_at_position( val, pNode, pos, f )) {
1409                         m_Stat.onInsertRetry();
1410                         continue;
1411                     }
1412
1413                     increase_height( nHeight );
1414                     ++m_ItemCounter;
1415                     m_Stat.onAddNode( nHeight );
1416                     m_Stat.onInsertSuccess();
1417                     scp.release();
1418                     bRet =  true;
1419                     break;
1420                 }
1421             }
1422
1423             return bRet;
1424         }
1425
1426         /// Ensures that the \p val exists in the set
1427         /**
1428             The operation performs inserting or changing data with lock-free manner.
1429
1430             If the item \p val is not found in the set, then \p val is inserted into the set.
1431             Otherwise, the functor \p func is called with item found.
1432             The functor signature is:
1433             \code
1434                 void func( bool bNew, value_type& item, value_type& val );
1435             \endcode
1436             with arguments:
1437             - \p bNew - \p true if the item has been inserted, \p false otherwise
1438             - \p item - item of the set
1439             - \p val - argument \p val passed into the \p %ensure() function
1440             If new item has been inserted (i.e. \p bNew is \p true) then \p item and \p val arguments
1441             refer to the same thing.
1442
1443             The functor can change non-key fields of the \p item; however, \p func must guarantee
1444             that during changing no any other modifications could be made on this item by concurrent threads.
1445
1446             RCU \p synchronize method can be called. RCU should not be locked.
1447
1448             Returns std::pair<bool, bool> where \p first is \p true if operation is successfull,
1449             \p second is \p true if new item has been added or \p false if the item with \p key
1450             already is in the set.
1451
1452             @warning See \ref cds_intrusive_item_creating "insert item troubleshooting"
1453         */
1454         template <typename Func>
1455         std::pair<bool, bool> ensure( value_type& val, Func func )
1456         {
1457             check_deadlock_policy::check();
1458
1459             position pos;
1460             std::pair<bool, bool> bRet( true, false );
1461
1462             {
1463                 node_type * pNode = node_traits::to_node_ptr( val );
1464                 scoped_node_ptr scp( pNode );
1465                 unsigned int nHeight = pNode->height();
1466                 bool bTowerOk = nHeight > 1 && pNode->get_tower() != nullptr;
1467                 bool bTowerMade = false;
1468
1469                 rcu_lock rcuLock;
1470                 while ( true )
1471                 {
1472                     bool bFound = find_position( val, pos, key_comparator(), true );
1473                     if ( bFound ) {
1474                         // scoped_node_ptr deletes the node tower if we create it before
1475                         if ( !bTowerMade )
1476                             scp.release();
1477
1478                         func( false, *node_traits::to_value_ptr(pos.pCur), val );
1479                         m_Stat.onEnsureExist();
1480                         break;
1481                     }
1482
1483                     if ( !bTowerOk ) {
1484                         build_node( pNode );
1485                         nHeight = pNode->height();
1486                         bTowerMade =
1487                             bTowerOk = true;
1488                     }
1489
1490                     if ( !insert_at_position( val, pNode, pos, [&func]( value_type& item ) { func( true, item, item ); })) {
1491                         m_Stat.onInsertRetry();
1492                         continue;
1493                     }
1494
1495                     increase_height( nHeight );
1496                     ++m_ItemCounter;
1497                     scp.release();
1498                     m_Stat.onAddNode( nHeight );
1499                     m_Stat.onEnsureNew();
1500                     bRet.second = true;
1501                     break;
1502                 }
1503             }
1504
1505             return bRet;
1506         }
1507
1508         /// Unlinks the item \p val from the set
1509         /**
1510             The function searches the item \p val in the set and unlink it from the set
1511             if it is found and is equal to \p val.
1512
1513             Difference between \p erase() and \p %unlink() functions: \p erase() finds <i>a key</i>
1514             and deletes the item found. \p %unlink() searches an item by key and deletes it
1515             only if \p val is an item of that set, i.e. the pointer to item found
1516             is equal to <tt> &val </tt>.
1517
1518             RCU \p synchronize method can be called. RCU should not be locked.
1519
1520             The \ref disposer specified in \p Traits class template parameter is called
1521             by garbage collector \p GC asynchronously.
1522
1523             The function returns \p true if success and \p false otherwise.
1524         */
1525         bool unlink( value_type& val )
1526         {
1527             check_deadlock_policy::check();
1528
1529             position pos;
1530             bool bRet;
1531
1532             {
1533                 rcu_lock l;
1534
1535                 if ( !find_position( val, pos, key_comparator(), false ) ) {
1536                     m_Stat.onUnlinkFailed();
1537                     bRet = false;
1538                 }
1539                 else {
1540                     node_type * pDel = pos.pCur;
1541                     assert( key_comparator()( *node_traits::to_value_ptr( pDel ), val ) == 0 );
1542
1543                     unsigned int nHeight = pDel->height();
1544
1545                     if ( node_traits::to_value_ptr( pDel ) == &val && try_remove_at( pDel, pos, [](value_type const&) {}, false )) {
1546                         --m_ItemCounter;
1547                         m_Stat.onRemoveNode( nHeight );
1548                         m_Stat.onUnlinkSuccess();
1549                         bRet = true;
1550                     }
1551                     else {
1552                         m_Stat.onUnlinkFailed();
1553                         bRet = false;
1554                     }
1555                 }
1556             }
1557
1558             return bRet;
1559         }
1560
1561         /// Extracts the item from the set with specified \p key
1562         /** \anchor cds_intrusive_SkipListSet_rcu_extract
1563             The function searches an item with key equal to \p key in the set,
1564             unlinks it from the set, and returns \ref cds::urcu::exempt_ptr "exempt_ptr" pointer to the item found.
1565             If the item with key equal to \p key is not found the function returns an empty \p exempt_ptr.
1566
1567             Note the compare functor should accept a parameter of type \p Q that can be not the same as \p value_type.
1568
1569             RCU \p synchronize method can be called. RCU should NOT be locked.
1570             The function does not call the disposer for the item found.
1571             The disposer will be implicitly invoked when the returned object is destroyed or when
1572             its \p release() member function is called.
1573             Example:
1574             \code
1575             typedef cds::intrusive::SkipListSet< cds::urcu::gc< cds::urcu::general_buffered<> >, foo, my_traits > skip_list;
1576             skip_list theList;
1577             // ...
1578
1579             typename skip_list::exempt_ptr ep( theList.extract( 5 ));
1580             if ( ep ) {
1581                 // Deal with ep
1582                 //...
1583
1584                 // Dispose returned item.
1585                 ep.release();
1586             }
1587             \endcode
1588         */
1589         template <typename Q>
1590         exempt_ptr extract( Q const& key )
1591         {
1592             return exempt_ptr( do_extract( key ));
1593         }
1594
1595         /// Extracts the item from the set with comparing functor \p pred
1596         /**
1597             The function is an analog of \p extract(Q const&) but \p pred predicate is used for key comparing.
1598             \p Less has the interface like \p std::less.
1599             \p pred must imply the same element order as the comparator used for building the set.
1600         */
1601         template <typename Q, typename Less>
1602         exempt_ptr extract_with( Q const& key, Less pred )
1603         {
1604             return exempt_ptr( do_extract_with( key, pred ));
1605         }
1606
1607         /// Extracts an item with minimal key from the list
1608         /**
1609             The function searches an item with minimal key, unlinks it, and returns \ref cds::urcu::exempt_ptr "exempt_ptr" pointer to the item.
1610             If the skip-list is empty the function returns an empty \p exempt_ptr.
1611
1612             RCU \p synchronize method can be called. RCU should NOT be locked.
1613             The function does not call the disposer for the item found.
1614             The disposer will be implicitly invoked when the returned object is destroyed or when
1615             its \p release() member function is manually called.
1616             Example:
1617             \code
1618             typedef cds::intrusive::SkipListSet< cds::urcu::gc< cds::urcu::general_buffered<> >, foo, my_traits > skip_list;
1619             skip_list theList;
1620             // ...
1621
1622             typename skip_list::exempt_ptr ep(theList.extract_min());
1623             if ( ep ) {
1624                 // Deal with ep
1625                 //...
1626
1627                 // Dispose returned item.
1628                 ep.release();
1629             }
1630             \endcode
1631
1632             @note Due the concurrent nature of the list, the function extracts <i>nearly</i> minimum key.
1633             It means that the function gets leftmost item and tries to unlink it.
1634             During unlinking, a concurrent thread may insert an item with key less than leftmost item's key.
1635             So, the function returns the item with minimum key at the moment of list traversing.
1636         */
1637         exempt_ptr extract_min()
1638         {
1639             return exempt_ptr( do_extract_min());
1640         }
1641
1642         /// Extracts an item with maximal key from the list
1643         /**
1644             The function searches an item with maximal key, unlinks it, and returns \ref cds::urcu::exempt_ptr "exempt_ptr" pointer to the item.
1645             If the skip-list is empty the function returns an empty \p exempt_ptr.
1646
1647             RCU \p synchronize method can be called. RCU should NOT be locked.
1648             The function does not call the disposer for the item found.
1649             The disposer will be implicitly invoked when the returned object is destroyed or when
1650             its \p release() member function is manually called.
1651             Example:
1652             \code
1653             typedef cds::intrusive::SkipListSet< cds::urcu::gc< cds::urcu::general_buffered<> >, foo, my_traits > skip_list;
1654             skip_list theList;
1655             // ...
1656
1657             typename skip_list::exempt_ptr ep( theList.extract_max() );
1658             if ( ep ) {
1659                 // Deal with ep
1660                 //...
1661                 // Dispose returned item.
1662                 ep.release();
1663             }
1664             \endcode
1665
1666             @note Due the concurrent nature of the list, the function extracts <i>nearly</i> maximal key.
1667             It means that the function gets rightmost item and tries to unlink it.
1668             During unlinking, a concurrent thread can insert an item with key greater than rightmost item's key.
1669             So, the function returns the item with maximum key at the moment of list traversing.
1670         */
1671         exempt_ptr extract_max()
1672         {
1673             return exempt_ptr( do_extract_max());
1674         }
1675
1676         /// Deletes the item from the set
1677         /** \anchor cds_intrusive_SkipListSet_rcu_erase
1678             The function searches an item with key equal to \p key in the set,
1679             unlinks it from the set, and returns \p true.
1680             If the item with key equal to \p key is not found the function return \p false.
1681
1682             Note the hash functor should accept a parameter of type \p Q that can be not the same as \p value_type.
1683
1684             RCU \p synchronize method can be called. RCU should not be locked.
1685         */
1686         template <typename Q>
1687         bool erase( const Q& key )
1688         {
1689             return do_erase( key, key_comparator(), [](value_type const&) {} );
1690         }
1691
1692         /// Delete the item from the set with comparing functor \p pred
1693         /**
1694             The function is an analog of \ref cds_intrusive_SkipListSet_rcu_erase "erase(Q const&)"
1695             but \p pred predicate is used for key comparing.
1696             \p Less has the interface like \p std::less.
1697             \p pred must imply the same element order as the comparator used for building the set.
1698         */
1699         template <typename Q, typename Less>
1700         bool erase_with( const Q& key, Less pred )
1701         {
1702             CDS_UNUSED( pred );
1703             return do_erase( key, cds::opt::details::make_comparator_from_less<Less>(), [](value_type const&) {} );
1704         }
1705
1706         /// Deletes the item from the set
1707         /** \anchor cds_intrusive_SkipListSet_rcu_erase_func
1708             The function searches an item with key equal to \p key in the set,
1709             call \p f functor with item found, unlinks it from the set, and returns \p true.
1710             The \ref disposer specified in \p Traits class template parameter is called
1711             by garbage collector \p GC asynchronously.
1712
1713             The \p Func interface is
1714             \code
1715             struct functor {
1716                 void operator()( value_type const& item );
1717             };
1718             \endcode
1719             If the item with key equal to \p key is not found the function return \p false.
1720
1721             Note the hash functor should accept a parameter of type \p Q that can be not the same as \p value_type.
1722
1723             RCU \p synchronize method can be called. RCU should not be locked.
1724         */
1725         template <typename Q, typename Func>
1726         bool erase( Q const& key, Func f )
1727         {
1728             return do_erase( key, key_comparator(), f );
1729         }
1730
1731         /// Delete the item from the set with comparing functor \p pred
1732         /**
1733             The function is an analog of \ref cds_intrusive_SkipListSet_rcu_erase_func "erase(Q const&, Func)"
1734             but \p pred predicate is used for key comparing.
1735             \p Less has the interface like \p std::less.
1736             \p pred must imply the same element order as the comparator used for building the set.
1737         */
1738         template <typename Q, typename Less, typename Func>
1739         bool erase_with( Q const& key, Less pred, Func f )
1740         {
1741             CDS_UNUSED( pred );
1742             return do_erase( key, cds::opt::details::make_comparator_from_less<Less>(), f );
1743         }
1744
1745         /// Finds \p key
1746         /** @anchor cds_intrusive_SkipListSet_rcu_find_func
1747             The function searches the item with key equal to \p key and calls the functor \p f for item found.
1748             The interface of \p Func functor is:
1749             \code
1750             struct functor {
1751                 void operator()( value_type& item, Q& key );
1752             };
1753             \endcode
1754             where \p item is the item found, \p key is the <tt>find</tt> function argument.
1755
1756             The functor can change non-key fields of \p item. Note that the functor is only guarantee
1757             that \p item cannot be disposed during functor is executing.
1758             The functor does not serialize simultaneous access to the set \p item. If such access is
1759             possible you must provide your own synchronization schema on item level to exclude unsafe item modifications.
1760
1761             The \p key argument is non-const since it can be used as \p f functor destination i.e., the functor
1762             can modify both arguments.
1763
1764             The function applies RCU lock internally.
1765
1766             The function returns \p true if \p key is found, \p false otherwise.
1767         */
1768         template <typename Q, typename Func>
1769         bool find( Q& key, Func f )
1770         {
1771             return do_find_with( key, key_comparator(), f );
1772         }
1773         //@cond
1774         template <typename Q, typename Func>
1775         bool find( Q const& key, Func f )
1776         {
1777             return do_find_with( key, key_comparator(), f );
1778         }
1779         //@endcond
1780
1781         /// Finds the key \p key with comparing functor \p pred
1782         /**
1783             The function is an analog of \ref cds_intrusive_SkipListSet_rcu_find_func "find(Q&, Func)"
1784             but \p cmp is used for key comparison.
1785             \p Less functor has the interface like \p std::less.
1786             \p cmp must imply the same element order as the comparator used for building the set.
1787         */
1788         template <typename Q, typename Less, typename Func>
1789         bool find_with( Q& key, Less pred, Func f )
1790         {
1791             CDS_UNUSED( pred );
1792             return do_find_with( key, cds::opt::details::make_comparator_from_less<Less>(), f );
1793         }
1794         //@cond
1795         template <typename Q, typename Less, typename Func>
1796         bool find_with( Q const& key, Less pred, Func f )
1797         {
1798             CDS_UNUSED( pred );
1799             return do_find_with( key, cds::opt::details::make_comparator_from_less<Less>(), f );
1800         }
1801         //@endcond
1802
1803         /// Finds \p key
1804         /** @anchor cds_intrusive_SkipListSet_rcu_find_val
1805             The function searches the item with key equal to \p key
1806             and returns \p true if it is found, and \p false otherwise.
1807
1808             The function applies RCU lock internally.
1809         */
1810         template <typename Q>
1811         bool find( Q const& key )
1812         {
1813             return do_find_with( key, key_comparator(), [](value_type& , Q const& ) {} );
1814         }
1815
1816         /// Finds \p key with comparing functor \p pred
1817         /**
1818             The function is an analog of \ref cds_intrusive_SkipListSet_rcu_find_val "find(Q const&)"
1819             but \p pred is used for key compare.
1820             \p Less functor has the interface like \p std::less.
1821             \p pred must imply the same element order as the comparator used for building the set.
1822         */
1823         template <typename Q, typename Less>
1824         bool find_with( Q const& key, Less pred )
1825         {
1826             CDS_UNUSED( pred );
1827             return do_find_with( key, cds::opt::details::make_comparator_from_less<Less>(), [](value_type& , Q const& ) {} );
1828         }
1829
1830         /// Finds \p key and return the item found
1831         /** \anchor cds_intrusive_SkipListSet_rcu_get
1832             The function searches the item with key equal to \p key and returns a \p raw_ptr object pointed to item found.
1833             If \p key is not found it returns empty \p raw_ptr.
1834
1835             Note the compare functor should accept a parameter of type \p Q that can be not the same as \p value_type.
1836
1837             RCU should be locked before call of this function.
1838             Returned item is valid only while RCU is locked:
1839             \code
1840             typedef cds::intrusive::SkipListSet< cds::urcu::gc< cds::urcu::general_buffered<> >, foo, my_traits > skip_list;
1841             skip_list theList;
1842             // ...
1843             typename skip_list::raw_ptr pVal;
1844             {
1845                 // Lock RCU
1846                 skip_list::rcu_lock lock;
1847
1848                 pVal = theList.get( 5 );
1849                 if ( pVal ) {
1850                     // Deal with pVal
1851                     //...
1852                 }
1853             }
1854             // You can manually release pVal after RCU-locked section
1855             pVal.release();
1856             \endcode
1857         */
1858         template <typename Q>
1859         raw_ptr get( Q const& key )
1860         {
1861             assert( gc::is_locked());
1862
1863             position pos;
1864             value_type * pFound;
1865             if ( do_find_with( key, key_comparator(), [&pFound](value_type& found, Q const& ) { pFound = &found; }, pos ))
1866                 return raw_ptr( pFound, raw_ptr_disposer( pos ));
1867             return raw_ptr( raw_ptr_disposer( pos ));
1868         }
1869
1870         /// Finds \p key and return the item found
1871         /**
1872             The function is an analog of \ref cds_intrusive_SkipListSet_rcu_get "get(Q const&)"
1873             but \p pred is used for comparing the keys.
1874
1875             \p Less functor has the semantics like \p std::less but should take arguments of type \ref value_type and \p Q
1876             in any order.
1877             \p pred must imply the same element order as the comparator used for building the set.
1878         */
1879         template <typename Q, typename Less>
1880         raw_ptr get_with( Q const& key, Less pred )
1881         {
1882             CDS_UNUSED( pred );
1883             assert( gc::is_locked());
1884
1885             value_type * pFound = nullptr;
1886             position pos;
1887             if ( do_find_with( key, cds::opt::details::make_comparator_from_less<Less>(),
1888                 [&pFound](value_type& found, Q const& ) { pFound = &found; }, pos ))
1889             {
1890                 return raw_ptr( pFound, raw_ptr_disposer( pos ));
1891             }
1892             return raw_ptr( raw_ptr_disposer( pos ));
1893         }
1894
1895         /// Returns item count in the set
1896         /**
1897             The value returned depends on item counter type provided by \p Traits template parameter.
1898             For \p atomicity::empty_item_counter the function always returns 0.
1899             Therefore, the function is not suitable for checking the set emptiness, use \p empty()
1900             member function for this purpose.
1901         */
1902         size_t size() const
1903         {
1904             return m_ItemCounter;
1905         }
1906
1907         /// Checks if the set is empty
1908         bool empty() const
1909         {
1910             return m_Head.head()->next( 0 ).load( memory_model::memory_order_relaxed ) == nullptr;
1911         }
1912
1913         /// Clears the set (not atomic)
1914         /**
1915             The function unlink all items from the set.
1916             The function is not atomic, thus, in multi-threaded environment with parallel insertions
1917             this sequence
1918             \code
1919             set.clear();
1920             assert( set.empty() );
1921             \endcode
1922             the assertion could be raised.
1923
1924             For each item the \p disposer will be called automatically after unlinking.
1925         */
1926         void clear()
1927         {
1928             exempt_ptr ep;
1929             while ( (ep = extract_min()) );
1930         }
1931
1932         /// Returns maximum height of skip-list. The max height is a constant for each object and does not exceed 32.
1933         static CDS_CONSTEXPR unsigned int max_height() CDS_NOEXCEPT
1934         {
1935             return c_nMaxHeight;
1936         }
1937
1938         /// Returns const reference to internal statistics
1939         stat const& statistics() const
1940         {
1941             return m_Stat;
1942         }
1943     };
1944
1945 }} // namespace cds::intrusive
1946
1947
1948 #endif // #ifndef CDSLIB_INTRUSIVE_SKIP_LIST_RCU_H