cds/container/fcdeque.h

   1 /*
   2     This file is a part of libcds - Concurrent Data Structures library
   3
   4     (C) Copyright Maxim Khizhinsky (libcds.dev@gmail.com) 2006-2016
   5
   6     Source code repo: http://github.com/khizmax/libcds/
   7     Download: http://sourceforge.net/projects/libcds/files/
   8
   9     Redistribution and use in source and binary forms, with or without
  10     modification, are permitted provided that the following conditions are met:
  11
  12     * Redistributions of source code must retain the above copyright notice, this
  13       list of conditions and the following disclaimer.
  14
  15     * Redistributions in binary form must reproduce the above copyright notice,
  16       this list of conditions and the following disclaimer in the documentation
  17       and/or other materials provided with the distribution.
  18
  19     THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
  20     AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
  21     IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
  22     DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE
  23     FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
  24     DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
  25     SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
  26     CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
  27     OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
  28     OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
  29 */
  30
  31 #ifndef CDSLIB_CONTAINER_FCDEQUE_H
  32 #define CDSLIB_CONTAINER_FCDEQUE_H
  33
  34 #include <cds/algo/flat_combining.h>
  35 #include <cds/algo/elimination_opt.h>
  36 #include <deque>
  37
  38 namespace cds { namespace container {
  39
  40     /// FCDeque related definitions
  41     /** @ingroup cds_nonintrusive_helper
  42     */
  43     namespace fcdeque {
  44
  45         /// FCDeque internal statistics
  46         template <typename Counter = cds::atomicity::event_counter >
  47         struct stat: public cds::algo::flat_combining::stat<Counter>
  48         {
  49             typedef cds::algo::flat_combining::stat<Counter>    flat_combining_stat; ///< Flat-combining statistics
  50             typedef typename flat_combining_stat::counter_type  counter_type;        ///< Counter type
  51
  52             counter_type    m_nPushFront     ;  ///< Count of push_front operations
  53             counter_type    m_nPushFrontMove ;  ///< Count of push_front operations with move semantics
  54             counter_type    m_nPushBack      ;  ///< Count of push_back operations
  55             counter_type    m_nPushBackMove  ;  ///< Count of push_back operations with move semantics
  56             counter_type    m_nPopFront      ;  ///< Count of success pop_front operations
  57             counter_type    m_nFailedPopFront;  ///< Count of failed pop_front operations (pop from empty deque)
  58             counter_type    m_nPopBack       ;  ///< Count of success pop_back operations
  59             counter_type    m_nFailedPopBack ;  ///< Count of failed pop_back operations (pop from empty deque)
  60             counter_type    m_nCollided      ;  ///< How many pairs of push/pop were collided, if elimination is enabled
  61
  62             //@cond
  63             void    onPushFront()             { ++m_nPushFront; }
  64             void    onPushFrontMove()         { ++m_nPushFrontMove; }
  65             void    onPushBack()              { ++m_nPushBack; }
  66             void    onPushBackMove()          { ++m_nPushBackMove; }
  67             void    onPopFront( bool bFailed ) { if ( bFailed ) ++m_nFailedPopFront; else ++m_nPopFront;  }
  68             void    onPopBack( bool bFailed ) { if ( bFailed ) ++m_nFailedPopBack; else ++m_nPopBack;  }
  69             void    onCollide()               { ++m_nCollided; }
  70             //@endcond
  71         };
  72
  73         /// FCDeque dummy statistics, no overhead
  74         struct empty_stat: public cds::algo::flat_combining::empty_stat
  75         {
  76             //@cond
  77             void    onPushFront()       {}
  78             void    onPushFrontMove()   {}
  79             void    onPushBack()        {}
  80             void    onPushBackMove()    {}
  81             void    onPopFront(bool)    {}
  82             void    onPopBack(bool)     {}
  83             void    onCollide()         {}
  84             //@endcond
  85         };
  86
  87         /// FCDeque type traits
  88         struct traits: public cds::algo::flat_combining::traits
  89         {
  90             typedef empty_stat      stat;   ///< Internal statistics
  91             static CDS_CONSTEXPR const bool enable_elimination = false; ///< Enable \ref cds_elimination_description "elimination"
  92         };
  93
  94         /// Metafunction converting option list to traits
  95         /**
  96             \p Options are:
  97             - \p opt::lock_type - mutex type, default is \p cds::sync::spin
  98             - \p opt::back_off - back-off strategy, defalt is \p cds::backoff::delay_of<2>
  99             - \p opt::allocator - allocator type, default is \ref CDS_DEFAULT_ALLOCATOR
 100             - \p opt::stat - internal statistics, possible type: \ref stat, \ref empty_stat (the default)
 101             - \p opt::memory_model - C++ memory ordering model.
 102                 List of all available memory ordering see opt::memory_model.
 103                 Default if cds::opt::v:relaxed_ordering
 104             - \p opt::enable_elimination - enable/disable operation \ref cds_elimination_description "elimination"
 105                 By default, the elimination is disabled. For queue, the elimination is possible if the queue
 106                 is empty.
 107         */
 108         template <typename... Options>
 109         struct make_traits {
 110 #   ifdef CDS_DOXYGEN_INVOKED
 111             typedef implementation_defined type ;   ///< Metafunction result
 112 #   else
 113             typedef typename cds::opt::make_options<
 114                 typename cds::opt::find_type_traits< traits, Options... >::type
 115                 ,Options...
 116             >::type   type;
 117 #   endif
 118         };
 119
 120     } // namespace fcqueue
 121
 122     /// Flat-combining deque
 123     /**
 124         @ingroup cds_nonintrusive_deque
 125         @ingroup cds_flat_combining_container
 126
 127         \ref cds_flat_combining_description "Flat combining" sequential deque.
 128         The class can be considered as a concurrent FC-based wrapper for \p std::deque.
 129
 130         Template parameters:
 131         - \p T - a value type stored in the deque
 132         - \p Deque - sequential deque implementation, for example, \p std::deque<T> (the default)
 133             or \p boost::container::deque
 134         - \p Trats - type traits of flat combining, default is \p fcdeque::traits.
 135             \p fcdeque::make_traits metafunction can be used to construct specialized \p %fcdeque::traits
 136     */
 137     template <typename T,
 138         class Deque = std::deque<T>,
 139         typename Traits = fcdeque::traits
 140     >
 141     class FCDeque
 142 #ifndef CDS_DOXYGEN_INVOKED
 143         : public cds::algo::flat_combining::container
 144 #endif
 145     {
 146     public:
 147         typedef T           value_type;     ///< Value type
 148         typedef Deque       deque_type;     ///< Sequential deque class
 149         typedef Traits      traits;         ///< Deque type traits
 150
 151         typedef typename traits::stat  stat;   ///< Internal statistics type
 152         static CDS_CONSTEXPR const bool c_bEliminationEnabled = traits::enable_elimination; ///< \p true if elimination is enabled
 153
 154     protected:
 155         //@cond
 156         /// Deque operation IDs
 157         enum fc_operation {
 158             op_push_front = cds::algo::flat_combining::req_Operation, ///< Push front
 159             op_push_front_move,     ///< Push front (move semantics)
 160             op_push_back,           ///< Push back
 161             op_push_back_move,      ///< Push back (move semantics)
 162             op_pop_front,           ///< Pop front
 163             op_pop_back,            ///< Pop back
 164             op_clear,               ///< Clear
 165             op_empty                ///< Empty
 166         };
 167
 168         /// Flat combining publication list record
 169         struct fc_record: public cds::algo::flat_combining::publication_record
 170         {
 171             union {
 172                 value_type const *  pValPush; ///< Value to push
 173                 value_type *        pValPop;  ///< Pop destination
 174             };
 175             bool            bEmpty; ///< \p true if the deque is empty
 176         };
 177         //@endcond
 178
 179         /// Flat combining kernel
 180         typedef cds::algo::flat_combining::kernel< fc_record, traits > fc_kernel;
 181
 182     protected:
 183         //@cond
 184         fc_kernel   m_FlatCombining;
 185         deque_type  m_Deque;
 186         //@endcond
 187
 188     public:
 189         /// Initializes empty deque object
 190         FCDeque()
 191         {}
 192
 193         /// Initializes empty deque object and gives flat combining parameters
 194         FCDeque(
 195             unsigned int nCompactFactor     ///< Flat combining: publication list compacting factor
 196             ,unsigned int nCombinePassCount ///< Flat combining: number of combining passes for combiner thread
 197             )
 198             : m_FlatCombining( nCompactFactor, nCombinePassCount )
 199         {}
 200
 201         /// Inserts a new element at the beginning of the deque container
 202         /**
 203             The function always returns \p true
 204         */
 205         bool push_front(
 206             value_type const& val ///< Value to be copied to inserted element
 207         )
 208         {
 209             fc_record * pRec = m_FlatCombining.acquire_record();
 210             pRec->pValPush = &val;
 211
 212             if ( c_bEliminationEnabled )
 213                 m_FlatCombining.batch_combine( op_push_front, pRec, *this );
 214             else
 215                 m_FlatCombining.combine( op_push_front, pRec, *this );
 216
 217             assert( pRec->is_done() );
 218             m_FlatCombining.release_record( pRec );
 219             m_FlatCombining.internal_statistics().onPushFront();
 220             return true;
 221         }
 222
 223         /// Inserts a new element at the beginning of the deque container (move semantics)
 224         /**
 225             The function always returns \p true
 226         */
 227         bool push_front(
 228             value_type&& val ///< Value to be moved to inserted element
 229         )
 230         {
 231             fc_record * pRec = m_FlatCombining.acquire_record();
 232             pRec->pValPush = &val;
 233
 234             if ( c_bEliminationEnabled )
 235                 m_FlatCombining.batch_combine( op_push_front_move, pRec, *this );
 236             else
 237                 m_FlatCombining.combine( op_push_front_move, pRec, *this );
 238
 239             assert( pRec->is_done() );
 240             m_FlatCombining.release_record( pRec );
 241             m_FlatCombining.internal_statistics().onPushFrontMove();
 242             return true;
 243         }
 244
 245         /// Inserts a new element at the end of the deque container
 246         /**
 247             The function always returns \p true
 248         */
 249         bool push_back(
 250             value_type const& val ///< Value to be copied to inserted element
 251         )
 252         {
 253             fc_record * pRec = m_FlatCombining.acquire_record();
 254             pRec->pValPush = &val;
 255
 256             if ( c_bEliminationEnabled )
 257                 m_FlatCombining.batch_combine( op_push_back, pRec, *this );
 258             else
 259                 m_FlatCombining.combine( op_push_back, pRec, *this );
 260
 261             assert( pRec->is_done() );
 262             m_FlatCombining.release_record( pRec );
 263             m_FlatCombining.internal_statistics().onPushBack();
 264             return true;
 265         }
 266
 267         /// Inserts a new element at the end of the deque container (move semantics)
 268         /**
 269             The function always returns \p true
 270         */
 271         bool push_back(
 272             value_type&& val ///< Value to be moved to inserted element
 273         )
 274         {
 275             fc_record * pRec = m_FlatCombining.acquire_record();
 276             pRec->pValPush = &val;
 277
 278             if ( c_bEliminationEnabled )
 279                 m_FlatCombining.batch_combine( op_push_back_move, pRec, *this );
 280             else
 281                 m_FlatCombining.combine( op_push_back_move, pRec, *this );
 282
 283             assert( pRec->is_done() );
 284             m_FlatCombining.release_record( pRec );
 285             m_FlatCombining.internal_statistics().onPushBackMove();
 286             return true;
 287         }
 288
 289         /// Removes the first element in the deque container
 290         /**
 291             The function returns \p false if the deque is empty, \p true otherwise.
 292             If the deque is empty \p val is not changed.
 293         */
 294         bool pop_front(
 295             value_type& val ///< Target to be received the copy of removed element
 296         )
 297         {
 298             fc_record * pRec = m_FlatCombining.acquire_record();
 299             pRec->pValPop = &val;
 300
 301             if ( c_bEliminationEnabled )
 302                 m_FlatCombining.batch_combine( op_pop_front, pRec, *this );
 303             else
 304                 m_FlatCombining.combine( op_pop_front, pRec, *this );
 305
 306             assert( pRec->is_done() );
 307             m_FlatCombining.release_record( pRec );
 308             m_FlatCombining.internal_statistics().onPopFront( pRec->bEmpty );
 309             return !pRec->bEmpty;
 310         }
 311
 312         /// Removes the last element in the deque container
 313         /**
 314             The function returns \p false if the deque is empty, \p true otherwise.
 315             If the deque is empty \p val is not changed.
 316         */
 317         bool pop_back(
 318             value_type& val ///< Target to be received the copy of removed element
 319         )
 320         {
 321             fc_record * pRec = m_FlatCombining.acquire_record();
 322             pRec->pValPop = &val;
 323
 324             if ( c_bEliminationEnabled )
 325                 m_FlatCombining.batch_combine( op_pop_back, pRec, *this );
 326             else
 327                 m_FlatCombining.combine( op_pop_back, pRec, *this );
 328
 329             assert( pRec->is_done() );
 330             m_FlatCombining.release_record( pRec );
 331             m_FlatCombining.internal_statistics().onPopBack( pRec->bEmpty );
 332             return !pRec->bEmpty;
 333         }
 334
 335         /// Clears the deque
 336         void clear()
 337         {
 338             fc_record * pRec = m_FlatCombining.acquire_record();
 339
 340             if ( c_bEliminationEnabled )
 341                 m_FlatCombining.batch_combine( op_clear, pRec, *this );
 342             else
 343                 m_FlatCombining.combine( op_clear, pRec, *this );
 344
 345             assert( pRec->is_done() );
 346             m_FlatCombining.release_record( pRec );
 347         }
 348
 349         /// Returns the number of elements in the deque.
 350         /**
 351             Note that <tt>size() == 0</tt> is not mean that the deque is empty because
 352             combining record can be in process.
 353             To check emptiness use \ref empty function.
 354         */
 355         size_t size() const
 356         {
 357             return m_Deque.size();
 358         }
 359
 360         /// Checks if the deque is empty
 361         /**
 362             If the combining is in process the function waits while combining done.
 363         */
 364         bool empty()
 365         {
 366             fc_record * pRec = m_FlatCombining.acquire_record();
 367
 368             if ( c_bEliminationEnabled )
 369                 m_FlatCombining.batch_combine( op_empty, pRec, *this );
 370             else
 371                 m_FlatCombining.combine( op_empty, pRec, *this );
 372
 373             assert( pRec->is_done() );
 374             m_FlatCombining.release_record( pRec );
 375             return pRec->bEmpty;
 376         }
 377
 378         /// Internal statistics
 379         stat const& statistics() const
 380         {
 381             return m_FlatCombining.statistics();
 382         }
 383
 384     public: // flat combining cooperation, not for direct use!
 385         //@cond
 386         /// Flat combining supporting function. Do not call it directly!
 387         /**
 388             The function is called by \ref cds::algo::flat_combining::kernel "flat combining kernel"
 389             object if the current thread becomes a combiner. Invocation of the function means that
 390             the deque should perform an action recorded in \p pRec.
 391         */
 392         void fc_apply( fc_record * pRec )
 393         {
 394             assert( pRec );
 395
 396             // this function is called under FC mutex, so switch TSan off
 397             CDS_TSAN_ANNOTATE_IGNORE_RW_BEGIN;
 398
 399             switch ( pRec->op() ) {
 400             case op_push_front:
 401                 assert( pRec->pValPush );
 402                 m_Deque.push_front( *(pRec->pValPush) );
 403                 break;
 404             case op_push_front_move:
 405                 assert( pRec->pValPush );
 406                 m_Deque.push_front( std::move( *(pRec->pValPush )) );
 407                 break;
 408             case op_push_back:
 409                 assert( pRec->pValPush );
 410                 m_Deque.push_back( *(pRec->pValPush) );
 411                 break;
 412             case op_push_back_move:
 413                 assert( pRec->pValPush );
 414                 m_Deque.push_back( std::move( *(pRec->pValPush )) );
 415                 break;
 416             case op_pop_front:
 417                 assert( pRec->pValPop );
 418                 pRec->bEmpty = m_Deque.empty();
 419                 if ( !pRec->bEmpty ) {
 420                     *(pRec->pValPop) = std::move( m_Deque.front());
 421                     m_Deque.pop_front();
 422                 }
 423                 break;
 424             case op_pop_back:
 425                 assert( pRec->pValPop );
 426                 pRec->bEmpty = m_Deque.empty();
 427                 if ( !pRec->bEmpty ) {
 428                     *(pRec->pValPop) = std::move( m_Deque.back());
 429                     m_Deque.pop_back();
 430                 }
 431                 break;
 432             case op_clear:
 433                 while ( !m_Deque.empty() )
 434                     m_Deque.pop_front();
 435                 break;
 436             case op_empty:
 437                 pRec->bEmpty = m_Deque.empty();
 438                 break;
 439             default:
 440                 assert(false);
 441                 break;
 442             }
 443             CDS_TSAN_ANNOTATE_IGNORE_RW_END;
 444         }
 445
 446         /// Batch-processing flat combining
 447         void fc_process( typename fc_kernel::iterator itBegin, typename fc_kernel::iterator itEnd )
 448         {
 449             typedef typename fc_kernel::iterator fc_iterator;
 450
 451             // this function is called under FC mutex, so switch TSan off
 452             CDS_TSAN_ANNOTATE_IGNORE_RW_BEGIN;
 453
 454             for ( fc_iterator it = itBegin, itPrev = itEnd; it != itEnd; ++it ) {
 455                 switch ( it->op() ) {
 456                 case op_push_front:
 457                     if ( itPrev != itEnd
 458                         && (itPrev->op() == op_pop_front || (m_Deque.empty() && itPrev->op() == op_pop_back)) )
 459                     {
 460                         collide( *it, *itPrev );
 461                         itPrev = itEnd;
 462                     }
 463                     else
 464                         itPrev = it;
 465                     break;
 466                 case op_push_front_move:
 467                     if ( itPrev != itEnd
 468                       && (itPrev->op() == op_pop_front || ( m_Deque.empty() && itPrev->op() == op_pop_back )))
 469                     {
 470                         collide_move( *it, *itPrev );
 471                         itPrev = itEnd;
 472                     }
 473                     else
 474                         itPrev = it;
 475                     break;
 476                 case op_push_back:
 477                     if ( itPrev != itEnd
 478                         && (itPrev->op() == op_pop_back || (m_Deque.empty() && itPrev->op() == op_pop_front)) )
 479                     {
 480                         collide( *it, *itPrev );
 481                         itPrev = itEnd;
 482                     }
 483                     else
 484                         itPrev = it;
 485                     break;
 486                 case op_push_back_move:
 487                     if ( itPrev != itEnd
 488                         && (itPrev->op() == op_pop_back || ( m_Deque.empty() && itPrev->op() == op_pop_front )))
 489                     {
 490                         collide_move( *it, *itPrev );
 491                         itPrev = itEnd;
 492                     }
 493                     else
 494                         itPrev = it;
 495                     break;
 496                 case op_pop_front:
 497                     if ( itPrev != itEnd ) {
 498                         if ( m_Deque.empty() ) {
 499                             switch ( itPrev->op() ) {
 500                             case op_push_back:
 501                                 collide( *itPrev, *it );
 502                                 itPrev = itEnd;
 503                                 break;
 504                             case op_push_back_move:
 505                                 collide_move( *itPrev, *it );
 506                                 itPrev = itEnd;
 507                                 break;
 508                             default:
 509                                 itPrev = it;
 510                                 break;
 511                             }
 512                         }
 513                         else {
 514                             switch ( itPrev->op() ) {
 515                             case op_push_front:
 516                                 collide( *itPrev, *it );
 517                                 itPrev = itEnd;
 518                                 break;
 519                             case op_push_front_move:
 520                                 collide_move( *itPrev, *it );
 521                                 itPrev = itEnd;
 522                                 break;
 523                             default:
 524                                 itPrev = it;
 525                                 break;
 526                             }
 527                         }
 528                     }
 529                     else
 530                         itPrev = it;
 531                     break;
 532                 case op_pop_back:
 533                     if ( itPrev != itEnd ) {
 534                         if ( m_Deque.empty() ) {
 535                             switch ( itPrev->op() ) {
 536                             case op_push_front:
 537                                 collide( *itPrev, *it );
 538                                 itPrev = itEnd;
 539                                 break;
 540                             case op_push_front_move:
 541                                 collide_move( *itPrev, *it );
 542                                 itPrev = itEnd;
 543                                 break;
 544                             default:
 545                                 itPrev = it;
 546                                 break;
 547                             }
 548                         }
 549                         else {
 550                             switch ( itPrev->op() ) {
 551                             case op_push_back:
 552                                 collide( *itPrev, *it );
 553                                 itPrev = itEnd;
 554                                 break;
 555                             case op_push_back_move:
 556                                 collide_move( *itPrev, *it );
 557                                 itPrev = itEnd;
 558                                 break;
 559                             default:
 560                                 itPrev = it;
 561                                 break;
 562                             }
 563                         }
 564                     }
 565                     else
 566                         itPrev = it;
 567                     break;
 568                 }
 569             }
 570             CDS_TSAN_ANNOTATE_IGNORE_RW_END;
 571         }
 572         //@endcond
 573
 574     private:
 575         //@cond
 576         void collide( fc_record& recPush, fc_record& recPop )
 577         {
 578             *(recPop.pValPop) = *(recPush.pValPush);
 579             recPop.bEmpty = false;
 580             m_FlatCombining.operation_done( recPush );
 581             m_FlatCombining.operation_done( recPop );
 582             m_FlatCombining.internal_statistics().onCollide();
 583         }
 584
 585         void collide_move( fc_record& recPush, fc_record& recPop )
 586         {
 587             *(recPop.pValPop) = std::move( *(recPush.pValPush));
 588             recPop.bEmpty = false;
 589             m_FlatCombining.operation_done( recPush );
 590             m_FlatCombining.operation_done( recPop );
 591             m_FlatCombining.internal_statistics().onCollide();
 592         }
 593         //@endcond
 594     };
 595
 596 }} // namespace cds::container
 597
 598 #endif // #ifndef CDSLIB_CONTAINER_FCDEQUE_H