cds/container/fcpriority_queue.h

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