cds/intrusive/optimistic_queue.h

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