Chris@102: //  (C) Copyright 2008-10 Anthony Williams
Chris@102: //  (C) Copyright 2011-2015 Vicente J. Botet Escriba
Chris@102: //
Chris@102: //  Distributed under the Boost Software License, Version 1.0. (See
Chris@102: //  accompanying file LICENSE_1_0.txt or copy at
Chris@102: //  http://www.boost.org/LICENSE_1_0.txt)
Chris@102: 
Chris@102: #ifndef BOOST_THREAD_FUTURES_WAIT_FOR_ANY_HPP
Chris@102: #define BOOST_THREAD_FUTURES_WAIT_FOR_ANY_HPP
Chris@102: 
Chris@102: #include <boost/thread/detail/config.hpp>
Chris@102: 
Chris@102: #include <boost/thread/detail/move.hpp>
Chris@102: #include <boost/thread/futures/is_future_type.hpp>
Chris@102: #include <boost/thread/lock_algorithms.hpp>
Chris@102: #include <boost/thread/mutex.hpp>
Chris@102: 
Chris@102: #include <boost/core/enable_if.hpp>
Chris@102: #include <boost/next_prior.hpp>
Chris@102: #include <boost/scoped_array.hpp>
Chris@102: 
Chris@102: #include <iterator>
Chris@102: #include <vector>
Chris@102: 
Chris@102: namespace boost
Chris@102: {
Chris@102:   namespace detail
Chris@102:   {
Chris@102:     template <class Future>
Chris@102:     class waiter_for_any_in_seq
Chris@102:     {
Chris@102:       struct registered_waiter;
Chris@102:       typedef std::vector<int>::size_type count_type;
Chris@102: 
Chris@102:       struct registered_waiter
Chris@102:       {
Chris@102:         typedef Future future_type;
Chris@102:         future_type* future_;
Chris@102:         typedef typename Future::notify_when_ready_handle notify_when_ready_handle;
Chris@102:         notify_when_ready_handle handle;
Chris@102:         count_type index;
Chris@102: 
Chris@102:         registered_waiter(future_type & a_future,
Chris@102:             notify_when_ready_handle handle_, count_type index_) :
Chris@102:           future_(&a_future), handle(handle_), index(index_)
Chris@102:         {
Chris@102:         }
Chris@102:       };
Chris@102: 
Chris@102:       struct all_futures_lock
Chris@102:       {
Chris@102: #ifdef _MANAGED
Chris@102:         typedef std::ptrdiff_t count_type_portable;
Chris@102: #else
Chris@102:         typedef count_type count_type_portable;
Chris@102: #endif
Chris@102:         count_type_portable count;
Chris@102:         boost::scoped_array<boost::unique_lock<boost::mutex> > locks;
Chris@102: 
Chris@102:         all_futures_lock(std::vector<registered_waiter>& waiters) :
Chris@102:           count(waiters.size()), locks(new boost::unique_lock<boost::mutex>[count])
Chris@102:         {
Chris@102:           for (count_type_portable i = 0; i < count; ++i)
Chris@102:           {
Chris@102:             locks[i] = BOOST_THREAD_MAKE_RV_REF(boost::unique_lock<boost::mutex>(waiters[i].future_->mutex()));
Chris@102:           }
Chris@102:         }
Chris@102: 
Chris@102:         void lock()
Chris@102:         {
Chris@102:           boost::lock(locks.get(), locks.get() + count);
Chris@102:         }
Chris@102: 
Chris@102:         void unlock()
Chris@102:         {
Chris@102:           for (count_type_portable i = 0; i < count; ++i)
Chris@102:           {
Chris@102:             locks[i].unlock();
Chris@102:           }
Chris@102:         }
Chris@102:       };
Chris@102: 
Chris@102:       boost::condition_variable_any cv;
Chris@102:       std::vector<registered_waiter> waiters_;
Chris@102:       count_type future_count;
Chris@102: 
Chris@102:     public:
Chris@102:       waiter_for_any_in_seq() :
Chris@102:         future_count(0)
Chris@102:       {
Chris@102:       }
Chris@102: 
Chris@102:       template <typename F>
Chris@102:       void add(F& f)
Chris@102:       {
Chris@102:         if (f.valid())
Chris@102:         {
Chris@102:           registered_waiter waiter(f, f.notify_when_ready(cv), future_count);
Chris@102:           try
Chris@102:           {
Chris@102:             waiters_.push_back(waiter);
Chris@102:           }
Chris@102:           catch (...)
Chris@102:           {
Chris@102:             f.future_->unnotify_when_ready(waiter.handle);
Chris@102:             throw;
Chris@102:           }
Chris@102:           ++future_count;
Chris@102:         }
Chris@102:       }
Chris@102: 
Chris@102: #ifndef BOOST_NO_CXX11_VARIADIC_TEMPLATES
Chris@102:       template <typename F1, typename ... Fs>
Chris@102:       void add(F1& f1, Fs&... fs)
Chris@102:       {
Chris@102:         add(f1);
Chris@102:         add(fs...);
Chris@102:       }
Chris@102: #endif
Chris@102: 
Chris@102:       count_type wait()
Chris@102:       {
Chris@102:         all_futures_lock lk(waiters_);
Chris@102:         for (;;)
Chris@102:         {
Chris@102:           for (count_type i = 0; i < waiters_.size(); ++i)
Chris@102:           {
Chris@102:             if (waiters_[i].future_->is_ready(lk.locks[i]))
Chris@102:             {
Chris@102:               return waiters_[i].index;
Chris@102:             }
Chris@102:           }
Chris@102:           cv.wait(lk);
Chris@102:         }
Chris@102:       }
Chris@102: 
Chris@102:       ~waiter_for_any_in_seq()
Chris@102:       {
Chris@102:         for (count_type i = 0; i < waiters_.size(); ++i)
Chris@102:         {
Chris@102:           waiters_[i].future_->unnotify_when_ready(waiters_[i].handle);
Chris@102:         }
Chris@102:       }
Chris@102:     };
Chris@102:   }
Chris@102: 
Chris@102:   template <typename Iterator>
Chris@102:   typename boost::disable_if<is_future_type<Iterator> , Iterator>::type wait_for_any(Iterator begin, Iterator end)
Chris@102:   {
Chris@102:     if (begin == end) return end;
Chris@102: 
Chris@102:     detail::waiter_for_any_in_seq<typename std::iterator_traits<Iterator>::value_type> waiter;
Chris@102:     for (Iterator current = begin; current != end; ++current)
Chris@102:     {
Chris@102:       waiter.add(*current);
Chris@102:     }
Chris@102:     return boost::next(begin, waiter.wait());
Chris@102:   }
Chris@102: }
Chris@102: 
Chris@102: #endif // header