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         /// Inserts a new element in the priority queue (move semantics)
 171         /**
 172             The function always returns \p true
 173         */
 174         bool push(
 175             value_type&& val ///< Value to be moved to inserted element
 176         )
 177         {
 178             fc_record * pRec = m_FlatCombining.acquire_record();
 179             pRec->pValPush = &val;
 180
 181             m_FlatCombining.combine( op_push_move, pRec, *this );
 182
 183             assert( pRec->is_done() );
 184             m_FlatCombining.release_record( pRec );
 185             m_FlatCombining.internal_statistics().onPushMove();
 186             return true;
 187         }
 188
 189         /// Removes the top element from priority queue
 190         /**
 191             The function returns \p false if the queue is empty, \p true otherwise.
 192             If the queue is empty \p val is not changed.
 193         */
 194         bool pop(
 195             value_type& val ///< Target to be received the copy of top element
 196         )
 197         {
 198             fc_record * pRec = m_FlatCombining.acquire_record();
 199             pRec->pValPop = &val;
 200
 201             m_FlatCombining.combine( op_pop, pRec, *this );
 202
 203             assert( pRec->is_done() );
 204             m_FlatCombining.release_record( pRec );
 205             m_FlatCombining.internal_statistics().onPop( pRec->bEmpty );
 206             return !pRec->bEmpty;
 207         }
 208
 209         /// Clears the priority queue
 210         void clear()
 211         {
 212             fc_record * pRec = m_FlatCombining.acquire_record();
 213
 214            m_FlatCombining.combine( op_clear, pRec, *this );
 215
 216             assert( pRec->is_done() );
 217             m_FlatCombining.release_record( pRec );
 218         }
 219
 220         /// Returns the number of elements in the priority queue.
 221         /**
 222             Note that <tt>size() == 0</tt> does not mean that the queue is empty because
 223             combining record can be in process.
 224             To check emptiness use \ref empty function.
 225         */
 226         size_t size() const
 227         {
 228             return m_PQueue.size();
 229         }
 230
 231         /// Checks if the priority queue is empty
 232         /**
 233             If the combining is in process the function waits while combining done.
 234         */
 235         bool empty() const
 236         {
 237             m_FlatCombining.wait_while_combining();
 238             return m_PQueue.empty();
 239         }
 240
 241         /// Internal statistics
 242         stat const& statistics() const
 243         {
 244             return m_FlatCombining.statistics();
 245         }
 246
 247     public: // flat combining cooperation, not for direct use!
 248         //@cond
 249         /*
 250             The function is called by \ref cds::algo::flat_combining::kernel "flat combining kernel"
 251             object if the current thread becomes a combiner. Invocation of the function means that
 252             the priority queue should perform an action recorded in \p pRec.
 253         */
 254         void fc_apply( fc_record * pRec )
 255         {
 256             assert( pRec );
 257
 258             switch ( pRec->op() ) {
 259             case op_push:
 260                 assert( pRec->pValPush );
 261                 m_PQueue.push( *(pRec->pValPush) );
 262                 break;
 263             case op_push_move:
 264                 assert( pRec->pValPush );
 265                 m_PQueue.push( std::move( *(pRec->pValPush )) );
 266                 break;
 267             case op_pop:
 268                 assert( pRec->pValPop );
 269                 pRec->bEmpty = m_PQueue.empty();
 270                 if ( !pRec->bEmpty ) {
 271                     *(pRec->pValPop) = m_PQueue.top();
 272                     m_PQueue.pop();
 273                 }
 274                 break;
 275             case op_clear:
 276                 while ( !m_PQueue.empty() )
 277                     m_PQueue.pop();
 278                 break;
 279             default:
 280                 assert(false);
 281                 break;
 282             }
 283         }
 284         //@endcond
 285     };
 286
 287 }} // namespace cds::container
 288
 289 #endif // #ifndef __CDS_CONTAINER_FCPRIORITY_QUEUE_H