thread/futures/wait_for_any.hpp

0001 //  (C) Copyright 2008-10 Anthony Williams
0002 //  (C) Copyright 2011-2015 Vicente J. Botet Escriba
0003 //
0004 //  Distributed under the Boost Software License, Version 1.0. (See
0005 //  accompanying file LICENSE_1_0.txt or copy at
0006 //  http://www.boost.org/LICENSE_1_0.txt)
0007
0008 #ifndef BOOST_THREAD_FUTURES_WAIT_FOR_ANY_HPP
0009 #define BOOST_THREAD_FUTURES_WAIT_FOR_ANY_HPP
0010
0011 #include <boost/thread/detail/config.hpp>
0012
0013 #include <boost/thread/detail/move.hpp>
0014 #include <boost/thread/futures/is_future_type.hpp>
0015 #include <boost/thread/lock_algorithms.hpp>
0016 #include <boost/thread/mutex.hpp>
0017 #include <boost/thread/condition_variable.hpp>
0018
0019 #include <boost/core/enable_if.hpp>
0020 #include <boost/scoped_array.hpp>
0021
0022 #include <iterator>
0023 #include <vector>
0024
0025 namespace boost
0026 {
0027   namespace detail
0028   {
0029     template <class Future>
0030     class waiter_for_any_in_seq
0031     {
0032       struct registered_waiter;
0033       typedef std::vector<int>::size_type count_type;
0034
0035       struct registered_waiter
0036       {
0037         typedef Future future_type;
0038         future_type* future_;
0039         typedef typename Future::notify_when_ready_handle notify_when_ready_handle;
0040         notify_when_ready_handle handle;
0041         count_type index;
0042
0043         registered_waiter(future_type & a_future,
0044             notify_when_ready_handle handle_, count_type index_) :
0045           future_(&a_future), handle(handle_), index(index_)
0046         {
0047         }
0048       };
0049
0050       struct all_futures_lock
0051       {
0052 #ifdef _MANAGED
0053         typedef std::ptrdiff_t count_type_portable;
0054 #else
0055         typedef count_type count_type_portable;
0056 #endif
0057         count_type_portable count;
0058         boost::scoped_array<boost::unique_lock<boost::mutex> > locks;
0059
0060         all_futures_lock(std::vector<registered_waiter>& waiters) :
0061           count(waiters.size()), locks(new boost::unique_lock<boost::mutex>[count])
0062         {
0063           for (count_type_portable i = 0; i < count; ++i)
0064           {
0065             locks[i] = BOOST_THREAD_MAKE_RV_REF(boost::unique_lock<boost::mutex>(waiters[i].future_->mutex()));
0066           }
0067         }
0068
0069         void lock()
0070         {
0071           boost::lock(locks.get(), locks.get() + count);
0072         }
0073
0074         void unlock()
0075         {
0076           for (count_type_portable i = 0; i < count; ++i)
0077           {
0078             locks[i].unlock();
0079           }
0080         }
0081       };
0082
0083       boost::condition_variable_any cv;
0084       std::vector<registered_waiter> waiters_;
0085       count_type future_count;
0086
0087     public:
0088       waiter_for_any_in_seq() :
0089         future_count(0)
0090       {
0091       }
0092
0093       template <typename F>
0094       void add(F& f)
0095       {
0096         if (f.valid())
0097         {
0098           registered_waiter waiter(f, f.notify_when_ready(cv), future_count);
0099           try
0100           {
0101             waiters_.push_back(waiter);
0102           }
0103           catch (...)
0104           {
0105             f.future_->unnotify_when_ready(waiter.handle);
0106             throw;
0107           }
0108           ++future_count;
0109         }
0110       }
0111
0112 #ifndef BOOST_NO_CXX11_VARIADIC_TEMPLATES
0113       template <typename F1, typename ... Fs>
0114       void add(F1& f1, Fs&... fs)
0115       {
0116         add(f1);
0117         add(fs...);
0118       }
0119 #endif
0120
0121       count_type wait()
0122       {
0123         all_futures_lock lk(waiters_);
0124         for (;;)
0125         {
0126           for (count_type i = 0; i < waiters_.size(); ++i)
0127           {
0128             if (waiters_[i].future_->is_ready(lk.locks[i]))
0129             {
0130               return waiters_[i].index;
0131             }
0132           }
0133           cv.wait(lk);
0134         }
0135       }
0136
0137       ~waiter_for_any_in_seq()
0138       {
0139         for (count_type i = 0; i < waiters_.size(); ++i)
0140         {
0141           waiters_[i].future_->unnotify_when_ready(waiters_[i].handle);
0142         }
0143       }
0144     };
0145   }
0146
0147   template <typename Iterator>
0148   typename boost::disable_if<is_future_type<Iterator> , Iterator>::type wait_for_any(Iterator begin, Iterator end)
0149   {
0150     if (begin == end) return end;
0151
0152     detail::waiter_for_any_in_seq<typename std::iterator_traits<Iterator>::value_type> waiter;
0153     for (Iterator current = begin; current != end; ++current)
0154     {
0155       waiter.add(*current);
0156     }
0157
0158     std::advance( begin, waiter.wait() );
0159     return begin;
0160   }
0161 }
0162
0163 #endif // header