cds/intrusive/optimistic_queue.h

   1 //$$CDS-header$$
   2
   3 #ifndef CDSLIB_INTRUSIVE_OPTIMISTIC_QUEUE_H
   4 #define CDSLIB_INTRUSIVE_OPTIMISTIC_QUEUE_H
   5
   6 #include <type_traits>
   7 #include <cds/intrusive/details/base.h>
   8 #include <cds/algo/atomic.h>
   9 #include <cds/gc/default_gc.h>
  10
  11 namespace cds { namespace intrusive {
  12
  13     /// OptimisticQueue related definitions
  14     /** @ingroup cds_intrusive_helper
  15     */
  16     namespace optimistic_queue {
  17
  18         /// Optimistic queue node
  19         /**
  20             Template parameters:
  21             - \p GC - garbage collector
  22             - \p Tag - a \ref cds_intrusive_hook_tag "tag"
  23         */
  24         template <class GC, typename Tag = opt::none>
  25         struct node
  26         {
  27             typedef GC  gc  ;   ///< Garbage collector
  28             typedef Tag tag ;   ///< tag
  29
  30             typedef typename gc::template atomic_ref<node>    atomic_node_ptr    ;    ///< atomic pointer
  31
  32             atomic_node_ptr m_pNext ;   ///< Pointer to next node
  33             atomic_node_ptr m_pPrev ;   ///< Pointer to previous node
  34
  35             CDS_CONSTEXPR node() CDS_NOEXCEPT
  36                 : m_pNext( nullptr )
  37                 , m_pPrev( nullptr )
  38             {}
  39         };
  40
  41         //@cond
  42         struct default_hook {
  43             typedef cds::gc::default_gc gc;
  44             typedef opt::none           tag;
  45         };
  46         //@endcond
  47
  48         //@cond
  49         template < typename HookType, typename... Options>
  50         struct hook
  51         {
  52             typedef typename opt::make_options< default_hook, Options...>::type  options;
  53             typedef typename options::gc    gc;
  54             typedef typename options::tag   tag;
  55             typedef node<gc, tag> node_type;
  56             typedef HookType     hook_type;
  57         };
  58         //@endcond
  59
  60         /// Base hook
  61         /**
  62             \p Options are:
  63             - opt::gc - garbage collector used.
  64             - opt::tag - a \ref cds_intrusive_hook_tag "tag"
  65         */
  66         template < typename... Options >
  67         struct base_hook: public hook< opt::base_hook_tag, Options... >
  68         {};
  69
  70         /// Member hook
  71         /**
  72             \p MemberOffset specifies offset in bytes of \ref node member into your structure.
  73             Use \p offsetof macro to define \p MemberOffset
  74
  75             \p Options are:
  76             - opt::gc - garbage collector used.
  77             - opt::tag - a \ref cds_intrusive_hook_tag "tag"
  78         */
  79         template < size_t MemberOffset, typename... Options >
  80         struct member_hook: public hook< opt::member_hook_tag, Options... >
  81         {
  82             //@cond
  83             static const size_t c_nMemberOffset = MemberOffset;
  84             //@endcond
  85         };
  86
  87         /// Traits hook
  88         /**
  89             \p NodeTraits defines type traits for node.
  90             See \ref node_traits for \p NodeTraits interface description
  91
  92             \p Options are:
  93             - opt::gc - garbage collector used.
  94             - opt::tag - a \ref cds_intrusive_hook_tag "tag"
  95         */
  96         template <typename NodeTraits, typename... Options >
  97         struct traits_hook: public hook< opt::traits_hook_tag, Options... >
  98         {
  99             //@cond
 100             typedef NodeTraits node_traits;
 101             //@endcond
 102         };
 103
 104         /// Check link
 105         template <typename Node>
 106         struct link_checker {
 107             //@cond
 108             typedef Node node_type;
 109             //@endcond
 110
 111             /// Checks if the link fields of node \p pNode is \p nullptr
 112             /**
 113                 An asserting is generated if \p pNode link fields is not \p nullptr
 114             */
 115             static void is_empty( const node_type * pNode )
 116             {
 117                 assert( pNode->m_pNext.load( atomics::memory_order_relaxed ) == nullptr );
 118                 assert( pNode->m_pPrev.load( atomics::memory_order_relaxed ) == nullptr );
 119                 CDS_UNUSED( pNode );
 120             }
 121         };
 122
 123         /// Metafunction for selecting appropriate link checking policy
 124         template < typename Node, opt::link_check_type LinkType >
 125         struct get_link_checker
 126         {
 127             //@cond
 128             typedef intrusive::opt::v::empty_link_checker<Node>  type;
 129             //@endcond
 130         };
 131
 132         //@cond
 133         template < typename Node >
 134         struct get_link_checker< Node, opt::always_check_link >
 135         {
 136             typedef link_checker<Node>  type;
 137         };
 138         template < typename Node >
 139         struct get_link_checker< Node, opt::debug_check_link >
 140         {
 141 #       ifdef _DEBUG
 142             typedef link_checker<Node>  type;
 143 #       else
 144             typedef intrusive::opt::v::empty_link_checker<Node>  type;
 145 #       endif
 146         };
 147         //@endcond
 148
 149         /// OptimisticQueue internal statistics. May be used for debugging or profiling
 150         /**
 151             Template argument \p Counter defines type of counter.
 152             Default is \p cds::atomicity::event_counter, that is weak, i.e. it is not guaranteed
 153             strict event counting.
 154             You may use stronger type of counter like as \p cds::atomicity::item_counter,
 155             or even integral type, for example, \p int.
 156         */
 157         template <typename Counter = cds::atomicity::event_counter >
 158         struct stat
 159         {
 160             typedef Counter     counter_type;   ///< Counter type
 161
 162             counter_type m_EnqueueCount;    ///< Enqueue call count
 163             counter_type m_DequeueCount;    ///< Dequeue call count
 164             counter_type m_EnqueueRace;     ///< Count of enqueue race conditions encountered
 165             counter_type m_DequeueRace;     ///< Count of dequeue race conditions encountered
 166             counter_type m_AdvanceTailError; ///< Count of "advance tail failed" events
 167             counter_type m_BadTail;         ///< Count of events "Tail is not pointed to the last item in the queue"
 168             counter_type m_FixListCount;    ///< Count of fix list event
 169             counter_type m_EmptyDequeue;    ///< Count of dequeue from empty queue
 170
 171             /// Register enqueue call
 172             void onEnqueue() { ++m_EnqueueCount; }
 173             /// Register dequeue call
 174             void onDequeue() { ++m_DequeueCount; }
 175             /// Register enqueue race event
 176             void onEnqueueRace() { ++m_EnqueueRace; }
 177             /// Register dequeue race event
 178             void onDequeueRace() { ++m_DequeueRace; }
 179             /// Register "advance tail failed" event
 180             void onAdvanceTailFailed() { ++m_AdvanceTailError; }
 181             /// Register event "Tail is not pointed to last item in the queue"
 182             void onBadTail() { ++m_BadTail; }
 183             /// Register fix list event
 184             void onFixList()    { ++m_FixListCount; }
 185             /// Register dequeuing from empty queue
 186             void onEmptyDequeue() { ++m_EmptyDequeue; }
 187
 188             //@cond
 189             void reset()
 190             {
 191                 m_EnqueueCount.reset();
 192                 m_DequeueCount.reset();
 193                 m_EnqueueRace.reset();
 194                 m_DequeueRace.reset();
 195                 m_AdvanceTailError.reset();
 196                 m_BadTail.reset();
 197                 m_FixListCount.reset();
 198                 m_EmptyDequeue.reset();
 199             }
 200
 201             stat& operator +=( stat const& s )
 202             {
 203                 m_EnqueueCount += s.m_EnqueueCount.get();
 204                 m_DequeueCount += s.m_DequeueCount.get();
 205                 m_EnqueueRace += s.m_EnqueueRace.get();
 206                 m_DequeueRace += s.m_DequeueRace.get();
 207                 m_AdvanceTailError += s.m_AdvanceTailError.get();
 208                 m_BadTail += s.m_BadTail.get();
 209                 m_FixListCount += s.m_FixListCount.get();
 210                 m_EmptyDequeue += s.m_EmptyDequeue.get();
 211
 212                 return *this;
 213             }
 214             //@endcond
 215         };
 216
 217         /// Dummy OptimisticQueue statistics - no counting is performed. Support interface like \ref optimistic_queue::stat
 218         struct empty_stat
 219         {
 220             //@cond
 221             void onEnqueue()            const {}
 222             void onDequeue()            const {}
 223             void onEnqueueRace()        const {}
 224             void onDequeueRace()        const {}
 225             void onAdvanceTailFailed()  const {}
 226             void onBadTail()            const {}
 227             void onFixList()            const {}
 228             void onEmptyDequeue()       const {}
 229
 230             void reset() {}
 231             empty_stat& operator +=( empty_stat const& )
 232             {
 233                 return *this;
 234             }
 235             //@endcond
 236         };
 237
 238         /// OptimisticQueue default type traits
 239         struct traits
 240         {
 241             /// Back-off strategy
 242             typedef cds::backoff::empty             back_off;
 243
 244             /// Hook, possible types are \p optimistic_queue::base_hook, \p optimistic_queue::member_hook, \p optimistic_queue::traits_hook
 245             typedef optimistic_queue::base_hook<>   hook;
 246
 247             /// The functor used for dispose removed items. Default is \p opt::v::empty_disposer. This option is used for dequeuing
 248             typedef opt::v::empty_disposer          disposer;
 249
 250             /// Item counting feature; by default, disabled. Use \p cds::atomicity::item_counter to enable item counting
 251             typedef cds::atomicity::empty_item_counter   item_counter;
 252
 253             /// C++ memory ordering model
 254             /**
 255                 Can be \p opt::v::relaxed_ordering (relaxed memory model, the default)
 256                 or \p opt::v::sequential_consistent (sequentially consisnent memory model).
 257             */
 258             typedef opt::v::relaxed_ordering        memory_model;
 259
 260             /// Internal statistics (by default, disabled)
 261             /**
 262                 Possible option value are: \p optimistic_queue::stat, \p optimistic_queue::empty_stat (the default),
 263                 user-provided class that supports \p %optimistic_queue::stat interface.
 264             */
 265             typedef optimistic_queue::empty_stat    stat;
 266
 267             /// Link checking, see \p cds::opt::link_checker
 268             static CDS_CONSTEXPR const opt::link_check_type link_checker = opt::debug_check_link;
 269
 270             /// Padding for internal critical atomic data. Default is \p opt::cache_line_padding
 271             enum { padding = opt::cache_line_padding };
 272         };
 273
 274         /// Metafunction converting option list to \p optimistic_queue::traits
 275         /**
 276             Supported \p Options are:
 277
 278             - \p opt::hook - hook used. Possible hooks are: \p optimistic_queue::base_hook, \p optimistic_queue::member_hook, \p optimistic_queue::traits_hook.
 279                 If the option is not specified, \p %optimistic_queue::base_hook<> is used.
 280             - \p opt::back_off - back-off strategy used, default is \p cds::backoff::empty.
 281             - \p opt::disposer - the functor used for dispose removed items. Default is \p opt::v::empty_disposer. This option is used
 282                 when dequeuing.
 283             - \p opt::link_checker - the type of node's link fields checking. Default is \p opt::debug_check_link
 284             - \p opt::item_counter - the type of item counting feature. Default is \p cds::atomicity::empty_item_counter (item counting disabled)
 285                 To enable item counting use \p cds::atomicity::item_counter
 286             - \p opt::stat - the type to gather internal statistics.
 287                 Possible statistics types are: \p optimistic_queue::stat, \p optimistic_queue::empty_stat,
 288                 user-provided class that supports \p %optimistic_queue::stat interface.
 289                 Default is \p %optimistic_queue::empty_stat (internal statistics disabled).
 290             - \p opt::padding - padding for internal critical atomic data. Default is \p opt::cache_line_padding
 291             - \p opt::memory_model - C++ memory ordering model. Can be \p opt::v::relaxed_ordering (relaxed memory model, the default)
 292                 or \p opt::v::sequential_consistent (sequentially consisnent memory model).
 293
 294             Example: declare \p %OptimisticQueue with item counting and internal statistics
 295             \code
 296             typedef cds::intrusive::OptimisticQueue< cds::gc::HP, Foo,
 297                 typename cds::intrusive::optimistic_queue::make_traits<
 298                     cds::intrusive::opt:hook< cds::intrusive::optimistic_queue::base_hook< cds::opt::gc<cds:gc::HP> >>,
 299                     cds::opt::item_counte< cds::atomicity::item_counter >,
 300                     cds::opt::stat< cds::intrusive::optimistic_queue::stat<> >
 301                 >::type
 302             > myQueue;
 303             \endcode
 304         */
 305         template <typename... Options>
 306         struct make_traits {
 307 #   ifdef CDS_DOXYGEN_INVOKED
 308             typedef implementation_defined type;   ///< Metafunction result
 309 #   else
 310             typedef typename cds::opt::make_options<
 311                 typename cds::opt::find_type_traits< traits, Options... >::type
 312                 , Options...
 313             >::type type;
 314 #   endif
 315         };
 316     }   // namespace optimistic_queue
 317
 318     /// Optimistic intruive lock-free queue
 319     /** @ingroup cds_intrusive_queue
 320         Implementation of Ladan-Mozes & Shavit optimistic queue algorithm.
 321             [2008] Edya Ladan-Mozes, Nir Shavit "An Optimistic Approach to Lock-Free FIFO Queues"
 322
 323         Template arguments:
 324         - \p GC - garbage collector type: \p gc::HP, \p gc::DHP
 325         - \p T - type of value to be stored in the queue. A value of type \p T must be derived from \p optimistic_queue::node for \p optimistic_queue::base_hook,
 326             or it should have a member of type \p %optimistic_queue::node for \p optimistic_queue::member_hook,
 327             or it should be convertible to \p %optimistic_queue::node for \p optimistic_queue::traits_hook.
 328         - \p Traits - queue traits, default is \p optimistic_queue::traits. You can use \p optimistic_queue::make_traits
 329             metafunction to make your traits or just derive your traits from \p %optimistic_queue::traits:
 330             \code
 331             struct myTraits: public cds::intrusive::optimistic_queue::traits {
 332                 typedef cds::intrusive::optimistic_queue::stat<> stat;
 333                 typedef cds::atomicity::item_counter    item_counter;
 334             };
 335             typedef cds::intrusive::OptimisticQueue< cds::gc::HP, Foo, myTraits > myQueue;
 336
 337             // Equivalent make_traits example:
 338             typedef cds::intrusive::OptimisticQueue< cds::gc::HP, Foo,
 339                 typename cds::intrusive::optimistic_queue::make_traits<
 340                     cds::opt::stat< cds::intrusive::optimistic_queue::stat<> >,
 341                     cds::opt::item_counter< cds::atomicity::item_counter >
 342                 >::type
 343             > myQueue;
 344             \endcode
 345
 346         Garbage collecting schema \p GC must be consistent with the optimistic_queue::node GC.
 347
 348         \par About item disposing
 349         The optimistic queue algo has a key feature: even if the queue is empty it contains one item that is "dummy" one from
 350         the standpoint of the algo. See \p dequeue() function for explanation.
 351
 352         \par Examples
 353         \code
 354         #include <cds/gc/hp.h>
 355         #include <cds/intrusive/optimistic_queue.h>
 356
 357         namespace ci = cds::inrtusive;
 358         typedef cds::gc::HP hp_gc;
 359
 360         // Optimistic queue with Hazard Pointer garbage collector, base hook + item counter:
 361         struct Foo: public ci::optimistic_queue::node< hp_gc >
 362         {
 363             // Your data
 364             ...
 365         };
 366
 367         typedef ci::OptimisticQueue< hp_gc,
 368             Foo,
 369             typename ci::optimistic_queue::make_traits<
 370                 ci::opt::hook<
 371                     ci::optimistic_queue::base_hook< ci::opt::gc< hp_gc > >
 372                 >
 373                 ,cds::opt::item_counter< cds::atomicity::item_counter >
 374             >::type
 375         > FooQueue;
 376
 377         // Optimistic queue with Hazard Pointer garbage collector, member hook, no item counter:
 378         struct Bar
 379         {
 380             // Your data
 381             ...
 382             ci::optimistic_queue::node< hp_gc > hMember;
 383         };
 384
 385         typedef ci::OptimisticQueue< hp_gc,
 386             Bar,
 387             typename ci::optimistic_queue::make_traits<
 388                 ci::opt::hook<
 389                     ci::optimistic_queue::member_hook<
 390                         offsetof(Bar, hMember)
 391                         ,ci::opt::gc< hp_gc >
 392                     >
 393                 >
 394             >::type
 395         > BarQueue;
 396         \endcode
 397     */
 398     template <typename GC, typename T, typename Traits = optimistic_queue::traits >
 399     class OptimisticQueue
 400     {
 401     public:
 402         typedef GC gc;          ///< Garbage collector
 403         typedef T  value_type;  ///< type of value to be stored in the queue
 404         typedef Traits traits;  ///< Queue traits
 405
 406         typedef typename traits::hook       hook;       ///< hook type
 407         typedef typename hook::node_type    node_type;  ///< node type
 408         typedef typename traits::disposer   disposer;   ///< disposer used
 409         typedef typename get_node_traits< value_type, node_type, hook>::type node_traits;    ///< node traits
 410         typedef typename optimistic_queue::get_link_checker< node_type, traits::link_checker >::type link_checker;   ///< link checker
 411         typedef typename traits::back_off  back_off;        ///< back-off strategy
 412         typedef typename traits::item_counter item_counter; ///< Item counting policy used
 413         typedef typename traits::memory_model  memory_model;///< Memory ordering. See cds::opt::memory_model option
 414         typedef typename traits::stat      stat;            ///< Internal statistics policy used
 415
 416         /// Rebind template arguments
 417         template <typename GC2, typename T2, typename Traits2>
 418         struct rebind {
 419             typedef OptimisticQueue< GC2, T2, Traits2 > other   ;   ///< Rebinding result
 420         };
 421
 422     protected:
 423         //@cond
 424         typedef typename node_type::atomic_node_ptr atomic_node_ptr;
 425
 426         // GC and node_type::gc must be the same
 427         static_assert((std::is_same<gc, typename node_type::gc>::value), "GC and node_type::gc must be the same");
 428         //@endcond
 429
 430         atomic_node_ptr     m_pTail;   ///< Pointer to tail node
 431         typename opt::details::apply_padding< atomic_node_ptr, traits::padding >::padding_type pad1_;
 432         atomic_node_ptr     m_pHead;   ///< Pointer to head node
 433         typename opt::details::apply_padding< atomic_node_ptr, traits::padding >::padding_type pad2_;
 434         node_type           m_Dummy ;   ///< dummy node
 435         typename opt::details::apply_padding< atomic_node_ptr, traits::padding >::padding_type pad3_;
 436         item_counter        m_ItemCounter   ;   ///< Item counter
 437         stat                m_Stat          ;   ///< Internal statistics
 438
 439         static CDS_CONSTEXPR const size_t c_nHazardPtrCount = 5 ; ///< Count of hazard pointer required for the algorithm
 440
 441     protected:
 442         //@cond
 443         static void clear_links( node_type * pNode )
 444         {
 445             pNode->m_pNext.store( nullptr, memory_model::memory_order_release );
 446             pNode->m_pPrev.store( nullptr, memory_model::memory_order_release );
 447         }
 448
 449         struct dequeue_result {
 450             typename gc::template GuardArray<3>  guards;
 451
 452             node_type * pHead;
 453             node_type * pNext;
 454         };
 455
 456         bool do_dequeue( dequeue_result& res )
 457         {
 458             node_type * pTail;
 459             node_type * pHead;
 460             node_type * pFirstNodePrev;
 461             back_off bkoff;
 462
 463             while ( true ) { // Try till success or empty
 464                 pHead = res.guards.protect( 0, m_pHead, [](node_type * p) -> value_type * {return node_traits::to_value_ptr(p);});
 465                 pTail = res.guards.protect( 1, m_pTail, [](node_type * p) -> value_type * {return node_traits::to_value_ptr(p);});
 466                 assert( pHead != nullptr );
 467                 pFirstNodePrev = res.guards.protect( 2, pHead->m_pPrev, [](node_type * p) -> value_type * {return node_traits::to_value_ptr(p);});
 468
 469                 if ( pHead == m_pHead.load(memory_model::memory_order_acquire)) {
 470                     if ( pTail != pHead ) {
 471                         if ( pFirstNodePrev == nullptr
 472                           || pFirstNodePrev->m_pNext.load(memory_model::memory_order_acquire) != pHead )
 473                         {
 474                             fix_list( pTail, pHead );
 475                             continue;
 476                         }
 477                         if ( m_pHead.compare_exchange_weak( pHead, pFirstNodePrev, memory_model::memory_order_acquire, atomics::memory_order_relaxed )) {
 478                             // dequeue success
 479                             break;
 480                         }
 481                     }
 482                     else {
 483                         // the queue is empty
 484                         m_Stat.onEmptyDequeue();
 485                         return false;
 486                     }
 487                 }
 488
 489                 m_Stat.onDequeueRace();
 490                 bkoff();
 491             }
 492
 493             --m_ItemCounter;
 494             m_Stat.onDequeue();
 495
 496             res.pHead = pHead;
 497             res.pNext = pFirstNodePrev;
 498             return true;
 499         }
 500
 501
 502         /// Helper function for optimistic queue. Corrects \p prev pointer of queue's nodes if it is needed
 503         void fix_list( node_type * pTail, node_type * pHead )
 504         {
 505             // pTail and pHead are already guarded
 506
 507             node_type * pCurNode;
 508             node_type * pCurNodeNext;
 509
 510             typename gc::template GuardArray<2> guards;
 511
 512             pCurNode = pTail;
 513             while ( pCurNode != pHead ) { // While not at head
 514                 pCurNodeNext = guards.protect(0, pCurNode->m_pNext, [](node_type * p) -> value_type * { return node_traits::to_value_ptr(p);});
 515                 if ( pHead != m_pHead.load(memory_model::memory_order_acquire))
 516                     break;
 517                 pCurNodeNext->m_pPrev.store( pCurNode, memory_model::memory_order_release );
 518                 guards.assign( 1, node_traits::to_value_ptr( pCurNode = pCurNodeNext ));
 519             }
 520
 521             m_Stat.onFixList();
 522         }
 523
 524         void dispose_result( dequeue_result& res )
 525         {
 526             dispose_node( res.pHead );
 527         }
 528
 529         void dispose_node( node_type * p )
 530         {
 531             assert( p != nullptr );
 532
 533             if ( p != &m_Dummy ) {
 534                 struct internal_disposer
 535                 {
 536                     void operator ()( value_type * p )
 537                     {
 538                         assert( p != nullptr );
 539
 540                         OptimisticQueue::clear_links( node_traits::to_node_ptr( *p ) );
 541                         disposer()(p);
 542                     }
 543                 };
 544                 gc::template retire<internal_disposer>( node_traits::to_value_ptr(p) );
 545             }
 546         }
 547
 548         //@endcond
 549
 550     public:
 551         /// Constructor creates empty queue
 552         OptimisticQueue()
 553             : m_pTail( &m_Dummy )
 554             , m_pHead( &m_Dummy )
 555         {}
 556
 557         ~OptimisticQueue()
 558         {
 559             clear();
 560             node_type * pHead = m_pHead.load(memory_model::memory_order_relaxed);
 561             CDS_DEBUG_ONLY( node_type * pTail = m_pTail.load(memory_model::memory_order_relaxed); )
 562             CDS_DEBUG_ONLY( assert( pHead == pTail ); )
 563             assert( pHead != nullptr );
 564
 565             m_pHead.store( nullptr, memory_model::memory_order_relaxed );
 566             m_pTail.store( nullptr, memory_model::memory_order_relaxed );
 567
 568             dispose_node( pHead );
 569         }
 570
 571         /// @anchor cds_intrusive_OptimisticQueue_enqueue Enqueues \p data in lock-free manner. Always return \a true
 572         bool enqueue( value_type& val )
 573         {
 574             node_type * pNew = node_traits::to_node_ptr( val );
 575             link_checker::is_empty( pNew );
 576
 577             typename gc::template GuardArray<2> guards;
 578             back_off bkoff;
 579
 580             guards.assign( 1, &val );
 581             node_type * pTail = guards.protect( 0, m_pTail, [](node_type * p) -> value_type * {return node_traits::to_value_ptr(p);} );   // Read the tail
 582             while( true ) {
 583                 pNew->m_pNext.store( pTail, memory_model::memory_order_release );
 584                 if ( m_pTail.compare_exchange_strong( pTail, pNew, memory_model::memory_order_release, atomics::memory_order_relaxed )) { // Try to CAS the tail
 585                     pTail->m_pPrev.store( pNew, memory_model::memory_order_release ); // Success, write prev
 586                     ++m_ItemCounter;
 587                     m_Stat.onEnqueue();
 588                     break;     // Enqueue done!
 589                 }
 590                 guards.assign( 0, node_traits::to_value_ptr( pTail ));  // pTail has been changed by CAS above
 591                 m_Stat.onEnqueueRace();
 592                 bkoff();
 593             }
 594             return true;
 595         }
 596
 597         /// Dequeues a value from the queue
 598         /** @anchor cds_intrusive_OptimisticQueue_dequeue
 599             If the queue is empty the function returns \p nullptr
 600
 601             \par Warning
 602             The queue algorithm has following feature: when \p dequeue is called,
 603             the item returning is still queue's top, and previous top is disposed:
 604
 605             \code
 606             before dequeuing         Dequeue               after dequeuing
 607             +------------------+                           +------------------+
 608       Top ->|      Item 1      |  -> Dispose Item 1        |      Item 2      | <- Top
 609             +------------------+                           +------------------+
 610             |      Item 2      |  -> Return Item 2         |       ...        |
 611             +------------------+
 612             |       ...        |
 613             \endcode
 614
 615             \p dequeue function returns Item 2, that becomes new top of queue, and calls
 616             the disposer for Item 1, that was queue's top on function entry.
 617             Thus, you cannot manually delete item returned because it is still included in
 618             item sequence and it has valuable link field that must not be zeroed.
 619             The item may be deleted only in disposer call.
 620         */
 621         value_type * dequeue()
 622         {
 623             dequeue_result res;
 624             if ( do_dequeue( res )) {
 625                 dispose_result( res );
 626
 627                 return node_traits::to_value_ptr( *res.pNext );
 628             }
 629             return nullptr;
 630         }
 631
 632         /// Synonym for \p enqueue()
 633         bool push( value_type& val )
 634         {
 635             return enqueue( val );
 636         }
 637
 638         /// Synonym for \p dequeue()
 639         value_type * pop()
 640         {
 641             return dequeue();
 642         }
 643
 644         /// Checks if the queue is empty
 645         bool empty() const
 646         {
 647             return m_pTail.load(memory_model::memory_order_relaxed) == m_pHead.load(memory_model::memory_order_relaxed);
 648         }
 649
 650         /// Clear the stack
 651         /**
 652             The function repeatedly calls \ref dequeue until it returns \p nullptr.
 653             The disposer defined in template \p Traits is called for each item
 654             that can be safely disposed.
 655         */
 656         void clear()
 657         {
 658             value_type * pv;
 659             while ( (pv = dequeue()) != nullptr );
 660         }
 661
 662         /// Returns queue's item count
 663         /**
 664             The value returned depends on \p optimistic_queue::traits::item_counter.
 665             For \p atomicity::empty_item_counter, this function always returns 0.
 666
 667             @note Even if you use real item counter and it returns 0, this fact is not mean that the queue
 668             is empty. To check queue emptyness use \p empty() method.
 669         */
 670         size_t size() const
 671         {
 672             return m_ItemCounter.value();
 673         }
 674
 675         /// Returns refernce to internal statistics
 676         const stat& statistics() const
 677         {
 678             return m_Stat;
 679         }
 680     };
 681
 682 }}  // namespace cds::intrusive
 683
 684 #endif // #ifndef CDSLIB_INTRUSIVE_OPTIMISTIC_QUEUE_H