thread/concurrent_queues/sync_deque.hpp

0001 #ifndef BOOST_THREAD_CONCURRENT_QUEUES_SYNC_DEQUE_HPP
0002 #define BOOST_THREAD_CONCURRENT_QUEUES_SYNC_DEQUE_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/concurrent_queues/detail/sync_queue_base.hpp>
0016 #include <boost/thread/concurrent_queues/queue_op_status.hpp>
0017 #include <boost/thread/condition_variable.hpp>
0018 #include <boost/thread/csbl/devector.hpp>
0019 #include <boost/thread/detail/move.hpp>
0020 #include <boost/thread/mutex.hpp>
0021
0022 #include <boost/throw_exception.hpp>
0023 #include <boost/smart_ptr/shared_ptr.hpp>
0024 #include <boost/smart_ptr/make_shared.hpp>
0025
0026 #include <boost/config/abi_prefix.hpp>
0027
0028 namespace boost
0029 {
0030 namespace concurrent
0031 {
0032   template <class ValueType, class Container = csbl::devector<ValueType> >
0033   class sync_deque
0034     : public detail::sync_queue_base<ValueType, Container >
0035   {
0036     typedef detail::sync_queue_base<ValueType, Container >  super;
0037
0038   public:
0039     typedef ValueType value_type;
0040     //typedef typename super::value_type value_type; // fixme
0041     typedef typename super::underlying_queue_type underlying_queue_type;
0042     typedef typename super::size_type size_type;
0043     typedef typename super::op_status op_status;
0044
0045     // Constructors/Assignment/Destructors
0046     BOOST_THREAD_NO_COPYABLE(sync_deque)
0047     inline sync_deque();
0048     //template <typename Range>
0049     //inline explicit sync_deque(Range range);
0050     inline ~sync_deque();
0051
0052     // Modifiers
0053     inline void push_back(const value_type& x);
0054     inline queue_op_status try_push_back(const value_type& x);
0055     inline queue_op_status nonblocking_push_back(const value_type& x);
0056     inline queue_op_status wait_push_back(const value_type& x);
0057     inline void push_back(BOOST_THREAD_RV_REF(value_type) x);
0058     inline queue_op_status try_push_back(BOOST_THREAD_RV_REF(value_type) x);
0059     inline queue_op_status nonblocking_push_back(BOOST_THREAD_RV_REF(value_type) x);
0060     inline queue_op_status wait_push_back(BOOST_THREAD_RV_REF(value_type) x);
0061
0062     // Observers/Modifiers
0063     inline void pull_front(value_type&);
0064     // enable_if is_nothrow_copy_movable<value_type>
0065     inline value_type pull_front();
0066
0067     inline queue_op_status try_pull_front(value_type&);
0068     inline queue_op_status nonblocking_pull_front(value_type&);
0069     inline queue_op_status wait_pull_front(ValueType& elem);
0070
0071   private:
0072
0073     inline queue_op_status try_pull_front(value_type& x, unique_lock<mutex>& lk);
0074     inline queue_op_status wait_pull_front(value_type& x, unique_lock<mutex>& lk);
0075     inline queue_op_status try_push_back(const value_type& x, unique_lock<mutex>& lk);
0076     inline queue_op_status wait_push_back(const value_type& x, unique_lock<mutex>& lk);
0077     inline queue_op_status try_push_back(BOOST_THREAD_RV_REF(value_type) x, unique_lock<mutex>& lk);
0078     inline queue_op_status wait_push_back(BOOST_THREAD_RV_REF(value_type) x, unique_lock<mutex>& lk);
0079
0080     inline void pull_front(value_type& elem, unique_lock<mutex>& )
0081     {
0082       elem = boost::move(super::data_.front());
0083       super::data_.pop_front();
0084     }
0085     inline value_type pull_front(unique_lock<mutex>& )
0086     {
0087       value_type e = boost::move(super::data_.front());
0088       super::data_.pop_front();
0089       return boost::move(e);
0090     }
0091
0092     inline void push_back(const value_type& elem, unique_lock<mutex>& lk)
0093     {
0094       super::data_.push_back(elem);
0095       super::notify_elem_added(lk);
0096     }
0097
0098     inline void push_back(BOOST_THREAD_RV_REF(value_type) elem, unique_lock<mutex>& lk)
0099     {
0100       super::data_.push_back(boost::move(elem));
0101       super::notify_elem_added(lk);
0102     }
0103   };
0104
0105   template <class ValueType, class Container>
0106   sync_deque<ValueType, Container>::sync_deque() :
0107     super()
0108   {
0109   }
0110
0111 //  template <class ValueType, class Container>
0112 //  template <class Range>
0113 //  explicit sync_deque<ValueType, Container>::sync_deque(Range range) :
0114 //    data_(), closed_(false)
0115 //  {
0116 //    try
0117 //    {
0118 //      typedef typename Range::iterator iterator_t;
0119 //      iterator_t first = boost::begin(range);
0120 //      iterator_t end = boost::end(range);
0121 //      for (iterator_t cur = first; cur != end; ++cur)
0122 //      {
0123 //        data_.push(boost::move(*cur));;
0124 //      }
0125 //      notify_elem_added(lk);
0126 //    }
0127 //    catch (...)
0128 //    {
0129 //      delete[] data_;
0130 //    }
0131 //  }
0132
0133   template <class ValueType, class Container>
0134   sync_deque<ValueType, Container>::~sync_deque()
0135   {
0136   }
0137
0138   template <class ValueType, class Container>
0139   queue_op_status sync_deque<ValueType, Container>::try_pull_front(ValueType& elem, unique_lock<mutex>& lk)
0140   {
0141     if (super::empty(lk))
0142     {
0143       if (super::closed(lk)) return queue_op_status::closed;
0144       return queue_op_status::empty;
0145     }
0146     pull_front(elem, lk);
0147     return queue_op_status::success;
0148   }
0149   template <class ValueType, class Container>
0150   queue_op_status sync_deque<ValueType, Container>::wait_pull_front(ValueType& elem, unique_lock<mutex>& lk)
0151   {
0152     const bool has_been_closed = super::wait_until_not_empty_or_closed(lk);
0153     if (has_been_closed) return queue_op_status::closed;
0154     pull_front(elem, lk);
0155     return queue_op_status::success;
0156   }
0157
0158   template <class ValueType, class Container>
0159   queue_op_status sync_deque<ValueType, Container>::try_pull_front(ValueType& elem)
0160   {
0161     unique_lock<mutex> lk(super::mtx_);
0162     return try_pull_front(elem, lk);
0163   }
0164
0165   template <class ValueType, class Container>
0166   queue_op_status sync_deque<ValueType, Container>::wait_pull_front(ValueType& elem)
0167   {
0168     unique_lock<mutex> lk(super::mtx_);
0169     return wait_pull_front(elem, lk);
0170   }
0171
0172   template <class ValueType, class Container>
0173   queue_op_status sync_deque<ValueType, Container>::nonblocking_pull_front(ValueType& elem)
0174   {
0175     unique_lock<mutex> lk(super::mtx_, try_to_lock);
0176     if (!lk.owns_lock())
0177     {
0178       return queue_op_status::busy;
0179     }
0180     return try_pull_front(elem, lk);
0181   }
0182
0183   template <class ValueType, class Container>
0184   void sync_deque<ValueType, Container>::pull_front(ValueType& elem)
0185   {
0186       unique_lock<mutex> lk(super::mtx_);
0187       const bool has_been_closed = super::wait_until_not_empty_or_closed(lk);
0188       if (has_been_closed) super::throw_if_closed(lk);
0189       pull_front(elem, lk);
0190   }
0191
0192   // enable if ValueType is nothrow movable
0193   template <class ValueType, class Container>
0194   ValueType sync_deque<ValueType, Container>::pull_front()
0195   {
0196       unique_lock<mutex> lk(super::mtx_);
0197       const bool has_been_closed = super::wait_until_not_empty_or_closed(lk);
0198       if (has_been_closed) super::throw_if_closed(lk);
0199       return pull_front(lk);
0200   }
0201
0202   template <class ValueType, class Container>
0203   queue_op_status sync_deque<ValueType, Container>::try_push_back(const ValueType& elem, unique_lock<mutex>& lk)
0204   {
0205     if (super::closed(lk)) return queue_op_status::closed;
0206     push_back(elem, lk);
0207     return queue_op_status::success;
0208   }
0209
0210   template <class ValueType, class Container>
0211   queue_op_status sync_deque<ValueType, Container>::try_push_back(const ValueType& elem)
0212   {
0213     unique_lock<mutex> lk(super::mtx_);
0214     return try_push_back(elem, lk);
0215   }
0216
0217   template <class ValueType, class Container>
0218   queue_op_status sync_deque<ValueType, Container>::wait_push_back(const ValueType& elem, unique_lock<mutex>& lk)
0219   {
0220     if (super::closed(lk)) return queue_op_status::closed;
0221     push_back(elem, lk);
0222     return queue_op_status::success;
0223   }
0224
0225   template <class ValueType, class Container>
0226   queue_op_status sync_deque<ValueType, Container>::wait_push_back(const ValueType& elem)
0227   {
0228     unique_lock<mutex> lk(super::mtx_);
0229     return wait_push_back(elem, lk);
0230   }
0231
0232   template <class ValueType, class Container>
0233   queue_op_status sync_deque<ValueType, Container>::nonblocking_push_back(const ValueType& elem)
0234   {
0235     unique_lock<mutex> lk(super::mtx_, try_to_lock);
0236     if (!lk.owns_lock()) return queue_op_status::busy;
0237     return try_push_back(elem, lk);
0238   }
0239
0240   template <class ValueType, class Container>
0241   void sync_deque<ValueType, Container>::push_back(const ValueType& elem)
0242   {
0243       unique_lock<mutex> lk(super::mtx_);
0244       super::throw_if_closed(lk);
0245       push_back(elem, lk);
0246   }
0247
0248   template <class ValueType, class Container>
0249   queue_op_status sync_deque<ValueType, Container>::try_push_back(BOOST_THREAD_RV_REF(ValueType) elem, unique_lock<mutex>& lk)
0250   {
0251     if (super::closed(lk)) return queue_op_status::closed;
0252     push_back(boost::move(elem), lk);
0253     return queue_op_status::success;
0254   }
0255
0256   template <class ValueType, class Container>
0257   queue_op_status sync_deque<ValueType, Container>::try_push_back(BOOST_THREAD_RV_REF(ValueType) elem)
0258   {
0259     unique_lock<mutex> lk(super::mtx_);
0260     return try_push_back(boost::move(elem), lk);
0261   }
0262
0263   template <class ValueType, class Container>
0264   queue_op_status sync_deque<ValueType, Container>::wait_push_back(BOOST_THREAD_RV_REF(ValueType) elem, unique_lock<mutex>& lk)
0265   {
0266     if (super::closed(lk)) return queue_op_status::closed;
0267     push_back(boost::move(elem), lk);
0268     return queue_op_status::success;
0269   }
0270
0271   template <class ValueType, class Container>
0272   queue_op_status sync_deque<ValueType, Container>::wait_push_back(BOOST_THREAD_RV_REF(ValueType) elem)
0273   {
0274     unique_lock<mutex> lk(super::mtx_);
0275     return wait_push_back(boost::move(elem), lk);
0276   }
0277
0278   template <class ValueType, class Container>
0279   queue_op_status sync_deque<ValueType, Container>::nonblocking_push_back(BOOST_THREAD_RV_REF(ValueType) elem)
0280   {
0281     unique_lock<mutex> lk(super::mtx_, try_to_lock);
0282     if (!lk.owns_lock())
0283     {
0284       return queue_op_status::busy;
0285     }
0286     return try_push_back(boost::move(elem), lk);
0287   }
0288
0289   template <class ValueType, class Container>
0290   void sync_deque<ValueType, Container>::push_back(BOOST_THREAD_RV_REF(ValueType) elem)
0291   {
0292       unique_lock<mutex> lk(super::mtx_);
0293       super::throw_if_closed(lk);
0294       push_back(boost::move(elem), lk);
0295   }
0296
0297   template <class ValueType, class Container>
0298   sync_deque<ValueType, Container>& operator<<(sync_deque<ValueType, Container>& sbq, BOOST_THREAD_RV_REF(ValueType) elem)
0299   {
0300     sbq.push_back(boost::move(elem));
0301     return sbq;
0302   }
0303
0304   template <class ValueType, class Container>
0305   sync_deque<ValueType, Container>& operator<<(sync_deque<ValueType, Container>& sbq, ValueType const&elem)
0306   {
0307     sbq.push_back(elem);
0308     return sbq;
0309   }
0310
0311   template <class ValueType, class Container>
0312   sync_deque<ValueType, Container>& operator>>(sync_deque<ValueType, Container>& sbq, ValueType &elem)
0313   {
0314     sbq.pull_front(elem);
0315     return sbq;
0316   }
0317
0318 }
0319 using concurrent::sync_deque;
0320
0321 }
0322
0323 #include <boost/config/abi_suffix.hpp>
0324
0325 #endif