thread/concurrent_queues/sync_bounded_queue.hpp

0001 #ifndef BOOST_THREAD_CONCURRENT_QUEUES_SYNC_BOUNDED_QUEUE_HPP
0002 #define BOOST_THREAD_CONCURRENT_QUEUES_SYNC_BOUNDED_QUEUE_HPP
0003
0004 //////////////////////////////////////////////////////////////////////////////
0005 //
0006 // (C) Copyright Vicente J. Botet Escriba 2013-2014. Distributed under the Boost
0007 // Software License, Version 1.0. (See accompanying file
0008 // LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
0009 //
0010 // See http://www.boost.org/libs/thread for documentation.
0011 //
0012 //////////////////////////////////////////////////////////////////////////////
0013
0014 #include <boost/thread/detail/config.hpp>
0015 #include <boost/thread/condition_variable.hpp>
0016 #include <boost/thread/mutex.hpp>
0017 #include <boost/thread/detail/move.hpp>
0018 #include <boost/throw_exception.hpp>
0019 #include <boost/thread/concurrent_queues/queue_op_status.hpp>
0020
0021 #ifndef BOOST_THREAD_QUEUE_DEPRECATE_OLD
0022 #include <boost/smart_ptr/shared_ptr.hpp>
0023 #include <boost/smart_ptr/make_shared.hpp>
0024 #endif
0025 #include <boost/config/abi_prefix.hpp>
0026
0027 namespace boost
0028 {
0029 namespace concurrent
0030 {
0031   template <typename ValueType>
0032   class sync_bounded_queue
0033   {
0034   public:
0035     typedef ValueType value_type;
0036     typedef std::size_t size_type;
0037
0038     // Constructors/Assignment/Destructors
0039     BOOST_THREAD_NO_COPYABLE(sync_bounded_queue)
0040     explicit sync_bounded_queue(size_type max_elems);
0041     template <typename Range>
0042     sync_bounded_queue(size_type max_elems, Range range);
0043     ~sync_bounded_queue();
0044
0045     // Observers
0046     inline bool empty() const;
0047     inline bool full() const;
0048     inline size_type capacity() const;
0049     inline size_type size() const;
0050     inline bool closed() const;
0051
0052     // Modifiers
0053     inline void close();
0054
0055 #ifndef BOOST_THREAD_QUEUE_DEPRECATE_OLD
0056     inline void push(const value_type& x);
0057     inline void push(BOOST_THREAD_RV_REF(value_type) x);
0058     inline bool try_push(const value_type& x);
0059     inline bool try_push(BOOST_THREAD_RV_REF(value_type) x);
0060     inline bool try_push(no_block_tag, const value_type& x);
0061     inline bool try_push(no_block_tag, BOOST_THREAD_RV_REF(value_type) x);
0062 #endif
0063     inline void push_back(const value_type& x);
0064     inline void push_back(BOOST_THREAD_RV_REF(value_type) x);
0065     inline queue_op_status try_push_back(const value_type& x);
0066     inline queue_op_status try_push_back(BOOST_THREAD_RV_REF(value_type) x);
0067     inline queue_op_status nonblocking_push_back(const value_type& x);
0068     inline queue_op_status nonblocking_push_back(BOOST_THREAD_RV_REF(value_type) x);
0069     inline queue_op_status wait_push_back(const value_type& x);
0070     inline queue_op_status wait_push_back(BOOST_THREAD_RV_REF(value_type) x);
0071
0072     // Observers/Modifiers
0073 #ifndef BOOST_THREAD_QUEUE_DEPRECATE_OLD
0074     inline void pull(value_type&);
0075     // enable_if is_nothrow_copy_movable<value_type>
0076     inline value_type pull();
0077     inline shared_ptr<ValueType> ptr_pull();
0078     inline bool try_pull(value_type&);
0079     inline bool try_pull(no_block_tag,value_type&);
0080     inline shared_ptr<ValueType> try_pull();
0081 #endif
0082     inline void pull_front(value_type&);
0083     // enable_if is_nothrow_copy_movable<value_type>
0084     inline value_type pull_front();
0085     inline queue_op_status try_pull_front(value_type&);
0086     inline queue_op_status nonblocking_pull_front(value_type&);
0087
0088     inline queue_op_status wait_pull_front(ValueType& elem);
0089
0090   private:
0091     mutable mutex mtx_;
0092     condition_variable not_empty_;
0093     condition_variable not_full_;
0094     size_type waiting_full_;
0095     size_type waiting_empty_;
0096     value_type* data_;
0097     size_type in_;
0098     size_type out_;
0099     size_type capacity_;
0100     bool closed_;
0101
0102     inline size_type inc(size_type idx) const BOOST_NOEXCEPT
0103     {
0104       return (idx + 1) % capacity_;
0105     }
0106
0107     inline bool empty(unique_lock<mutex>& ) const BOOST_NOEXCEPT
0108     {
0109       return in_ == out_;
0110     }
0111     inline bool empty(lock_guard<mutex>& ) const BOOST_NOEXCEPT
0112     {
0113       return in_ == out_;
0114     }
0115     inline bool full(unique_lock<mutex>& ) const BOOST_NOEXCEPT
0116     {
0117       return (inc(in_) == out_);
0118     }
0119     inline bool full(lock_guard<mutex>& ) const BOOST_NOEXCEPT
0120     {
0121       return (inc(in_) == out_);
0122     }
0123     inline size_type capacity(lock_guard<mutex>& ) const BOOST_NOEXCEPT
0124     {
0125       return capacity_-1;
0126     }
0127     inline size_type size(lock_guard<mutex>& lk) const BOOST_NOEXCEPT
0128     {
0129       if (full(lk)) return capacity(lk);
0130       return ((in_+capacity(lk)-out_) % capacity(lk));
0131     }
0132
0133     inline void throw_if_closed(unique_lock<mutex>&);
0134     inline bool closed(unique_lock<mutex>&) const;
0135
0136 #ifndef BOOST_THREAD_QUEUE_DEPRECATE_OLD
0137     inline bool try_pull(value_type& x, unique_lock<mutex>& lk);
0138     inline shared_ptr<value_type> try_pull(unique_lock<mutex>& lk);
0139     inline bool try_push(const value_type& x, unique_lock<mutex>& lk);
0140     inline bool try_push(BOOST_THREAD_RV_REF(value_type) x, unique_lock<mutex>& lk);
0141 #endif
0142     inline queue_op_status try_pull_front(value_type& x, unique_lock<mutex>& lk);
0143     inline queue_op_status try_push_back(const value_type& x, unique_lock<mutex>& lk);
0144     inline queue_op_status try_push_back(BOOST_THREAD_RV_REF(value_type) x, unique_lock<mutex>& lk);
0145
0146     inline queue_op_status wait_pull_front(value_type& x, unique_lock<mutex>& lk);
0147     inline queue_op_status wait_push_back(const value_type& x, unique_lock<mutex>& lk);
0148     inline queue_op_status wait_push_back(BOOST_THREAD_RV_REF(value_type) x, unique_lock<mutex>& lk);
0149
0150     inline void wait_until_not_empty(unique_lock<mutex>& lk);
0151     inline void wait_until_not_empty(unique_lock<mutex>& lk, bool&);
0152     inline size_type wait_until_not_full(unique_lock<mutex>& lk);
0153     inline size_type wait_until_not_full(unique_lock<mutex>& lk, bool&);
0154
0155
0156     inline void notify_not_empty_if_needed(unique_lock<mutex>& lk)
0157     {
0158       if (waiting_empty_ > 0)
0159       {
0160         --waiting_empty_;
0161         lk.unlock();
0162         not_empty_.notify_one();
0163       }
0164     }
0165     inline void notify_not_full_if_needed(unique_lock<mutex>& lk)
0166     {
0167       if (waiting_full_ > 0)
0168       {
0169         --waiting_full_;
0170         lk.unlock();
0171         not_full_.notify_one();
0172       }
0173     }
0174
0175 #ifndef BOOST_THREAD_QUEUE_DEPRECATE_OLD
0176     inline void pull(value_type& elem, unique_lock<mutex>& lk)
0177     {
0178       elem = boost::move(data_[out_]);
0179       out_ = inc(out_);
0180       notify_not_full_if_needed(lk);
0181     }
0182     inline value_type pull(unique_lock<mutex>& lk)
0183     {
0184       value_type elem = boost::move(data_[out_]);
0185       out_ = inc(out_);
0186       notify_not_full_if_needed(lk);
0187       return boost::move(elem);
0188     }
0189     inline boost::shared_ptr<value_type> ptr_pull(unique_lock<mutex>& lk)
0190     {
0191       shared_ptr<value_type> res = make_shared<value_type>(boost::move(data_[out_]));
0192       out_ = inc(out_);
0193       notify_not_full_if_needed(lk);
0194       return res;
0195     }
0196 #endif
0197     inline void pull_front(value_type& elem, unique_lock<mutex>& lk)
0198     {
0199       elem = boost::move(data_[out_]);
0200       out_ = inc(out_);
0201       notify_not_full_if_needed(lk);
0202     }
0203     inline value_type pull_front(unique_lock<mutex>& lk)
0204     {
0205       value_type elem = boost::move(data_[out_]);
0206       out_ = inc(out_);
0207       notify_not_full_if_needed(lk);
0208       return boost::move(elem);
0209     }
0210
0211     inline void set_in(size_type in, unique_lock<mutex>& lk)
0212     {
0213       in_ = in;
0214       notify_not_empty_if_needed(lk);
0215     }
0216
0217     inline void push_at(const value_type& elem, size_type in_p_1, unique_lock<mutex>& lk)
0218     {
0219       data_[in_] = elem;
0220       set_in(in_p_1, lk);
0221     }
0222
0223     inline void push_at(BOOST_THREAD_RV_REF(value_type) elem, size_type in_p_1, unique_lock<mutex>& lk)
0224     {
0225       data_[in_] = boost::move(elem);
0226       set_in(in_p_1, lk);
0227     }
0228   };
0229
0230   template <typename ValueType>
0231   sync_bounded_queue<ValueType>::sync_bounded_queue(typename sync_bounded_queue<ValueType>::size_type max_elems) :
0232     waiting_full_(0), waiting_empty_(0), data_(new value_type[max_elems + 1]), in_(0), out_(0), capacity_(max_elems + 1),
0233         closed_(false)
0234   {
0235     BOOST_ASSERT_MSG(max_elems >= 1, "number of elements must be > 1");
0236   }
0237
0238 //  template <typename ValueType>
0239 //  template <typename Range>
0240 //  sync_bounded_queue<ValueType>::sync_bounded_queue(size_type max_elems, Range range) :
0241 //    waiting_full_(0), waiting_empty_(0), data_(new value_type[max_elems + 1]), in_(0), out_(0), capacity_(max_elems + 1),
0242 //        closed_(false)
0243 //  {
0244 //    BOOST_ASSERT_MSG(max_elems >= 1, "number of elements must be > 1");
0245 //    BOOST_ASSERT_MSG(max_elems == size(range), "number of elements must match range's size");
0246 //    try
0247 //    {
0248 //      typedef typename Range::iterator iterator_t;
0249 //      iterator_t first = boost::begin(range);
0250 //      iterator_t end = boost::end(range);
0251 //      size_type in = 0;
0252 //      for (iterator_t cur = first; cur != end; ++cur, ++in)
0253 //      {
0254 //        data_[in] = *cur;
0255 //      }
0256 //      set_in(in);
0257 //    }
0258 //    catch (...)
0259 //    {
0260 //      delete[] data_;
0261 //    }
0262 //  }
0263
0264   template <typename ValueType>
0265   sync_bounded_queue<ValueType>::~sync_bounded_queue()
0266   {
0267     delete[] data_;
0268   }
0269
0270   template <typename ValueType>
0271   void sync_bounded_queue<ValueType>::close()
0272   {
0273     {
0274       lock_guard<mutex> lk(mtx_);
0275       closed_ = true;
0276     }
0277     not_empty_.notify_all();
0278     not_full_.notify_all();
0279   }
0280
0281   template <typename ValueType>
0282   bool sync_bounded_queue<ValueType>::closed() const
0283   {
0284     lock_guard<mutex> lk(mtx_);
0285     return closed_;
0286   }
0287   template <typename ValueType>
0288   bool sync_bounded_queue<ValueType>::closed(unique_lock<mutex>& ) const
0289   {
0290     return closed_;
0291   }
0292
0293   template <typename ValueType>
0294   bool sync_bounded_queue<ValueType>::empty() const
0295   {
0296     lock_guard<mutex> lk(mtx_);
0297     return empty(lk);
0298   }
0299   template <typename ValueType>
0300   bool sync_bounded_queue<ValueType>::full() const
0301   {
0302     lock_guard<mutex> lk(mtx_);
0303     return full(lk);
0304   }
0305
0306   template <typename ValueType>
0307   typename sync_bounded_queue<ValueType>::size_type sync_bounded_queue<ValueType>::capacity() const
0308   {
0309     lock_guard<mutex> lk(mtx_);
0310     return capacity(lk);
0311   }
0312
0313   template <typename ValueType>
0314   typename sync_bounded_queue<ValueType>::size_type sync_bounded_queue<ValueType>::size() const
0315   {
0316     lock_guard<mutex> lk(mtx_);
0317     return size(lk);
0318   }
0319
0320 #ifndef BOOST_THREAD_QUEUE_DEPRECATE_OLD
0321   template <typename ValueType>
0322   bool sync_bounded_queue<ValueType>::try_pull(ValueType& elem, unique_lock<mutex>& lk)
0323   {
0324     if (empty(lk))
0325     {
0326       throw_if_closed(lk);
0327       return false;
0328     }
0329     pull(elem, lk);
0330     return true;
0331   }
0332   template <typename ValueType>
0333   shared_ptr<ValueType> sync_bounded_queue<ValueType>::try_pull(unique_lock<mutex>& lk)
0334   {
0335     if (empty(lk))
0336     {
0337       throw_if_closed(lk);
0338       return shared_ptr<ValueType>();
0339     }
0340     return ptr_pull(lk);
0341   }
0342   template <typename ValueType>
0343   bool sync_bounded_queue<ValueType>::try_pull(ValueType& elem)
0344   {
0345       unique_lock<mutex> lk(mtx_);
0346       return try_pull(elem, lk);
0347   }
0348 #endif
0349
0350   template <typename ValueType>
0351   queue_op_status sync_bounded_queue<ValueType>::try_pull_front(ValueType& elem, unique_lock<mutex>& lk)
0352   {
0353     if (empty(lk))
0354     {
0355       if (closed(lk)) return queue_op_status::closed;
0356       return queue_op_status::empty;
0357     }
0358     pull_front(elem, lk);
0359     return queue_op_status::success;
0360   }
0361
0362   template <typename ValueType>
0363   queue_op_status sync_bounded_queue<ValueType>::try_pull_front(ValueType& elem)
0364   {
0365       unique_lock<mutex> lk(mtx_);
0366       return try_pull_front(elem, lk);
0367   }
0368
0369 #ifndef BOOST_THREAD_QUEUE_DEPRECATE_OLD
0370   template <typename ValueType>
0371   bool sync_bounded_queue<ValueType>::try_pull(no_block_tag,ValueType& elem)
0372   {
0373       unique_lock<mutex> lk(mtx_, try_to_lock);
0374       if (!lk.owns_lock())
0375       {
0376         return false;
0377       }
0378       return try_pull(elem, lk);
0379   }
0380   template <typename ValueType>
0381   boost::shared_ptr<ValueType> sync_bounded_queue<ValueType>::try_pull()
0382   {
0383       unique_lock<mutex> lk(mtx_);
0384       return try_pull(lk);
0385   }
0386 #endif
0387
0388   template <typename ValueType>
0389   queue_op_status sync_bounded_queue<ValueType>::nonblocking_pull_front(ValueType& elem)
0390   {
0391       unique_lock<mutex> lk(mtx_, try_to_lock);
0392       if (!lk.owns_lock())
0393       {
0394         return queue_op_status::busy;
0395       }
0396       return try_pull_front(elem, lk);
0397   }
0398
0399   template <typename ValueType>
0400   void sync_bounded_queue<ValueType>::throw_if_closed(unique_lock<mutex>&)
0401   {
0402     if (closed_)
0403     {
0404       BOOST_THROW_EXCEPTION( sync_queue_is_closed() );
0405     }
0406   }
0407
0408   template <typename ValueType>
0409   void sync_bounded_queue<ValueType>::wait_until_not_empty(unique_lock<mutex>& lk)
0410   {
0411     for (;;)
0412     {
0413       if (out_ != in_) break;
0414       throw_if_closed(lk);
0415       ++waiting_empty_;
0416       not_empty_.wait(lk);
0417     }
0418   }
0419   template <typename ValueType>
0420   void sync_bounded_queue<ValueType>::wait_until_not_empty(unique_lock<mutex>& lk, bool & closed)
0421   {
0422     for (;;)
0423     {
0424       if (out_ != in_) break;
0425       if (closed_) {closed=true; return;}
0426       ++waiting_empty_;
0427       not_empty_.wait(lk);
0428     }
0429   }
0430
0431 #ifndef BOOST_THREAD_QUEUE_DEPRECATE_OLD
0432   template <typename ValueType>
0433   void sync_bounded_queue<ValueType>::pull(ValueType& elem)
0434   {
0435       unique_lock<mutex> lk(mtx_);
0436       wait_until_not_empty(lk);
0437       pull(elem, lk);
0438   }
0439 //  template <typename ValueType>
0440 //  void sync_bounded_queue<ValueType>::pull(ValueType& elem, bool & closed)
0441 //  {
0442 //      unique_lock<mutex> lk(mtx_);
0443 //      wait_until_not_empty(lk, closed);
0444 //      if (closed) {return;}
0445 //      pull(elem, lk);
0446 //  }
0447
0448   // enable if ValueType is nothrow movable
0449   template <typename ValueType>
0450   ValueType sync_bounded_queue<ValueType>::pull()
0451   {
0452       unique_lock<mutex> lk(mtx_);
0453       wait_until_not_empty(lk);
0454       return pull(lk);
0455   }
0456   template <typename ValueType>
0457   boost::shared_ptr<ValueType> sync_bounded_queue<ValueType>::ptr_pull()
0458   {
0459       unique_lock<mutex> lk(mtx_);
0460       wait_until_not_empty(lk);
0461       return ptr_pull(lk);
0462   }
0463
0464 #endif
0465
0466   template <typename ValueType>
0467   void sync_bounded_queue<ValueType>::pull_front(ValueType& elem)
0468   {
0469       unique_lock<mutex> lk(mtx_);
0470       wait_until_not_empty(lk);
0471       pull_front(elem, lk);
0472   }
0473
0474   // enable if ValueType is nothrow movable
0475   template <typename ValueType>
0476   ValueType sync_bounded_queue<ValueType>::pull_front()
0477   {
0478       unique_lock<mutex> lk(mtx_);
0479       wait_until_not_empty(lk);
0480       return pull_front(lk);
0481   }
0482
0483   template <typename ValueType>
0484   queue_op_status sync_bounded_queue<ValueType>::wait_pull_front(ValueType& elem, unique_lock<mutex>& lk)
0485   {
0486       if (empty(lk) && closed(lk)) {return queue_op_status::closed;}
0487       bool is_closed = false;
0488       wait_until_not_empty(lk, is_closed);
0489       if (is_closed) {return queue_op_status::closed;}
0490       pull_front(elem, lk);
0491       return queue_op_status::success;
0492   }
0493   template <typename ValueType>
0494   queue_op_status sync_bounded_queue<ValueType>::wait_pull_front(ValueType& elem)
0495   {
0496     unique_lock<mutex> lk(mtx_);
0497     return wait_pull_front(elem, lk);
0498   }
0499
0500 #ifndef BOOST_THREAD_QUEUE_DEPRECATE_OLD
0501   template <typename ValueType>
0502   bool sync_bounded_queue<ValueType>::try_push(const ValueType& elem, unique_lock<mutex>& lk)
0503   {
0504     throw_if_closed(lk);
0505     size_type in_p_1 = inc(in_);
0506     if (in_p_1 == out_)  // full()
0507     {
0508       return false;
0509     }
0510     push_at(elem, in_p_1, lk);
0511     return true;
0512   }
0513   template <typename ValueType>
0514   bool sync_bounded_queue<ValueType>::try_push(const ValueType& elem)
0515   {
0516       unique_lock<mutex> lk(mtx_);
0517       return try_push(elem, lk);
0518   }
0519
0520 #endif
0521
0522   template <typename ValueType>
0523   queue_op_status sync_bounded_queue<ValueType>::try_push_back(const ValueType& elem, unique_lock<mutex>& lk)
0524   {
0525     if (closed(lk)) return queue_op_status::closed;
0526     size_type in_p_1 = inc(in_);
0527     if (in_p_1 == out_)  // full()
0528     {
0529       return queue_op_status::full;
0530     }
0531     push_at(elem, in_p_1, lk);
0532     return queue_op_status::success;
0533   }
0534
0535   template <typename ValueType>
0536   queue_op_status sync_bounded_queue<ValueType>::try_push_back(const ValueType& elem)
0537   {
0538     unique_lock<mutex> lk(mtx_);
0539     return try_push_back(elem, lk);
0540   }
0541
0542   template <typename ValueType>
0543   queue_op_status sync_bounded_queue<ValueType>::wait_push_back(const ValueType& elem, unique_lock<mutex>& lk)
0544   {
0545     if (closed(lk)) return queue_op_status::closed;
0546     push_at(elem, wait_until_not_full(lk), lk);
0547     return queue_op_status::success;
0548   }
0549   template <typename ValueType>
0550   queue_op_status sync_bounded_queue<ValueType>::wait_push_back(const ValueType& elem)
0551   {
0552     unique_lock<mutex> lk(mtx_);
0553     return wait_push_back(elem, lk);
0554   }
0555
0556
0557 #ifndef BOOST_THREAD_QUEUE_DEPRECATE_OLD
0558   template <typename ValueType>
0559   bool sync_bounded_queue<ValueType>::try_push(no_block_tag, const ValueType& elem)
0560   {
0561       unique_lock<mutex> lk(mtx_, try_to_lock);
0562       if (!lk.owns_lock()) return false;
0563       return try_push(elem, lk);
0564   }
0565 #endif
0566
0567   template <typename ValueType>
0568   queue_op_status sync_bounded_queue<ValueType>::nonblocking_push_back(const ValueType& elem)
0569   {
0570     unique_lock<mutex> lk(mtx_, try_to_lock);
0571     if (!lk.owns_lock()) return queue_op_status::busy;
0572     return try_push_back(elem, lk);
0573   }
0574
0575   template <typename ValueType>
0576   typename sync_bounded_queue<ValueType>::size_type sync_bounded_queue<ValueType>::wait_until_not_full(unique_lock<mutex>& lk)
0577   {
0578     for (;;)
0579     {
0580       throw_if_closed(lk);
0581       size_type in_p_1 = inc(in_);
0582       if (in_p_1 != out_) // ! full()
0583       {
0584         return in_p_1;
0585       }
0586       ++waiting_full_;
0587       not_full_.wait(lk);
0588     }
0589   }
0590
0591 #ifndef BOOST_THREAD_QUEUE_DEPRECATE_OLD
0592   template <typename ValueType>
0593   void sync_bounded_queue<ValueType>::push(const ValueType& elem)
0594   {
0595       unique_lock<mutex> lk(mtx_);
0596       push_at(elem, wait_until_not_full(lk), lk);
0597   }
0598 #endif
0599   template <typename ValueType>
0600   void sync_bounded_queue<ValueType>::push_back(const ValueType& elem)
0601   {
0602       unique_lock<mutex> lk(mtx_);
0603       push_at(elem, wait_until_not_full(lk), lk);
0604   }
0605
0606 #ifndef BOOST_THREAD_QUEUE_DEPRECATE_OLD
0607   template <typename ValueType>
0608   bool sync_bounded_queue<ValueType>::try_push(BOOST_THREAD_RV_REF(ValueType) elem, unique_lock<mutex>& lk)
0609   {
0610     throw_if_closed(lk);
0611     size_type in_p_1 = inc(in_);
0612     if (in_p_1 == out_) // full()
0613     {
0614       return false;
0615     }
0616     push_at(boost::move(elem), in_p_1, lk);
0617     return true;
0618   }
0619
0620   template <typename ValueType>
0621   bool sync_bounded_queue<ValueType>::try_push(BOOST_THREAD_RV_REF(ValueType) elem)
0622   {
0623       unique_lock<mutex> lk(mtx_);
0624       return try_push(boost::move(elem), lk);
0625   }
0626 #endif
0627
0628   template <typename ValueType>
0629   queue_op_status sync_bounded_queue<ValueType>::try_push_back(BOOST_THREAD_RV_REF(ValueType) elem, unique_lock<mutex>& lk)
0630   {
0631     if (closed(lk)) return queue_op_status::closed;
0632     size_type in_p_1 = inc(in_);
0633     if (in_p_1 == out_) // full()
0634     {
0635       return queue_op_status::full;
0636     }
0637     push_at(boost::move(elem), in_p_1, lk);
0638     return queue_op_status::success;
0639   }
0640   template <typename ValueType>
0641   queue_op_status sync_bounded_queue<ValueType>::try_push_back(BOOST_THREAD_RV_REF(ValueType) elem)
0642   {
0643       unique_lock<mutex> lk(mtx_);
0644       return try_push_back(boost::move(elem), lk);
0645   }
0646
0647   template <typename ValueType>
0648   queue_op_status sync_bounded_queue<ValueType>::wait_push_back(BOOST_THREAD_RV_REF(ValueType) elem, unique_lock<mutex>& lk)
0649   {
0650     if (closed(lk)) return queue_op_status::closed;
0651     push_at(boost::move(elem), wait_until_not_full(lk), lk);
0652     return queue_op_status::success;
0653   }
0654   template <typename ValueType>
0655   queue_op_status sync_bounded_queue<ValueType>::wait_push_back(BOOST_THREAD_RV_REF(ValueType) elem)
0656   {
0657       unique_lock<mutex> lk(mtx_);
0658       return wait_push_back(boost::move(elem), lk);
0659   }
0660
0661
0662 #ifndef BOOST_THREAD_QUEUE_DEPRECATE_OLD
0663   template <typename ValueType>
0664   bool sync_bounded_queue<ValueType>::try_push(no_block_tag, BOOST_THREAD_RV_REF(ValueType) elem)
0665   {
0666       unique_lock<mutex> lk(mtx_, try_to_lock);
0667       if (!lk.owns_lock())
0668       {
0669         return false;
0670       }
0671       return try_push(boost::move(elem), lk);
0672   }
0673 #endif
0674   template <typename ValueType>
0675   queue_op_status sync_bounded_queue<ValueType>::nonblocking_push_back(BOOST_THREAD_RV_REF(ValueType) elem)
0676   {
0677       unique_lock<mutex> lk(mtx_, try_to_lock);
0678       if (!lk.owns_lock())
0679       {
0680         return queue_op_status::busy;
0681       }
0682       return try_push_back(boost::move(elem), lk);
0683   }
0684
0685 #ifndef BOOST_THREAD_QUEUE_DEPRECATE_OLD
0686   template <typename ValueType>
0687   void sync_bounded_queue<ValueType>::push(BOOST_THREAD_RV_REF(ValueType) elem)
0688   {
0689       unique_lock<mutex> lk(mtx_);
0690       push_at(boost::move(elem), wait_until_not_full(lk), lk);
0691   }
0692 #endif
0693   template <typename ValueType>
0694   void sync_bounded_queue<ValueType>::push_back(BOOST_THREAD_RV_REF(ValueType) elem)
0695   {
0696       unique_lock<mutex> lk(mtx_);
0697       push_at(boost::move(elem), wait_until_not_full(lk), lk);
0698   }
0699
0700   template <typename ValueType>
0701   sync_bounded_queue<ValueType>& operator<<(sync_bounded_queue<ValueType>& sbq, BOOST_THREAD_RV_REF(ValueType) elem)
0702   {
0703     sbq.push_back(boost::move(elem));
0704     return sbq;
0705   }
0706
0707   template <typename ValueType>
0708   sync_bounded_queue<ValueType>& operator<<(sync_bounded_queue<ValueType>& sbq, ValueType const&elem)
0709   {
0710     sbq.push_back(elem);
0711     return sbq;
0712   }
0713
0714   template <typename ValueType>
0715   sync_bounded_queue<ValueType>& operator>>(sync_bounded_queue<ValueType>& sbq, ValueType &elem)
0716   {
0717     sbq.pull_front(elem);
0718     return sbq;
0719   }
0720 }
0721 using concurrent::sync_bounded_queue;
0722
0723 }
0724
0725 #include <boost/config/abi_suffix.hpp>
0726
0727 #endif