cds/intrusive/optimistic_queue.h

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