cds/algo/flat_combining/wait_strategy.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_ALGO_FLAT_COMBINING_WAIT_STRATEGY_H
  32 #define CDSLIB_ALGO_FLAT_COMBINING_WAIT_STRATEGY_H
  33
  34 #include <cds/algo/flat_combining/defs.h>
  35 #include <cds/algo/backoff_strategy.h>
  36 #include <mutex>
  37 #include <condition_variable>
  38 #include <boost/thread/tss.hpp>  // thread_specific_ptr
  39
  40
  41 namespace cds { namespace opt {
  42
  43     /// Wait strategy option for \p flat_combining::kernel
  44     template <typename Strategy>
  45     struct wait_strategy {
  46         //@cond
  47         template <typename Base> struct pack: public Base
  48         {
  49             typedef Strategy wait_strategy;
  50         };
  51         //@endcond
  52     };
  53
  54 }} // namespace cds::opt
  55
  56 namespace cds { namespace algo { namespace flat_combining {
  57
  58     /// Wait strategies for \p flat_combining technique
  59     /**
  60         Wait strategy specifies how a thread waits until its request is performed by the combiner.
  61         See \p wait_strategy::empty wait strategy to explain the interface.
  62     */
  63     namespace wait_strategy {
  64
  65         /// Empty wait strategy
  66         /**
  67             Empty wait strategy is just spinning on request field.
  68             All functions are empty.
  69         */
  70         struct empty
  71         {
  72             /// Metafunction for defining a publication record for flat combining technique
  73             /**
  74                 Any wait strategy may expand the publication record for storing
  75                 its own private data.
  76                 \p PublicationRecord is the type specified by \p flat_combining::kernel.
  77                 - If the strategy has no thread-private data, it should typedef \p PublicationRecord
  78                   as a return \p type of metafunction.
  79                 - Otherwise, if the strategy wants to store anything in thread-local data,
  80                   it should expand \p PublicationRecord, for example:
  81                   \code
  82                   template <typename PublicationRecord>
  83                   struct make_publication_record {
  84                     struct type: public PublicationRecord
  85                     {
  86                         int strategy_data;
  87                     };
  88                   };
  89                   \endcode
  90             */
  91             template <typename PublicationRecord>
  92             struct make_publication_record {
  93                 typedef PublicationRecord type; ///< Metafunction result
  94             };
  95
  96             /// Prepares the strategy
  97             /**
  98                 This function is called before enter to waiting cycle.
  99                 Some strategies need to prepare its thread-local data in \p rec.
 100
 101                 \p PublicationRecord is thread's publication record of type \p make_publication_record::type
 102             */
 103             template <typename PublicationRecord>
 104             void prepare( PublicationRecord& rec )
 105             {
 106                 CDS_UNUSED( rec );
 107             }
 108
 109             /// Waits for the combiner
 110             /**
 111                 The thread calls this function to wait for the combiner process
 112                 the request.
 113                 The function returns \p true if the thread was waked up by the combiner,
 114                 otherwise it should return \p false.
 115
 116                 \p FCKernel is a \p flat_combining::kernel object,
 117                 \p PublicationRecord is thread's publication record of type \p make_publication_record::type
 118             */
 119             template <typename FCKernel, typename PublicationRecord>
 120             bool wait( FCKernel& fc, PublicationRecord& rec )
 121             {
 122                 CDS_UNUSED( fc );
 123                 CDS_UNUSED( rec );
 124                 return false;
 125             }
 126
 127             /// Wakes up the thread
 128             /**
 129                 The combiner calls \p %notify() when it has been processed the request.
 130
 131                 \p FCKernel is a \p flat_combining::kernel object,
 132                 \p PublicationRecord is thread's publication record of type \p make_publication_record::type
 133             */
 134             template <typename FCKernel, typename PublicationRecord>
 135             void notify( FCKernel& fc, PublicationRecord& rec )
 136             {
 137                 CDS_UNUSED( fc );
 138                 CDS_UNUSED( rec );
 139             }
 140
 141             /// Moves control to other thread
 142             /**
 143                 This function is called when the thread becomes the combiner
 144                 but the request of the thread is already processed.
 145                 The strategy may call \p fc.wakeup_any() instructs the kernel
 146                 to wake up any pending thread.
 147
 148                 \p FCKernel is a \p flat_combining::kernel object,
 149             */
 150             template <typename FCKernel>
 151             void wakeup( FCKernel& fc )
 152             {
 153                 CDS_UNUSED( fc );
 154             }
 155         };
 156
 157         /// Back-off wait strategy
 158         /**
 159             Template argument \p Backoff specifies back-off strategy, default is cds::backoff::delay_of<2>
 160         */
 161         template <typename BackOff = cds::backoff::delay_of<2>>
 162         struct backoff
 163         {
 164             typedef BackOff back_off;   ///< Back-off strategy
 165
 166             /// Incorporates back-off strategy into publication record
 167             template <typename PublicationRecord>
 168             struct make_publication_record
 169             {
 170                 //@cond
 171                 struct type: public PublicationRecord
 172                 {
 173                     back_off bkoff;
 174                 };
 175                 //@endcond
 176             };
 177
 178             /// Resets back-off strategy in \p rec
 179             template <typename PublicationRecord>
 180             void prepare( PublicationRecord& rec )
 181             {
 182                 rec.bkoff.reset();
 183             }
 184
 185             /// Calls back-off strategy
 186             template <typename FCKernel, typename PublicationRecord>
 187             bool wait( FCKernel& /*fc*/, PublicationRecord& rec )
 188             {
 189                 rec.bkoff();
 190                 return false;
 191             }
 192
 193             /// Does nothing
 194             template <typename FCKernel, typename PublicationRecord>
 195             void notify( FCKernel& /*fc*/, PublicationRecord& /*rec*/ )
 196             {}
 197
 198             /// Does nothing
 199             template <typename FCKernel>
 200             void wakeup( FCKernel& )
 201             {}
 202         };
 203
 204         /// Wait strategy based on the single mutex and the condition variable
 205         /**
 206             The strategy shares the mutex and conditional variable for all thread.
 207
 208             Template parameter \p Milliseconds specifies waiting duration;
 209             the minimal value is 1.
 210         */
 211         template <int Milliseconds = 2>
 212         class single_mutex_single_condvar
 213         {
 214         //@cond
 215             std::mutex m_mutex;
 216             std::condition_variable m_condvar;
 217
 218             typedef std::unique_lock< std::mutex > unique_lock;
 219         //@endcond
 220
 221         public:
 222             enum {
 223                 c_nWaitMilliseconds = Milliseconds < 1 ? 1 : Milliseconds ///< Waiting duration
 224             };
 225
 226             /// Empty metafunction
 227             template <typename PublicationRecord>
 228             struct make_publication_record {
 229                 typedef PublicationRecord type; ///< publication record type
 230             };
 231
 232             /// Does nothing
 233             template <typename PublicationRecord>
 234             void prepare( PublicationRecord& /*rec*/ )
 235             {}
 236
 237             /// Sleeps on condition variable waiting for notification from combiner
 238             template <typename FCKernel, typename PublicationRecord>
 239             bool wait( FCKernel& fc, PublicationRecord& rec )
 240             {
 241                 if ( fc.get_operation( rec ) >= req_Operation ) {
 242                     unique_lock lock( m_mutex );
 243                     if ( fc.get_operation( rec ) >= req_Operation )
 244                         return m_condvar.wait_for( lock, std::chrono::milliseconds( c_nWaitMilliseconds )) == std::cv_status::no_timeout;
 245                 }
 246                 return false;
 247             }
 248
 249             /// Calls condition variable function \p notify_all()
 250             template <typename FCKernel, typename PublicationRecord>
 251             void notify( FCKernel& /*fc*/, PublicationRecord& /*rec*/ )
 252             {
 253                 m_condvar.notify_all();
 254             }
 255
 256             /// Calls condition variable function \p notify_all()
 257             template <typename FCKernel>
 258             void wakeup( FCKernel& /*fc*/ )
 259             {
 260                 m_condvar.notify_all();
 261             }
 262         };
 263
 264         /// Wait strategy based on the single mutex and thread-local condition variables
 265         /**
 266             The strategy shares the mutex, but each thread has its own conditional variable
 267
 268             Template parameter \p Milliseconds specifies waiting duration;
 269             the minimal value is 1.
 270         */
 271         template <int Milliseconds = 2>
 272         class single_mutex_multi_condvar
 273         {
 274         //@cond
 275             std::mutex m_mutex;
 276
 277             typedef std::unique_lock< std::mutex > unique_lock;
 278         //@endcond
 279
 280         public:
 281             enum {
 282                 c_nWaitMilliseconds = Milliseconds < 1 ? 1 : Milliseconds  ///< Waiting duration
 283             };
 284
 285             /// Incorporates a condition variable into \p PublicationRecord
 286             template <typename PublicationRecord>
 287             struct make_publication_record {
 288                 /// Metafunction result
 289                 struct type: public PublicationRecord
 290                 {
 291                     //@cond
 292                     std::condition_variable m_condvar;
 293                     //@endcond
 294                 };
 295             };
 296
 297             /// Does nothing
 298             template <typename PublicationRecord>
 299             void prepare( PublicationRecord& /*rec*/ )
 300             {}
 301
 302             /// Sleeps on condition variable waiting for notification from combiner
 303             template <typename FCKernel, typename PublicationRecord>
 304             bool wait( FCKernel& fc, PublicationRecord& rec )
 305             {
 306                 if ( fc.get_operation( rec ) >= req_Operation ) {
 307                     unique_lock lock( m_mutex );
 308                     if ( fc.get_operation( rec ) >= req_Operation )
 309                         return rec.m_condvar.wait_for( lock, std::chrono::milliseconds( c_nWaitMilliseconds )) == std::cv_status::no_timeout;
 310                 }
 311                 return false;
 312             }
 313
 314             /// Calls condition variable function \p notify_one()
 315             template <typename FCKernel, typename PublicationRecord>
 316             void notify( FCKernel& /*fc*/, PublicationRecord& rec )
 317             {
 318                 rec.m_condvar.notify_one();
 319             }
 320
 321             /// Calls \p fc.wakeup_any() to wake up any pending thread
 322             template <typename FCKernel>
 323             void wakeup( FCKernel& fc )
 324             {
 325                 fc.wakeup_any();
 326             }
 327         };
 328
 329         /// Wait strategy where each thread has a mutex and a condition variable
 330         /**
 331             Template parameter \p Milliseconds specifies waiting duration;
 332             the minimal value is 1.
 333         */
 334         template <int Milliseconds = 2>
 335         class multi_mutex_multi_condvar
 336         {
 337         //@cond
 338             typedef std::unique_lock< std::mutex > unique_lock;
 339         //@endcond
 340         public:
 341             enum {
 342                 c_nWaitMilliseconds = Milliseconds < 1 ? 1 : Milliseconds   ///< Waiting duration
 343             };
 344
 345             /// Incorporates a condition variable and a mutex into \p PublicationRecord
 346             template <typename PublicationRecord>
 347             struct make_publication_record {
 348                 /// Metafunction result
 349                 struct type: public PublicationRecord
 350                 {
 351                     //@cond
 352                     std::mutex              m_mutex;
 353                     std::condition_variable m_condvar;
 354                     //@endcond
 355                 };
 356             };
 357
 358             /// Does nothing
 359             template <typename PublicationRecord>
 360             void prepare( PublicationRecord& /*rec*/ )
 361             {}
 362
 363             /// Sleeps on condition variable waiting for notification from combiner
 364             template <typename FCKernel, typename PublicationRecord>
 365             bool wait( FCKernel& fc, PublicationRecord& rec )
 366             {
 367                 if ( fc.get_operation( rec ) >= req_Operation ) {
 368                     unique_lock lock( rec.m_mutex );
 369                     if ( fc.get_operation( rec ) >= req_Operation )
 370                         return rec.m_condvar.wait_for( lock, std::chrono::milliseconds( c_nWaitMilliseconds )) == std::cv_status::no_timeout;
 371                 }
 372                 return false;
 373             }
 374
 375             /// Calls condition variable function \p notify_one()
 376             template <typename FCKernel, typename PublicationRecord>
 377             void notify( FCKernel& /*fc*/, PublicationRecord& rec )
 378             {
 379                 rec.m_condvar.notify_one();
 380             }
 381
 382             /// Calls \p fc.wakeup_any() to wake up any pending thread
 383             template <typename FCKernel>
 384             void wakeup( FCKernel& fc )
 385             {
 386                 fc.wakeup_any();
 387             }
 388         };
 389
 390     } // namespace wait_strategy
 391 }}} // namespace cds::algo::flat_combining
 392
 393 #endif //CDSLIB_ALGO_FLAT_COMBINING_WAIT_STRATEGY_H