cds/container/fcpriority_queue.h

   1 //$$CDS-header$$
   2
   3 #ifndef __CDS_CONTAINER_FCPRIORITY_QUEUE_H
   4 #define __CDS_CONTAINER_FCPRIORITY_QUEUE_H
   5
   6 #include <cds/algo/flat_combining.h>
   7 #include <cds/algo/elimination_opt.h>
   8 #include <queue>
   9
  10 namespace cds { namespace container {
  11
  12     /// FCPriorityQueue related definitions
  13     /** @ingroup cds_nonintrusive_helper
  14     */
  15     namespace fcpqueue {
  16
  17         /// FCPriorityQueue internal statistics
  18         template <typename Counter = cds::atomicity::event_counter >
  19         struct stat: public cds::algo::flat_combining::stat<Counter>
  20         {
  21             typedef cds::algo::flat_combining::stat<Counter>    flat_combining_stat; ///< Flat-combining statistics
  22             typedef typename flat_combining_stat::counter_type  counter_type;        ///< Counter type
  23
  24             counter_type    m_nPush     ;  ///< Count of push operations
  25             counter_type    m_nPushMove ;  ///< Count of push operations with move semantics
  26             counter_type    m_nPop      ;  ///< Count of success pop operations
  27             counter_type    m_nFailedPop;  ///< Count of failed pop operations (pop from empty queue)
  28
  29             //@cond
  30             void    onPush()             { ++m_nPush; }
  31             void    onPushMove()         { ++m_nPushMove; }
  32             void    onPop( bool bFailed ) { if ( bFailed ) ++m_nFailedPop; else ++m_nPop;  }
  33             //@endcond
  34         };
  35
  36         /// FCPriorityQueue dummy statistics, no overhead
  37         struct empty_stat: public cds::algo::flat_combining::empty_stat
  38         {
  39             //@cond
  40             void    onPush()       {}
  41             void    onPushMove()   {}
  42             void    onPop(bool)    {}
  43             //@endcond
  44         };
  45
  46         /// FCPriorityQueue type traits
  47         struct type_traits: public cds::algo::flat_combining::type_traits
  48         {
  49             typedef empty_stat      stat;   ///< Internal statistics
  50         };
  51
  52         /// Metafunction converting option list to traits
  53         /**
  54             This is a wrapper for <tt> cds::opt::make_options< type_traits, Options...> </tt>
  55             \p Options are:
  56             - \p opt::lock_type - mutex type, default is \p cds::lock::Spin
  57             - \p opt::back_off - back-off strategy, defalt is \p cds::backoff::Default
  58             - \p opt::allocator - allocator type, default is \ref CDS_DEFAULT_ALLOCATOR
  59             - \p opt::stat - internal statistics, possible type: \ref stat, \ref empty_stat (the default)
  60             - \p opt::memory_model - C++ memory ordering model.
  61                 List of all available memory ordering see opt::memory_model.
  62                 Default is cds::opt::v:relaxed_ordering
  63         */
  64         template <CDS_DECL_OPTIONS7>
  65         struct make_traits {
  66 #   ifdef CDS_DOXYGEN_INVOKED
  67             typedef implementation_defined type ;   ///< Metafunction result
  68 #   else
  69             typedef typename cds::opt::make_options<
  70                 typename cds::opt::find_type_traits< type_traits, CDS_OPTIONS7 >::type
  71                 ,CDS_OPTIONS7
  72             >::type   type;
  73 #   endif
  74         };
  75
  76     } // namespace fcpqueue
  77
  78     /// Flat-combining priority queue
  79     /**
  80         @ingroup cds_nonintrusive_priority_queue
  81         @ingroup cds_flat_combining_container
  82
  83         \ref cds_flat_combining_description "Flat combining" sequential priority queue.
  84         The class can be considered as a concurrent FC-based wrapper for \p std::priority_queue.
  85
  86         Template parameters:
  87         - \p T - a value type stored in the queue
  88         - \p PriorityQueue - sequential priority queue implementation, default is \p std::priority_queue<T>
  89         - \p Traits - type traits of flat combining, default is \p fcpqueue::type_traits.
  90             \p fcpqueue::make_traits metafunction can be used to construct specialized \p %type_traits
  91     */
  92     template <typename T,
  93         class PriorityQueue = std::priority_queue<T>,
  94         typename Traits = fcpqueue::type_traits
  95     >
  96     class FCPriorityQueue
  97 #ifndef CDS_DOXYGEN_INVOKED
  98         : public cds::algo::flat_combining::container
  99 #endif
 100     {
 101     public:
 102         typedef T               value_type;          ///< Value type
 103         typedef PriorityQueue   priority_queue_type; ///< Sequential priority queue class
 104         typedef Traits          type_traits;         ///< Priority queue type traits
 105
 106         typedef typename type_traits::stat  stat;    ///< Internal statistics type
 107
 108     protected:
 109         //@cond
 110         // Priority queue operation IDs
 111         enum fc_operation {
 112             op_push = cds::algo::flat_combining::req_Operation,
 113             op_push_move,
 114             op_pop,
 115             op_clear
 116         };
 117
 118         // Flat combining publication list record
 119         struct fc_record: public cds::algo::flat_combining::publication_record
 120         {
 121             union {
 122                 value_type const *  pValPush; // Value to push
 123                 value_type *        pValPop;  // Pop destination
 124             };
 125             bool            bEmpty; // true if the queue is empty
 126         };
 127         //@endcond
 128
 129         /// Flat combining kernel
 130         typedef cds::algo::flat_combining::kernel< fc_record, type_traits > fc_kernel;
 131
 132     protected:
 133         //@cond
 134         fc_kernel               m_FlatCombining;
 135         priority_queue_type     m_PQueue;
 136         //@endcond
 137
 138     public:
 139         /// Initializes empty priority queue object
 140         FCPriorityQueue()
 141         {}
 142
 143         /// Initializes empty priority queue object and gives flat combining parameters
 144         FCPriorityQueue(
 145             unsigned int nCompactFactor     ///< Flat combining: publication list compacting factor
 146             ,unsigned int nCombinePassCount ///< Flat combining: number of combining passes for combiner thread
 147             )
 148             : m_FlatCombining( nCompactFactor, nCombinePassCount )
 149         {}
 150
 151         /// Inserts a new element in the priority queue
 152         /**
 153             The function always returns \p true
 154         */
 155         bool push(
 156             value_type const& val ///< Value to be copied to inserted element
 157         )
 158         {
 159             fc_record * pRec = m_FlatCombining.acquire_record();
 160             pRec->pValPush = &val;
 161
 162             m_FlatCombining.combine( op_push, pRec, *this );
 163
 164             assert( pRec->is_done() );
 165             m_FlatCombining.release_record( pRec );
 166             m_FlatCombining.internal_statistics().onPush();
 167             return true;
 168         }
 169
 170 #   ifdef CDS_MOVE_SEMANTICS_SUPPORT
 171         /// Inserts a new element in the priority queue (move semantics)
 172         /**
 173             The function always returns \p true
 174         */
 175         bool push(
 176             value_type&& val ///< Value to be moved to inserted element
 177         )
 178         {
 179             fc_record * pRec = m_FlatCombining.acquire_record();
 180             pRec->pValPush = &val;
 181
 182             m_FlatCombining.combine( op_push_move, pRec, *this );
 183
 184             assert( pRec->is_done() );
 185             m_FlatCombining.release_record( pRec );
 186             m_FlatCombining.internal_statistics().onPushMove();
 187             return true;
 188         }
 189 #   endif
 190
 191         /// Removes the top element from priority queue
 192         /**
 193             The function returns \p false if the queue is empty, \p true otherwise.
 194             If the queue is empty \p val is not changed.
 195         */
 196         bool pop(
 197             value_type& val ///< Target to be received the copy of top element
 198         )
 199         {
 200             fc_record * pRec = m_FlatCombining.acquire_record();
 201             pRec->pValPop = &val;
 202
 203             m_FlatCombining.combine( op_pop, pRec, *this );
 204
 205             assert( pRec->is_done() );
 206             m_FlatCombining.release_record( pRec );
 207             m_FlatCombining.internal_statistics().onPop( pRec->bEmpty );
 208             return !pRec->bEmpty;
 209         }
 210
 211         /// Clears the priority queue
 212         void clear()
 213         {
 214             fc_record * pRec = m_FlatCombining.acquire_record();
 215
 216            m_FlatCombining.combine( op_clear, pRec, *this );
 217
 218             assert( pRec->is_done() );
 219             m_FlatCombining.release_record( pRec );
 220         }
 221
 222         /// Returns the number of elements in the priority queue.
 223         /**
 224             Note that <tt>size() == 0</tt> does not mean that the queue is empty because
 225             combining record can be in process.
 226             To check emptiness use \ref empty function.
 227         */
 228         size_t size() const
 229         {
 230             return m_PQueue.size();
 231         }
 232
 233         /// Checks if the priority queue is empty
 234         /**
 235             If the combining is in process the function waits while combining done.
 236         */
 237         bool empty() const
 238         {
 239             m_FlatCombining.wait_while_combining();
 240             return m_PQueue.empty();
 241         }
 242
 243         /// Internal statistics
 244         stat const& statistics() const
 245         {
 246             return m_FlatCombining.statistics();
 247         }
 248
 249     public: // flat combining cooperation, not for direct use!
 250         //@cond
 251         /*
 252             The function is called by \ref cds::algo::flat_combining::kernel "flat combining kernel"
 253             object if the current thread becomes a combiner. Invocation of the function means that
 254             the priority queue should perform an action recorded in \p pRec.
 255         */
 256         void fc_apply( fc_record * pRec )
 257         {
 258             assert( pRec );
 259
 260             switch ( pRec->op() ) {
 261             case op_push:
 262                 assert( pRec->pValPush );
 263                 m_PQueue.push( *(pRec->pValPush) );
 264                 break;
 265 #       ifdef CDS_MOVE_SEMANTICS_SUPPORT
 266             case op_push_move:
 267                 assert( pRec->pValPush );
 268                 m_PQueue.push( std::move( *(pRec->pValPush )) );
 269                 break;
 270 #       endif
 271             case op_pop:
 272                 assert( pRec->pValPop );
 273                 pRec->bEmpty = m_PQueue.empty();
 274                 if ( !pRec->bEmpty ) {
 275                     *(pRec->pValPop) = m_PQueue.top();
 276                     m_PQueue.pop();
 277                 }
 278                 break;
 279             case op_clear:
 280                 while ( !m_PQueue.empty() )
 281                     m_PQueue.pop();
 282                 break;
 283             default:
 284                 assert(false);
 285                 break;
 286             }
 287         }
 288         //@endcond
 289     };
 290
 291 }} // namespace cds::container
 292
 293 #endif // #ifndef __CDS_CONTAINER_FCPRIORITY_QUEUE_H