cds/container/fcdeque.h

   1 //$$CDS-header$$
   2
   3 #ifndef CDSLIB_CONTAINER_FCDEQUE_H
   4 #define CDSLIB_CONTAINER_FCDEQUE_H
   5
   6 #include <cds/algo/flat_combining.h>
   7 #include <cds/algo/elimination_opt.h>
   8 #include <deque>
   9
  10 namespace cds { namespace container {
  11
  12     /// FCDeque related definitions
  13     /** @ingroup cds_nonintrusive_helper
  14     */
  15     namespace fcdeque {
  16
  17         /// FCDeque 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_nPushFront     ;  ///< Count of push_front operations
  25             counter_type    m_nPushFrontMove ;  ///< Count of push_front operations with move semantics
  26             counter_type    m_nPushBack      ;  ///< Count of push_back operations
  27             counter_type    m_nPushBackMove  ;  ///< Count of push_back operations with move semantics
  28             counter_type    m_nPopFront      ;  ///< Count of success pop_front operations
  29             counter_type    m_nFailedPopFront;  ///< Count of failed pop_front operations (pop from empty deque)
  30             counter_type    m_nPopBack       ;  ///< Count of success pop_back operations
  31             counter_type    m_nFailedPopBack ;  ///< Count of failed pop_back operations (pop from empty deque)
  32             counter_type    m_nCollided      ;  ///< How many pairs of push/pop were collided, if elimination is enabled
  33
  34             //@cond
  35             void    onPushFront()             { ++m_nPushFront; }
  36             void    onPushFrontMove()         { ++m_nPushFrontMove; }
  37             void    onPushBack()              { ++m_nPushBack; }
  38             void    onPushBackMove()          { ++m_nPushBackMove; }
  39             void    onPopFront( bool bFailed ) { if ( bFailed ) ++m_nFailedPopFront; else ++m_nPopFront;  }
  40             void    onPopBack( bool bFailed ) { if ( bFailed ) ++m_nFailedPopBack; else ++m_nPopBack;  }
  41             void    onCollide()               { ++m_nCollided; }
  42             //@endcond
  43         };
  44
  45         /// FCDeque dummy statistics, no overhead
  46         struct empty_stat: public cds::algo::flat_combining::empty_stat
  47         {
  48             //@cond
  49             void    onPushFront()       {}
  50             void    onPushFrontMove()   {}
  51             void    onPushBack()        {}
  52             void    onPushBackMove()    {}
  53             void    onPopFront(bool)    {}
  54             void    onPopBack(bool)     {}
  55             void    onCollide()         {}
  56             //@endcond
  57         };
  58
  59         /// FCDeque type traits
  60         struct traits: public cds::algo::flat_combining::traits
  61         {
  62             typedef empty_stat      stat;   ///< Internal statistics
  63             static CDS_CONSTEXPR const bool enable_elimination = false; ///< Enable \ref cds_elimination_description "elimination"
  64         };
  65
  66         /// Metafunction converting option list to traits
  67         /**
  68             \p Options are:
  69             - \p opt::lock_type - mutex type, default is \p cds::sync::spin
  70             - \p opt::back_off - back-off strategy, defalt is \p cds::backoff::delay_of<2>
  71             - \p opt::allocator - allocator type, default is \ref CDS_DEFAULT_ALLOCATOR
  72             - \p opt::stat - internal statistics, possible type: \ref stat, \ref empty_stat (the default)
  73             - \p opt::memory_model - C++ memory ordering model.
  74                 List of all available memory ordering see opt::memory_model.
  75                 Default if cds::opt::v:relaxed_ordering
  76             - \p opt::enable_elimination - enable/disable operation \ref cds_elimination_description "elimination"
  77                 By default, the elimination is disabled. For queue, the elimination is possible if the queue
  78                 is empty.
  79         */
  80         template <typename... Options>
  81         struct make_traits {
  82 #   ifdef CDS_DOXYGEN_INVOKED
  83             typedef implementation_defined type ;   ///< Metafunction result
  84 #   else
  85             typedef typename cds::opt::make_options<
  86                 typename cds::opt::find_type_traits< traits, Options... >::type
  87                 ,Options...
  88             >::type   type;
  89 #   endif
  90         };
  91
  92     } // namespace fcqueue
  93
  94     /// Flat-combining deque
  95     /**
  96         @ingroup cds_nonintrusive_deque
  97         @ingroup cds_flat_combining_container
  98
  99         \ref cds_flat_combining_description "Flat combining" sequential deque.
 100         The class can be considered as a concurrent FC-based wrapper for \p std::deque.
 101
 102         Template parameters:
 103         - \p T - a value type stored in the deque
 104         - \p Deque - sequential deque implementation, for example, \p std::deque<T> (the default)
 105             or \p boost::container::deque
 106         - \p Trats - type traits of flat combining, default is \p fcdeque::traits.
 107             \p fcdeque::make_traits metafunction can be used to construct specialized \p %fcdeque::traits
 108     */
 109     template <typename T,
 110         class Deque = std::deque<T>,
 111         typename Traits = fcdeque::traits
 112     >
 113     class FCDeque
 114 #ifndef CDS_DOXYGEN_INVOKED
 115         : public cds::algo::flat_combining::container
 116 #endif
 117     {
 118     public:
 119         typedef T           value_type;     ///< Value type
 120         typedef Deque       deque_type;     ///< Sequential deque class
 121         typedef Traits      traits;         ///< Deque type traits
 122
 123         typedef typename traits::stat  stat;   ///< Internal statistics type
 124         static CDS_CONSTEXPR const bool c_bEliminationEnabled = traits::enable_elimination; ///< \p true if elimination is enabled
 125
 126     protected:
 127         //@cond
 128         /// Deque operation IDs
 129         enum fc_operation {
 130             op_push_front = cds::algo::flat_combining::req_Operation, ///< Push front
 131             op_push_front_move,     ///< Push front (move semantics)
 132             op_push_back,           ///< Push back
 133             op_push_back_move,      ///< Push back (move semantics)
 134             op_pop_front,           ///< Pop front
 135             op_pop_back,            ///< Pop back
 136             op_clear                ///< Clear
 137         };
 138
 139         /// Flat combining publication list record
 140         struct fc_record: public cds::algo::flat_combining::publication_record
 141         {
 142             union {
 143                 value_type const *  pValPush; ///< Value to push
 144                 value_type *        pValPop;  ///< Pop destination
 145             };
 146             bool            bEmpty; ///< \p true if the deque is empty
 147         };
 148         //@endcond
 149
 150         /// Flat combining kernel
 151         typedef cds::algo::flat_combining::kernel< fc_record, traits > fc_kernel;
 152
 153     protected:
 154         //@cond
 155         fc_kernel   m_FlatCombining;
 156         deque_type  m_Deque;
 157         //@endcond
 158
 159     public:
 160         /// Initializes empty deque object
 161         FCDeque()
 162         {}
 163
 164         /// Initializes empty deque object and gives flat combining parameters
 165         FCDeque(
 166             unsigned int nCompactFactor     ///< Flat combining: publication list compacting factor
 167             ,unsigned int nCombinePassCount ///< Flat combining: number of combining passes for combiner thread
 168             )
 169             : m_FlatCombining( nCompactFactor, nCombinePassCount )
 170         {}
 171
 172         /// Inserts a new element at the beginning of the deque container
 173         /**
 174             The function always returns \p true
 175         */
 176         bool push_front(
 177             value_type const& val ///< Value to be copied to inserted element
 178         )
 179         {
 180             fc_record * pRec = m_FlatCombining.acquire_record();
 181             pRec->pValPush = &val;
 182
 183             if ( c_bEliminationEnabled )
 184                 m_FlatCombining.batch_combine( op_push_front, pRec, *this );
 185             else
 186                 m_FlatCombining.combine( op_push_front, pRec, *this );
 187
 188             assert( pRec->is_done() );
 189             m_FlatCombining.release_record( pRec );
 190             m_FlatCombining.internal_statistics().onPushFront();
 191             return true;
 192         }
 193
 194         /// Inserts a new element at the beginning of the deque container (move semantics)
 195         /**
 196             The function always returns \p true
 197         */
 198         bool push_front(
 199             value_type&& val ///< Value to be moved to inserted element
 200         )
 201         {
 202             fc_record * pRec = m_FlatCombining.acquire_record();
 203             pRec->pValPush = &val;
 204
 205             if ( c_bEliminationEnabled )
 206                 m_FlatCombining.batch_combine( op_push_front_move, pRec, *this );
 207             else
 208                 m_FlatCombining.combine( op_push_front_move, pRec, *this );
 209
 210             assert( pRec->is_done() );
 211             m_FlatCombining.release_record( pRec );
 212             m_FlatCombining.internal_statistics().onPushFrontMove();
 213             return true;
 214         }
 215
 216         /// Inserts a new element at the end of the deque container
 217         /**
 218             The function always returns \p true
 219         */
 220         bool push_back(
 221             value_type const& val ///< Value to be copied to inserted element
 222         )
 223         {
 224             fc_record * pRec = m_FlatCombining.acquire_record();
 225             pRec->pValPush = &val;
 226
 227             if ( c_bEliminationEnabled )
 228                 m_FlatCombining.batch_combine( op_push_back, pRec, *this );
 229             else
 230                 m_FlatCombining.combine( op_push_back, pRec, *this );
 231
 232             assert( pRec->is_done() );
 233             m_FlatCombining.release_record( pRec );
 234             m_FlatCombining.internal_statistics().onPushBack();
 235             return true;
 236         }
 237
 238         /// Inserts a new element at the end of the deque container (move semantics)
 239         /**
 240             The function always returns \p true
 241         */
 242         bool push_back(
 243             value_type&& val ///< Value to be moved to inserted element
 244         )
 245         {
 246             fc_record * pRec = m_FlatCombining.acquire_record();
 247             pRec->pValPush = &val;
 248
 249             if ( c_bEliminationEnabled )
 250                 m_FlatCombining.batch_combine( op_push_back_move, pRec, *this );
 251             else
 252                 m_FlatCombining.combine( op_push_back_move, pRec, *this );
 253
 254             assert( pRec->is_done() );
 255             m_FlatCombining.release_record( pRec );
 256             m_FlatCombining.internal_statistics().onPushBackMove();
 257             return true;
 258         }
 259
 260         /// Removes the first element in the deque container
 261         /**
 262             The function returns \p false if the deque is empty, \p true otherwise.
 263             If the deque is empty \p val is not changed.
 264         */
 265         bool pop_front(
 266             value_type& val ///< Target to be received the copy of removed element
 267         )
 268         {
 269             fc_record * pRec = m_FlatCombining.acquire_record();
 270             pRec->pValPop = &val;
 271
 272             if ( c_bEliminationEnabled )
 273                 m_FlatCombining.batch_combine( op_pop_front, pRec, *this );
 274             else
 275                 m_FlatCombining.combine( op_pop_front, pRec, *this );
 276
 277             assert( pRec->is_done() );
 278             m_FlatCombining.release_record( pRec );
 279             m_FlatCombining.internal_statistics().onPopFront( pRec->bEmpty );
 280             return !pRec->bEmpty;
 281         }
 282
 283         /// Removes the last element in the deque container
 284         /**
 285             The function returns \p false if the deque is empty, \p true otherwise.
 286             If the deque is empty \p val is not changed.
 287         */
 288         bool pop_back(
 289             value_type& val ///< Target to be received the copy of removed element
 290         )
 291         {
 292             fc_record * pRec = m_FlatCombining.acquire_record();
 293             pRec->pValPop = &val;
 294
 295             if ( c_bEliminationEnabled )
 296                 m_FlatCombining.batch_combine( op_pop_back, pRec, *this );
 297             else
 298                 m_FlatCombining.combine( op_pop_back, pRec, *this );
 299
 300             assert( pRec->is_done() );
 301             m_FlatCombining.release_record( pRec );
 302             m_FlatCombining.internal_statistics().onPopBack( pRec->bEmpty );
 303             return !pRec->bEmpty;
 304         }
 305
 306         /// Clears the deque
 307         void clear()
 308         {
 309             fc_record * pRec = m_FlatCombining.acquire_record();
 310
 311             if ( c_bEliminationEnabled )
 312                 m_FlatCombining.batch_combine( op_clear, pRec, *this );
 313             else
 314                 m_FlatCombining.combine( op_clear, pRec, *this );
 315
 316             assert( pRec->is_done() );
 317             m_FlatCombining.release_record( pRec );
 318         }
 319
 320         /// Returns the number of elements in the deque.
 321         /**
 322             Note that <tt>size() == 0</tt> is not mean that the deque is empty because
 323             combining record can be in process.
 324             To check emptiness use \ref empty function.
 325         */
 326         size_t size() const
 327         {
 328             return m_Deque.size();
 329         }
 330
 331         /// Checks if the deque is empty
 332         /**
 333             If the combining is in process the function waits while combining done.
 334         */
 335         bool empty() const
 336         {
 337             m_FlatCombining.wait_while_combining();
 338             return m_Deque.empty();
 339         }
 340
 341         /// Internal statistics
 342         stat const& statistics() const
 343         {
 344             return m_FlatCombining.statistics();
 345         }
 346
 347     public: // flat combining cooperation, not for direct use!
 348         //@cond
 349         /// Flat combining supporting function. Do not call it directly!
 350         /**
 351             The function is called by \ref cds::algo::flat_combining::kernel "flat combining kernel"
 352             object if the current thread becomes a combiner. Invocation of the function means that
 353             the deque should perform an action recorded in \p pRec.
 354         */
 355         void fc_apply( fc_record * pRec )
 356         {
 357             assert( pRec );
 358
 359             // this function is called under FC mutex, so switch TSan off
 360             CDS_TSAN_ANNOTATE_IGNORE_RW_BEGIN;
 361
 362             switch ( pRec->op() ) {
 363             case op_push_front:
 364                 assert( pRec->pValPush );
 365                 m_Deque.push_front( *(pRec->pValPush) );
 366                 break;
 367             case op_push_front_move:
 368                 assert( pRec->pValPush );
 369                 m_Deque.push_front( std::move( *(pRec->pValPush )) );
 370                 break;
 371             case op_push_back:
 372                 assert( pRec->pValPush );
 373                 m_Deque.push_back( *(pRec->pValPush) );
 374                 break;
 375             case op_push_back_move:
 376                 assert( pRec->pValPush );
 377                 m_Deque.push_back( std::move( *(pRec->pValPush )) );
 378                 break;
 379             case op_pop_front:
 380                 assert( pRec->pValPop );
 381                 pRec->bEmpty = m_Deque.empty();
 382                 if ( !pRec->bEmpty ) {
 383                     *(pRec->pValPop) = m_Deque.front();
 384                     m_Deque.pop_front();
 385                 }
 386                 break;
 387             case op_pop_back:
 388                 assert( pRec->pValPop );
 389                 pRec->bEmpty = m_Deque.empty();
 390                 if ( !pRec->bEmpty ) {
 391                     *(pRec->pValPop) = m_Deque.back();
 392                     m_Deque.pop_back();
 393                 }
 394                 break;
 395             case op_clear:
 396                 while ( !m_Deque.empty() )
 397                     m_Deque.pop_front();
 398                 break;
 399             default:
 400                 assert(false);
 401                 break;
 402             }
 403             CDS_TSAN_ANNOTATE_IGNORE_RW_END;
 404         }
 405
 406         /// Batch-processing flat combining
 407         void fc_process( typename fc_kernel::iterator itBegin, typename fc_kernel::iterator itEnd )
 408         {
 409             typedef typename fc_kernel::iterator fc_iterator;
 410
 411             // this function is called under FC mutex, so switch TSan off
 412             CDS_TSAN_ANNOTATE_IGNORE_RW_BEGIN;
 413
 414             for ( fc_iterator it = itBegin, itPrev = itEnd; it != itEnd; ++it ) {
 415                 switch ( it->op() ) {
 416                 case op_push_front:
 417                 case op_push_front_move:
 418                     if ( itPrev != itEnd
 419                       && (itPrev->op() == op_pop_front || ( m_Deque.empty() && itPrev->op() == op_pop_back )))
 420                     {
 421                         collide( *it, *itPrev );
 422                         itPrev = itEnd;
 423                     }
 424                     else
 425                         itPrev = it;
 426                     break;
 427                 case op_push_back:
 428                 case op_push_back_move:
 429                     if ( itPrev != itEnd
 430                         && (itPrev->op() == op_pop_back || ( m_Deque.empty() && itPrev->op() == op_pop_front )))
 431                     {
 432                         collide( *it, *itPrev );
 433                         itPrev = itEnd;
 434                     }
 435                     else
 436                         itPrev = it;
 437                     break;
 438                 case op_pop_front:
 439                     if ( itPrev != itEnd
 440                         && ( itPrev->op() == op_push_front || itPrev->op() == op_push_front_move
 441                           || ( m_Deque.empty() && ( itPrev->op() == op_push_back || itPrev->op() == op_push_back_move ))))
 442                     {
 443                         collide( *itPrev, *it );
 444                         itPrev = itEnd;
 445                     }
 446                     else
 447                         itPrev = it;
 448                     break;
 449                 case op_pop_back:
 450                     if ( itPrev != itEnd
 451                         && ( itPrev->op() == op_push_back || itPrev->op() == op_push_back_move
 452                         || ( m_Deque.empty() && ( itPrev->op() == op_push_front || itPrev->op() == op_push_front_move ))))
 453                     {
 454                         collide( *itPrev, *it );
 455                         itPrev = itEnd;
 456                     }
 457                     else
 458                         itPrev = it;
 459                     break;
 460                 }
 461             }
 462             CDS_TSAN_ANNOTATE_IGNORE_RW_END;
 463         }
 464         //@endcond
 465
 466     private:
 467         //@cond
 468         void collide( fc_record& recPush, fc_record& recPop )
 469         {
 470             *(recPop.pValPop) = *(recPush.pValPush);
 471             recPop.bEmpty = false;
 472             m_FlatCombining.operation_done( recPush );
 473             m_FlatCombining.operation_done( recPop );
 474             m_FlatCombining.internal_statistics().onCollide();
 475         }
 476         //@endcond
 477     };
 478
 479 }} // namespace cds::container
 480
 481 #endif // #ifndef CDSLIB_CONTAINER_FCDEQUE_H