cds/intrusive/treiber_stack.h

   1 //$$CDS-header$$
   2
   3 #ifndef __CDS_INTRUSIVE_TREIBER_STACK_H
   4 #define __CDS_INTRUSIVE_TREIBER_STACK_H
   5
   6 #include <type_traits>
   7 #include <functional>   // ref
   8 #include <mutex>        // unique_lock
   9 #include <cds/intrusive/details/single_link_struct.h>
  10 #include <cds/algo/elimination.h>
  11 #include <cds/opt/buffer.h>
  12 #include <cds/lock/spinlock.h>
  13 #include <cds/details/type_padding.h>
  14
  15 namespace cds { namespace intrusive {
  16
  17     /// TreiberStack related definitions
  18     /** @ingroup cds_intrusive_helper
  19     */
  20     namespace treiber_stack {
  21
  22         //@cond
  23         /// Operation id for the \ref cds_elimination_description "elimination back-off"
  24         enum operation_id {
  25             op_push,    ///< push op id
  26             op_pop      ///< pop op id
  27         };
  28
  29         /// Operation descriptor for the \ref cds_elimination_description "elimination back-off"
  30         template <typename T>
  31         struct operation: public cds::algo::elimination::operation_desc
  32         {
  33             operation_id    idOp;   ///< Op id
  34             T *             pVal;   ///< for push: pointer to argument; for pop: accepts a return value
  35             atomics::atomic<unsigned int> nStatus; ///< Internal elimination status
  36
  37             operation()
  38                 : pVal( nullptr )
  39                 , nStatus(0)
  40             {}
  41         };
  42         //@endcond
  43
  44         /// Stack internal statistics. May be useful for debugging or profiling
  45         /**
  46             Template argument \p Counter defines type of counter.
  47             Default is cds::atomicity::event_counter.
  48             You may use stronger type of counter like as cds::atomicity::item_counter,
  49             or even an integral type, for example, \p int
  50         */
  51         template <typename Counter = cds::atomicity::event_counter >
  52         struct stat
  53         {
  54             typedef Counter     counter_type    ;   ///< Counter type
  55
  56             counter_type m_PushCount        ;  ///< Push call count
  57             counter_type m_PopCount         ;  ///< Pop call count
  58             counter_type m_PushRace         ;  ///< Count of push race conditions encountered
  59             counter_type m_PopRace          ;  ///< Count of pop race conditions encountered
  60             counter_type m_ActivePushCollision  ; ///< Count of active push collision for elimination back-off
  61             counter_type m_ActivePopCollision   ; ///< Count of active pop collision for elimination back-off
  62             counter_type m_PassivePushCollision ; ///< Count of passive push collision for elimination back-off
  63             counter_type m_PassivePopCollision  ; ///< Count of passive pop collision for elimination back-off
  64             counter_type m_EliminationFailed    ; ///< Count of unsuccessful elimination back-off
  65
  66             //@cond
  67             void onPush()               { ++m_PushCount; }
  68             void onPop()                { ++m_PopCount; }
  69             void onPushRace()           { ++m_PushRace; }
  70             void onPopRace()            { ++m_PopRace; }
  71             void onActiveCollision( operation_id opId )
  72             {
  73                 if ( opId == treiber_stack::op_push )
  74                     ++m_ActivePushCollision;
  75                 else
  76                     ++m_ActivePopCollision;
  77             }
  78             void onPassiveCollision( operation_id opId )
  79             {
  80                 if ( opId == treiber_stack::op_push )
  81                     ++m_PassivePushCollision;
  82                 else
  83                     ++m_PassivePopCollision;
  84             }
  85             void onEliminationFailed()          { ++m_EliminationFailed;}
  86             //@endcond
  87         };
  88
  89         /// Empty (no overhead) stack statistics. Support interface like treiber_stack::stat
  90         struct empty_stat
  91         {
  92             //@cond
  93             void onPush()       {}
  94             void onPop()        {}
  95             void onPushRace()   {}
  96             void onPopRace()    {}
  97             void onActiveCollision( operation_id )  {}
  98             void onPassiveCollision( operation_id ) {}
  99             void onEliminationFailed() {}
 100             //@endcond
 101         };
 102
 103         //@cond
 104         namespace details {
 105
 106             template <bool EnableElimination, typename T, typename Traits>
 107             class elimination_backoff;
 108
 109             template <typename T, typename Traits>
 110             class elimination_backoff<false, T, Traits>
 111             {
 112                 typedef typename Traits::back_off   back_off;
 113
 114                 back_off    m_bkoff;
 115             public:
 116                 elimination_backoff()
 117                 {}
 118
 119                 elimination_backoff( size_t )
 120                 {}
 121
 122                 void reset()
 123                 {
 124                     m_bkoff.reset();
 125                 }
 126
 127                 template <typename Stat>
 128                 bool backoff(treiber_stack::operation< T >&, Stat& )
 129                 {
 130                     m_bkoff();
 131                     return false;
 132                 }
 133             };
 134
 135             template <typename T, typename Traits>
 136             class elimination_backoff<true, T, Traits>
 137             {
 138                 typedef typename Traits::back_off   back_off;
 139
 140                 /// Back-off for elimination (usually delay)
 141                 typedef typename Traits::elimination_backoff elimination_backoff_type;
 142                 /// Lock type used in elimination back-off
 143                 typedef typename Traits::lock_type elimination_lock_type;
 144                 /// Random engine used in elimination back-off
 145                 typedef typename Traits::random_engine elimination_random_engine;
 146
 147                 /// Per-thread elimination record
 148                 typedef cds::algo::elimination::record  elimination_rec;
 149
 150                 /// Collision array record
 151                 struct collision_array_record {
 152                     elimination_rec *     pRec;
 153                     elimination_lock_type lock;
 154                 };
 155
 156                 /// Collision array used in elimination-backoff; each item is optimized for cache-line size
 157                 typedef typename Traits::buffer::template rebind<
 158                     typename cds::details::type_padding<collision_array_record, cds::c_nCacheLineSize >::type
 159                 >::other collision_array;
 160
 161                 /// Operation descriptor used in elimination back-off
 162                 typedef treiber_stack::operation< T >  operation_desc;
 163
 164                 /// Elimination back-off data
 165                 struct elimination_data {
 166                     elimination_random_engine   randEngine; ///< random engine
 167                     collision_array             collisions; ///< collision array
 168
 169                     elimination_data()
 170                     {
 171                         //TODO: check Traits::buffer must be static!
 172                     }
 173                     elimination_data( size_t nCollisionCapacity )
 174                         : collisions( nCollisionCapacity )
 175                     {}
 176                 };
 177
 178                 elimination_data m_Elimination;
 179
 180                 enum operation_status {
 181                     op_free = 0,
 182                     op_busy = 1,
 183                     op_collided = 2
 184                 };
 185
 186                 typedef std::unique_lock< elimination_lock_type > slot_scoped_lock;
 187
 188             public:
 189                 elimination_backoff()
 190                 {
 191                     m_Elimination.collisions.zeroize();
 192                 }
 193
 194                 elimination_backoff( size_t nCollisionCapacity )
 195                     : m_Elimination( nCollisionCapacity )
 196                 {
 197                     m_Elimination.collisions.zeroize();
 198                 }
 199
 200                 void reset()
 201                 {}
 202
 203                 template <typename Stat>
 204                 bool backoff( operation_desc& op, Stat& stat )
 205                 {
 206                     elimination_backoff_type bkoff;
 207                     op.nStatus.store( op_busy, atomics::memory_order_relaxed );
 208
 209                     elimination_rec * myRec = cds::algo::elimination::init_record( op );
 210
 211                     collision_array_record& slot = m_Elimination.collisions[m_Elimination.randEngine() % m_Elimination.collisions.capacity()];
 212                     {
 213                         slot.lock.lock();
 214                         elimination_rec * himRec = slot.pRec;
 215                         if ( himRec ) {
 216                             operation_desc * himOp = static_cast<operation_desc *>( himRec->pOp );
 217                             assert( himOp );
 218                             if ( himOp->idOp != op.idOp ) {
 219                                 if ( op.idOp == treiber_stack::op_push )
 220                                     himOp->pVal = op.pVal;
 221                                 else
 222                                     op.pVal = himOp->pVal;
 223                                 slot.pRec = nullptr;
 224                                 slot.lock.unlock();
 225
 226                                 himOp->nStatus.store( op_collided, atomics::memory_order_release );
 227                                 cds::algo::elimination::clear_record();
 228                                 stat.onActiveCollision( op.idOp );
 229                                 return true;
 230                             }
 231                             himOp->nStatus.store( op_free, atomics::memory_order_release );
 232                         }
 233                         slot.pRec = myRec;
 234                         slot.lock.unlock();
 235                     }
 236
 237                     // Wait for colliding operation
 238                     bkoff( [&op]() -> bool { return op.nStatus.load( atomics::memory_order_acquire ) != op_busy; } );
 239                     {
 240                         slot_scoped_lock l( slot.lock );
 241                         if ( slot.pRec == myRec )
 242                             slot.pRec = nullptr;
 243                     }
 244
 245                     bool bCollided = op.nStatus.load( atomics::memory_order_acquire ) == op_collided;
 246
 247                     if ( !bCollided )
 248                         stat.onEliminationFailed();
 249                     else
 250                         stat.onPassiveCollision( op.idOp );
 251
 252                     cds::algo::elimination::clear_record();
 253                     return bCollided;
 254                 }
 255             };
 256
 257         } // namespace details
 258         //@endcond
 259     } // namespace treiber_stack
 260
 261     /// Treiber stack
 262     /** @ingroup cds_intrusive_stack
 263         Intrusive implementation of well-known Treiber's stack algorithm:
 264         - R. K. Treiber. Systems programming: Coping with parallelism. Technical Report RJ 5118, IBM Almaden Research Center, April 1986.
 265
 266         \ref cds_elimination_description "Elimination back-off technique" can be used optionally.
 267         The idea of elimination algorithm is taken from:
 268         - [2004] Danny Hendler, Nir Shavit, Lena Yerushalmi "A Scalable Lock-free Stack Algorithm"
 269
 270         The elimination algorithm uses a single elimination array as a back-off schema
 271         on a shared lock-free stack. If the threads fail on the stack, they attempt to eliminate
 272         on the array, and if they fail in eliminating, they attempt to access the stack again and so on.
 273
 274         @note Hendler's et al paper describes a lock-free implementation of elimination back-off which is quite complex.
 275         The main difficulty is the managing life-time of elimination record.
 276         Our implementation uses simplified lock-based (spin-based) approach which allows
 277         the elimination record allocation on thread's stack.
 278         This approach demonstrates sufficient performance under high load.
 279
 280         Template arguments:
 281         - \p GC - garbage collector type: gc::HP, gc::PTB
 282         - \p T - type to be inserted into the stack
 283         - \p Options - options
 284
 285         \p Options are:
 286         - opt::hook - hook used. Possible values are: single_link::base_hook, single_link::member_hook, single_link::traits_hook.
 287             If the option is not specified, <tt>single_link::base_hook<></tt> is used.
 288         - opt::back_off - back-off strategy used. If the option is not specified, the cds::backoff::Default is used.
 289         - opt::disposer - the functor used for dispose removed items. Default is opt::v::empty_disposer. This option is used only
 290             in \ref clear function.
 291         - opt::link_checker - the type of node's link fields checking. Default is \ref opt::debug_check_link.
 292         - opt::memory_model - C++ memory ordering model. Can be opt::v::relaxed_ordering (relaxed memory model, the default)
 293             or opt::v::sequential_consistent (sequentially consisnent memory model).
 294         - opt::item_counter - the type of item counting feature. Default is \ref atomicity::empty_item_counter
 295         - opt::stat - the type to gather internal statistics.
 296             Possible option value are: \ref treiber_stack::stat, \ref treiber_stack::empty_stat (the default),
 297             user-provided class that supports treiber_stack::stat interface.
 298         - opt::enable_elimination - enable elimination back-off for the stack. Default value is \p valse.
 299
 300         If elimination back-off is enabled (\p %cds::opt::enable_elimination< true >) additional options can be specified:
 301         - opt::buffer - a buffer type for elimination array, see \p opt::v::static_buffer, \p opt::v::dynamic_buffer.
 302             The buffer can be any size: \p Exp2 template parameter of those classes can be \p false.
 303             The size should be selected empirically for your application and hardware, there are no common rules for that.
 304             Default is <tt> %opt::v::static_buffer< any_type, 4 > </tt>.
 305         - opt::random_engine - a random engine to generate a random position in elimination array.
 306             Default is opt::v::c_rand.
 307         - opt::elimination_backoff - back-off strategy to wait for elimination, default is cds::backoff::delay<>
 308         - opt::lock_type - a lock type used in elimination back-off, default is cds::lock::Spin.
 309
 310         Garbage collecting schema \p GC must be consistent with the single_link::node GC.
 311
 312         Be careful when you want destroy an item popped, see \ref cds_intrusive_item_destroying "Destroying items of intrusive containers".
 313
 314         @anchor cds_intrusive_TreiberStack_examples
 315         \par Examples
 316
 317         Example of how to use \p single_link::base_hook.
 318         Your class that objects will be pushed on \p %TreiberStack should be based on \p single_link::node class
 319         \code
 320         #include <cds/intrusive/stack/treiber_stack.h>
 321         #include <cds/gc/hp.h>
 322
 323         namespace ci = cds::intrusive;
 324         typedef cds::gc::HP gc;
 325
 326         struct myData: public ci::single_link::node< gc >
 327         {
 328             // ...
 329         };
 330
 331         // Stack type
 332         typedef ci::TreiberStack< gc,
 333             myData,
 334             ci::opt::hook< ci::single_link::base_hook< gc > >
 335         > stack_t;
 336
 337         // Stack with elimination back-off enabled
 338         typedef ci::TreiberStack< gc,
 339             myData,
 340             ci::opt::hook< ci::single_link::base_hook< gc > >,
 341             cds::opt::enable_elimination<true>
 342         > elimination_stack_t;
 343         \endcode
 344
 345         Example of how to use \p base_hook with different tags.
 346         \code
 347         #include <cds/intrusive/stack/treiber_stack.h>
 348         #include <cds/gc/hp.h>
 349
 350         namespace ci = cds::intrusive;
 351         typedef cds::gc::HP gc;
 352
 353         // It is not necessary to declare complete type for tags
 354         struct tag1;
 355         struct tag2;
 356
 357         struct myData
 358             : public ci::single_link::node< gc, tag1 >
 359             , public ci::single_link::node< gc, tag2 >
 360         {
 361             // ...
 362         };
 363
 364         typedef ci::TreiberStack< gc, myData, ci::opt::hook< ci::single_link::base_hook< gc, tag1 > > stack1_t;
 365         typedef ci::TreiberStack< gc, myData, ci::opt::hook< ci::single_link::base_hook< gc, tag2 > > stack2_t;
 366
 367         // You may add myData objects in the objects of type stack1_t and stack2_t independently
 368         void foo() {
 369             stack1_t    s1;
 370             stack2_t    s2;
 371
 372             myData i1, i2;
 373             s1.push( i1 );
 374             s2.push( i2 );
 375             s2.push( i1 )   ;   // i1 is now contained in s1 and s2.
 376
 377             myData * p;
 378
 379             p = s1.pop()    ;   // pop i1 from s1
 380             p = s1.pop()    ;   // p == nullptr, s1 is empty
 381             p = s2.pop()    ;   // pop i1 from s2
 382             p = s2.pop()    ;   // pop i2 from s2
 383             p = s2.pop()    ;   // p == nullptr, s2 is empty
 384         }
 385         \endcode
 386
 387         Example of how to use \p member_hook.
 388         Your class that will be pushed on \p %TreiberStack should have a member of type \p single_link::node
 389         \code
 390         #include <cds/intrusive/stack/treiber_stack.h>
 391         #include <cds/gc/hp.h>
 392         #include <stddef.h>     // offsetof macro
 393
 394         namespace ci = cds::intrusive;
 395         typedef cds::gc::HP gc;
 396
 397         struct myData
 398         {
 399             // ...
 400             ci::single_link::node< gc >      member_hook_;
 401             // ...
 402         };
 403
 404         typedef ci::TreiberStack< gc, myData,
 405             ci::opt::hook<
 406                 ci::single_link::member_hook< offsetof(myData, member_hook_),
 407                 gc
 408             >
 409         > stack_t;
 410         \endcode
 411     */
 412     template <typename GC, typename T, typename... Options>
 413     class TreiberStack
 414     {
 415         //@cond
 416         struct default_options
 417         {
 418             typedef cds::backoff::Default           back_off;
 419             typedef single_link::base_hook<>        hook;
 420             typedef opt::v::empty_disposer          disposer;
 421             typedef atomicity::empty_item_counter   item_counter;
 422             typedef opt::v::relaxed_ordering        memory_model;
 423             typedef treiber_stack::empty_stat       stat;
 424             static CDS_CONSTEXPR_CONST opt::link_check_type link_checker = opt::debug_check_link;
 425
 426             // Elimination back-off options
 427             static CDS_CONSTEXPR_CONST bool enable_elimination = false;
 428             typedef cds::backoff::delay<>          elimination_backoff;
 429             typedef opt::v::static_buffer< int, 4 > buffer;
 430             typedef opt::v::c_rand                  random_engine;
 431             typedef cds::lock::Spin                 lock_type;
 432         };
 433         //@endcond
 434
 435     public:
 436         //@cond
 437         typedef typename opt::make_options<
 438             typename cds::opt::find_type_traits< default_options, Options... >::type
 439             ,Options...
 440         >::type   options;
 441         //@endcond
 442
 443     public:
 444         /// Rebind template arguments
 445         template <typename GC2, typename T2, typename... Options2>
 446         struct rebind {
 447             typedef TreiberStack< GC2, T2, Options2...> other   ;   ///< Rebinding result
 448         };
 449
 450     public:
 451         typedef T  value_type   ;   ///< type of value stored in the stack
 452         typedef typename options::hook      hook        ;   ///< hook type
 453         typedef typename hook::node_type    node_type   ;   ///< node type
 454         typedef typename options::disposer  disposer    ;   ///< disposer used
 455         typedef typename get_node_traits< value_type, node_type, hook>::type node_traits ;    ///< node traits
 456         typedef typename single_link::get_link_checker< node_type, options::link_checker >::type link_checker   ;   ///< link checker
 457         typedef typename options::memory_model  memory_model      ;   ///< Memory ordering. See cds::opt::memory_model option
 458         typedef typename options::item_counter item_counter ;   ///< Item counting policy used
 459         typedef typename options::stat      stat        ;   ///< Internal statistics policy used
 460
 461         typedef GC  gc          ;   ///< Garbage collector
 462         typedef typename options::back_off  back_off    ;   ///< back-off strategy
 463
 464     public: // related to elimination back-off
 465
 466         /// Elimination back-off is enabled or not
 467         static CDS_CONSTEXPR_CONST bool enable_elimination = options::enable_elimination;
 468         /// back-off strategy used to wait for elimination
 469         typedef typename options::elimination_backoff elimination_backoff_type;
 470         /// Lock type used in elimination back-off
 471         typedef typename options::lock_type elimination_lock_type;
 472         /// Random engine used in elimination back-off
 473         typedef typename options::random_engine elimination_random_engine;
 474
 475
 476     protected:
 477         typename node_type::atomic_node_ptr m_Top       ;   ///< Top of the stack
 478         item_counter        m_ItemCounter   ;   ///< Item counter
 479         stat                m_stat          ;   ///< Internal statistics
 480
 481         //@cond
 482         treiber_stack::details::elimination_backoff<enable_elimination, value_type, options> m_Backoff;
 483
 484         typedef intrusive::node_to_value<TreiberStack>  node_to_value;
 485         typedef treiber_stack::operation< value_type >  operation_desc;
 486         //@endcond
 487
 488     protected:
 489         //@cond
 490         void clear_links( node_type * pNode ) CDS_NOEXCEPT
 491         {
 492             pNode->m_pNext.store( nullptr, memory_model::memory_order_relaxed );
 493         }
 494
 495         template <bool EnableElimination>
 496         struct elimination_backoff_impl;
 497
 498         void init()
 499         {
 500             // GC and node_type::gc must be the same
 501             static_assert(( std::is_same<gc, typename node_type::gc>::value ), "GC and node_type::gc must be the same");
 502
 503             static_assert( (!enable_elimination || std::is_same<typename elimination_random_engine::result_type, unsigned int>::value),
 504                 "Random engine result type must be unsigned int" );
 505         }
 506
 507         //@endcond
 508
 509     public:
 510         /// Constructs empty stack
 511         TreiberStack()
 512             : m_Top( nullptr )
 513         {
 514             init();
 515         }
 516
 517         /// Constructs empty stack and initializes elimination back-off data
 518         /**
 519             This form should be used if you use elimination back-off with dynamically allocated collision array, i.e
 520             \p Options... contains cds::opt::buffer< cds::opt::v::dynamic_buffer >.
 521             \p nCollisionCapacity parameter specifies the capacity of collision array.
 522         */
 523         TreiberStack( size_t nCollisionCapacity )
 524             : m_Top( nullptr )
 525             , m_Backoff( nCollisionCapacity )
 526         {
 527             init();
 528         }
 529
 530         TreiberStack( TreiberStack const& ) = delete;
 531
 532         /// Destructor calls \ref cds_intrusive_TreiberStack_clear "clear" member function
 533         ~TreiberStack()
 534         {
 535             clear();
 536         }
 537
 538         /// Push the item \p val on the stack
 539         /**
 540             No copying is made since it is intrusive stack.
 541         */
 542         bool push( value_type& val )
 543         {
 544             node_type * pNew = node_traits::to_node_ptr( val );
 545             link_checker::is_empty( pNew );
 546
 547             m_Backoff.reset();
 548
 549             operation_desc op;
 550             if ( enable_elimination ) {
 551                 op.idOp = treiber_stack::op_push;
 552                 op.pVal = &val;
 553             }
 554
 555             node_type * t = m_Top.load(memory_model::memory_order_relaxed);
 556             while ( true ) {
 557                 pNew->m_pNext.store( t, memory_model::memory_order_relaxed );
 558                 if ( m_Top.compare_exchange_weak( t, pNew, memory_model::memory_order_release, atomics::memory_order_relaxed )) {     // #1 sync-with #2
 559                     ++m_ItemCounter;
 560                     m_stat.onPush();
 561                     return true;
 562                 }
 563                 m_stat.onPushRace();
 564
 565                 if ( m_Backoff.backoff( op, m_stat ))
 566                     return true;
 567             }
 568         }
 569
 570         /// Pop an item from the stack
 571         /**
 572             If stack is empty, returns \p nullptr.
 573             The disposer is <b>not</b> called for popped item.
 574             See \ref cds_intrusive_item_destroying "Destroying items of intrusive containers".
 575         */
 576         value_type * pop()
 577         {
 578             m_Backoff.reset();
 579             typename gc::Guard  guard;
 580
 581             operation_desc op;
 582             if ( enable_elimination ) {
 583                 op.idOp = treiber_stack::op_pop;
 584             }
 585
 586             while ( true ) {
 587                 node_type * t = guard.protect( m_Top, node_to_value() );
 588                 if ( t == nullptr )
 589                     return nullptr;    // stack is empty
 590
 591                 node_type * pNext = t->m_pNext.load(memory_model::memory_order_relaxed);
 592                 if ( m_Top.compare_exchange_weak( t, pNext, memory_model::memory_order_acquire, atomics::memory_order_relaxed )) {              // #2
 593                     clear_links( t );
 594                     --m_ItemCounter;
 595                     m_stat.onPop();
 596                     return node_traits::to_value_ptr( *t );
 597                 }
 598
 599                 m_stat.onPopRace();
 600                 if ( m_Backoff.backoff( op, m_stat )) {
 601                     // may return nullptr if stack is empty
 602                     return op.pVal;
 603                 }
 604             }
 605         }
 606
 607         /// Check if stack is empty
 608         bool empty() const
 609         {
 610             // http://www.manning-sandbox.com/thread.jspa?threadID=46245&tstart=0
 611             return m_Top.load( memory_model::memory_order_relaxed ) == nullptr;
 612         }
 613
 614         /// Clear the stack
 615         /** @anchor cds_intrusive_TreiberStack_clear
 616             For each removed item the disposer is called.
 617
 618             <b>Caution</b>
 619             It is possible that after <tt>clear()</tt> the <tt>empty()</tt> returns \p false
 620             if some other thread pushes an item into the stack during \p clear works
 621         */
 622         void clear()
 623         {
 624             back_off bkoff;
 625             node_type * pTop;
 626             while ( true ) {
 627                 pTop = m_Top.load( memory_model::memory_order_relaxed );
 628                 if ( pTop == nullptr )
 629                     return;
 630                 if ( m_Top.compare_exchange_weak( pTop, nullptr, memory_model::memory_order_acq_rel, atomics::memory_order_relaxed ) ) {    // sync-with #1 and #2
 631                     m_ItemCounter.reset();
 632                     break;
 633                 }
 634                 bkoff();
 635             }
 636
 637             while( pTop ) {
 638                 node_type * p = pTop;
 639                 pTop = p->m_pNext.load(memory_model::memory_order_relaxed);
 640                 clear_links( p );
 641                 gc::template retire<disposer>( node_traits::to_value_ptr( *p ) );
 642             }
 643         }
 644
 645         /// Returns stack's item count
 646         /**
 647             The value returned depends on opt::item_counter option. For atomicity::empty_item_counter,
 648             this function always returns 0.
 649
 650             <b>Warning</b>: even if you use real item counter and it returns 0, this fact is not mean that the stack
 651             is empty. To check emptyness use \ref empty() method.
 652         */
 653         size_t    size() const
 654         {
 655             return m_ItemCounter.value();
 656         }
 657
 658         /// Returns reference to internal statistics
 659         stat const& statistics() const
 660         {
 661             return m_stat;
 662         }
 663     };
 664
 665 }} // namespace cds::intrusive
 666
 667 #endif  // #ifndef __CDS_INTRUSIVE_TREIBER_STACK_H