cds/container/segmented_queue.h

   1 //$$CDS-header$$
   2
   3 #ifndef CDSLIB_CONTAINER_SEGMENTED_QUEUE_H
   4 #define CDSLIB_CONTAINER_SEGMENTED_QUEUE_H
   5
   6 #include <memory>
   7 #include <functional>   // ref
   8 #include <cds/intrusive/segmented_queue.h>
   9 #include <cds/details/trivial_assign.h>
  10
  11 namespace cds { namespace container {
  12
  13     /// SegmentedQueue -related declarations
  14     namespace segmented_queue {
  15
  16 #   ifdef CDS_DOXYGEN_INVOKED
  17         /// SegmentedQueue internal statistics
  18         typedef cds::intrusive::segmented_queue::stat stat;
  19 #   else
  20         using cds::intrusive::segmented_queue::stat;
  21 #   endif
  22
  23         /// SegmentedQueue empty internal statistics (no overhead)
  24         typedef cds::intrusive::segmented_queue::empty_stat empty_stat;
  25
  26         /// SegmentedQueue default type traits
  27         struct traits {
  28
  29             /// Item allocator. Default is \ref CDS_DEFAULT_ALLOCATOR
  30             typedef CDS_DEFAULT_ALLOCATOR   node_allocator;
  31
  32             /// Item counter, default is atomicity::item_counter
  33             /**
  34                 The item counting is an essential part of segmented queue algorithm.
  35                 The \p empty() member function is based on checking <tt>size() == 0</tt>.
  36                 Therefore, dummy item counter like atomicity::empty_item_counter is not the proper counter.
  37             */
  38             typedef atomicity::item_counter item_counter;
  39
  40             /// Internal statistics, possible predefined types are \ref stat, \ref empty_stat (the default)
  41             typedef segmented_queue::empty_stat        stat;
  42
  43             /// Memory model, default is opt::v::relaxed_ordering. See cds::opt::memory_model for the full list of possible types
  44             typedef opt::v::relaxed_ordering  memory_model;
  45
  46             /// Alignment of critical data, default is cache line alignment. See cds::opt::alignment option specification
  47             enum { alignment = opt::cache_line_alignment };
  48
  49             /// Padding of segment data, default is no special padding
  50             /**
  51                 The segment is just an array of atomic data pointers,
  52                 so, the high load leads to false sharing and performance degradation.
  53                 A padding of segment data can eliminate false sharing issue.
  54                 On the other hand, the padding leads to increase segment size.
  55             */
  56             enum { padding = cds::intrusive::segmented_queue::traits::padding };
  57
  58             /// Segment allocator. Default is \ref CDS_DEFAULT_ALLOCATOR
  59             typedef CDS_DEFAULT_ALLOCATOR allocator;
  60
  61             /// Lock type used to maintain an internal list of allocated segments
  62             typedef cds::sync::spin lock_type;
  63
  64             /// Random \ref cds::opt::permutation_generator "permutation generator" for sequence [0, quasi_factor)
  65             typedef cds::opt::v::random2_permutation<int>    permutation_generator;
  66         };
  67
  68          /// Metafunction converting option list to traits for SegmentedQueue
  69         /**
  70             The metafunction can be useful if a few fields in \p segmented_queue::traits should be changed.
  71             For example:
  72             \code
  73             typedef cds::container::segmented_queue::make_traits<
  74                 cds::opt::item_counter< cds::atomicity::item_counter >
  75             >::type my_segmented_queue_traits;
  76             \endcode
  77             This code creates \p %SegmentedQueue type traits with item counting feature,
  78             all other \p segmented_queue::traits members left unchanged.
  79
  80             \p Options are:
  81             - \p opt::node_allocator - node allocator.
  82             - \p opt::stat - internal statistics, possible type: \p segmented_queue::stat, \p segmented_queue::empty_stat (the default)
  83             - \p opt::item_counter - item counting feature. Note that \p atomicity::empty_item_counetr is not suitable
  84                 for segmented queue.
  85             - \p opt::memory_model - memory model, default is \p opt::v::relaxed_ordering.
  86                 See option description for the full list of possible models
  87             - \p opt::alignment - the alignment of critical data, see option description for explanation
  88             - \p opt::padding - the padding of segment data, default no special padding.
  89                 See \p traits::padding for explanation.
  90             - \p opt::allocator - the allocator used to maintain segments.
  91             - \p opt::lock_type - a mutual exclusion lock type used to maintain internal list of allocated
  92                 segments. Default is \p cds::opt::Spin, \p std::mutex is also suitable.
  93             - \p opt::permutation_generator - a random permutation generator for sequence [0, quasi_factor),
  94                 default is \p cds::opt::v::random2_permutation<int>
  95         */
  96         template <typename... Options>
  97         struct make_traits {
  98 #   ifdef CDS_DOXYGEN_INVOKED
  99             typedef implementation_defined type ;   ///< Metafunction result
 100 #   else
 101             typedef typename cds::opt::make_options<
 102                 typename cds::opt::find_type_traits< traits, Options... >::type
 103                 ,Options...
 104             >::type   type;
 105 #   endif
 106         };
 107
 108     } // namespace segmented_queue
 109
 110     //@cond
 111     namespace details {
 112
 113         template <typename GC, typename T, typename Traits>
 114         struct make_segmented_queue
 115         {
 116             typedef GC      gc;
 117             typedef T       value_type;
 118             typedef Traits  original_type_traits;
 119
 120             typedef cds::details::Allocator< T, typename original_type_traits::node_allocator > cxx_node_allocator;
 121             struct node_disposer {
 122                 void operator()( T * p )
 123                 {
 124                     cxx_node_allocator().Delete( p );
 125                 }
 126             };
 127
 128             struct intrusive_type_traits: public original_type_traits
 129             {
 130                 typedef node_disposer   disposer;
 131             };
 132
 133             typedef cds::intrusive::SegmentedQueue< gc, value_type, intrusive_type_traits > type;
 134         };
 135
 136     } // namespace details
 137     //@endcond
 138
 139     /// Segmented queue
 140     /** @ingroup cds_nonintrusive_queue
 141
 142         The queue is based on work
 143         - [2010] Afek, Korland, Yanovsky "Quasi-Linearizability: relaxed consistency for improved concurrency"
 144
 145         In this paper the authors offer a relaxed version of linearizability, so-called quasi-linearizability,
 146         that preserves some of the intuition, provides a flexible way to control the level of relaxation
 147         and supports th implementation of more concurrent and scalable data structure.
 148         Intuitively, the linearizability requires each run to be equivalent in some sense to a serial run
 149         of the algorithm. This equivalence to some serial run imposes strong synchronization requirements
 150         that in many cases results in limited scalability and synchronization bottleneck.
 151
 152         The general idea is that the queue maintains a linked list of segments, each segment is an array of
 153         nodes in the size of the quasi factor, and each node has a deleted boolean marker, which states
 154         if it has been dequeued. Each producer iterates over last segment in the linked list in some random
 155         permutation order. Whet it finds an empty cell it performs a CAS operation attempting to enqueue its
 156         new element. In case the entire segment has been scanned and no available cell is found (implying
 157         that the segment is full), then it attempts to add a new segment to the list.
 158
 159         The dequeue operation is similar: the consumer iterates over the first segment in the linked list
 160         in some random permutation order. When it finds an item which has not yet been dequeued, it performs
 161         CAS on its deleted marker in order to "delete" it, if succeeded this item is considered dequeued.
 162         In case the entire segment was scanned and all the nodes have already been dequeued (implying that
 163         the segment is empty), then it attempts to remove this segment from the linked list and starts
 164         the same process on the next segment. If there is no next segment, the queue is considered empty.
 165
 166         Based on the fact that most of the time threads do not add or remove segments, most of the work
 167         is done in parallel on different cells in the segments. This ensures a controlled contention
 168         depending on the segment size, which is quasi factor.
 169
 170         The segmented queue is an <i>unfair</i> queue since it violates the strong FIFO order but no more than
 171         quasi factor. It means that the consumer dequeues any item from the current first segment.
 172
 173         Template parameters:
 174         - \p GC - a garbage collector, possible types are cds::gc::HP, cds::gc::DHP
 175         - \p T - the type of values stored in the queue
 176         - \p Traits - queue type traits, default is \p segmented_queue::traits.
 177             \p segmented_queue::make_traits metafunction can be used to construct your
 178             type traits.
 179     */
 180     template <class GC, typename T, typename Traits = segmented_queue::traits >
 181     class SegmentedQueue:
 182 #ifdef CDS_DOXYGEN_INVOKED
 183         public cds::intrusive::SegmentedQueue< GC, T, Traits >
 184 #else
 185         public details::make_segmented_queue< GC, T, Traits >::type
 186 #endif
 187     {
 188         //@cond
 189         typedef details::make_segmented_queue< GC, T, Traits > maker;
 190         typedef typename maker::type base_class;
 191         //@endcond
 192     public:
 193         typedef GC  gc;         ///< Garbage collector
 194         typedef T   value_type; ///< type of the value stored in the queue
 195         typedef Traits traits;  ///< Queue traits
 196
 197         typedef typename traits::node_allocator node_allocator;   ///< Node allocator
 198         typedef typename base_class::memory_model  memory_model;   ///< Memory ordering. See cds::opt::memory_model option
 199         typedef typename base_class::item_counter  item_counter;   ///< Item counting policy, see cds::opt::item_counter option setter
 200         typedef typename base_class::stat          stat        ;   ///< Internal statistics policy
 201         typedef typename base_class::lock_type     lock_type   ;   ///< Type of mutex for maintaining an internal list of allocated segments.
 202         typedef typename base_class::permutation_generator permutation_generator; ///< Random permutation generator for sequence [0, quasi-factor)
 203
 204         static const size_t c_nHazardPtrCount = base_class::c_nHazardPtrCount ; ///< Count of hazard pointer required for the algorithm
 205
 206     protected:
 207         //@cond
 208         typedef typename maker::cxx_node_allocator  cxx_node_allocator;
 209         typedef std::unique_ptr< value_type, typename maker::node_disposer >  scoped_node_ptr;
 210
 211         static value_type * alloc_node( value_type const& v )
 212         {
 213             return cxx_node_allocator().New( v );
 214         }
 215
 216         static value_type * alloc_node()
 217         {
 218             return cxx_node_allocator().New();
 219         }
 220
 221         template <typename... Args>
 222         static value_type * alloc_node_move( Args&&... args )
 223         {
 224             return cxx_node_allocator().MoveNew( std::forward<Args>( args )... );
 225         }
 226         //@endcond
 227
 228     public:
 229         /// Initializes the empty queue
 230         SegmentedQueue(
 231             size_t nQuasiFactor     ///< Quasi factor. If it is not a power of 2 it is rounded up to nearest power of 2. Minimum is 2.
 232             )
 233             : base_class( nQuasiFactor )
 234         {}
 235
 236         /// Clears the queue and deletes all internal data
 237         ~SegmentedQueue()
 238         {}
 239
 240         /// Inserts a new element at last segment of the queue
 241         /**
 242             The function makes queue node in dynamic memory calling copy constructor for \p val
 243             and then it calls intrusive::SEgmentedQueue::enqueue.
 244             Returns \p true if success, \p false otherwise.
 245         */
 246         bool enqueue( value_type const& val )
 247         {
 248             scoped_node_ptr p( alloc_node(val) );
 249             if ( base_class::enqueue( *p ) ) {
 250                 p.release();
 251                 return true;
 252             }
 253             return false;
 254         }
 255
 256         /// Enqueues data to the queue using a functor
 257         /**
 258             \p Func is a functor called to create node.
 259             The functor \p f takes one argument - a reference to a new node of type \ref value_type :
 260             \code
 261             cds::container::SegmentedQueue< cds::gc::HP, Foo > myQueue;
 262             Bar bar;
 263             myQueue.enqueue_with( [&bar]( Foo& dest ) { dest = bar; } );
 264             \endcode
 265         */
 266         template <typename Func>
 267         bool enqueue_with( Func f )
 268         {
 269             scoped_node_ptr p( alloc_node() );
 270             f( *p );
 271             if ( base_class::enqueue( *p ) ) {
 272                 p.release();
 273                 return true;
 274             }
 275             return false;
 276         }
 277
 278
 279         /// Synonym for \p enqueue() member function
 280         bool push( value_type const& val )
 281         {
 282             return enqueue( val );
 283         }
 284
 285         /// Synonym for \p enqueue_with() member function
 286         template <typename Func>
 287         bool push_with( Func f )
 288         {
 289             return enqueue_with( f );
 290         }
 291
 292         /// Enqueues data of type \ref value_type constructed with <tt>std::forward<Args>(args)...</tt>
 293         template <typename... Args>
 294         bool emplace( Args&&... args )
 295         {
 296             scoped_node_ptr p( alloc_node_move( std::forward<Args>(args)... ) );
 297             if ( base_class::enqueue( *p )) {
 298                 p.release();
 299                 return true;
 300             }
 301             return false;
 302         }
 303
 304         /// Dequeues a value from the queue
 305         /**
 306             If queue is not empty, the function returns \p true, \p dest contains copy of
 307             dequeued value. The assignment operator for type \ref value_type is invoked.
 308             If queue is empty, the function returns \p false, \p dest is unchanged.
 309         */
 310         bool dequeue( value_type& dest )
 311         {
 312             return dequeue_with( [&dest]( value_type& src ) { dest = src; });
 313         }
 314
 315         /// Dequeues a value using a functor
 316         /**
 317             \p Func is a functor called to copy dequeued value.
 318             The functor takes one argument - a reference to removed node:
 319             \code
 320             cds:container::MSQueue< cds::gc::HP, Foo > myQueue;
 321             Bar bar;
 322             myQueue.dequeue_with( [&bar]( Foo& src ) { bar = std::move( src );});
 323             \endcode
 324             The functor is called only if the queue is not empty.
 325         */
 326         template <typename Func>
 327         bool dequeue_with( Func f )
 328         {
 329             value_type * p = base_class::dequeue();
 330             if ( p ) {
 331                 f( *p );
 332                 gc::template retire< typename maker::node_disposer >( p );
 333                 return true;
 334             }
 335             return false;
 336         }
 337
 338         /// Synonym for \p dequeue_with() function
 339         template <typename Func>
 340         bool pop_with( Func f )
 341         {
 342             return dequeue_with( f );
 343         }
 344
 345         /// Synonym for \p dequeue() function
 346         bool pop( value_type& dest )
 347         {
 348             return dequeue( dest );
 349         }
 350
 351         /// Checks if the queue is empty
 352         /**
 353             The original segmented queue algorithm does not allow to check emptiness accurately
 354             because \p empty() is unlinearizable.
 355             This function tests queue's emptiness checking <tt>size() == 0</tt>,
 356             so, the item counting feature is an essential part of queue's algorithm.
 357         */
 358         bool empty() const
 359         {
 360             return base_class::empty();
 361         }
 362
 363         /// Clear the queue
 364         /**
 365             The function repeatedly calls \ref dequeue until it returns \p nullptr.
 366             The disposer specified in \p Traits template argument is called for each removed item.
 367         */
 368         void clear()
 369         {
 370             base_class::clear();
 371         }
 372
 373         /// Returns queue's item count
 374         size_t size() const
 375         {
 376             return base_class::size();
 377         }
 378
 379         /// Returns reference to internal statistics
 380         /**
 381             The type of internal statistics is specified by \p Traits template argument.
 382         */
 383         const stat& statistics() const
 384         {
 385             return base_class::statistics();
 386         }
 387
 388         /// Returns quasi factor, a power-of-two number
 389         size_t quasi_factor() const
 390         {
 391             return base_class::quasi_factor();
 392         }
 393     };
 394
 395 }} // namespace cds::container
 396
 397 #endif // #ifndef CDSLIB_CONTAINER_SEGMENTED_QUEUE_H