ceph/src/boost/libs/thread/include/boost/thread/concurrent_queues/sync_queue.hpp

   1 #ifndef BOOST_THREAD_CONCURRENT_QUEUES_SYNC_QUEUE_HPP
   2 #define BOOST_THREAD_CONCURRENT_QUEUES_SYNC_QUEUE_HPP
   3
   4 //////////////////////////////////////////////////////////////////////////////
   5 //
   6 // (C) Copyright Vicente J. Botet Escriba 2013-2014. Distributed under the Boost
   7 // Software License, Version 1.0. (See accompanying file
   8 // LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
   9 //
  10 // See http://www.boost.org/libs/thread for documentation.
  11 //
  12 //////////////////////////////////////////////////////////////////////////////
  13 #include <iostream>
  14
  15 #include <boost/thread/detail/config.hpp>
  16 #include <boost/thread/concurrent_queues/detail/sync_queue_base.hpp>
  17 #include <boost/thread/concurrent_queues/queue_op_status.hpp>
  18 #include <boost/thread/condition_variable.hpp>
  19 #include <boost/thread/csbl/devector.hpp>
  20 #include <boost/thread/detail/move.hpp>
  21 #include <boost/thread/mutex.hpp>
  22
  23 #include <boost/throw_exception.hpp>
  24 #include <boost/smart_ptr/shared_ptr.hpp>
  25 #include <boost/smart_ptr/make_shared.hpp>
  26
  27 #include <boost/config/abi_prefix.hpp>
  28
  29 namespace boost
  30 {
  31 namespace concurrent
  32 {
  33   template <class ValueType, class Container = csbl::devector<ValueType> >
  34   class sync_queue
  35     : public detail::sync_queue_base<ValueType, Container >
  36   {
  37     typedef detail::sync_queue_base<ValueType, Container >  super;
  38
  39   public:
  40     typedef ValueType value_type;
  41     //typedef typename super::value_type value_type; // fixme
  42     typedef typename super::underlying_queue_type underlying_queue_type;
  43     typedef typename super::size_type size_type;
  44     typedef typename super::op_status op_status;
  45
  46     // Constructors/Assignment/Destructors
  47     BOOST_THREAD_NO_COPYABLE(sync_queue)
  48     inline sync_queue();
  49     //template <class Range>
  50     //inline explicit sync_queue(Range range);
  51     inline ~sync_queue();
  52
  53     // Modifiers
  54
  55     inline void push(const value_type& x);
  56     inline queue_op_status try_push(const value_type& x);
  57     inline queue_op_status nonblocking_push(const value_type& x);
  58     inline queue_op_status wait_push(const value_type& x);
  59     inline void push(BOOST_THREAD_RV_REF(value_type) x);
  60     inline queue_op_status try_push(BOOST_THREAD_RV_REF(value_type) x);
  61     inline queue_op_status nonblocking_push(BOOST_THREAD_RV_REF(value_type) x);
  62     inline queue_op_status wait_push(BOOST_THREAD_RV_REF(value_type) x);
  63
  64     // Observers/Modifiers
  65     inline void pull(value_type&);
  66     // enable_if is_nothrow_copy_movable<value_type>
  67     inline value_type pull();
  68
  69     inline queue_op_status try_pull(value_type&);
  70     inline queue_op_status nonblocking_pull(value_type&);
  71     inline queue_op_status wait_pull(ValueType& elem);
  72
  73   private:
  74
  75     inline queue_op_status try_pull(value_type& x, unique_lock<mutex>& lk);
  76     inline queue_op_status wait_pull(value_type& x, unique_lock<mutex>& lk);
  77     inline queue_op_status try_push(const value_type& x, unique_lock<mutex>& lk);
  78     inline queue_op_status wait_push(const value_type& x, unique_lock<mutex>& lk);
  79     inline queue_op_status try_push(BOOST_THREAD_RV_REF(value_type) x, unique_lock<mutex>& lk);
  80     inline queue_op_status wait_push(BOOST_THREAD_RV_REF(value_type) x, unique_lock<mutex>& lk);
  81
  82     inline void pull(value_type& elem, unique_lock<mutex>& )
  83     {
  84       elem = boost::move(super::data_.front());
  85       super::data_.pop_front();
  86     }
  87     inline value_type pull(unique_lock<mutex>& )
  88     {
  89       value_type e = boost::move(super::data_.front());
  90       super::data_.pop_front();
  91       return boost::move(e);
  92     }
  93
  94     inline void push(const value_type& elem, unique_lock<mutex>& lk)
  95     {
  96       super::data_.push_back(elem);
  97       super::notify_not_empty_if_needed(lk);
  98     }
  99
 100     inline void push(BOOST_THREAD_RV_REF(value_type) elem, unique_lock<mutex>& lk)
 101     {
 102       super::data_.push_back(boost::move(elem));
 103       super::notify_not_empty_if_needed(lk);
 104     }
 105   };
 106
 107   template <class ValueType, class Container>
 108   sync_queue<ValueType, Container>::sync_queue() :
 109     super()
 110   {
 111   }
 112
 113 //  template <class ValueType, class Container>
 114 //  template <class Range>
 115 //  explicit sync_queue<ValueType, Container>::sync_queue(Range range) :
 116 //    data_(), closed_(false)
 117 //  {
 118 //    try
 119 //    {
 120 //      typedef typename Range::iterator iterator_t;
 121 //      iterator_t first = boost::begin(range);
 122 //      iterator_t end = boost::end(range);
 123 //      for (iterator_t cur = first; cur != end; ++cur)
 124 //      {
 125 //        data_.push(boost::move(*cur));;
 126 //      }
 127 //      notify_not_empty_if_needed(lk);
 128 //    }
 129 //    catch (...)
 130 //    {
 131 //      delete[] data_;
 132 //    }
 133 //  }
 134
 135   template <class ValueType, class Container>
 136   sync_queue<ValueType, Container>::~sync_queue()
 137   {
 138   }
 139
 140   template <class ValueType, class Container>
 141   queue_op_status sync_queue<ValueType, Container>::try_pull(ValueType& elem, unique_lock<mutex>& lk)
 142   {
 143     if (super::empty(lk))
 144     {
 145       if (super::closed(lk)) return queue_op_status::closed;
 146       return queue_op_status::empty;
 147     }
 148     pull(elem, lk);
 149     return queue_op_status::success;
 150   }
 151   template <class ValueType, class Container>
 152   queue_op_status sync_queue<ValueType, Container>::wait_pull(ValueType& elem, unique_lock<mutex>& lk)
 153   {
 154     //std::cout << __FILE__ << "[" << __LINE__ << "]" << std::endl;
 155     if (super::empty(lk))
 156     {
 157       //std::cout << __FILE__ << "[" << __LINE__ << "]" << std::endl;
 158       if (super::closed(lk)) return queue_op_status::closed;
 159     }
 160     //std::cout << __FILE__ << "[" << __LINE__ << "]" << std::endl;
 161     bool has_been_closed = super::wait_until_not_empty_or_closed(lk);
 162     //std::cout << __FILE__ << "[" << __LINE__ << "]" << std::endl;
 163     if (has_been_closed) return queue_op_status::closed;
 164     //std::cout << __FILE__ << "[" << __LINE__ << "]" << std::endl;
 165     pull(elem, lk);
 166     //std::cout << __FILE__ << "[" << __LINE__ << "]" << std::endl;
 167     return queue_op_status::success;
 168   }
 169
 170   template <class ValueType, class Container>
 171   queue_op_status sync_queue<ValueType, Container>::try_pull(ValueType& elem)
 172   {
 173     unique_lock<mutex> lk(super::mtx_);
 174     return try_pull(elem, lk);
 175   }
 176
 177   template <class ValueType, class Container>
 178   queue_op_status sync_queue<ValueType, Container>::wait_pull(ValueType& elem)
 179   {
 180     unique_lock<mutex> lk(super::mtx_);
 181     return wait_pull(elem, lk);
 182   }
 183
 184   template <class ValueType, class Container>
 185   queue_op_status sync_queue<ValueType, Container>::nonblocking_pull(ValueType& elem)
 186   {
 187     unique_lock<mutex> lk(super::mtx_, try_to_lock);
 188     if (!lk.owns_lock())
 189     {
 190       return queue_op_status::busy;
 191     }
 192     return try_pull(elem, lk);
 193   }
 194
 195   template <class ValueType, class Container>
 196   void sync_queue<ValueType, Container>::pull(ValueType& elem)
 197   {
 198       unique_lock<mutex> lk(super::mtx_);
 199       super::wait_until_not_empty(lk);
 200       pull(elem, lk);
 201   }
 202
 203   // enable if ValueType is nothrow movable
 204   template <class ValueType, class Container>
 205   ValueType sync_queue<ValueType, Container>::pull()
 206   {
 207       unique_lock<mutex> lk(super::mtx_);
 208       super::wait_until_not_empty(lk);
 209       return pull(lk);
 210   }
 211
 212   template <class ValueType, class Container>
 213   queue_op_status sync_queue<ValueType, Container>::try_push(const ValueType& elem, unique_lock<mutex>& lk)
 214   {
 215     if (super::closed(lk)) return queue_op_status::closed;
 216     push(elem, lk);
 217     return queue_op_status::success;
 218   }
 219
 220   template <class ValueType, class Container>
 221   queue_op_status sync_queue<ValueType, Container>::try_push(const ValueType& elem)
 222   {
 223     unique_lock<mutex> lk(super::mtx_);
 224     return try_push(elem, lk);
 225   }
 226
 227   template <class ValueType, class Container>
 228   queue_op_status sync_queue<ValueType, Container>::wait_push(const ValueType& elem, unique_lock<mutex>& lk)
 229   {
 230     if (super::closed(lk)) return queue_op_status::closed;
 231     push(elem, lk);
 232     return queue_op_status::success;
 233   }
 234
 235   template <class ValueType, class Container>
 236   queue_op_status sync_queue<ValueType, Container>::wait_push(const ValueType& elem)
 237   {
 238     unique_lock<mutex> lk(super::mtx_);
 239     return wait_push(elem, lk);
 240   }
 241
 242   template <class ValueType, class Container>
 243   queue_op_status sync_queue<ValueType, Container>::nonblocking_push(const ValueType& elem)
 244   {
 245     unique_lock<mutex> lk(super::mtx_, try_to_lock);
 246     if (!lk.owns_lock()) return queue_op_status::busy;
 247     return try_push(elem, lk);
 248   }
 249
 250   template <class ValueType, class Container>
 251   void sync_queue<ValueType, Container>::push(const ValueType& elem)
 252   {
 253       unique_lock<mutex> lk(super::mtx_);
 254       super::throw_if_closed(lk);
 255       push(elem, lk);
 256   }
 257
 258   template <class ValueType, class Container>
 259   queue_op_status sync_queue<ValueType, Container>::try_push(BOOST_THREAD_RV_REF(ValueType) elem, unique_lock<mutex>& lk)
 260   {
 261     if (super::closed(lk)) return queue_op_status::closed;
 262     push(boost::move(elem), lk);
 263     return queue_op_status::success;
 264   }
 265
 266   template <class ValueType, class Container>
 267   queue_op_status sync_queue<ValueType, Container>::try_push(BOOST_THREAD_RV_REF(ValueType) elem)
 268   {
 269     unique_lock<mutex> lk(super::mtx_);
 270     return try_push(boost::move(elem), lk);
 271   }
 272
 273   template <class ValueType, class Container>
 274   queue_op_status sync_queue<ValueType, Container>::wait_push(BOOST_THREAD_RV_REF(ValueType) elem, unique_lock<mutex>& lk)
 275   {
 276     if (super::closed(lk)) return queue_op_status::closed;
 277     push(boost::move(elem), lk);
 278     return queue_op_status::success;
 279   }
 280
 281   template <class ValueType, class Container>
 282   queue_op_status sync_queue<ValueType, Container>::wait_push(BOOST_THREAD_RV_REF(ValueType) elem)
 283   {
 284     unique_lock<mutex> lk(super::mtx_);
 285     return wait_push(boost::move(elem), lk);
 286   }
 287
 288   template <class ValueType, class Container>
 289   queue_op_status sync_queue<ValueType, Container>::nonblocking_push(BOOST_THREAD_RV_REF(ValueType) elem)
 290   {
 291     unique_lock<mutex> lk(super::mtx_, try_to_lock);
 292     if (!lk.owns_lock())
 293     {
 294       return queue_op_status::busy;
 295     }
 296     return try_push(boost::move(elem), lk);
 297   }
 298
 299   template <class ValueType, class Container>
 300   void sync_queue<ValueType, Container>::push(BOOST_THREAD_RV_REF(ValueType) elem)
 301   {
 302       unique_lock<mutex> lk(super::mtx_);
 303       super::throw_if_closed(lk);
 304       push(boost::move(elem), lk);
 305   }
 306
 307   template <class ValueType, class Container>
 308   sync_queue<ValueType, Container>& operator<<(sync_queue<ValueType, Container>& sbq, BOOST_THREAD_RV_REF(ValueType) elem)
 309   {
 310     sbq.push(boost::move(elem));
 311     return sbq;
 312   }
 313
 314   template <class ValueType, class Container>
 315   sync_queue<ValueType, Container>& operator<<(sync_queue<ValueType, Container>& sbq, ValueType const&elem)
 316   {
 317     sbq.push(elem);
 318     return sbq;
 319   }
 320
 321   template <class ValueType, class Container>
 322   sync_queue<ValueType, Container>& operator>>(sync_queue<ValueType, Container>& sbq, ValueType &elem)
 323   {
 324     sbq.pull(elem);
 325     return sbq;
 326   }
 327
 328 }
 329 using concurrent::sync_queue;
 330
 331 }
 332
 333 #include <boost/config/abi_suffix.hpp>
 334
 335 #endif