sync/spin/condition.hpp

0001 //////////////////////////////////////////////////////////////////////////////
0002 //
0003 // (C) Copyright Ion Gaztanaga 2005-2012. Distributed under the Boost
0004 // Software License, Version 1.0. (See accompanying file
0005 // LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
0006 //
0007 // See http://www.boost.org/libs/interprocess for documentation.
0008 //
0009 //////////////////////////////////////////////////////////////////////////////
0010
0011 #ifndef BOOST_INTERPROCESS_DETAIL_SPIN_CONDITION_HPP
0012 #define BOOST_INTERPROCESS_DETAIL_SPIN_CONDITION_HPP
0013
0014 #ifndef BOOST_CONFIG_HPP
0015 #  include <boost/config.hpp>
0016 #endif
0017 0018 ">#
0019 #if defined(BOOST_HAS_PRAGMA_ONCE)
0020 #  pragma once
0021 #endif
0022
0023 #include <boost/interprocess/detail/config_begin.hpp>
0024 #include <boost/interprocess/detail/workaround.hpp>
0025
0026 #include <boost/interprocess/sync/cv_status.hpp>
0027 #include <boost/interprocess/sync/spin/mutex.hpp>
0028 #include <boost/interprocess/detail/atomic.hpp>
0029 #include <boost/interprocess/sync/scoped_lock.hpp>
0030 #include <boost/interprocess/exceptions.hpp>
0031 #include <boost/interprocess/detail/os_thread_functions.hpp>
0032 #include <boost/interprocess/timed_utils.hpp>
0033 #include <boost/interprocess/sync/spin/wait.hpp>
0034 #include <boost/move/utility_core.hpp>
0035 #include <boost/cstdint.hpp>
0036
0037 namespace boost {
0038 namespace interprocess {
0039 namespace ipcdetail {
0040
0041 class spin_condition
0042 {
0043    spin_condition(const spin_condition &);
0044    spin_condition &operator=(const spin_condition &);
0045
0046    public:
0047    spin_condition()
0048    {
0049       //Note that this class is initialized to zero.
0050       //So zeroed memory can be interpreted as an initialized
0051       //condition variable
0052       m_command      = SLEEP;
0053       m_num_waiters  = 0;
0054    }
0055
0056    ~spin_condition()
0057    {
0058       //Notify all waiting threads
0059       //to allow POSIX semantics on condition destruction
0060       this->notify_all();
0061    }
0062
0063    void notify_one()
0064    {  this->notify(NOTIFY_ONE);  }
0065
0066    void notify_all()
0067    {  this->notify(NOTIFY_ALL);  }
0068
0069    template <typename L>
0070    void wait(L& lock)
0071    {
0072       if (!lock)
0073          throw lock_exception();
0074       this->do_timed_wait_impl<false>(ustime(0u), *lock.mutex());
0075    }
0076
0077    template <typename L, typename Pr>
0078    void wait(L& lock, Pr pred)
0079    {
0080       if (!lock)
0081          throw lock_exception();
0082
0083       while (!pred())
0084          this->do_timed_wait_impl<false>(ustime(0u), *lock.mutex());
0085    }
0086
0087    template <typename L, typename TimePoint>
0088    bool timed_wait(L& lock, const TimePoint &abs_time)
0089    {
0090       if (!lock)
0091          throw lock_exception();
0092       //Handle infinity absolute time here to avoid complications in do_timed_wait
0093       if(is_pos_infinity(abs_time)){
0094          this->wait(lock);
0095          return true;
0096       }
0097       return this->do_timed_wait_impl<true>(abs_time, *lock.mutex());
0098    }
0099
0100    template <typename L, typename TimePoint, typename Pr>
0101    bool timed_wait(L& lock, const TimePoint &abs_time, Pr pred)
0102    {
0103       if (!lock)
0104          throw lock_exception();
0105       //Handle infinity absolute time here to avoid complications in do_timed_wait
0106       if(is_pos_infinity(abs_time)){
0107          this->wait(lock, pred);
0108          return true;
0109       }
0110       while (!pred()){
0111          if (!this->do_timed_wait_impl<true>(abs_time, *lock.mutex()))
0112             return pred();
0113       }
0114       return true;
0115    }
0116
0117    template <typename L, class TimePoint>
0118    cv_status wait_until(L& lock, const TimePoint &abs_time)
0119    {  return this->timed_wait(lock, abs_time) ? cv_status::no_timeout : cv_status::timeout; }
0120
0121    template <typename L, class TimePoint, typename Pr>
0122    bool wait_until(L& lock, const TimePoint &abs_time, Pr pred)
0123    {  return this->timed_wait(lock, abs_time, pred); }
0124
0125    template <typename L, class Duration>
0126    cv_status wait_for(L& lock, const Duration &dur)
0127    {  return this->wait_until(lock, duration_to_ustime(dur)); }
0128
0129    template <typename L, class Duration, typename Pr>
0130    bool wait_for(L& lock, const Duration &dur, Pr pred)
0131    {  return this->wait_until(lock, duration_to_ustime(dur), pred); }
0132
0133    private:
0134
0135    template<bool TimeoutEnabled, class InterprocessMutex, class TimePoint>
0136    bool do_timed_wait_impl(const TimePoint &abs_time, InterprocessMutex &mut)
0137    {
0138       typedef boost::interprocess::scoped_lock<spin_mutex> InternalLock;
0139       //The enter mutex guarantees that while executing a notification,
0140       //no other thread can execute the do_timed_wait method.
0141       {
0142          //---------------------------------------------------------------
0143          InternalLock lock;
0144          get_lock(bool_<TimeoutEnabled>(), m_enter_mut, lock, abs_time);
0145
0146          if(!lock)
0147             return false;
0148          //---------------------------------------------------------------
0149          //We increment the waiting thread count protected so that it will be
0150          //always constant when another thread enters the notification logic.
0151          //The increment marks this thread as "waiting on spin_condition"
0152          atomic_inc32(const_cast<boost::uint32_t*>(&m_num_waiters));
0153
0154          //We unlock the external mutex atomically with the increment
0155          mut.unlock();
0156       }
0157
0158       //By default, we suppose that no timeout has happened
0159       bool timed_out  = false, unlock_enter_mut= false;
0160
0161       //Loop until a notification indicates that the thread should
0162       //exit or timeout occurs
0163       while(1){
0164          //The thread sleeps/spins until a spin_condition commands a notification
0165          //Notification occurred, we will lock the checking mutex so that
0166          spin_wait swait;
0167          while(atomic_read32(&m_command) == SLEEP){
0168             swait.yield();
0169
0170             //Check for timeout
0171             if(TimeoutEnabled){
0172                typedef typename microsec_clock<TimePoint>::time_point time_point;
0173                time_point now = get_now<TimePoint>(bool_<TimeoutEnabled>());
0174
0175                if(now >= abs_time){
0176                   //If we can lock the mutex it means that no notification
0177                   //is being executed in this spin_condition variable
0178                   timed_out = m_enter_mut.try_lock();
0179
0180                   //If locking fails, indicates that another thread is executing
0181                   //notification, so we play the notification game
0182                   if(!timed_out){
0183                      //There is an ongoing notification, we will try again later
0184                      continue;
0185                   }
0186                   //No notification in execution, since enter mutex is locked.
0187                   //We will execute time-out logic, so we will decrement count,
0188                   //release the enter mutex and return false.
0189                   break;
0190                }
0191             }
0192          }
0193
0194          //If a timeout occurred, the mutex will not execute checking logic
0195          if(TimeoutEnabled && timed_out){
0196             //Decrement wait count
0197             atomic_dec32(const_cast<boost::uint32_t*>(&m_num_waiters));
0198             unlock_enter_mut = true;
0199             break;
0200          }
0201          else{
0202             boost::uint32_t result = atomic_cas32
0203                            (const_cast<boost::uint32_t*>(&m_command), SLEEP, NOTIFY_ONE);
0204             if(result == SLEEP){
0205                //Other thread has been notified and since it was a NOTIFY one
0206                //command, this thread must sleep again
0207                continue;
0208             }
0209             else if(result == NOTIFY_ONE){
0210                //If it was a NOTIFY_ONE command, only this thread should
0211                //exit. This thread has atomically marked command as sleep before
0212                //so no other thread will exit.
0213                //Decrement wait count.
0214                unlock_enter_mut = true;
0215                atomic_dec32(const_cast<boost::uint32_t*>(&m_num_waiters));
0216                break;
0217             }
0218             else{
0219                //If it is a NOTIFY_ALL command, all threads should return
0220                //from do_timed_wait function. Decrement wait count.
0221                unlock_enter_mut = 1 == atomic_dec32(const_cast<boost::uint32_t*>(&m_num_waiters));
0222                //Check if this is the last thread of notify_all waiters
0223                //Only the last thread will release the mutex
0224                if(unlock_enter_mut){
0225                   atomic_cas32(const_cast<boost::uint32_t*>(&m_command), SLEEP, NOTIFY_ALL);
0226                }
0227                break;
0228             }
0229          }
0230       }
0231
0232       //Unlock the enter mutex if it is a single notification, if this is
0233       //the last notified thread in a notify_all or a timeout has occurred
0234       if(unlock_enter_mut){
0235          m_enter_mut.unlock();
0236       }
0237
0238       //Lock external again before returning from the method
0239       mut.lock();
0240       return !timed_out;
0241    }
0242
0243    template <class TimePoint>
0244    static typename microsec_clock<TimePoint>::time_point get_now(bool_<true>)
0245    {  return microsec_clock<TimePoint>::universal_time();  }
0246
0247    template <class TimePoint>
0248    static typename microsec_clock<TimePoint>::time_point get_now(bool_<false>)
0249    {  return typename microsec_clock<TimePoint>::time_point();  }
0250
0251    template <class Mutex, class Lock, class TimePoint>
0252    static void  get_lock(bool_<true>, Mutex &m, Lock &lck, const TimePoint &abs_time)
0253    {
0254       Lock dummy(m, abs_time);
0255       lck = boost::move(dummy);
0256    }
0257
0258    template <class Mutex, class Lock, class TimePoint>
0259    static void get_lock(bool_<false>, Mutex &m, Lock &lck, const TimePoint &)
0260    {
0261       Lock dummy(m);
0262       lck = boost::move(dummy);
0263    }
0264
0265    void notify(boost::uint32_t command)
0266    {
0267       //This mutex guarantees that no other thread can enter to the
0268       //do_timed_wait method logic, so that thread count will be
0269       //constant until the function writes a NOTIFY_ALL command.
0270       //It also guarantees that no other notification can be signaled
0271       //on this spin_condition before this one ends
0272       m_enter_mut.lock();
0273
0274       //Return if there are no waiters
0275       if(!atomic_read32(&m_num_waiters)) {
0276          m_enter_mut.unlock();
0277          return;
0278       }
0279
0280       //Notify that all threads should execute wait logic
0281       spin_wait swait;
0282       while(SLEEP != atomic_cas32(const_cast<boost::uint32_t*>(&m_command), command, SLEEP)){
0283          swait.yield();
0284       }
0285       //The enter mutex will rest locked until the last waiting thread unlocks it
0286    }
0287
0288    enum { SLEEP = 0, NOTIFY_ONE, NOTIFY_ALL };
0289    spin_mutex  m_enter_mut;
0290    volatile boost::uint32_t    m_command;
0291    volatile boost::uint32_t    m_num_waiters;
0292 };
0293
0294 }  //namespace ipcdetail
0295 }  //namespace interprocess
0296 }  //namespace boost
0297
0298 #include <boost/interprocess/detail/config_end.hpp>
0299
0300 #endif   //BOOST_INTERPROCESS_DETAIL_SPIN_CONDITION_HPP